Agent 记忆系统

Agent 记忆系统#

为什么 Agent 需要记忆？#

大模型本身是无状态的：每次调用都从零开始，不记得上一轮说了什么。 Agent 需要记忆才能处理长期任务、维持上下文连贯性。

记忆的四个层次#

层次	类型	适用场景	容量
1	对话历史（滑动窗口）	短期对话	最近 N 条
2	摘要记忆	长期对话	压缩后的摘要
3	实体记忆	人物/事实跟踪	结构化键值对
4	向量记忆	语义检索	全量历史

本章将逐一实现这四种记忆，最后组合使用。

import json
import os
from collections import deque
import litellm
litellm.drop_params = True
from dotenv import load_dotenv

load_dotenv()

MODEL = os.getenv("LLM_MODEL")
EMBED_MODEL = os.getenv("EMBED_MODEL", "openai/text-embedding-3-small")

print(f"对话模型：{MODEL}")
print(f"嵌入模型：{EMBED_MODEL}")
# gpt-5/o系列不支持自定义temperature值，统一用安全wrapper
def _c(**kw):
    _m = kw.get('model', MODEL)
    if any(_m.startswith(p) for p in ('openai/gpt-5','openai/o1','openai/o3','openai/o4')):
        kw.pop('temperature', None)
    return litellm.completion(**kw)

对话模型：openai/gpt-5-mini
嵌入模型：openai/text-embedding-3-small

Section 1：对话历史（滑动窗口）#

最简单的记忆方式：保留最近的 N 条消息。

优点：实现简单，延迟低缺点：窗口之外的信息完全丢失

使用 collections.deque 的 maxlen 参数，自动丢弃最旧的消息。

class ConversationMemory:
    """
    滑动窗口对话记忆。
    只保留最近 window_size 条消息，超出时自动丢弃最旧的。
    """

    def __init__(self, window_size: int = 6, system_prompt: str = "你是一个友好的助手。"):
        self.window_size = window_size
        self.system_prompt = system_prompt
        # deque maxlen 会自动丢弃超出部分（先进先出）
        self._history = deque(maxlen=window_size)
        self._total_turns = 0  # 统计总对话轮数

    def chat(self, user_message: str) -> str:
        """发送消息并获取回复，自动维护滑动窗口"""
        self._history.append({"role": "user", "content": user_message})
        self._total_turns += 1

        messages = [
            {"role": "system", "content": self.system_prompt},
            *list(self._history),
        ]

        response = _c(
            model=MODEL,
            messages=messages,
            temperature=0.7,
            max_tokens=256,
        )
        reply = response.choices[0].message.content
        self._history.append({"role": "assistant", "content": reply})

        return reply

    @property
    def current_window_size(self) -> int:
        return len(self._history)

    def show_stats(self):
        print(f"总对话轮数：{self._total_turns}")
        print(f"窗口大小：{self.window_size} 条（当前保留 {self.current_window_size} 条）")
        print(f"丢弃轮次：{max(0, self._total_turns * 2 - self.window_size)} 条消息")


# ── 演示：超长对话，观察窗口截断 ────────────────────────────
memory = ConversationMemory(window_size=4)  # 只保留最近 4 条消息

conversations = [
    "我叫小明，今年 25 岁",
    "我喜欢打篮球",
    "我最近在学 Python",
    "我有一只猫叫小白",
    "我的名字是什么？",  # 测试：能否记住第 1 条消息中的名字？
]

print("滑动窗口记忆演示（窗口大小=4）")
print("="*50)

for i, msg in enumerate(conversations):
    print(f"\n[第 {i+1} 轮] 用户：{msg}")
    reply = memory.chat(msg)
    print(f"        助手：{reply}")
    print(f"        当前窗口：{memory.current_window_size} 条消息")

print()
memory.show_stats()

滑动窗口记忆演示（窗口大小=4）
==================================================

[第 1 轮] 用户：我叫小明，今年 25 岁

        助手：
        当前窗口：2 条消息

[第 2 轮] 用户：我喜欢打篮球

        助手：
        当前窗口：4 条消息

[第 3 轮] 用户：我最近在学 Python

        助手：
        当前窗口：4 条消息

[第 4 轮] 用户：我有一只猫叫小白

        助手：
        当前窗口：4 条消息

[第 5 轮] 用户：我的名字是什么？

        助手：
        当前窗口：4 条消息

总对话轮数：5
窗口大小：4 条（当前保留 4 条）
丢弃轮次：6 条消息

观察上面的输出：当问「我的名字是什么？」时，模型很可能不记得了。因为包含名字的消息已经被滑出了窗口。

这就是摘要记忆要解决的问题。

Section 2：摘要记忆#

核心思路：当历史消息超过阈值时，调用 LLM 将旧消息压缩成摘要，只保留摘要 + 最近 N 条消息。

旧消息 (1~n)  →  LLM 摘要  →  [摘要] + 最近 k 条

压缩率通常能达到 5:1 甚至更高，大幅降低 token 消耗。

class SummaryMemory:
    """
    摘要记忆：超过阈值后自动压缩旧消息为摘要。

    工作流程：
    1. 正常累积对话历史
    2. 当历史超过 compress_threshold 条时，触发压缩
    3. 将旧的 (总量 - keep_recent) 条消息发给 LLM 生成摘要
    4. 用摘要替换旧消息，保留最近 keep_recent 条
    """

    def __init__(
        self,
        system_prompt: str = "你是一个友好的助手。",
        compress_threshold: int = 8,  # 超过这个数量触发压缩
        keep_recent: int = 4,          # 压缩后保留最近 N 条
    ):
        self.system_prompt = system_prompt
        self.compress_threshold = compress_threshold
        self.keep_recent = keep_recent
        self._history = []       # 完整当前历史
        self._summary = ""       # 压缩后的摘要
        self._compression_count = 0
        self._original_tokens = 0
        self._summary_tokens = 0

    def _compress(self):
        """将旧消息压缩为摘要"""
        # 要压缩的消息（保留最近 keep_recent 条）
        to_compress = self._history[:-self.keep_recent] if self.keep_recent > 0 else self._history[:]
        recent = self._history[-self.keep_recent:] if self.keep_recent > 0 else []

        # 估算原始 token 数（简单用字符数 / 2 近似）
        old_text = "\n".join(f"{m['role']}: {m['content']}" for m in to_compress)
        self._original_tokens += len(old_text) // 2

        # 构建压缩提示
        compress_prompt = f"""请将以下对话历史压缩为简洁的摘要，保留所有重要信息（人名、事实、决定等）。

已有摘要：{self._summary or '（无）'}

新增对话：
{old_text}

请输出更新后的摘要（100 字以内）："""

        response = _c(
            model=MODEL,
            messages=[{"role": "user", "content": compress_prompt}],
            temperature=0,
            max_tokens=200,
        )
        self._summary = response.choices[0].message.content
        self._summary_tokens += len(self._summary) // 2
        self._history = recent
        self._compression_count += 1

        print(f"  [摘要记忆] 第 {self._compression_count} 次压缩完成")
        print(f"  摘要：{self._summary[:80]}..." if len(self._summary) > 80 else f"  摘要：{self._summary}")

    def _build_messages(self) -> list:
        """构建发给模型的消息列表"""
        system = self.system_prompt
        if self._summary:
            system += f"\n\n对话摘要（之前的对话记录）：\n{self._summary}"
        return [{"role": "system", "content": system}, *self._history]

    def chat(self, user_message: str) -> str:
        self._history.append({"role": "user", "content": user_message})

        # 检查是否需要压缩
        if len(self._history) >= self.compress_threshold:
            self._compress()

        response = _c(
            model=MODEL,
            messages=self._build_messages(),
            temperature=0.7,
            max_tokens=256,
        )
        reply = response.choices[0].message.content
        self._history.append({"role": "assistant", "content": reply})
        return reply

    def show_compression_stats(self):
        ratio = self._original_tokens / max(self._summary_tokens, 1)
        print(f"压缩次数：{self._compression_count}")
        print(f"原始 token（估算）：{self._original_tokens}")
        print(f"摘要 token（估算）：{self._summary_tokens}")
        print(f"压缩率：约 {ratio:.1f}:1")


# ── 演示：长对话摘要压缩 ──────────────────────────────────
smem = SummaryMemory(compress_threshold=6, keep_recent=2)

long_conversation = [
    "你好，我叫张伟，是一名软件工程师",
    "我正在做一个 AI 聊天机器人项目",
    "项目使用 Python 和 FastAPI",
    "我的团队有 5 个人",
    "我们预计下个月上线",
    "我最喜欢的编程语言是 Rust",
    "我住在北京朝阳区",
    "我叫什么名字？我在做什么项目？",  # 测试记忆
]

print("摘要记忆演示（压缩阈值=6，保留最近=2）")
print("="*50)

for i, msg in enumerate(long_conversation):
    print(f"\n[第 {i+1} 轮] 用户：{msg}")
    reply = smem.chat(msg)
    print(f"        助手：{reply}")

print()
smem.show_compression_stats()

摘要记忆演示（压缩阈值=6，保留最近=2）
==================================================

[第 1 轮] 用户：你好，我叫张伟，是一名软件工程师

        助手：你好，张伟！很高兴认识你。你是软件工程师——太棒了。我要怎么帮助你？例如：

- 代码问题（调试、性能优化、重构）
- 技术选型或架构设计
- 面试准备（算法题、系统设计、行为面试）
- 简历或岗位申请建议
- 学习路线或资源推荐（前端/后端/云/机器学习等）
- 项目点子或具体实现方案

告诉我你主要使用的编程语言/技术栈以及当前最想解决的事情，我就能给出更具体的建议或示例。

[第 2 轮] 用户：我正在做一个 AI 聊天机器人项目

        助手：

[第 3 轮] 用户：项目使用 Python 和 FastAPI

        助手：

[第 4 轮] 用户：我的团队有 5 个人

  [摘要记忆] 第 1 次压缩完成
  摘要：

        助手：

[第 5 轮] 用户：我们预计下个月上线

        助手：

[第 6 轮] 用户：我最喜欢的编程语言是 Rust

  [摘要记忆] 第 2 次压缩完成
  摘要：

        助手：

[第 7 轮] 用户：我住在北京朝阳区

        助手：

[第 8 轮] 用户：我叫什么名字？我在做什么项目？

  [摘要记忆] 第 3 次压缩完成
  摘要：

        助手：

压缩次数：3
原始 token（估算）：202
摘要 token（估算）：0
压缩率：约 202.0:1

Section 3：实体记忆#

实体记忆专门追踪对话中提到的关键实体：人名、地名、事实、偏好等。

工作流程：

每轮对话后，调用 LLM 从新消息中提取实体
更新实体字典（新信息覆盖或补充旧信息）
将相关实体注入系统提示，让模型「记住」它们

优势：精确跟踪用户信息，不会因窗口截断而遗忘关键事实

class EntityMemory:
    """
    实体记忆：提取并存储对话中的关键实体信息。

    实体以 JSON 字典形式存储，例如：
    {
        "用户名字": "小明",
        "用户的猫": "叫小白",
        "用户职业": "工程师"
    }
    """

    def __init__(self, system_prompt: str = "你是一个友好的助手。"):
        self.system_prompt = system_prompt
        self._history = []       # 最近几条消息（小窗口）
        self._entities = {}      # 实体字典
        self._window = 6        # 保留最近 6 条

    def _extract_entities(self, new_messages: list) -> dict:
        """从新消息中提取实体，返回新发现的实体字典"""
        text = "\n".join(f"{m['role']}: {m['content']}" for m in new_messages)
        existing = json.dumps(self._entities, ensure_ascii=False)

        prompt = f"""从以下对话中提取关键实体信息（人名、地点、宠物、偏好、事实等）。
已知实体：{existing}

新对话：
{text}

请以 JSON 格式输出新发现或更新的实体（只输出 JSON，无需其他文字）：
例如：{{"用户名字": "小明", "用户的猫": "叫小白"}}
如果没有新实体，输出：{{}}"""

        response = _c(
            model=MODEL,
            messages=[{"role": "user", "content": prompt}],
            temperature=0,
            max_tokens=200,
        )
        content = response.choices[0].message.content.strip()

        # 尝试解析 JSON
        try:
            # 去掉可能的 Markdown 代码块
            if content.startswith("```"):
                content = content.split("\n", 1)[1].rsplit("\n", 1)[0]
            return json.loads(content)
        except json.JSONDecodeError:
            return {}

    def chat(self, user_message: str) -> str:
        self._history.append({"role": "user", "content": user_message})

        # 构建包含实体信息的系统提示
        system = self.system_prompt
        if self._entities:
            entity_str = "\n".join(f"- {k}：{v}" for k, v in self._entities.items())
            system += f"\n\n已知用户信息：\n{entity_str}"

        messages = [
            {"role": "system", "content": system},
            *self._history[-self._window:],
        ]

        response = _c(
            model=MODEL,
            messages=messages,
            temperature=0.7,
            max_tokens=256,
        )
        reply = response.choices[0].message.content
        self._history.append({"role": "assistant", "content": reply})

        # 异步提取实体（实际应用中可以放后台线程）
        new_entities = self._extract_entities(self._history[-2:])
        if new_entities:
            self._entities.update(new_entities)
            print(f"  [实体记忆] 更新实体：{new_entities}")

        return reply

    def show_entities(self):
        print("当前已知实体：")
        for k, v in self._entities.items():
            print(f"  {k}: {v}")


# ── 演示：跨轮次实体记忆 ─────────────────────────────────
emem = EntityMemory()

entity_conversation = [
    "我叫李雷，是一名医生",
    "我有一只猫叫小白，它非常喜欢吃鱼",
    "我住在上海",
    "今天天气真好",  # 无关消息
    "我的宠物喜欢吃什么？",  # 需要记忆：小白喜欢吃鱼
    "我是做什么工作的？",    # 需要记忆：医生
]

print("实体记忆演示")
print("="*50)

for i, msg in enumerate(entity_conversation):
    print(f"\n[第 {i+1} 轮] 用户：{msg}")
    reply = emem.chat(msg)
    print(f"        助手：{reply}")

print()
emem.show_entities()

实体记忆演示
==================================================

[第 1 轮] 用户：我叫李华，是一名医生

        助手：

[第 2 轮] 用户：我有一只猫叫小白，它非常喜欢吃鱼

        助手：

[第 3 轮] 用户：我住在上海

  [实体记忆] 更新实体：{'居住地': '上海'}
        助手：

[第 4 轮] 用户：今天天气真好

        助手：

[第 5 轮] 用户：我的宠物喜欢吃什么？

        助手：

[第 6 轮] 用户：我是做什么工作的？

        助手：

当前已知实体：
  居住地: 上海

Section 4：向量记忆（轻量实现）#

向量记忆将历史对话转换为嵌入向量，通过语义相似度检索最相关的历史片段。

不需要 ChromaDB 等向量数据库——用 NumPy 实现余弦相似度即可。

适用场景：历史记录很长，但当前问题只与部分历史相关。

import numpy as np


def get_embedding(text: str) -> np.ndarray:
    """获取文本的嵌入向量"""
    response = litellm.embedding(
        model=EMBED_MODEL,
        input=[text],
    )
    return np.array(response.data[0]["embedding"])


def cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
    """计算两个向量的余弦相似度（-1 到 1，越大越相似）"""
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) + 1e-8))


class VectorMemory:
    """
    向量记忆：基于语义相似度检索相关历史对话。

    每条历史记录存储：
    - text: 原始文本
    - embedding: 嵌入向量
    - role: 角色 (user/assistant)
    """

    def __init__(
        self,
        system_prompt: str = "你是一个智能助手。",
        top_k: int = 3,              # 检索最相关的 K 条历史
        similarity_threshold: float = 0.5,  # 相似度阈值
    ):
        self.system_prompt = system_prompt
        self.top_k = top_k
        self.similarity_threshold = similarity_threshold
        self._store = []   # [{"text": ..., "embedding": ..., "role": ...}]

    def _store_turn(self, role: str, text: str):
        """将一条消息存入向量存储"""
        embedding = get_embedding(text)
        self._store.append({
            "role": role,
            "text": text,
            "embedding": embedding,
        })

    def _retrieve_relevant(self, query: str) -> list:
        """根据查询语义检索最相关的历史条目"""
        if not self._store:
            return []

        query_emb = get_embedding(query)
        scored = [
            (cosine_similarity(query_emb, item["embedding"]), item)
            for item in self._store
        ]
        # 过滤低于阈值，按相似度排序
        scored = [(s, item) for s, item in scored if s >= self.similarity_threshold]
        scored.sort(key=lambda x: x[0], reverse=True)
        return [item for _, item in scored[:self.top_k]]

    def chat(self, user_message: str) -> str:
        # 检索相关历史
        relevant = self._retrieve_relevant(user_message)

        # 构建消息
        system = self.system_prompt
        if relevant:
            context = "\n".join(
                f"[历史] {item['role']}: {item['text']}" for item in relevant
            )
            system += f"\n\n相关历史对话：\n{context}"
            print(f"  [向量记忆] 检索到 {len(relevant)} 条相关历史")

        messages = [
            {"role": "system", "content": system},
            {"role": "user", "content": user_message},
        ]

        response = _c(
            model=MODEL,
            messages=messages,
            temperature=0.7,
            max_tokens=256,
        )
        reply = response.choices[0].message.content

        # 存储本轮对话
        self._store_turn("user", user_message)
        self._store_turn("assistant", reply)

        return reply

    def show_store_size(self):
        print(f"向量存储条目数：{len(self._store)}")
        if self._store:
            dim = len(self._store[0]['embedding'])
            print(f"嵌入维度：{dim}")


# ── 演示：语义检索历史 ─────────────────────────────────────
vmem = VectorMemory(top_k=2, similarity_threshold=0.4)

# 预先存入一些历史对话
seed_history = [
    ("user", "我喜欢打羽毛球，每周打三次"),
    ("assistant", "羽毛球是很好的有氧运动！"),
    ("user", "我的猫叫花花，是一只橘猫"),
    ("assistant", "橘猫很可爱！花花听起来很讨人喜欢。"),
    ("user", "我在学习机器学习，最近在研究 Transformer"),
    ("assistant", "Transformer 是现代 NLP 的基础架构。"),
]

print("存入历史对话到向量记忆...")
for role, text in seed_history:
    vmem._store_turn(role, text)
print(f"已存入 {len(vmem._store)} 条历史")

# 现在测试语义检索
print("\n向量记忆语义检索演示")
print("="*50)

test_questions = [
    "我平时有什么运动爱好？",  # 应检索到羽毛球相关
    "我的宠物叫什么名字？",    # 应检索到花花相关
]

for q in test_questions:
    print(f"\n用户：{q}")
    reply = vmem.chat(q)
    print(f"助手：{reply}")

vmem.show_store_size()

存入历史对话到向量记忆...

已存入 6 条历史

向量记忆语义检索演示
==================================================

用户：我平时有什么运动爱好？

  [向量记忆] 检索到 2 条相关历史

助手：你平时喜欢打羽毛球，大约每周打三次。要不要我帮你制定训练计划或给些练习建议？

用户：我的宠物叫什么名字？

  [向量记忆] 检索到 2 条相关历史

助手：你的宠物叫花花，是一只橘猫。
向量存储条目数：10
嵌入维度：1536

Section 5：组合记忆（HybridMemory）#

实际应用中，单一记忆类型往往不够。 HybridMemory 将滑动窗口和实体记忆结合：

滑动窗口：保证最近对话的连贯性
实体记忆：跨越窗口持续追踪关键信息

演示场景：一个关于项目管理的多轮对话。

class HybridMemory:
    """
    混合记忆：滑动窗口 + 实体记忆。

    - 短期：滑动窗口保证最近对话连贯
    - 长期：实体记忆跨越窗口保持关键信息
    """

    def __init__(
        self,
        system_prompt: str = "你是一个专业的项目管理助手。",
        window_size: int = 6,
    ):
        self.system_prompt = system_prompt
        self._window = deque(maxlen=window_size)
        self._entities = {}
        self._turn_count = 0

    def _update_entities(self, user_msg: str, assistant_msg: str):
        """从最新一轮对话中提取实体"""
        prompt = f"""从以下对话中提取关键信息（项目名称、截止日期、负责人、状态、决定等）。
已知信息：{json.dumps(self._entities, ensure_ascii=False)}

新对话：
用户：{user_msg}
助手：{assistant_msg}

以 JSON 输出新发现的信息，没有则输出 {{}}："""

        try:
            response = _c(
                model=MODEL,
                messages=[{"role": "user", "content": prompt}],
                temperature=0,
                max_tokens=150,
            )
            content = response.choices[0].message.content.strip()
            if content.startswith("```"):
                content = content.split("\n", 1)[1].rsplit("\n", 1)[0]
            new_entities = json.loads(content)
            if new_entities:
                self._entities.update(new_entities)
                print(f"  [实体更新] {new_entities}")
        except Exception:
            pass  # 解析失败时静默跳过

    def _build_system_prompt(self) -> str:
        system = self.system_prompt
        if self._entities:
            info = "\n".join(f"- {k}：{v}" for k, v in self._entities.items())
            system += f"\n\n当前已知项目信息：\n{info}"
        return system

    def chat(self, user_message: str) -> str:
        self._turn_count += 1
        self._window.append({"role": "user", "content": user_message})

        messages = [
            {"role": "system", "content": self._build_system_prompt()},
            *list(self._window),
        ]

        response = _c(
            model=MODEL,
            messages=messages,
            temperature=0.7,
            max_tokens=256,
        )
        reply = response.choices[0].message.content
        self._window.append({"role": "assistant", "content": reply})

        # 每轮更新实体
        self._update_entities(user_message, reply)
        return reply

    def status(self):
        print(f"对话轮次：{self._turn_count}")
        print(f"窗口消息数：{len(self._window)}")
        print(f"实体数量：{len(self._entities)}")
        if self._entities:
            print("已知实体：")
            for k, v in self._entities.items():
                print(f"  {k}: {v}")


# ── 演示：项目管理多轮对话 ───────────────────────────────
hmem = HybridMemory(system_prompt="你是一个专业的项目管理助手，帮助用户追踪项目进度。")

project_conversation = [
    "我们有个项目叫'星辰计划'，负责人是王芳",
    "项目截止日期是 3 月 31 日",
    "目前完成了需求分析和设计阶段",
    "开发团队遇到了一个技术难题：数据库性能问题",
    "我们决定改用 PostgreSQL 替换 MySQL",
    "好的，现在说说今天的天气",  # 无关消息
    "回到项目：星辰计划现在进展如何？负责人是谁？截止日期是什么时候？",
]

print("混合记忆演示：项目管理对话")
print("="*50)

for i, msg in enumerate(project_conversation):
    print(f"\n[第 {i+1} 轮] 用户：{msg}")
    reply = hmem.chat(msg)
    print(f"        助手：{reply}")

print("\n" + "="*50)
print("记忆系统状态：")
hmem.status()

混合记忆演示：项目管理对话
==================================================

[第 1 轮] 用户：我们有个项目叫'星辰计划'，负责人是王芳

        助手：

[第 2 轮] 用户：项目截止日期是 3 月 31 日

        助手：

[第 3 轮] 用户：目前完成了需求分析和设计阶段

        助手：

[第 4 轮] 用户：开发团队遇到了一个技术难题：数据库性能问题

        助手：

[第 5 轮] 用户：我们决定改用 PostgreSQL 替换 MySQL

        助手：

[第 6 轮] 用户：好的，现在说说今天的天气

        助手：

[第 7 轮] 用户：回到项目：星辰计划现在进展如何？负责人是谁？截止日期是什么时候？

        助手：

==================================================
记忆系统状态：
对话轮次：7
窗口消息数：6
实体数量：0

进阶：记忆持久化#

实际部署中，Agent 重启后需要恢复记忆。下面展示如何将实体记忆序列化到 JSON 文件。

import json
from pathlib import Path


def save_memory(entities: dict, filepath: str = "/tmp/agent_memory.json"):
    """将实体记忆保存到 JSON 文件"""
    with open(filepath, "w", encoding="utf-8") as f:
        json.dump(entities, f, ensure_ascii=False, indent=2)
    print(f"记忆已保存到：{filepath}")


def load_memory(filepath: str = "/tmp/agent_memory.json") -> dict:
    """从 JSON 文件加载实体记忆"""
    path = Path(filepath)
    if not path.exists():
        print("没有找到持久化记忆文件，从空白开始")
        return {}
    with open(filepath, encoding="utf-8") as f:
        entities = json.load(f)
    print(f"从 {filepath} 加载了 {len(entities)} 条实体记忆")
    return entities


# 保存 HybridMemory 的实体到文件
save_memory(hmem._entities)

# 模拟重启后加载
loaded = load_memory()
print("\n加载的实体信息：")
for k, v in loaded.items():
    print(f"  {k}: {v}")

记忆已保存到：/tmp/agent_memory.json
从 /tmp/agent_memory.json 加载了 0 条实体记忆

加载的实体信息：

练习题#

改进滑动窗口：当前按条数截断，试着改为按 token 数截断（提示：litellm.token_counter()）。
摘要质量评估：实现一个函数，对比摘要前后的信息保留率（让 LLM 判断摘要是否遗漏了关键信息）。
实体冲突处理：如果用户说「我的猫叫小白」，后来又说「我没有猫」，实体记忆应该如何处理矛盾？修改 EntityMemory._extract_entities 来支持实体删除。
向量记忆优化：当前每次都对所有历史计算相似度（O(n) 复杂度）。研究如何用 FAISS 或 ScaNN 加速检索。

总结#

四种记忆类型对比#

类型	实现复杂度	容量	适用场景	主要缺点
滑动窗口	简单	最近 N 条	短期对话、问答	窗口外信息丢失
摘要记忆	中等	压缩历史 + N 条	长对话、客服	摘要可能丢细节
实体记忆	中等	无限（键值对）	用户画像、CRM	只记结构化信息
向量记忆	较高	全量历史	知识库问答	需要嵌入模型

选择建议#

客服机器人：摘要记忆 + 实体记忆
个人助手：滑动窗口 + 实体记忆（HybridMemory）
知识库 QA：向量记忆
生产环境：配合 Redis/PostgreSQL 做持久化存储

下一章我们将学习多 Agent 协作，让多个专门的 Agent 分工合作完成复杂任务。

# 知识检验：内存类型速查
summary_table = {
    "ConversationMemory": {
        "核心机制": "deque(maxlen=N)",
        "适用场景": "短期对话",
        "LLM 调用次数": "每轮 1 次",
    },
    "SummaryMemory": {
        "核心机制": "超阈值后调用 LLM 摘要",
        "适用场景": "长对话",
        "LLM 调用次数": "正常 1 次 + 压缩时额外 1 次",
    },
    "EntityMemory": {
        "核心机制": "每轮提取实体注入 system prompt",
        "适用场景": "用户画像跟踪",
        "LLM 调用次数": "每轮 2 次（回答 + 提取）",
    },
    "VectorMemory": {
        "核心机制": "余弦相似度检索相关历史",
        "适用场景": "大量历史中语义检索",
        "LLM 调用次数": "每轮 1 次 + 嵌入调用",
    },
}

print("记忆系统速查表")
print("="*60)
for name, info in summary_table.items():
    print(f"\n{name}")
    for k, v in info.items():
        print(f"  {k}: {v}")

记忆系统速查表
============================================================

ConversationMemory
  核心机制: deque(maxlen=N)
  适用场景: 短期对话
  LLM 调用次数: 每轮 1 次

SummaryMemory
  核心机制: 超阈值后调用 LLM 摘要
  适用场景: 长对话
  LLM 调用次数: 正常 1 次 + 压缩时额外 1 次

EntityMemory
  核心机制: 每轮提取实体注入 system prompt
  适用场景: 用户画像跟踪
  LLM 调用次数: 每轮 2 次（回答 + 提取）

VectorMemory
  核心机制: 余弦相似度检索相关历史
  适用场景: 大量历史中语义检索
  LLM 调用次数: 每轮 1 次 + 嵌入调用