长期记忆
记忆的概念
长期记忆赋予代理学习和改进的能力。与短期记忆（单次对话中的消息历史）不同，长期记忆跨越多个对话线程持久化，使代理能够：
- 记住用户的偏好和交互历史
- 在不同的对话中应用学到的知识
- 逐步改进其理解和响应
- 维护持久的知识库
记忆的架构
对话线程 1
├─ 消息历史（短期）
├─ 长期记忆
│ ├─ 用户偏好
│ ├─ 学到的规则
│ └─ 知识库
└─ 文件系统（临时）
对话线程 2（新用户或新对话）
├─ 消息历史（短期）
├─ 长期记忆（可访问线程 1 的记忆）
│ └─ 累积的全局知识
└─ 文件系统（临时）
记忆的类型
1. 用户偏好记忆
# AGENT_PREFERENCES.md 文件内容示例
---
# 用户偏好
## 沟通风格
- 语言：中文
- 正式程度：正式
- 代码注释语言：中文
- 响应长度：详细
## 技术偏好
- 首选编程语言：Python
- 框架偏好：[LangChain, FastAPI]
- 数据库：PostgreSQL
- 部署平台：Docker
## 速度偏好
- 优先级：速度 > 完整性
- 预算限制：使用质量较好但不是最昂贵的模型
2. 知识库记忆
# KNOWLEDGE_BASE.md 文件
---
# 知识库
## 关于用户组织
- 公司名称：TechCorp Inc
- 行业：AI/ML SaaS
- 团队规模：50 人
- 主要产品：数据分析平台
## 关于项目
- 项目名称：Analytics Dashboard 2.0
- 技术栈：React + Python + PostgreSQL
- 上一个版本的痛点：
1. 页面加载很慢（>3s）
2. 实时数据更新延迟（>30s）
3. 缩放性问题
## 最佳实践（自动化学习）
- API 端点的命名规范
- 测试覆盖率目标：>80%
- 代码审查检查清单
3. 交互历史记忆
# INTERACTION_HISTORY.md
---
# 交互历史
## 最近的项目
- 2024-01-15：帮助调试认证模块。关键学习：JWT token 过期处理很重要
- 2024-01-12：性能优化建议。关键学习：查询缓存可以减少 70% 的数据库命中
## 常见问题
- 问题：如何处理并发用户？
- 答案：使用连接池 + Redis 缓存
- 说明：见项目文档第 3.2 章
创建和管理长期记忆
基本设置
from deepagents import create_deep_agent
from deepagents.backends import CompositeBackend, StateBackend, StoreBackend
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
agent = create_deep_agent(
model="anthropic:claude-3-5-sonnet-20241022",
memory=[
"/memories/AGENT_PREFERENCES.md",
"/memories/KNOWLEDGE_BASE.md",
],
backend=CompositeBackend(
default=StateBackend(),
routes={
"/memories/": StoreBackend(
store=store,
namespace=lambda ctx: ("assistant",), # 全局命名空间
),
}
),
store=store,
)
初始化记忆
from deepagents.backends.utils import create_file_data
# 创建初始记忆文件
initial_preferences = """
# 代理偏好
## 回应方式
- 专业性：高
- 代码示例：详细
"""
initial_knowledge = """
# 知识库
## 项目信息
- 名称：MyProject
- 技术栈：Python, FastAPI
"""
# 写入到 Store
store.put(
("assistant",),
"/memories/AGENT_PREFERENCES.md",
create_file_data(initial_preferences),
)
store.put(
("assistant",),
"/memories/KNOWLEDGE_BASE.md",
create_file_data(initial_knowledge),
)
作用域记忆
全局记忆（所有用户共享）
# 代理学习所有用户的最佳实践
agent = create_deep_agent(
memory=["/memories/GLOBAL_BEST_PRACTICES.md"],
backend=CompositeBackend(
default=StateBackend(),
routes={
"/memories/": StoreBackend(
namespace=lambda ctx: ("global",), # 全局命名空间
),
}
),
)
# 所有用户和会话都可以访问这个记忆
用户级记忆（每个用户隔离）
agent = create_deep_agent(
memory=["/memories/USER_PROFILE.md"],
backend=CompositeBackend(
default=StateBackend(),
routes={
"/memories/": StoreBackend(
namespace=lambda ctx: (
ctx.runtime.server_info.user_id, # 按用户隔离
),
),
}
),
store=store,
)
# 每个用户只能看到自己的记忆
助手级记忆（每个代理实例隔离）
agent = create_deep_agent(
memory=["/memories/ASSISTANT_PERSONALITY.md"],
backend=CompositeBackend(
default=StateBackend(),
routes={
"/memories/": StoreBackend(
namespace=lambda ctx: (
ctx.runtime.server_info.assistant_id, # 按助手隔离
),
),
}
),
store=store,
)
# 每个代理实例有自己的记忆和性格
记忆更新
代理自动更新记忆
代理可以在对话过程中更新长期记忆：
# 系统提示中告诉代理何时更新记忆
system_prompt = """你是一个专业助手。
当以下情况发生时，使用 edit_file 工具更新记忆：
1. 学到用户的新偏好
2. 发现新的解决方案或最佳实践
3. 交互历史需要记录
更新 /memories/KNOWLEDGE_BASE.md 的"最佳实践"部分。"""
agent = create_deep_agent(
system_prompt=system_prompt,
memory=["/memories/KNOWLEDGE_BASE.md"],
)
# 代理会在适当时候自动更新记忆
result = agent.invoke({
"messages": [{
"role": "user",
"content": "我们已经解决了那个性能问题。缓存策略非常有效。"
}]
})
手动更新记忆
def update_agent_memory(store, assistant_id: str, section: str, content: str):
"""手动更新代理记忆"""
# 读取当前记忆
current = store.get(
(assistant_id,),
"/memories/KNOWLEDGE_BASE.md"
)
if current:
# 解析 Markdown 并更新特定部分
# 这是伪代码，实际需要 Markdown 解析
updated_content = update_markdown_section(
current.content,
section,
content
)
else:
updated_content = f"# {section}\n\n{content}"
# 存储更新
store.put(
(assistant_id,),
"/memories/KNOWLEDGE_BASE.md",
create_file_data(updated_content),
)
实际例子："学习型"代理
from deepagents import create_deep_agent
from deepagents.backends import CompositeBackend, StateBackend, StoreBackend
from langgraph.store.memory import InMemoryStore
# 创建学习型代理
class LearningAgent:
def __init__(self):
self.store = InMemoryStore()
# 初始化记忆文件
self._init_memory()
self.agent = create_deep_agent(
model="anthropic:claude-3-5-sonnet-20241022",
memory=[
"/memories/SESSION_NOTES.md",
"/memories/LEARNED_SOLUTIONS.md",
],
backend=CompositeBackend(
default=StateBackend(),
routes={
"/memories/": StoreBackend(
store=self.store,
namespace=lambda ctx: ("learning-agent",),
),
}
),
store=self.store,
system_prompt="""你是一个不断学习的编程助手。
你的职责：
1. 帮助用户解决编程问题
2. 记录每次交互的要点
3. 学习并记住解决方案
4. 在将来类似问题中应用学到的知识
当解决问题时：
1. 解决：提供解决方案
2. 学习：更新 /memories/LEARNED_SOLUTIONS.md
3. 验证：确认解决方案有效
4. 总结：添加要点到 /memories/SESSION_NOTES.md
格式要求：
- 保持 Markdown 格式
- 使用清晰的标题和列表
- 包含日期和关键词以便检索""",
)
def _init_memory(self):
"""初始化记忆文件"""
from deepagents.backends.utils import create_file_data
self.store.put(
("learning-agent",),
"/memories/LEARNED_SOLUTIONS.md",
create_file_data("# 学到的解决方案\n"),
)
self.store.put(
("learning-agent",),
"/memories/SESSION_NOTES.md",
create_file_data("# 会话记录\n"),
)
def solve_problem(self, problem: str) -> str:
"""解决问题并学习"""
result = self.agent.invoke({
"messages": [{
"role": "user",
"content": problem
}]
})
return result["messages"][-1].content
def get_memory(self, file_name: str) -> str:
"""获取特定的记忆文件"""
result = self.store.get(
("learning-agent",),
f"/memories/{file_name}"
)
return result.content if result else ""
# 使用示例
agent = LearningAgent()
# 第一次交互
response1 = agent.solve_problem(
"如何在 Python 中实现递归二分查找？"
)
print(response1)
# 代理会自动记录解决方案和要点
# 查看学到的内容
solutions = agent.get_memory("LEARNED_SOLUTIONS.md")
print("\n学到的解决方案：")
print(solutions)
后台整合
对于长时间运行的代理，可以在会话间自动总结和压缩记忆：
import asyncio
from datetime import datetime
async def consolidate_memory_background(agent_config, interval_hours=24):
"""定期整合记忆（汇总、压缩、清理）"""
while True:
try:
# 等待指定间隔
await asyncio.sleep(interval_hours * 3600)
# 读取当前记忆
store = agent_config.store
sessions = store.get_all(("learning-agent",))
# 汇总会话记录
summary = summarize_sessions(sessions)
# 更新汇总文件
store.put(
("learning-agent",),
"/memories/MONTHLY_SUMMARY.md",
create_file_data(summary),
)
# 清理旧的详细记录（保留最近 100 条）
# 实现具体的清理逻辑
print(f"[{datetime.now()}] 记忆已整合")
except Exception as e:
print(f"整合记忆时出错：{e}")
# 在应用启动时 运行
# asyncio.create_task(consolidate_memory_background(agent_config))
记忆的最佳实践
1. 清晰的组织结构
/* ❌ 杂乱无章 */
# 杂项笔记
- 用户喜欢 Python...
- 公司有 50 人...
- 问题：缓存...
- 解决方案：使用 Redis...
/* ✅ 有组织的 */
# 用户偏好和配置文件
## 技术偏好
- 首选语言：Python
- ...
# 组织信息
## 基本信息
- 团队规模：50 人
- ...
# 已解决的问题和解决方案
## 缓存问题
原始问题：...
解决方案：使用 Redis...
2. 可搜索和可索引
/* ✅ 好的做法：使用标签便于搜索 */
# 已解决的问题
## 问题：缓存性能 `#performance #caching #redis`
- 日期：2024-01-15
- 关键字：Redis, 缓存, 性能
- 解决方案：...
## 问题：并发处理 `#concurrency #threading #python`
- 日期：2024-01-10
- 关键字：并发, 线程, Python
- 解决方案：...
3. 定期维护
def maintain_memory(store, namespace):
"""定期维护记忆清洁度"""
# 检查大小（记忆不应过大）
memory_files = store.get_all(namespace)
total_size = sum(len(f.content) for f in memory_files.values())
if total_size > 1_000_000: # 1MB 限制
print("警告：记忆文件过大，需要清理")
# 实现清理逻辑
# 检查过期内容（超过 1 年的）
# 检查格式一致性