🧠 模块1：Agent 记忆机制实战

1.1 三种记忆类型：通俗理解

记忆类型	通俗比喻	生命周期	存储方式	典型容量
短期记忆	大脑的"工作台"——当前对话的上下文	当前会话有效	Python 列表（内存中）	几千字（受模型上下文限制）
长期记忆	"笔记本"——重要信息永久保存	跨会话持久化	JSON 文件 / SQLite / 向量数据库	几乎无限
会话记忆	"本次聊天记录"——这次对话的所有内容	一次对话生命周期	内存 + 可选持久化	整个对话过程

1.2 增强版记忆 Agent 完整代码

在项目目录下创建 memory_agent.py。这个 Agent 在第一阶段 MiniAgent 的基础上增加了完整的短期+长期+会话记忆系统。

#!/usr/bin/env python3
"""
MemoryAgent — 带完整记忆系统的 AI Agent
==========================================
记忆系统架构：
  📝 短期记忆 (ShortTermMemory)  — 当前上下文的最近N轮对话
  💾 长期记忆 (LongTermMemory)   — JSON文件持久化的关键信息
  📋 会话记忆 (SessionMemory)    — 本次会话的完整记录

运行：python3 memory_agent.py
"""

import json
import os
import re
from datetime import datetime
from pathlib import Path
from typing import Any

from dotenv import load_dotenv
from anthropic import Anthropic

load_dotenv()

# ================================================================
# 第1部分：记忆系统类
# ================================================================

class ShortTermMemory:
    """
    短期记忆 — Agent 的"工作台"
    存储最近 N 轮对话，直接拼接到每次 API 调用的上下文中
    """

    def __init__(self, max_turns: int = 10):
        self.max_turns = max_turns
        self.turns: list[dict] = []

    def add(self, role: str, content: str) -> None:
        """添加一轮对话"""
        self.turns.append({"role": role, "content": content, "time": datetime.now().isoformat()})
        # 超出上限时，移除最早的对话
        if len(self.turns) > self.max_turns * 2:  # user + assistant 各算一轮
            self.turns = self.turns[-(self.max_turns * 2):]

    def get_context(self) -> list[dict]:
        """获取可发送给 API 的消息格式"""
        return [{"role": t["role"], "content": t["content"]} for t in self.turns]

    def clear(self) -> None:
        """清空短期记忆"""
        self.turns = []

    def summarize(self) -> str:
        """生成短期记忆摘要（用于在记忆不足时压缩信息）"""
        if not self.turns:
            return "（无对话记录）"
        summary_parts = []
        for t in self.turns[-6:]:  # 只总结最近6条
            role_icon = "👤" if t["role"] == "user" else "🤖"
            short = t["content"][:80].replace("\n", " ")
            summary_parts.append(f"{role_icon} {short}...")
        return "\n".join(summary_parts)


class LongTermMemory:
    """
    长期记忆 — Agent 的"笔记本"
    将重要信息持久化到 JSON 文件，跨会话可用
    """

    def __init__(self, storage_path: str = "agent_memory.json"):
        self.storage_path = Path(storage_path)
        self.data: dict[str, Any] = self._load()

    def _load(self) -> dict:
        """从文件加载长期记忆"""
        if self.storage_path.exists():
            try:
                return json.loads(self.storage_path.read_text("utf-8"))
            except json.JSONDecodeError:
                print(f"⚠️ 记忆文件损坏，使用空记忆重新开始")
                return self._init_structure()
        return self._init_structure()

    def _init_structure(self) -> dict:
        return {
            "user_profile": {},       # 用户信息：姓名、偏好、习惯等
            "facts": [],              # 用户告诉过 Agent 的事实
            "preferences": {},        # 用户偏好设置
            "learned_skills": [],     # Agent 学到的新能力
            "important_notes": [],    # 用户标记为重要的笔记
            "stats": {                # 统计信息
                "total_conversations": 0,
                "total_turns": 0,
                "created_at": datetime.now().isoformat(),
                "last_updated": datetime.now().isoformat(),
            }
        }

    def save(self) -> None:
        """保存长期记忆到文件"""
        self.data["stats"]["last_updated"] = datetime.now().isoformat()
        self.storage_path.write_text(
            json.dumps(self.data, ensure_ascii=False, indent=2),
            encoding="utf-8"
        )

    def remember_fact(self, fact: str, category: str = "general") -> None:
        """记住一个事实"""
        self.data["facts"].append({
            "content": fact,
            "category": category,
            "remembered_at": datetime.now().isoformat()
        })
        self.save()

    def remember_preference(self, key: str, value: str) -> None:
        """记住用户偏好"""
        self.data["preferences"][key] = value
        self.save()

    def add_note(self, note: str, importance: str = "normal") -> None:
        """添加重要笔记"""
        self.data["important_notes"].append({
            "content": note,
            "importance": importance,
            "created_at": datetime.now().isoformat()
        })
        self.save()

    def recall_facts(self, keyword: str = None, category: str = None) -> list:
        """回忆相关事实（简单关键词匹配）"""
        results = self.data["facts"]
        if keyword:
            results = [f for f in results if keyword.lower() in f["content"].lower()]
        if category:
            results = [f for f in results if f["category"] == category]
        return results

    def recall_preferences(self) -> dict:
        """获取所有用户偏好"""
        return self.data["preferences"]

    def get_all_notes(self) -> list:
        """获取所有笔记"""
        return self.data["important_notes"]

    def forget_old_facts(self, days: int = 30) -> int:
        """遗忘超过指定天数的旧事实（记忆清理）"""
        cutoff = datetime.now().timestamp() - days * 86400
        old_count = len(self.data["facts"])
        self.data["facts"] = [
            f for f in self.data["facts"]
            if datetime.fromisoformat(f["remembered_at"]).timestamp() > cutoff
        ]
        removed = old_count - len(self.data["facts"])
        if removed > 0:
            self.save()
        return removed


class SessionMemory:
    """
    会话记忆 — 本次对话的完整记录
    用于会话结束后的总结和分析
    """

    def __init__(self):
        self.session_id = datetime.now().strftime("%Y%m%d_%H%M%S")
        self.turns: list[dict] = []
        self.start_time = datetime.now()

    def record(self, user_input: str, agent_response: str, metadata: dict = None) -> None:
        """记录一轮对话"""
        self.turns.append({
            "user": user_input,
            "agent": agent_response,
            "metadata": metadata or {},
            "timestamp": datetime.now().isoformat(),
            "turn_number": len(self.turns) + 1,
        })

    def get_duration(self) -> str:
        """获取会话持续时间"""
        delta = datetime.now() - self.start_time
        minutes = int(delta.total_seconds() / 60)
        return f"{minutes} 分钟"

    def get_summary(self) -> dict:
        """获取会话摘要"""
        return {
            "session_id": self.session_id,
            "total_turns": len(self.turns),
            "duration": self.get_duration(),
            "start_time": self.start_time.isoformat(),
            "topics": self._extract_topics(),
        }

    def _extract_topics(self) -> list:
        """简单提取会话主题（基于高频词）"""
        all_text = " ".join([t["user"] for t in self.turns])
        # 简单的关键词提取（生产环境可用 NLP 库）
        keywords = ["计算", "文件", "时间", "天气", "代码", "笔记", "记忆", "工具"]
        found = [kw for kw in keywords if kw in all_text]
        return found if found else ["一般对话"]

    def export(self, filepath: str = None) -> str:
        """导出会话记录为 Markdown"""
        if filepath is None:
            filepath = f"session_{self.session_id}.md"

        lines = [
            f"# 会话记录 — {self.session_id}",
            f"时间: {self.start_time.isoformat()} | 轮次: {len(self.turns)} | 时长: {self.get_duration()}",
            "",
        ]
        for t in self.turns:
            lines.append(f"## 第{t['turn_number']}轮")
            lines.append(f"**👤 用户:** {t['user']}")
            lines.append(f"**🤖 Agent:** {t['agent']}")
            lines.append("")

        Path(filepath).write_text("\n".join(lines), encoding="utf-8")
        return filepath


# ================================================================
# 第2部分：带记忆的 Agent
# ================================================================

class MemoryAgent:
    """具备完整记忆系统的 AI Agent"""

    def __init__(self, name: str = "MemoryAgent"):
        self.name = name
        self.client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
        self.model = "claude-sonnet-4-6"

        # 三层记忆系统
        self.short_term = ShortTermMemory(max_turns=10)
        self.long_term = LongTermMemory()
        self.session = SessionMemory()

        # 注册工具
        self.tools = {
            "remember": self._tool_remember,
            "recall": self._tool_recall,
            "note": self._tool_note,
            "my_preferences": self._tool_show_preferences,
        }

        self._load_long_term_into_short_term()

    def _load_long_term_into_short_term(self):
        """将长期记忆中的关键信息加载到短期记忆中"""
        prefs = self.long_term.recall_preferences()
        if prefs:
            prefs_text = "已知用户偏好: " + ", ".join(
                [f"{k}={v}" for k, v in prefs.items()]
            )
            self.short_term.add("system", prefs_text)

    # --- 工具定义 ---

    def _tool_remember(self, fact: str) -> str:
        """记住用户说的事实"""
        self.long_term.remember_fact(fact)
        return f"✅ 已记住: {fact}"

    def _tool_recall(self, keyword: str) -> str:
        """回忆相关事实"""
        results = self.long_term.recall_facts(keyword=keyword)
        if not results:
            return f"🔍 没有找到与 '{keyword}' 相关的记忆"
        items = [f"{i+1}. {r['content']} (记于 {r['remembered_at'][:10]})"
                 for i, r in enumerate(results[:10])]
        return "📋 相关记忆:\n" + "\n".join(items)

    def _tool_note(self, content: str) -> str:
        """添加笔记"""
        self.long_term.add_note(content)
        return f"📝 笔记已保存: {content[:50]}..."

    def _tool_show_preferences(self, _=None) -> str:
        """查看用户偏好"""
        prefs = self.long_term.recall_preferences()
        if not prefs:
            return "📋 暂无用户偏好记录"
        items = [f"  • {k}: {v}" for k, v in prefs.items()]
        return "📋 用户偏好:\n" + "\n".join(items)

    # --- 核心运行 ---

    def _build_system_prompt(self) -> str:
        """构建包含记忆上下文的系统提示词"""
        memory_context = self.short_term.summarize()
        return f"""你是 {self.name}，一个具备记忆能力的智能助手。

【你的记忆能力】
- 短期记忆：你记得本次对话中最近的内容
- 长期记忆：你可以使用工具记住重要信息，跨对话保留
- 每次对话开始时会加载用户的偏好和重要记忆

【可用工具】
- remember — 记住一个事实（如 "用户喜欢简洁的回答"），参数: fact
- recall — 回忆相关事实，参数: keyword（搜索关键词）
- note — 保存一条笔记，参数: content
- my_preferences — 查看用户偏好，无需参数

【近期对话摘要】
{memory_context}

【重要规则】
- 用户告诉你重要信息时，主动使用 remember 工具记录下来
- 遇到相关问题，主动使用 recall 工具回忆之前的记录
- 优先使用中文回复
"""

    def run(self, user_input: str) -> str:
        """运行 Agent 主循环"""
        print(f"\n{'─'*50}")
        print(f"👤 用户: {user_input}")

        # 添加到短期记忆
        self.short_term.add("user", user_input)

        # 构建消息
        system_prompt = self._build_system_prompt()
        messages = self.short_term.get_context()

        # 调用 API
        try:
            response = self.client.messages.create(
                model=self.model,
                max_tokens=1024,
                system=system_prompt,
                messages=messages,
                temperature=0.3,
            )
            reply = response.content[0].text.strip()
        except Exception as e:
            reply = f"❌ Agent 调用失败: {str(e)}"

        # 处理工具调用
        result = self._process_tool_calls(reply)

        # 更新记忆
        self.short_term.add("assistant", result)
        self.session.record(user_input, result)

        # 检测用户偏好并自动记忆
        self._auto_detect_preferences(user_input)

        print(f"🤖 {self.name}: {result}")
        return result

    def _process_tool_calls(self, reply: str) -> str:
        """解析并执行工具调用"""
        try:
            json_str = reply
            for marker in ["```json", "```"]:
                if marker in reply:
                    json_str = reply.split(marker)[1].split("```")[0]
                    break
            tool_call = json.loads(json_str.strip())
            if "tool" in tool_call:
                tool_name = tool_call["tool"]
                args = tool_call.get("args", {})
                first_arg = list(args.values())[0] if args else ""
                if tool_name in self.tools:
                    return self.tools[tool_name](first_arg)
                return f"❌ 未知工具: {tool_name}"
        except (json.JSONDecodeError, KeyError, IndexError):
            pass
        return reply

    def _auto_detect_preferences(self, user_input: str):
        """自动检测用户偏好并记忆"""
        patterns = {
            "response_style": [
                (r"(简洁|简略|简短)(一点|些|的)", "简洁"),
                (r"(详细|详细点|具体)(一点|些|的)", "详细"),
            ],
            "language": [
                (r"(说|用|使用)(英文|英语|English)", "英文"),
                (r"(说|用|使用)(中文|汉语)", "中文"),
            ],
        }
        for pref_key, triggers in patterns.items():
            for pattern, value in triggers:
                if re.search(pattern, user_input):
                    self.long_term.remember_preference(pref_key, value)
                    print(f"  💡 自动记住偏好: {pref_key} = {value}")
                    break

    def show_stats(self) -> None:
        """显示记忆系统统计"""
        stats = self.long_term.data["stats"]
        print(f"\n📊 {self.name} 记忆系统统计:")
        print(f"  短期记忆轮次: {len(self.short_term.turns) // 2}")
        print(f"  长期记忆事实: {len(self.long_term.data['facts'])} 条")
        print(f"  用户偏好: {len(self.long_term.data['preferences'])} 项")
        print(f"  笔记: {len(self.long_term.data['important_notes'])} 条")
        print(f"  本次会话: {self.session.get_summary()['total_turns']} 轮 | {self.session.get_duration()}")
        print(f"  历史总对话: {stats['total_conversations']} 次 | 历史总轮次: {stats['total_turns']}")

    def cleanup_old_memories(self, days: int = 30) -> int:
        """清理旧记忆"""
        return self.long_term.forget_old_facts(days=days)

    def close(self) -> str:
        """关闭 Agent，保存会话"""
        self.long_term.data["stats"]["total_conversations"] += 1
        self.long_term.data["stats"]["total_turns"] += len(self.session.turns)
        self.long_term.save()
        export_path = self.session.export()
        return export_path


# ================================================================
# 第3部分：交互入口
# ================================================================

def main():
    print("=" * 50)
    print("🧠 MemoryAgent — 带记忆的 AI Agent")
    print("=" * 50)
    print("试试这些命令：")
    print("  • 告诉我一些关于你的事 → Agent 会记住")
    print("  • 回忆我刚才说的 → Agent 会查询记忆")
    print("  • 我喜欢xxx → Agent 自动记住偏好")
    print("  • 保存笔记：xxx → Agent 保存笔记")
    print("  • stats → 查看记忆统计")
    print("  • quit → 退出并导出会话记录")
    print("=" * 50)

    if not os.getenv("ANTHROPIC_API_KEY"):
        print("❌ 请先配置 ANTHROPIC_API_KEY！")
        return

    agent = MemoryAgent()
    print(f"\n🤖 {agent.name} 已就绪！\n")

    while True:
        try:
            user_input = input("👤 你: ").strip()
            if not user_input:
                continue
            if user_input.lower() in ("quit", "exit", "q"):
                break
            if user_input.lower() == "stats":
                agent.show_stats()
                continue
            if user_input.lower() == "cleanup":
                removed = agent.cleanup_old_memories(30)
                print(f"🧹 清理了 {removed} 条旧记忆")
                continue

            agent.run(user_input)

        except KeyboardInterrupt:
            break

    export_path = agent.close()
    print(f"\n📄 会话记录已导出: {export_path}")
    agent.show_stats()
    print(f"👋 {agent.name} 已退出。再见！")


if __name__ == "__main__":
    main()

1.3 记忆优化策略

策略1：冗余信息过滤

# 在 ShortTermMemory.add() 中添加去重逻辑
def add(self, role: str, content: str) -> None:
    # 去重：如果和最近一条内容相同，跳过
    if self.turns and self.turns[-1]["content"] == content:
        return
    # ... 原有逻辑

策略2：关键信息提取（用 Claude 帮你摘要）

def extract_key_info(self, conversation: str) -> str:
    """使用 Claude 提取对话中的关键信息"""
    response = self.client.messages.create(
        model=self.model,
        max_tokens=200,
        messages=[{"role": "user", "content":
            f"从以下对话中提取关键信息（人名、偏好、事实、任务），用一句话总结。\n\n{conversation}"}],
    )
    return response.content[0].text.strip()

策略3：分层记忆——热数据 vs 冷数据

• 热数据（短期记忆）：最近5轮对话，直接塞入 API 上下文
• 温数据（近期摘要）：5-20轮对话的摘要，可选塞入
• 冷数据（长期存储）：JSON 文件，需要时通过 recall 工具查询

1.4 记忆溢出与上下文超限解决方案

问题场景	症状	解决方案
上下文超过模型限制	API 报错 context_length_exceeded	1) 减少 max_turns 到 5 或更少 2) 用摘要代替原始对话 3) 截断每条消息的长度
记忆文件过大	加载慢、内存占用高	定期清理（forget_old_facts） 分文件存储不同类别的记忆
短期记忆"遗忘"重要信息	Agent 在长对话中丢失早期重要内容	在长期记忆中标记"重要"标志 重要信息不参与淘汰

📝 知识点总结

概念	一句话总结
短期记忆	内存中保存的最近对话，每次 API 调用时带上
长期记忆	JSON/SQLite 文件持久化的关键信息，跨会话保留
会话记忆	完整对话记录，用于总结导出和分析
记忆优化核心	不是记住一切，而是记住该记的、过滤冗余的