Agent 和人的一个关键相似点:没有记忆,什么都做不了。
想象一个 Agent 在帮你做一个为期三天的数据分析项目。第一天你告诉它数据源在哪、偏好什么图表风格。第二天你让它继续——如果它完全不记得第一天的事,你要把所有指令重新说一遍。
这就是记忆系统要解决的问题。本文拆解三种 Agent 记忆机制,从原理到代码。
| 类型 | 类比 | 时效 | 实现 |
|---|---|---|---|
| 短期记忆 | 工作记忆 | 当前会话 | 消息列表 |
| 长期记忆 | 笔记本 | 跨会话持久 | 数据库/文件 |
| RAG 记忆 | 图书馆 | 按需检索 | 向量数据库 |
短期记忆就是 messages 列表。我们前面写的 Agent 已经在用了:
messages = [
{"role": "system", "content": "你是…"},
{"role": "user", "content": user_input},
{"role": "assistant", "content": "我来搜索一下…"},
{"role": "tool", "content": "搜索结果…"},
{"role": "assistant", "content": "根据搜索结果…"}
]
问题是:模型的上下文窗口有限。如果对话太长,超出窗口怎么办?
1. 滑动窗口——只保留最近 N 条消息。简单粗暴,但会丢掉早期关键信息。
2. 智能摘要——用模型定期压缩历史对话,把「用户喜欢蓝色图表、数据在 data/ 目录」浓缩成一段系统提示词,替换掉冗长的原始对话:
def compress_history(messages, client):
"""将历史对话压缩为一句话摘要。"""
summary_prompt = "将以下对话的关键信息压缩成一段摘要:\n" + \
"\n".join([f"{m['role']}: {m['content'][:200]}" for m in messages[-20:]])
summary = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": summary_prompt}]
).choices[0].message.content
return summary
3. 分层窗口——最近的保留原始消息,稍远的只保留摘要,更远的丢掉。兼顾上下文和效率。
长期记忆让 Agent 记住跨会话的信息——你的偏好、项目结构、上次任务的结果。
import json, os
MEMORY_FILE = "agent_memory.json"
def load_memory() -> dict:
if os.path.exists(MEMORY_FILE):
with open(MEMORY_FILE) as f:
return json.load(f)
return {"facts": [], "preferences": {}}
def save_fact(key: str, value: str):
memory = load_memory()
memory["facts"].append({"key": key, "value": value, "time": datetime.now().isoformat()})
with open(MEMORY_FILE, "w") as f:
json.dump(memory, f, indent=2)
def get_relevant_context(query: str) -> str:
"""启动时,将记忆注入系统提示词。"""
memory = load_memory()
facts = "\n".join([f"- {f['key']}: {f['value']}" for f in memory["facts"]])
return f"用户已知信息:\n{facts}"
使用时,在每次对话开始时加载记忆注入 system 消息:
system_prompt = f"""你是一个有用的助手。
{get_relevant_context()}
请根据已知信息回答用户的问题。"""
对于更复杂的场景,可以用 SQLite 存储分类型的记忆:
CREATE TABLE memory (
id INTEGER PRIMARY KEY,
category TEXT, -- 'preference', 'project', 'person', 'fact'
key TEXT,
value TEXT,
importance REAL, -- 0.0 到 1.0,决定是否注入上下文
created_at TIMESTAMP,
last_accessed TIMESTAMP
);
注入时只选高重要度或最近访问的记忆,避免上下文膨胀。
前两层适合记录「元信息」——偏好、状态、事实。但如果你有大量文档——代码库、手册、研究报告——需要的是 RAG。
RAG 的核心流程:
import chromadb
from openai import OpenAI
client = OpenAI(base_url="...", api_key="...")
db = chromadb.PersistentClient(path="./agent_rag_db")
collection = db.get_or_create_collection("knowledge_base")
# 1. 索引 — 文档入库
def index_document(doc_id: str, content: str):
chunks = [content[i:i+500] for i in range(0, len(content), 500)]
for i, chunk in enumerate(chunks):
embedding = client.embeddings.create(
model="text-embedding-3-small", input=chunk
).data[0].embedding
collection.add(
ids=[f"{doc_id}_{i}"],
embeddings=[embedding],
documents=[chunk]
)
# 2. 检索 — 找到相关片段
def retrieve(query: str, top_k: int = 5) -> str:
query_embedding = client.embeddings.create(
model="text-embedding-3-small", input=query
).data[0].embedding
results = collection.query(query_embeddings=[query_embedding], n_results=top_k)
return "\n\n".join(results["documents"][0])
# 3. 注入 Agent — 在 system prompt 中追加检索结果
rag_context = retrieve(user_input)
system_prompt += f"\n\n参考知识库:\n{rag_context}"
实际项目中三层不是互斥的,而是各司其职:
| 场景 | 用哪层 |
|---|---|
| 记住用户这次说了什么 | 短期记忆(消息列表) |
| 记住用户偏好、项目路径 | 长期记忆(JSON/SQLite) |
| 查技术文档、研究资料 | RAG(向量数据库) |
| 代码库里找相关函数 | RAG + 目录树索引 |
| 总结已完成的任务 | 长期记忆 + 定期摘要 |
把三层整合进我们的 Agent 循环:
def run_agent_with_memory(user_input: str, user_id: str = "default"):
# 加载长期记忆
long_term = load_memory(user_id)
# RAG 检索
rag_context = retrieve(user_input)
# 构建系统提示词(三层信息融合)
system = f"""你是用户的 AI 助手。
## 用户偏好
{long_term}
## 知识库参考
{rag_context}
## 对话指南
- 优先使用知识库中的信息
- 记住用户的偏好用于后续对话"""
messages = [{"role": "system", "content": system}]
messages.extend(load_recent_history(user_id)[-20:]) # 短期记忆
messages.append({"role": "user", "content": user_input})
# ReAct 循环(不变)
for turn in range(10):
response = client.chat.completions.create(
model="gpt-4o", messages=messages, tools=TOOLS
)
msg = response.choices[0].message
if not msg.tool_calls:
# 主动保存新学到的事实
extract_and_save_facts(msg.content, user_id)
return msg.content
# ... 执行工具 ...
return "达到最大轮次"
📖 下一篇:Agent 错误恢复与自我纠错 — 让 Agent 在犯错后自己修好