Agent/agent/models/memory.py

"""
Memory 数据模型

Experience: 经验规则（条件 + 规则 + 证据）
Skill: 技能（从经验归纳的高层知识）
"""

from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, Any, List, Optional, Literal
import uuid


@dataclass
class Experience:
    """
    经验规则

    从执行过程或人工反馈中提取的规则，格式：
    - condition: 什么情况下适用
    - rule: 应该怎么做
    - evidence: 证据（step_ids）
    """
    exp_id: str
    scope: str  # "agent:{type}" 或 "user:{uid}"

    # 核心三元组
    condition: str
    rule: str
    evidence: List[str] = field(default_factory=list)  # step_ids

    # 元数据
    source: Literal["execution", "feedback", "manual"] = "feedback"
    confidence: float = 0.5
    usage_count: int = 0
    success_rate: float = 0.0

    # 时间
    created_at: datetime = field(default_factory=datetime.now)
    updated_at: datetime = field(default_factory=datetime.now)

    @classmethod
    def create(
        cls,
        scope: str,
        condition: str,
        rule: str,
        evidence: List[str] = None,
        source: Literal["execution", "feedback", "manual"] = "feedback",
        confidence: float = 0.5,
    ) -> "Experience":
        """创建新的 Experience"""
        now = datetime.now()
        return cls(
            exp_id=str(uuid.uuid4()),
            scope=scope,
            condition=condition,
            rule=rule,
            evidence=evidence or [],
            source=source,
            confidence=confidence,
            created_at=now,
            updated_at=now,
        )

    def to_dict(self) -> Dict[str, Any]:
        """转换为字典"""
        return {
            "exp_id": self.exp_id,
            "scope": self.scope,
            "condition": self.condition,
            "rule": self.rule,
            "evidence": self.evidence,
            "source": self.source,
            "confidence": self.confidence,
            "usage_count": self.usage_count,
            "success_rate": self.success_rate,
            "created_at": self.created_at.isoformat() if self.created_at else None,
            "updated_at": self.updated_at.isoformat() if self.updated_at else None,
        }

    def to_prompt_text(self) -> str:
        """转换为可注入 Prompt 的文本"""
        return f"当 {self.condition} 时，{self.rule}"


@dataclass
class Skill:
    """
    技能 - 从经验归纳的高层知识

    技能可以形成层次结构（通过 parent_id）
    """
    skill_id: str
    scope: str  # "agent:{type}" 或 "user:{uid}"

    name: str
    description: str
    category: str  # 分类，如 "search", "reasoning", "writing"

    # 层次结构
    parent_id: Optional[str] = None

    # 内容
    guidelines: List[str] = field(default_factory=list)
    derived_from: List[str] = field(default_factory=list)  # experience_ids

    # 版本
    version: int = 1

    # 时间
    created_at: datetime = field(default_factory=datetime.now)
    updated_at: datetime = field(default_factory=datetime.now)

    @classmethod
    def create(
        cls,
        scope: str,
        name: str,
        description: str,
        category: str = "general",
        guidelines: List[str] = None,
        derived_from: List[str] = None,
        parent_id: Optional[str] = None,
    ) -> "Skill":
        """创建新的 Skill"""
        now = datetime.now()
        return cls(
            skill_id=str(uuid.uuid4()),
            scope=scope,
            name=name,
            description=description,
            category=category,
            parent_id=parent_id,
            guidelines=guidelines or [],
            derived_from=derived_from or [],
            created_at=now,
            updated_at=now,
        )

    def to_dict(self) -> Dict[str, Any]:
        """转换为字典"""
        return {
            "skill_id": self.skill_id,
            "scope": self.scope,
            "name": self.name,
            "description": self.description,
            "category": self.category,
            "parent_id": self.parent_id,
            "guidelines": self.guidelines,
            "derived_from": self.derived_from,
            "version": self.version,
            "created_at": self.created_at.isoformat() if self.created_at else None,
            "updated_at": self.updated_at.isoformat() if self.updated_at else None,
        }

    def to_prompt_text(self) -> str:
        """转换为可注入 Prompt 的文本"""
        lines = [f"### {self.name}", self.description]
        if self.guidelines:
            lines.append("指导原则：")
            for g in self.guidelines:
                lines.append(f"- {g}")
        return "\n".join(lines)