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"
    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 = (
    "请基于项目全生命周期统计，输出中文管理层分析。"
    "包括总体判断、前三个风险、接下来一周优先动作。"
    "不要审批付款、不要最终定绩效、不要下投资交易指令。"
)
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.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]