Agent 开发实战（四）：记忆与上下文管理——让 Agent 记住用户、历史与知识
Agent 开发实战（四）：记忆与上下文管理——让 Agent 记住用户、历史与知识
第（三）篇解决了"工具稳不稳"。但还有一个明显问题：每次对话都是从零开始。用户上周说"我是做生信的，只看小鼠数据"，这周又来问，Agent 全忘了；一次工具返回了 5000 行数据，全塞进上下文，token 直接爆表。本篇讲清楚：短期记忆怎么截断、长结果怎么压缩、长期记忆怎么存、何时该注入。
1. 为什么 Agent 需要记忆
三类典型需求：
- 短期：多轮对话要连续。"再细化一下上一条"、"换个渠道看看"——得知道"上一条"是什么。
- 长期：用户画像、偏好、历史结论，跨会话保留。"这个用户只看小鼠数据"应长期生效。
- 知识：站点/业务知识（如某生信流程的字段含义），不必每次从文档现查，可常驻。
把它们分开处理，比"把一切塞进 prompt"靠谱得多。
2. 短期记忆：对话窗口与截断
最朴素的做法是保留最近 N 轮。但要注意：system 提示必须始终在头部，工具结果可能很长需要裁剪。
# memory/short_term.py
class BufferWindowMemory:
def __init__(self, window: int = 10):
self.window = window
self.turns: list[dict] = []
def add(self, role: str, content):
self.turns.append({"role": role, "content": content})
# 只保留最近 window 轮（一轮 = user+assistant 计 1）
if len(self.turns) > self.window * 2:
self.turns = self.turns[-self.window * 2:]
def get_messages(self, system_prompt: str) -> list[dict]:
return [{"role": "system", "content": system_prompt}] + self.turns
坑点：工具调用消息（
role: "tool"/ 带tool_calls的 assistant 消息）也必须原样保留，否则模型上下文对不上会报错。截断时要把"一对"工具消息整体保留或整体丢弃。
3. 上下文压缩：大工具结果摘要化
第（二）篇里我们把 5000 行塞进 tool 消息，既烧钱又容易超长。正确做法：工具返回原始数据，但回灌模型前先摘要。
# memory/compress.py
def compress_tool_result(raw: dict, max_chars: int = 1500) -> str:
"""把工具原始返回压成模型可读的摘要。"""
if raw.get("__error__"):
return f"[工具错误] {raw['__error__']}"
text = json.dumps(raw, ensure_ascii=False, default=str)
if len(text) <= max_chars:
return text
# 超长：抽取关键字段 + 统计信息，而不是整段粘贴
if "rows" in raw:
rows = raw["rows"]
summary = {
"row_count": len(rows),
"first_3": rows[:3],
"columns": list(rows[0].keys()) if rows else [],
}
return "[数据过大，仅返回摘要] " + json.dumps(summary, ensure_ascii=False, default=str)
return text[:max_chars] + "…(已截断)"
调用侧：
for call in msg.tool_calls:
res = execute_one(call)
messages.append({
"role": "tool",
"tool_call_id": call.id,
"content": compress_tool_result(res), # 摘要化回灌
})
经验值：给模型"足够做结论"的信息即可。5000 行原始数据，模型真正用到的往往是"总数 + Top N + 字段含义"。把原始数据留在服务端，只回灌摘要，成本和稳定性都更好。
4. 长期记忆：用户画像与知识（向量库）
跨会话要记住的东西，存数据库或向量库。检索时按当前问题召回相关片段，注入 prompt。
复用（二）里的 embedding 客户端：
# memory/long_term.py
import os, json, numpy as np
from openai import OpenAI
client = OpenAI(api_key=os.getenv("LLM_API_KEY"), base_url=os.getenv("LLM_BASE_URL"))
_store: list[tuple[str, np.ndarray]] = [] # 生产环境换成 Milvus/PGVector
def remember(text: str):
emb = client.embeddings.create(model="text-embedding-3-small", input=text).data[0].embedding
_store.append((text, np.array(emb)))
def recall(query: str, top_k: int = 3) -> list[str]:
if not _store:
return []
q = np.array(client.embeddings.create(model="text-embedding-3-small", input=query).data[0].embedding)
scored = [(t, np.dot(q, e) / (np.linalg.norm(q) * np.linalg.norm(e))) for t, e in _store]
scored.sort(key=lambda x: x[1], reverse=True)
return [t for t, _ in scored[:top_k]]
什么时候写入？别让模型随意写。稳妥做法是：每轮对话后，用一个轻量 LLM 调用判断"本条是否值得长期记住"，只存确认过的（用户偏好、关键结论）。避免把临时闲聊也塞进长期记忆。
5. 把记忆注入 Agent 循环
def run_agent(question, memory: BufferWindowMemory, user_id: str):
# 1) 召回长期记忆，拼进 system
facts = recall(question)
sys_prompt = SYSTEM_PROMPT
if facts:
sys_prompt += "\n已知用户背景：\n" + "\n".join(f"- {f}" for f in facts)
messages = memory.get_messages(sys_prompt)
messages.append({"role": "user", "content": question})
while True:
resp = client.chat.completions.create(
model=os.getenv("LLM_MODEL"), messages=messages,
tools=registry.openai_schemas(), tool_choice="auto")
msg = resp.choices[0].message
if not msg.tool_calls:
memory.add("assistant", msg.content)
return msg.content
messages.append(msg)
for call in msg.tool_calls:
res = execute_one(call)
messages.append({"role":"tool","tool_call_id":call.id,
"content": compress_tool_result(res)})
memory.add("user", question) # 实际应在开头加；此处示意
6. 隐私与成本注意
- 长期记忆是敏感区：用户提到的姓名、病历、密钥，别无脑入库。写入前做脱敏/过滤，或在
remember里加正则拦截明显敏感信息。 - 向量库要隔离：不同用户的长期记忆用
user_id分区，召回时只查自己的。 - 压缩不是删数据：原始结果仍可在服务端日志/数据库留痕（供审计），只是不回灌模型。
7. 小结
记忆管理三件事：
- 短期靠窗口截断，注意工具消息成对保留；
- 上下文靠压缩，把"原始数据"留在服务端、只回灌"模型够用"的摘要；
- 长期靠向量召回，且仅存经确认的用户背景与结论，注意隔离与脱敏。
做好这三层，Agent 才从一个"每次重启都失忆的问答机"，变成一个"记得你、不乱说话"的助手。
记忆越多越贵也越危险。默认不记，只在明确有价值时才记——这是工程上最稳的默认。