""" 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 # 内容 content: Optional[str] = None # 完整的 skill 内容(Markdown) 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", content: Optional[str] = None, 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, content=content, 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, "content": self.content, "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 的文本 优先使用完整的 content(如果有),否则使用 description + guidelines """ # 如果有完整的 content,直接使用 if self.content: return self.content.strip() # 否则使用旧的格式(向后兼容) lines = [f"### {self.name}", self.description] if self.guidelines: lines.append("指导原则:") for g in self.guidelines: lines.append(f"- {g}") return "\n".join(lines)