09d - Agent 记忆系统

from dataclasses import dataclass, field
from typing import Literal
import tiktoken
 
@dataclass
class Message:
    role: Literal["system", "user", "assistant", "tool"]
    content: str
    token_count: int = 0
    timestamp: float = 0.0
    importance: float = 0.5  # 0-1，用于优先级淘汰
 
    def __post_init__(self):
        if self.token_count == 0:
            enc = tiktoken.encoding_for_model("gpt-4o")
            self.token_count = len(enc.encode(self.content))
 
class ShortTermMemory:
    """短期记忆管理器——上下文窗口管理"""
 
    def __init__(
        self,
        max_tokens: int = 120000,
        system_budget: int = 4000,
        summary_threshold: float = 0.8,  # 80% 容量时触发压缩
    ):
        self.max_tokens = max_tokens
        self.system_budget = system_budget
        self.summary_threshold = summary_threshold
        self.messages: list[Message] = []
        self.summary: str | None = None  # 压缩后的历史摘要
 
    @property
    def total_tokens(self) -> int:
        return sum(m.token_count for m in self.messages)
 
    def add_message(self, message: Message):
        """添加消息，必要时触发压缩"""
        self.messages.append(message)
 
        # 检查是否需要压缩
        if self.total_tokens > self.max_tokens * self.summary_threshold:
            self._compress_history()
 
    def _compress_history(self):
        """压缩旧对话为摘要"""
        # 保留系统消息和最近 10 轮对话
        system_msgs = [m for m in self.messages if m.role == "system"]
        recent_msgs = [m for m in self.messages if m.role != "system"][-20:]
        old_msgs = [m for m in self.messages if m.role != "system"][:-20]
 
        if not old_msgs:
            return
 
        # 生成摘要（实际使用时调用 LLM）
        old_content = "\n".join(
            f"{m.role}: {m.content[:200]}" for m in old_msgs
        )
        summary_prompt = f"""请将以下对话历史压缩为简洁摘要，保留关键信息：
{old_content}"""
 
        # 这里应调用 LLM 生成摘要，示例用占位
        self.summary = f"[历史摘要] 之前讨论了 {len(old_msgs)} 条消息的内容"
 
        # 重建消息列表
        self.messages = system_msgs
        if self.summary:
            self.messages.append(Message(
                role="system",
                content=f"之前的对话摘要：{self.summary}",
                importance=0.8,
            ))
        self.messages.extend(recent_msgs)
 
    def get_context(self) -> list[dict]:
        """获取当前上下文（用于 LLM 调用）"""
        return [
            {"role": m.role, "content": m.content}
            for m in self.messages
        ]
 
    def get_stats(self) -> dict:
        return {
            "message_count": len(self.messages),
            "total_tokens": self.total_tokens,
            "utilization": f"{self.total_tokens / self.max_tokens:.1%}",
            "has_summary": self.summary is not None,
        }

interface ChatMessage {
  role: "system" | "user" | "assistant" | "tool";
  content: string;
  tokenCount: number;
  timestamp: number;
}
 
interface TokenBudget {
  system: number;
  history: number;
  tools: number;
  response: number;
}
 
class ShortTermMemory {
  private messages: ChatMessage[] = [];
  private maxTokens: number;
  private budget: TokenBudget;
 
  constructor(
    maxTokens: number = 120000,
    budget: TokenBudget = {
      system: 4000,
      history: 80000,
      tools: 16000,
      response: 20000,
    }
  ) {
    this.maxTokens = maxTokens;
    this.budget = budget;
  }
 
  addMessage(message: ChatMessage): void {
    this.messages.push(message);
    this.enforceTokenBudget();
  }
 
  private enforceTokenBudget(): void {
    const historyMessages = this.messages.filter(
      (m) => m.role !== "system"
    );
    let historyTokens = historyMessages.reduce(
      (sum, m) => sum + m.tokenCount, 0
    );
 
    // 从最旧的非系统消息开始淘汰
    while (historyTokens > this.budget.history && historyMessages.length > 2) {
      const removed = historyMessages.shift()!;
      historyTokens -= removed.tokenCount;
      const idx = this.messages.indexOf(removed);
      if (idx !== -1) this.messages.splice(idx, 1);
    }
  }
 
  getContext(): Array<{ role: string; content: string }> {
    return this.messages.map(({ role, content }) => ({ role, content }));
  }
 
  getTokenUsage(): {
    total: number;
    byRole: Record<string, number>;
    utilization: string;
  } {
    const byRole: Record<string, number> = {};
    let total = 0;
    for (const msg of this.messages) {
      byRole[msg.role] = (byRole[msg.role] ?? 0) + msg.tokenCount;
      total += msg.tokenCount;
    }
    return {
      total,
      byRole,
      utilization: `${((total / this.maxTokens) * 100).toFixed(1)}%`,
    };
  }
}

你是一个具有记忆管理能力的 AI 助手。

## 上下文管理规则
1. 当对话超过 [TOKEN_LIMIT] tokens 时，自动压缩早期对话
2. 保留所有包含 [关键决策/用户偏好/任务目标] 的消息
3. 工具调用结果只保留最终输出，丢弃中间步骤
4. 系统提示词始终保留在上下文最前面

## 之前的对话摘要
[CONVERSATION_SUMMARY]

## 当前任务
[CURRENT_TASK]

from dataclasses import dataclass, field
from typing import Any
from datetime import datetime
import json
 
@dataclass
class ScratchpadEntry:
    key: str
    value: Any
    entry_type: str  # "plan", "result", "state", "hypothesis"
    created_at: datetime = field(default_factory=datetime.now)
    updated_at: datetime = field(default_factory=datetime.now)
    ttl_seconds: int | None = None  # 自动过期时间
 
    def is_expired(self) -> bool:
        if self.ttl_seconds is None:
            return False
        elapsed = (datetime.now() - self.updated_at).total_seconds()
        return elapsed > self.ttl_seconds
 
@dataclass
class TaskPlan:
    goal: str
    steps: list[str]
    current_step: int = 0
    status: str = "in_progress"  # in_progress, completed, failed
 
class WorkingMemory:
    """Agent 工作记忆——任务执行期间的草稿本"""
 
    def __init__(self, max_entries: int = 100):
        self.max_entries = max_entries
        self._store: dict[str, ScratchpadEntry] = {}
        self._plan: TaskPlan | None = None
 
    # ---- 计划管理 ----
    def set_plan(self, goal: str, steps: list[str]):
        """设置当前任务计划"""
        self._plan = TaskPlan(goal=goal, steps=steps)
        self.write("current_plan", {
            "goal": goal,
            "steps": steps,
            "current_step": 0,
        }, entry_type="plan")
 
    def advance_plan(self) -> str | None:
        """推进到下一步"""
        if self._plan and self._plan.current_step < len(self._plan.steps):
            step = self._plan.steps[self._plan.current_step]
            self._plan.current_step += 1
            return step
        return None
 
    # ---- 读写操作 ----
    def write(
        self,
        key: str,
        value: Any,
        entry_type: str = "state",
        ttl_seconds: int | None = None,
    ):
        """写入工作记忆"""
        self._store[key] = ScratchpadEntry(
            key=key,
            value=value,
            entry_type=entry_type,
            ttl_seconds=ttl_seconds,
        )
        self._evict_if_needed()
 
    def read(self, key: str) -> Any | None:
        """读取工作记忆"""
        entry = self._store.get(key)
        if entry is None or entry.is_expired():
            if entry and entry.is_expired():
                del self._store[key]
            return None
        return entry.value
 
    def delete(self, key: str):
        """删除条目"""
        self._store.pop(key, None)
 
    # ---- 中间结果管理 ----
    def store_result(self, step_name: str, result: Any):
        """存储工具调用或推理的中间结果"""
        self.write(
            f"result_{step_name}",
            result,
            entry_type="result",
            ttl_seconds=3600,  # 1 小时后过期
        )
 
    def get_all_results(self) -> dict[str, Any]:
        """获取所有中间结果"""
        return {
            k: v.value
            for k, v in self._store.items()
            if v.entry_type == "result" and not v.is_expired()
        }
 
    # ---- 序列化（注入上下文窗口）----
    def to_context_string(self) -> str:
        """将工作记忆序列化为可注入上下文的字符串"""
        sections = []
 
        if self._plan:
            sections.append(
                f"## 当前计划\n"
                f"目标: {self._plan.goal}\n"
                f"进度: {self._plan.current_step}/{len(self._plan.steps)}\n"
                f"当前步骤: {self._plan.steps[self._plan.current_step] if self._plan.current_step < len(self._plan.steps) else '已完成'}"
            )
 
        results = self.get_all_results()
        if results:
            sections.append(
                "## 中间结果\n" +
                "\n".join(f"- {k}: {json.dumps(v, ensure_ascii=False)[:200]}" for k, v in results.items())
            )
 
        states = {
            k: v.value for k, v in self._store.items()
            if v.entry_type == "state" and not v.is_expired()
        }
        if states:
            sections.append(
                "## 任务状态\n" +
                "\n".join(f"- {k}: {v}" for k, v in states.items())
            )
 
        return "\n\n".join(sections)
 
    def _evict_if_needed(self):
        """淘汰过期和超量条目"""
        # 先清理过期条目
        expired = [k for k, v in self._store.items() if v.is_expired()]
        for k in expired:
            del self._store[k]
 
        # 超量时按创建时间淘汰
        if len(self._store) > self.max_entries:
            sorted_entries = sorted(
                self._store.items(),
                key=lambda x: x[1].created_at,
            )
            to_remove = len(self._store) - self.max_entries
            for k, _ in sorted_entries[:to_remove]:
                del self._store[k]
 
 
# 使用示例
memory = WorkingMemory()
 
# 设置任务计划
memory.set_plan(
    goal="分析用户上传的 CSV 文件并生成报告",
    steps=[
        "读取 CSV 文件",
        "数据清洗和验证",
        "统计分析",
        "生成可视化图表",
        "输出报告",
    ]
)
 
# 执行步骤并存储中间结果
step = memory.advance_plan()  # "读取 CSV 文件"
memory.store_result("csv_read", {"rows": 1500, "columns": 12})
 
step = memory.advance_plan()  # "数据清洗和验证"
memory.store_result("data_clean", {"removed_rows": 23, "valid_rows": 1477})
 
# 注入上下文
context = memory.to_context_string()
print(context)

interface PlanStep {
  name: string;
  status: "pending" | "in_progress" | "completed" | "failed";
  result?: unknown;
}
 
interface WorkingMemoryState {
  plan: {
    goal: string;
    steps: PlanStep[];
    currentStep: number;
  } | null;
  scratchpad: Map<string, { value: unknown; expiresAt?: number }>;
  hypotheses: Array<{ claim: string; confidence: number; evidence: string[] }>;
}
 
class AgentWorkingMemory {
  private state: WorkingMemoryState = {
    plan: null,
    scratchpad: new Map(),
    hypotheses: [],
  };
 
  // ---- 计划管理 ----
  setPlan(goal: string, stepNames: string[]): void {
    this.state.plan = {
      goal,
      steps: stepNames.map((name) => ({ name, status: "pending" })),
      currentStep: 0,
    };
  }
 
  advanceStep(result?: unknown): string | null {
    const plan = this.state.plan;
    if (!plan || plan.currentStep >= plan.steps.length) return null;
 
    const current = plan.steps[plan.currentStep];
    current.status = "completed";
    current.result = result;
    plan.currentStep++;
 
    const next = plan.steps[plan.currentStep];
    if (next) {
      next.status = "in_progress";
      return next.name;
    }
    return null;
  }
 
  // ---- 草稿本操作 ----
  write(key: string, value: unknown, ttlMs?: number): void {
    this.state.scratchpad.set(key, {
      value,
      expiresAt: ttlMs ? Date.now() + ttlMs : undefined,
    });
  }
 
  read(key: string): unknown | null {
    const entry = this.state.scratchpad.get(key);
    if (!entry) return null;
    if (entry.expiresAt && Date.now() > entry.expiresAt) {
      this.state.scratchpad.delete(key);
      return null;
    }
    return entry.value;
  }
 
  // ---- 假设管理（用于推理链）----
  addHypothesis(claim: string, confidence: number, evidence: string[]): void {
    this.state.hypotheses.push({ claim, confidence, evidence });
  }
 
  getBestHypothesis(): { claim: string; confidence: number } | null {
    if (this.state.hypotheses.length === 0) return null;
    return this.state.hypotheses.reduce((best, h) =>
      h.confidence > best.confidence ? h : best
    );
  }
 
  // ---- 序列化 ----
  toContextString(): string {
    const parts: string[] = [];
 
    if (this.state.plan) {
      const { goal, steps, currentStep } = this.state.plan;
      parts.push(
        `## 当前计划\n目标: ${goal}\n进度: ${currentStep}/${steps.length}\n` +
        steps.map((s, i) =>
          `${i === currentStep ? "→" : " "} [${s.status}] ${s.name}`
        ).join("\n")
      );
    }
 
    if (this.state.scratchpad.size > 0) {
      const entries = Array.from(this.state.scratchpad.entries())
        .filter(([, v]) => !v.expiresAt || Date.now() <= v.expiresAt)
        .map(([k, v]) => `- ${k}: ${JSON.stringify(v.value).slice(0, 200)}`);
      if (entries.length) {
        parts.push(`## 草稿本\n${entries.join("\n")}`);
      }
    }
 
    return parts.join("\n\n");
  }
}

import chromadb
from chromadb.config import Settings
from datetime import datetime
from typing import Any
import json
import hashlib
 
class LongTermMemory:
    """基于向量数据库的长期记忆系统"""
 
    def __init__(
        self,
        collection_name: str = "agent_memory",
        persist_directory: str = "./memory_store",
    ):
        self.client = chromadb.PersistentClient(
            path=persist_directory,
            settings=Settings(anonymized_telemetry=False),
        )
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            metadata={"hnsw:space": "cosine"},
        )
 
    def store(
        self,
        content: str,
        metadata: dict[str, Any] | None = None,
        memory_type: str = "knowledge",
    ) -> str:
        """存储记忆到向量数据库"""
        doc_id = hashlib.md5(content.encode()).hexdigest()[:16]
        meta = {
            "memory_type": memory_type,
            "created_at": datetime.now().isoformat(),
            "content_length": len(content),
            **(metadata or {}),
        }
 
        self.collection.upsert(
            ids=[doc_id],
            documents=[content],
            metadatas=[meta],
        )
        return doc_id
 
    def retrieve(
        self,
        query: str,
        n_results: int = 5,
        memory_type: str | None = None,
        min_relevance: float = 0.7,
    ) -> list[dict]:
        """语义检索相关记忆"""
        where_filter = None
        if memory_type:
            where_filter = {"memory_type": memory_type}
 
        results = self.collection.query(
            query_texts=[query],
            n_results=n_results,
            where=where_filter,
            include=["documents", "metadatas", "distances"],
        )
 
        memories = []
        for i, doc in enumerate(results["documents"][0]):
            distance = results["distances"][0][i]
            relevance = 1 - distance  # cosine distance → similarity
            if relevance >= min_relevance:
                memories.append({
                    "content": doc,
                    "metadata": results["metadatas"][0][i],
                    "relevance": round(relevance, 3),
                })
 
        return memories
 
    def forget(self, doc_id: str):
        """删除指定记忆"""
        self.collection.delete(ids=[doc_id])
 
    def search_by_metadata(
        self,
        filters: dict[str, Any],
        limit: int = 10,
    ) -> list[dict]:
        """按元数据过滤检索"""
        results = self.collection.get(
            where=filters,
            limit=limit,
            include=["documents", "metadatas"],
        )
        return [
            {"content": doc, "metadata": meta}
            for doc, meta in zip(results["documents"], results["metadatas"])
        ]
 
    def get_stats(self) -> dict:
        return {
            "total_memories": self.collection.count(),
            "collection_name": self.collection.name,
        }
 
 
# 使用示例
ltm = LongTermMemory()
 
# 存储用户偏好
ltm.store(
    "用户偏好使用 TypeScript 而非 JavaScript，喜欢函数式编程风格",
    metadata={"user_id": "user_001", "category": "preference"},
    memory_type="user_profile",
)
 
# 存储项目知识
ltm.store(
    "项目使用 Next.js 14 + tRPC + Prisma 技术栈，部署在 Vercel 上",
    metadata={"project": "my-saas", "category": "tech_stack"},
    memory_type="knowledge",
)
 
# 检索相关记忆
results = ltm.retrieve("用户喜欢什么编程语言？", n_results=3)
for r in results:
    print(f"[{r['relevance']}] {r['content']}")

import { Pool } from "pg";
import OpenAI from "openai";
 
interface MemoryRecord {
  id: string;
  content: string;
  embedding?: number[];
  memoryType: string;
  metadata: Record<string, unknown>;
  relevance?: number;
  createdAt: Date;
}
 
class PgVectorLongTermMemory {
  private pool: Pool;
  private openai: OpenAI;
  private tableName: string;
 
  constructor(connectionString: string, tableName = "agent_memories") {
    this.pool = new Pool({ connectionString });
    this.openai = new OpenAI();
    this.tableName = tableName;
  }
 
  async initialize(): Promise<void> {
    await this.pool.query("CREATE EXTENSION IF NOT EXISTS vector");
    await this.pool.query(`
      CREATE TABLE IF NOT EXISTS ${this.tableName} (
        id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
        content TEXT NOT NULL,
        embedding vector(1536),
        memory_type VARCHAR(50) NOT NULL DEFAULT 'knowledge',
        metadata JSONB DEFAULT '{}',
        created_at TIMESTAMPTZ DEFAULT NOW(),
        updated_at TIMESTAMPTZ DEFAULT NOW()
      )
    `);
    await this.pool.query(`
      CREATE INDEX IF NOT EXISTS idx_${this.tableName}_embedding
      ON ${this.tableName}
      USING ivfflat (embedding vector_cosine_ops)
      WITH (lists = 100)
    `);
  }
 
  private async embed(text: string): Promise<number[]> {
    const response = await this.openai.embeddings.create({
      model: "text-embedding-3-small",
      input: text,
    });
    return response.data[0].embedding;
  }
 
  async store(
    content: string,
    memoryType = "knowledge",
    metadata: Record<string, unknown> = {}
  ): Promise<string> {
    const embedding = await this.embed(content);
    const result = await this.pool.query(
      `INSERT INTO ${this.tableName} (content, embedding, memory_type, metadata)
       VALUES ($1, $2::vector, $3, $4)
       RETURNING id`,
      [content, JSON.stringify(embedding), memoryType, JSON.stringify(metadata)]
    );
    return result.rows[0].id;
  }
 
  async retrieve(
    query: string,
    limit = 5,
    memoryType?: string,
    minRelevance = 0.7
  ): Promise<MemoryRecord[]> {
    const queryEmbedding = await this.embed(query);
    const typeFilter = memoryType
      ? "AND memory_type = $3"
      : "";
    const params: unknown[] = [
      JSON.stringify(queryEmbedding),
      limit,
    ];
    if (memoryType) params.push(memoryType);
 
    const result = await this.pool.query(
      `SELECT id, content, memory_type, metadata, created_at,
              1 - (embedding <=> $1::vector) AS relevance
       FROM ${this.tableName}
       WHERE 1 - (embedding <=> $1::vector) >= ${minRelevance}
       ${typeFilter}
       ORDER BY embedding <=> $1::vector
       LIMIT $2`,
      params
    );
 
    return result.rows.map((row) => ({
      id: row.id,
      content: row.content,
      memoryType: row.memory_type,
      metadata: row.metadata,
      relevance: parseFloat(row.relevance),
      createdAt: row.created_at,
    }));
  }
 
  async forget(id: string): Promise<void> {
    await this.pool.query(
      `DELETE FROM ${this.tableName} WHERE id = $1`,
      [id]
    );
  }
}

from mem0 import Memory
 
# 初始化 Mem0
config = {
    "llm": {
        "provider": "openai",
        "config": {"model": "gpt-4o-mini"},
    },
    "vector_store": {
        "provider": "chroma",
        "config": {"collection_name": "agent_memory"},
    },
}
 
memory = Memory.from_config(config)
 
# 添加记忆（Mem0 自动提取关键信息）
memory.add(
    "我正在开发一个 SaaS 产品，使用 Next.js 和 Supabase。"
    "我偏好 TypeScript 和函数式编程风格。"
    "项目预算有限，需要控制 API 调用成本。",
    user_id="developer_001",
)
 
# 检索相关记忆
results = memory.search(
    "推荐什么技术栈？",
    user_id="developer_001",
)
for r in results:
    print(f"[{r['score']:.2f}] {r['memory']}")
 
# 获取用户所有记忆
all_memories = memory.get_all(user_id="developer_001")
for m in all_memories:
    print(f"- {m['memory']}")
 
# 更新记忆（Mem0 自动处理冲突和去重）
memory.add(
    "我决定从 Supabase 切换到 PlanetScale 作为数据库。",
    user_id="developer_001",
)

你是一个具有长期记忆的 AI 助手。在回答问题前，请先检索相关记忆。

## 记忆检索结果
以下是与当前问题相关的历史记忆（按相关度排序）：

[RETRIEVED_MEMORIES]

## 使用规则
1. 优先使用记忆中的信息回答问题
2. 如果记忆中的信息与当前上下文矛盾，以最新信息为准
3. 如果记忆不足以回答问题，明确告知用户并提供你的最佳推测
4. 回答后，判断是否需要将新信息存入长期记忆

## 记忆更新判断
回答完成后，请判断：
- 用户是否提供了新的偏好或需求？→ 存储为 user_profile
- 是否产生了新的项目知识？→ 存储为 knowledge
- 是否有值得记录的决策？→ 存储为 decision

from dataclasses import dataclass, field
from datetime import datetime
from typing import Any
import json
 
@dataclass
class Episode:
    """一次完整的 Agent 交互情景"""
    episode_id: str
    timestamp: datetime
    task_description: str
    actions_taken: list[dict]  # 执行的操作序列
    outcome: str  # "success", "partial", "failure"
    outcome_details: str
    lessons_learned: list[str]
    user_feedback: str | None = None
    context_tags: list[str] = field(default_factory=list)
    duration_seconds: float = 0.0
 
    def to_summary(self) -> str:
        """生成情景摘要（用于注入上下文）"""
        action_summary = ", ".join(
            a.get("tool", "unknown") for a in self.actions_taken[:5]
        )
        return (
            f"[{self.timestamp.strftime('%Y-%m-%d')}] "
            f"任务: {self.task_description[:100]} | "
            f"操作: {action_summary} | "
            f"结果: {self.outcome} | "
            f"教训: {'; '.join(self.lessons_learned[:3])}"
        )
 
 
class EpisodicMemory:
    """情景记忆系统——记录和检索 Agent 的历史经历"""
 
    def __init__(self, long_term_memory=None):
        self.episodes: list[Episode] = []
        self.ltm = long_term_memory  # 可选：连接长期记忆做语义检索
        self._pattern_cache: dict[str, list[str]] = {}
 
    def record_episode(
        self,
        task_description: str,
        actions: list[dict],
        outcome: str,
        outcome_details: str,
        lessons: list[str] | None = None,
        tags: list[str] | None = None,
    ) -> Episode:
        """记录一次完整的交互情景"""
        episode = Episode(
            episode_id=f"ep_{len(self.episodes):04d}",
            timestamp=datetime.now(),
            task_description=task_description,
            actions_taken=actions,
            outcome=outcome,
            outcome_details=outcome_details,
            lessons_learned=lessons or [],
            context_tags=tags or [],
        )
        self.episodes.append(episode)
 
        # 同步到长期记忆（如果可用）
        if self.ltm:
            self.ltm.store(
                episode.to_summary(),
                metadata={
                    "episode_id": episode.episode_id,
                    "outcome": outcome,
                    "tags": json.dumps(tags or []),
                },
                memory_type="episode",
            )
 
        return episode
 
    def recall_similar(
        self,
        current_task: str,
        n_results: int = 3,
        outcome_filter: str | None = None,
    ) -> list[Episode]:
        """检索与当前任务相似的历史情景"""
        if self.ltm:
            # 使用向量检索
            results = self.ltm.retrieve(
                current_task,
                n_results=n_results,
                memory_type="episode",
            )
            episode_ids = [
                r["metadata"].get("episode_id")
                for r in results
            ]
            return [
                ep for ep in self.episodes
                if ep.episode_id in episode_ids
                and (outcome_filter is None or ep.outcome == outcome_filter)
            ]
        else:
            # 简单的关键词匹配回退
            scored = []
            for ep in self.episodes:
                if outcome_filter and ep.outcome != outcome_filter:
                    continue
                # 简单相关度：共同词数
                task_words = set(current_task.lower().split())
                ep_words = set(ep.task_description.lower().split())
                overlap = len(task_words & ep_words)
                if overlap > 0:
                    scored.append((overlap, ep))
            scored.sort(key=lambda x: x[0], reverse=True)
            return [ep for _, ep in scored[:n_results]]
 
    def get_lessons_for_task(self, task_description: str) -> list[str]:
        """获取与当前任务相关的历史教训"""
        similar = self.recall_similar(task_description, n_results=5)
        lessons = []
        for ep in similar:
            lessons.extend(ep.lessons_learned)
        # 去重
        return list(dict.fromkeys(lessons))
 
    def get_success_patterns(self) -> dict[str, int]:
        """分析成功模式——哪些操作组合最常导致成功"""
        patterns: dict[str, int] = {}
        for ep in self.episodes:
            if ep.outcome == "success":
                tools_used = tuple(
                    sorted(set(a.get("tool", "") for a in ep.actions_taken))
                )
                key = " + ".join(tools_used)
                patterns[key] = patterns.get(key, 0) + 1
        return dict(sorted(patterns.items(), key=lambda x: x[1], reverse=True))
 
    def to_context_string(self, current_task: str, max_episodes: int = 3) -> str:
        """生成可注入上下文的情景记忆摘要"""
        similar = self.recall_similar(current_task, n_results=max_episodes)
        if not similar:
            return "## 相关历史经验\n暂无相关历史经验。"
 
        lines = ["## 相关历史经验"]
        for ep in similar:
            lines.append(f"\n### {ep.episode_id} ({ep.outcome})")
            lines.append(f"任务: {ep.task_description[:150]}")
            lines.append(f"结果: {ep.outcome_details[:150]}")
            if ep.lessons_learned:
                lines.append("教训:")
                for lesson in ep.lessons_learned[:3]:
                    lines.append(f"  - {lesson}")
        return "\n".join(lines)

interface AgentAction {
  tool: string;
  input: Record<string, unknown>;
  output: string;
  success: boolean;
  durationMs: number;
}
 
interface Episode {
  id: string;
  timestamp: Date;
  taskDescription: string;
  actions: AgentAction[];
  outcome: "success" | "partial" | "failure";
  outcomeDetails: string;
  lessonsLearned: string[];
  tags: string[];
}
 
class EpisodicMemory {
  private episodes: Episode[] = [];
  private maxEpisodes: number;
 
  constructor(maxEpisodes = 1000) {
    this.maxEpisodes = maxEpisodes;
  }
 
  record(params: {
    taskDescription: string;
    actions: AgentAction[];
    outcome: Episode["outcome"];
    outcomeDetails: string;
    lessons?: string[];
    tags?: string[];
  }): Episode {
    const episode: Episode = {
      id: `ep_${this.episodes.length.toString().padStart(4, "0")}`,
      timestamp: new Date(),
      taskDescription: params.taskDescription,
      actions: params.actions,
      outcome: params.outcome,
      outcomeDetails: params.outcomeDetails,
      lessonsLearned: params.lessons ?? [],
      tags: params.tags ?? [],
    };
 
    this.episodes.push(episode);
 
    // 超量时淘汰最旧的失败情景
    if (this.episodes.length > this.maxEpisodes) {
      const failIdx = this.episodes.findIndex((e) => e.outcome === "failure");
      if (failIdx !== -1) {
        this.episodes.splice(failIdx, 1);
      } else {
        this.episodes.shift();
      }
    }
 
    return episode;
  }
 
  recallByOutcome(outcome: Episode["outcome"], limit = 5): Episode[] {
    return this.episodes
      .filter((e) => e.outcome === outcome)
      .slice(-limit);
  }
 
  getFailurePatterns(): Array<{ tool: string; failCount: number; commonErrors: string[] }> {
    const toolFailures = new Map<string, { count: number; errors: string[] }>();
 
    for (const ep of this.episodes) {
      if (ep.outcome === "failure") {
        for (const action of ep.actions) {
          if (!action.success) {
            const existing = toolFailures.get(action.tool) ?? { count: 0, errors: [] };
            existing.count++;
            existing.errors.push(action.output.slice(0, 100));
            toolFailures.set(action.tool, existing);
          }
        }
      }
    }
 
    return Array.from(toolFailures.entries())
      .map(([tool, data]) => ({
        tool,
        failCount: data.count,
        commonErrors: [...new Set(data.errors)].slice(0, 3),
      }))
      .sort((a, b) => b.failCount - a.failCount);
  }
 
  generateReflection(): string {
    const total = this.episodes.length;
    if (total === 0) return "暂无历史经验可供反思。";
 
    const successRate = this.episodes.filter((e) => e.outcome === "success").length / total;
    const allLessons = this.episodes.flatMap((e) => e.lessonsLearned);
    const uniqueLessons = [...new Set(allLessons)].slice(0, 5);
    const failPatterns = this.getFailurePatterns().slice(0, 3);
 
    return [
      `## Agent 经验反思`,
      `总交互次数: ${total}，成功率: ${(successRate * 100).toFixed(1)}%`,
      ``,
      `### 关键教训`,
      ...uniqueLessons.map((l) => `- ${l}`),
      ``,
      `### 常见失败模式`,
      ...failPatterns.map(
        (p) => `- ${p.tool}: 失败 ${p.failCount} 次 (${p.commonErrors[0] ?? "未知错误"})`
      ),
    ].join("\n");
  }
}

基于以下近期经历，请提取 3-5 条高层次洞察：

## 近期经历
[RECENT_EPISODES]

## 反思要求
1. 识别重复出现的模式（成功模式和失败模式）
2. 提取可泛化的经验教训
3. 发现用户偏好的变化趋势
4. 建议未来行为的调整方向

请以 JSON 格式输出：
{
  "insights": [
    {
      "observation": "观察到的模式",
      "lesson": "从中学到的教训",
      "action": "建议的行为调整",
      "confidence": 0.0-1.0
    }
  ]
}

from dataclasses import dataclass
from typing import Any
 
@dataclass
class MemoryQuery:
    text: str
    memory_types: list[str] | None = None  # None = 搜索所有类型
    max_results: int = 5
    min_relevance: float = 0.6
 
@dataclass
class MemoryResult:
    content: str
    source: str  # "short_term", "working", "long_term", "episodic"
    relevance: float
    metadata: dict[str, Any]
 
class UnifiedMemoryManager:
    """统一记忆管理器——协调四层记忆"""
 
    def __init__(
        self,
        short_term: "ShortTermMemory",
        working: "WorkingMemory",
        long_term: "LongTermMemory",
        episodic: "EpisodicMemory",
    ):
        self.short_term = short_term
        self.working = working
        self.long_term = long_term
        self.episodic = episodic
 
    def build_context(self, user_query: str) -> str:
        """构建完整的记忆增强上下文"""
        sections = []
 
        # 1. 工作记忆（当前任务状态）
        working_ctx = self.working.to_context_string()
        if working_ctx:
            sections.append(working_ctx)
 
        # 2. 长期记忆检索
        ltm_results = self.long_term.retrieve(
            user_query, n_results=3, min_relevance=0.7
        )
        if ltm_results:
            sections.append(
                "## 相关知识\n" +
                "\n".join(f"- {r['content'][:200]}" for r in ltm_results)
            )
 
        # 3. 情景记忆检索
        episodic_ctx = self.episodic.to_context_string(
            user_query, max_episodes=2
        )
        if "暂无" not in episodic_ctx:
            sections.append(episodic_ctx)
 
        return "\n\n".join(sections)
 
    def on_session_end(self, session_summary: dict):
        """会话结束时的记忆巩固"""
        # 1. 将重要对话内容存入长期记忆
        if session_summary.get("key_decisions"):
            for decision in session_summary["key_decisions"]:
                self.long_term.store(
                    decision,
                    metadata={"type": "decision"},
                    memory_type="knowledge",
                )
 
        # 2. 记录情景
        self.episodic.record_episode(
            task_description=session_summary.get("task", ""),
            actions=session_summary.get("actions", []),
            outcome=session_summary.get("outcome", "unknown"),
            outcome_details=session_summary.get("details", ""),
            lessons=session_summary.get("lessons", []),
        )
 
        # 3. 清理工作记忆
        self.working = WorkingMemory()
 
    def search(self, query: MemoryQuery) -> list[MemoryResult]:
        """跨记忆层统一搜索"""
        results: list[MemoryResult] = []
        types = query.memory_types or [
            "long_term", "episodic"
        ]
 
        if "long_term" in types:
            for r in self.long_term.retrieve(
                query.text, n_results=query.max_results
            ):
                results.append(MemoryResult(
                    content=r["content"],
                    source="long_term",
                    relevance=r["relevance"],
                    metadata=r.get("metadata", {}),
                ))
 
        if "episodic" in types:
            episodes = self.episodic.recall_similar(
                query.text, n_results=query.max_results
            )
            for ep in episodes:
                results.append(MemoryResult(
                    content=ep.to_summary(),
                    source="episodic",
                    relevance=0.8,  # 情景记忆默认高相关度
                    metadata={"episode_id": ep.episode_id},
                ))
 
        # 按相关度排序
        results.sort(key=lambda r: r.relevance, reverse=True)
        return results[:query.max_results]

class CodingAssistantMemory:
    """编码助手的记忆系统"""
 
    def __init__(self):
        self.short_term = ShortTermMemory(max_tokens=120000)
        self.working = WorkingMemory()
        self.long_term = LongTermMemory(collection_name="coding_assistant")
        self.episodic = EpisodicMemory(long_term_memory=self.long_term)
 
    def on_new_session(self, user_id: str):
        """新会话开始——加载用户画像"""
        # 检索用户偏好
        prefs = self.long_term.retrieve(
            f"user:{user_id} 编程偏好和技术栈",
            memory_type="user_profile",
            n_results=5,
        )
        if prefs:
            pref_text = "\n".join(p["content"] for p in prefs)
            self.short_term.add_message(Message(
                role="system",
                content=f"用户偏好：\n{pref_text}",
                importance=0.9,
            ))
 
    def on_code_review(self, code: str, feedback: str, outcome: str):
        """代码审查完成——记录经验"""
        self.episodic.record_episode(
            task_description=f"代码审查: {code[:100]}...",
            actions=[{"tool": "code_review", "input": code[:200]}],
            outcome=outcome,
            outcome_details=feedback,
            lessons=[feedback] if outcome == "failure" else [],
            tags=["code_review"],
        )
 
    def get_review_hints(self, code_snippet: str) -> list[str]:
        """基于历史经验获取代码审查提示"""
        return self.episodic.get_lessons_for_task(
            f"代码审查: {code_snippet[:200]}"
        )