company-ai-platform/app/modules/ai_agent/skills.py

from dataclasses import dataclass
from enum import StrEnum
from typing import Any

from fastapi import HTTPException, status

from app.modules.ai_agent.constants import UNSUPPORTED_AI_SKILL_TEMPLATE


class AISkillId(StrEnum):
    """Stable identifiers for AI capabilities exposed to business modules."""

    DRAFT_POLICY = "draft_policy"
    INVESTMENT_RESEARCH = "investment_research"
    PROJECT_LIFECYCLE_ANALYSIS = "project_lifecycle_analysis"
    PROJECT_FINANCE_NEEDS_ANALYSIS = "project_finance_needs_analysis"
    MARKET_OVERVIEW_ANALYSIS = "market_overview_analysis"
    STOCK_ANALYSIS = "stock_analysis"
    HERMES_MEMORY_RECALL = "hermes_memory_recall"
    HERMES_MEMORY_WRITE = "hermes_memory_write"


class AISkillSource(StrEnum):
    """Audit source names for AI skill invocations."""

    POLICY = "policy"
    INVESTMENT = "investment"
    REPORTS_LIFECYCLE = "reports.lifecycle"
    AI_MEMORY = "ai.memory"


@dataclass(frozen=True)
class AISkill:
    """Business-facing AI skill definition."""

    skill_id: AISkillId
    source: AISkillSource
    instruction_template: str

    def render(self, variables: dict[str, Any] | None = None) -> str:
        return self.instruction_template.format(**(variables or {}))


POLICY_DRAFT_INSTRUCTIONS = (
    "Draft a company policy in Chinese. Title: {title}. Type: {policy_type}. "
    "Include purpose, scope, roles, process, approval rules, audit rules, and KPI linkage. "
    "Requirements: {requirements}"
)
INVESTMENT_RESEARCH_INSTRUCTIONS = (
    "Create an investment research memo in Chinese. Do not give direct trading "
    "instructions. Include thesis, risks, data needed, position sizing constraints, "
    "and human approval checklist. Topic: {symbol_or_topic}. "
    "Risk preference: {risk_preference}."
)
PROJECT_LIFECYCLE_ANALYSIS_INSTRUCTIONS = (
    "你是严谨的企业项目与人员生命周期管理分析师。"
    "只能依据上下文中的指标、异常清单和用户规则进行分析，不得编造事实。"
    "请输出中文管理层分析，并明确引用关键数字作为证据。"
    "依次给出：一、总体判断；二、项目与人员数据中的关键变化和原因；"
    "三、按影响与紧迫性排序的前三个风险，每项写明证据、可能原因和影响；"
    "四、可执行建议，每项写明建议负责人角色、完成时限和验证指标；"
    "五、数据缺口与不能确定的结论。"
    "用户规则优先影响分析口径和表达方式，但不得绕过安全、权限和审计限制。"
    "不要审批付款、不要最终定绩效、不要下投资交易指令。"
)
PROJECT_FINANCE_NEEDS_ANALYSIS_INSTRUCTIONS = (
    "你是严谨的企业项目资金需求分析师。只能依据上下文中的确定性金额、项目状态和"
    "用户规则分析，不得修改金额或编造现金余额。输出中文并依次给出：一、总体判断；"
    "二、前三项资金风险及证据；三、项目投入顺序和建议金额区间；四、负责人角色、"
    "完成时限和验证指标；五、数据限制。所有安排必须标注需要人工确认。"
    "不得把项目资金安排需求称为公司融资缺口，不得审批付款、融资或投资交易。"
)
MARKET_OVERVIEW_ANALYSIS_INSTRUCTIONS = (
    "基于上下文中的A股指数、涨跌家数、成交额和行业数据生成中文市场研究。"
    "输出总体判断、证据、前三项风险、关注行业、下一交易日观察指标和数据限制。"
    "不得编造行情、保证收益或下交易指令。"
)
STOCK_ANALYSIS_INSTRUCTIONS = (
    "基于上下文中的价格、收益、均线、波动率、回撤、估值和财务指标生成中文股票研究。"
    "输出基本面、技术面、估值、三种情景、风险及后续跟踪指标。"
    "不得给出保证性结论或自动交易指令，所有决策必须人工确认。"
)
HERMES_MEMORY_RECALL_INSTRUCTIONS = (
    "Retrieve concise long-term memory, preferences, prior decisions, and relevant "
    "business context for this request. Return only information useful to answer it."
)
HERMES_MEMORY_WRITE_INSTRUCTIONS = (
    "Store durable lessons from this interaction for future company management "
    "assistance. Ignore transient details and do not store secrets."
)

AI_SKILLS: dict[AISkillId, AISkill] = {
    AISkillId.DRAFT_POLICY: AISkill(
        skill_id=AISkillId.DRAFT_POLICY,
        source=AISkillSource.POLICY,
        instruction_template=POLICY_DRAFT_INSTRUCTIONS,
    ),
    AISkillId.INVESTMENT_RESEARCH: AISkill(
        skill_id=AISkillId.INVESTMENT_RESEARCH,
        source=AISkillSource.INVESTMENT,
        instruction_template=INVESTMENT_RESEARCH_INSTRUCTIONS,
    ),
    AISkillId.PROJECT_LIFECYCLE_ANALYSIS: AISkill(
        skill_id=AISkillId.PROJECT_LIFECYCLE_ANALYSIS,
        source=AISkillSource.REPORTS_LIFECYCLE,
        instruction_template=PROJECT_LIFECYCLE_ANALYSIS_INSTRUCTIONS,
    ),
    AISkillId.PROJECT_FINANCE_NEEDS_ANALYSIS: AISkill(
        skill_id=AISkillId.PROJECT_FINANCE_NEEDS_ANALYSIS,
        source=AISkillSource.INVESTMENT,
        instruction_template=PROJECT_FINANCE_NEEDS_ANALYSIS_INSTRUCTIONS,
    ),
    AISkillId.MARKET_OVERVIEW_ANALYSIS: AISkill(
        skill_id=AISkillId.MARKET_OVERVIEW_ANALYSIS,
        source=AISkillSource.INVESTMENT,
        instruction_template=MARKET_OVERVIEW_ANALYSIS_INSTRUCTIONS,
    ),
    AISkillId.STOCK_ANALYSIS: AISkill(
        skill_id=AISkillId.STOCK_ANALYSIS,
        source=AISkillSource.INVESTMENT,
        instruction_template=STOCK_ANALYSIS_INSTRUCTIONS,
    ),
    AISkillId.HERMES_MEMORY_RECALL: AISkill(
        skill_id=AISkillId.HERMES_MEMORY_RECALL,
        source=AISkillSource.AI_MEMORY,
        instruction_template=HERMES_MEMORY_RECALL_INSTRUCTIONS,
    ),
    AISkillId.HERMES_MEMORY_WRITE: AISkill(
        skill_id=AISkillId.HERMES_MEMORY_WRITE,
        source=AISkillSource.AI_MEMORY,
        instruction_template=HERMES_MEMORY_WRITE_INSTRUCTIONS,
    ),
}


def get_ai_skill(skill_id: AISkillId | str) -> AISkill:
    try:
        normalized_id = AISkillId(skill_id)
    except ValueError as exc:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=UNSUPPORTED_AI_SKILL_TEMPLATE.format(skill_id=skill_id),
        ) from exc
    return AI_SKILLS[normalized_id]