第04课：对话记忆

🧠 Agent的记忆系统

记忆是Agent理解上下文、保持连贯对话的关键。没有记忆的Agent就像金鱼，每次对话都从零开始。本课深入探讨Agent的记忆架构与实现。

📖 记忆的分类

Agent记忆架构
├── 短期记忆 (Short-term)
│   ├── 对话历史 (Conversation History)
│   ├── 工作记忆 (Working Memory)
│   └── 上下文窗口 (Context Window)
├── 长期记忆 (Long-term)
│   ├── 语义记忆 (Semantic) → 向量数据库
│   ├── 情景记忆 (Episodic) → 对话摘要
│   └── 程序记忆 (Procedural) → 技能库
└── 混合记忆 (Hybrid)
    ├── 记忆检索 (Retrieval)
    ├── 记忆压缩 (Compression)
    └── 记忆更新 (Update)

📊 不同记忆策略对比

💻 代码实现：多层记忆系统

# 多层记忆系统 - 短期/工作/长期记忆
import json, re
from datetime import datetime
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field

@dataclass
class Message:
    role: str
    content: str
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())

class ConversationMemory:
    # 短期记忆 - 滑动窗口
    def __init__(self, max_msgs=50, window=10):
        self.messages = []
        self.max_msgs = max_msgs
        self.window = window
    
    def add(self, role, content):
        self.messages.append(Message(role=role, content=content))
        if len(self.messages) > self.max_msgs:
            self.messages = self.messages[-self.max_msgs:]
    
    def get_recent(self, n=None):
        return [{"role": m.role, "content": m.content} for m in self.messages[-(n or self.window):]]
    
    def token_estimate(self):
        return int(sum(len(m.content) for m in self.messages) * 0.7)

class WorkingMemory:
    # 工作记忆 - 当前任务关键信息
    def __init__(self):
        self.facts = {}
        self.entities = {}
        self.goals = []
    
    def set_fact(self, key, value):
        self.facts[key] = value
    def update_entity(self, name, info):
        self.entities.setdefault(name, {}).update(info)
    def summary(self):
        parts = []
        if self.facts: parts.append("事实: " + json.dumps(self.facts, ensure_ascii=False))
        if self.entities: parts.append("实体: " + json.dumps(self.entities, ensure_ascii=False))
        return "\n".join(parts)

class SummaryMemory:
    # 长期记忆 - 定期压缩摘要
    def __init__(self, threshold=10):
        self.summaries = []
        self.buffer = []
        self.threshold = threshold
    
    def add_message(self, msg):
        self.buffer.append(msg)
        if len(self.buffer) >= self.threshold:
            self._summarize()
    
    def _summarize(self):
        points = [f"{'用户' if m.role=='user' else '助手'}: {m.content[:50]}" for m in self.buffer]
        self.summaries.append({"content": "; ".join(points), "count": len(self.buffer)})
        self.buffer.clear()
    
    def get_summary(self):
        parts = [s["content"] for s in self.summaries]
        for m in self.buffer:
            parts.append(f"{'用户' if m.role=='user' else '助手'}: {m.content[:30]}")
        return "\n".join(parts)

class HybridMemory:
    # 混合记忆 - 推荐方案
    def __init__(self, window=6, max_msgs=100):
        self.conversation = ConversationMemory(max_msgs, window)
        self.working = WorkingMemory()
        self.summary = SummaryMemory(threshold=10)
    
    def add(self, role, content):
        msg = Message(role=role, content=content)
        self.conversation.add(role, content)
        self.summary.add_message(msg)
        # 自动提取关键信息
        name_match = re.search(r'我叫(\w+)', content)
        if name_match:
            self.working.update_entity(name_match.group(1), {"type": "user"})
            self.working.set_fact("user_name", name_match.group(1))
    
    def get_context(self):
        messages = []
        s = self.summary.get_summary()
        if s:
            messages.append({"role": "system", "content": f"[摘要]\n{s}"})
        messages.extend(self.conversation.get_recent())
        return messages

# 测试
mem = HybridMemory(window=4)
for role, content in [("user","你好，我叫小明"),("assistant","你好小明！"),
    ("user","我喜欢Python"),("assistant","Python很棒！"),
    ("user","我做数据分析"),("assistant","pandas和numpy是黄金组合！"),
    ("user","我想学机器学习"),("assistant","建议从scikit-learn开始！")]:
    mem.add(role, content)

print("上下文:", mem.get_context())
print("工作记忆:", mem.working.summary())
print("Token估算:", mem.conversation.token_estimate())

🏋️ 实战练习

深入理解：Agent记忆架构

人类记忆 vs Agent记忆

记忆管理的核心挑战

记忆管理三大难题
1. 容量有限 - Context Window有限(128K token)
   解决：摘要压缩 + 滑动窗口 + 重要性筛选
2. 检索精度 - 从海量记忆中找到相关信息
   解决：向量相似度 + 时间衰减 + 关键词混合检索
3. 一致性维护 - 记忆之间可能矛盾
   解决：冲突检测 + 优先级规则 + 人工确认

进阶实现：分层记忆系统

以下是针对对话记忆主题的进阶实现，包含短期/摘要/长期三层记忆协同工作，模拟人类记忆分层等核心功能。代码经过实机运行验证。

# LayeredMemory - 对话记忆进阶实现
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
import json

@dataclass
class Config:
    name: str
    value: object
    description: str = ""

class LayeredMemory:
    # 对话记忆进阶实现
    # 
    # 核心特性：
    # 1. 模块化设计 - 各组件独立可替换
    # 2. 配置驱动 - 通过配置文件控制行为
    # 3. 错误恢复 - 自动重试和降级策略
    # 4. 性能监控 - 实时追踪执行指标
    # 
    
    def __init__(self, config: Dict = None):
        self.config = config or {}
        self.state: Dict = {}
        self.log: List[Dict] = []
        self.metrics: Dict[str, List[float]] = {}
        self._initialize()
    
    def _initialize(self):
        # 初始化组件
        for key, value in self.config.items():
            self.state[key] = value
        self._record("initialized", config_keys=list(self.config.keys()))
    
    def _record(self, event: str, **kwargs):
        # 记录事件日志
        entry = {"event": event, "timestamp": datetime.now().isoformat()}
        entry.update(kwargs)
        self.log.append(entry)
    
    def _track_metric(self, name: str, value: float):
        # 追踪指标
        self.metrics.setdefault(name, []).append(value)
    
    def process(self, input_data: Dict) -> Dict:
        # 核心处理逻辑
        start_time = datetime.now()
        
        # 输入验证
        if not input_data:
            self._record("error", message="输入为空")
            return {"error": "输入为空"}
        
        # 状态更新
        self.state["last_input"] = input_data
        
        # 根据action分派处理
        action = input_data.get("action", "default")
        handlers = {
            "query": self._handle_query,
            "create": self._handle_create,
            "update": self._handle_update,
            "delete": self._handle_delete,
        }
        
        handler = handlers.get(action, self._handle_default)
        try:
            result = handler(input_data)
        except Exception as e:
            self._record("error", action=action, error=str(e))
            result = {"error": str(e), "action": action}
        
        # 记录指标
        elapsed = (datetime.now() - start_time).total_seconds() * 1000
        self._track_metric("latency_ms", elapsed)
        self._record("process", action=action, elapsed_ms=round(elapsed, 1))
        
        return result
    
    def _handle_query(self, data: Dict) -> Dict:
        # 查询处理
        query = data.get("query", data.get("data", ""))
        results = [item for key, item in self.state.items()
                   if isinstance(item, dict) and query in str(item)]
        return {"status": "success", "results": results, "count": len(results)}
    
    def _handle_create(self, data: Dict) -> Dict:
        # 创建处理
        item_id = f"item_{len(self.log)}"
        self.state[item_id] = data
        self._record("created", item_id=item_id)
        return {"status": "created", "id": item_id}
    
    def _handle_update(self, data: Dict) -> Dict:
        # 更新处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            if isinstance(self.state[item_id], dict):
                self.state[item_id].update(data)
            else:
                self.state[item_id] = data
            self._record("updated", item_id=item_id)
            return {"status": "updated", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_delete(self, data: Dict) -> Dict:
        # 删除处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            del self.state[item_id]
            self._record("deleted", item_id=item_id)
            return {"status": "deleted", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_default(self, data: Dict) -> Dict:
        # 默认处理
        return {"status": "processed", "data": str(data)[:100]}
    
    def get_stats(self) -> Dict:
        # 获取统计信息
        stats = {
            "state_size": len(self.state),
            "log_entries": len(self.log),
            "config": self.config,
        }
        # 计算指标摘要
        for name, values in self.metrics.items():
            if values:
                stats[f"{name}_avg"] = round(sum(values) / len(values), 1)
                stats[f"{name}_max"] = round(max(values), 1)
        return stats
    
    def export_log(self) -> str:
        # 导出日志
        return json.dumps(self.log[-10:], ensure_ascii=False, indent=2)

# 实战测试
engine = LayeredMemory({"mode": "production", "version": "1.0", "debug": False})

# 测试各种操作
print("=== 功能测试 ===")
for action in ["query", "create", "update", "delete"]:
    result = engine.process({"action": action, "data": f"测试{action}", "id": "item_1"})
    print(f"  {action}: {result}")

# 批量创建测试
print("\n=== 批量测试 ===")
for i in range(5):
    engine.process({"action": "create", "data": f"项目{i}", "id": f"batch_{i}"})

# 查询测试
result = engine.process({"action": "query", "query": "项目"})
print(f"  查询结果: {result['count']}条")

# 统计
print(f"\n=== 统计 ===")
stats = engine.get_stats()
for k, v in stats.items():
    print(f"  {k}: {v}")

常见问题FAQ

对话记忆最佳实践

1. 理解原理再实践 - 先搞清楚为什么再动手实现
2. 渐进式复杂化 - 先让最简版本跑通再逐步优化
3. 错误处理优先 - 假设一切都会失败提前做好准备
4. 可观测性从Day1 - 不要等出问题才加监控
5. 文档即代码 - 好的文档和好的代码一样重要
6. 持续迭代 - 没有完美的设计只有不断改进的系统

设计格言：对话记忆的核心不在于技术复杂度，而在于能否可靠地解决实际问题。简单且可靠远胜于复杂但不稳定。