Jean's Blog

一个专注软件测试开发技术的个人博客

0%

DeepAgents为智能体打造长期记忆

框架长期记忆:存在的问题

官方代码

1
2
3
4
5
6
7
8
store = InMemoryStore(index={"embed": embed, "dims": 1536})
agent = create_deep_agent(
store=store,
backend=lambda rt: CompositeBackend(
StateBackend(rt),
{"/memories/": StoreBackend(rt)}
)
)

这是一个Agent 记忆模块的官方极简示例代码,核心逻辑很简单:

  1. InMemoryStore:创建一个内存型的向量存储,用来保存 Agent 的记忆数据。
    • embed 是嵌入模型,dims:1536 是向量维度(典型 OpenAI text-embedding-ada-002 的维度)。
  2. create_deep_agent:创建 Agent 实例,传入记忆存储和后端处理逻辑。
  3. CompositeBackend:组合了两种后端:
    • StateBackend:管理 Agent 的会话状态。
    • StoreBackend:专门处理 /memories/ 路径下的记忆读写请求。

这个看起来很简单,但完全不适用生产级别,有以下8大生产级的问题

问题 通俗解释 为什么是痛点?
1. 文件系统语义错位 ls/read_file 这类文件操作来管理记忆,和 LLM 对话式的上下文格格不入 LLM 的记忆本质是 “语义片段 + 关联关系”,不是 “文件 + 目录”。强行套文件系统逻辑,会导致 Agent 理解记忆的效率极低,甚至产生错误关联。
2. 跨会话命名空间混乱 /memories/user123/memories/thread456这类路径,没有清晰的多用户 / 多会话隔离规则 上线后多用户、多线程并发时,记忆数据会互相污染、泄露,完全不可控。
3. 语义搜索能力缺失 仅支持简单的语义搜索,无法做复杂的过滤、召回、排序 真实场景里,Agent 需要按时间、用户、话题、重要性等多维度筛选记忆,简单向量相似度根本不够用。
4. 权限控制真空 没有对记忆数据的访问控制机制 任何人都能读写 / 删除他人的记忆数据,完全没有数据安全可言,无法满足合规要求。
5. 数据清理机制空白 记忆数据没有自动清理、归档、整理的策略 长期运行后,记忆数据会无限膨胀,导致向量库性能雪崩,也无法区分有效 / 无效记忆。
6. 监控能力为零 无法跟踪记忆的使用情况、访问频率、性能瓶颈 出了问题不知道是哪条记忆出了错,也不知道向量库的性能瓶颈在哪,排查完全靠猜。
7. 调试工具匮乏 难以排查记忆相关的问题 比如 “Agent 为什么记错了?”“为什么没调用到关键记忆?”,没有工具可以追踪记忆的匹配、召回、使用链路。
8. 架构迁移困难 更换存储方案(比如从内存向量库换成生产级向量数据库)需要大量修改代码 代码和具体存储实现强耦合,想升级 / 替换存储方案就得重构,维护成本极高。

生产级方案:自定义架构

核心思想:分层记忆体系

这是这套方案的灵魂:

把 Agent 的记忆按 “信息类型、稳定性、使用场景” 拆成不同层级,而不是一股脑塞在一个向量库里。

示例代码只是示意,真实生产环境会用中间件实现记忆的智能管理和注入。

1. 三层记忆体系结构

层级 代码里的对象 作用 特点
静态事实层 self.static_facts 存储用户画像、偏好、固定不变的信息 稳定、不常变,比如用户的姓名、生日、长期偏好
提取事实层 self.extracted_facts 从对话中提取的关键信息 半结构化、有价值的事实,比如用户说过的 “我下周要出差”“我不喜欢吃辣”
动态对话层 self.conversation_history 原始的对话历史 高频、易过期,比如最近的聊天记录
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
class HierarchicalMemory:
"""
真实生产环境使用多层记忆体系:
1. 静态事实层: 用户画像、偏好、固定信息
2. 提取事实层: 从对话中提取的关键信息
3. 动态对话层: 原始历史对话
"""
def __init__(self):
# 静态事实: 用户个人信息、偏好
self.static_facts = StaticMemoryLayer()
# 提取事实: 从对话中提取的关键信息
self.extracted_facts = ExtractedMemoryLayer()
# 动态对话: 原始历史对话
self.conversation_history = DynamicMemoryLayer()

def get_context(self, user_id, query):
# 智能整合三层记忆
static = self.static_facts.get(user_id) # 用户画像
extracted = self.extracted_facts.search(query) # 相关事实
recent = self.conversation_history.get_recent(
user_id,
max_tokens=2000, # 按token维度
max_messages=10, # 按数量维度
time_window="7d" # 按时间维度
)
return self.merge_memories(static, extracted, recent)

这是这套架构最关键的地方,它解决了 “怎么把合适的记忆喂给大模型” 的问题:

  1. 静态事实层:直接按 user_id 拉取用户画像,这部分是固定不变的,直接拼接即可。
  2. 提取事实层:根据用户当前的 query 做语义搜索,召回和当前问题相关的关键事实。
  3. 动态对话层:不是把所有历史对话都塞进去,而是通过三重限制做 “精准裁剪”:
    • max_tokens=2000:按 token 数量限制,避免超出模型上下文长度。
    • max_messages=10:按消息条数限制,只保留最近的对话。
    • time_window="7d":按时间窗口限制,只保留 7 天内的对话。
  4. merge_memories:把三层记忆按优先级和格式整合起来,拼成最终喂给大模型的上下文。

2.中间件注入多层记忆

这段代码是Agent 执行流程中的一个前置中间件(Hook),作用是在 Agent 正式处理用户请求前,把之前提到的「三层分层记忆」自动注入到对话上下文中。

它解决了 “怎么把分层记忆喂给 Agent” 的问题,是整个分层记忆架构的接入层

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
@before_agent(can_jump_to=["end"])
def inject_hierarchical_memory(state: AgentState, runtime: Runtime):

# 获取三层记忆
user_id = state.user_id
last_message = state.messages[-1].content

# 整合静态事实 + 提取事实 + 动态对话
context = hierarchical_memory.get_context(
user_id=user_id,
query=last_message
)

return {
"messages": [
SystemMessage(content=f"""
可用记忆结构:
📊 用户画像: {context['static_facts']}
💡 相关事实: {context['extracted_facts']}
💬 最近对话: {context['conversation_history']}
"""),
*state["messages"]
]
}

代码解读

1
2
@before_agent(can_jump_to=["end"])
def inject_hierarchical_memory(state: AgentState, runtime: Runtime):
  • @before_agent:这是一个装饰器(Hook),表示这个函数会在 Agent 执行主逻辑之前运行。
  • can_jump_to=["end"]:可选配置,意思是这个中间件可以直接跳转到流程的end节点(比如记忆注入失败时直接结束流程)。
  • state: AgentState:Agent 的状态对象,包含当前对话的所有信息(用户 ID、历史消息等)。
  • runtime: Runtime:运行时对象,提供环境上下文、工具调用等能力。
1
2
3
# 获取三层记忆
user_id = state.user_id
last_message = state.messages[-1].content

这一步是从当前对话状态里,拿到后续召回记忆需要的关键信息:

  • user_id:用来隔离用户,拉取该用户独有的静态事实层记忆。
  • last_message:用户最新的提问,用来做语义召回,匹配提取事实层的相关记忆。
1
2
3
4
5
# 整合静态事实 + 提取事实 + 动态对话
context = hierarchical_memory.get_context(
user_id=user_id,
query=last_message
)
  • 调用上一张图里定义的 HierarchicalMemory.get_context() 方法,一次性拿到整合好的三层记忆:

    1. 静态事实层(用户画像)
    2. 提取事实层(相关关键信息)
    3. 动态对话层(最近对话)
  • 这个方法会自动帮你做好用户隔离、语义召回、对话裁剪,返回一个结构化的 context 对象。

1
2
3
4
5
6
7
8
9
10
11
    return {
"messages": [
SystemMessage(content=f"""
可用记忆结构:
📊 用户画像: {context['static_facts']}
💡 相关事实: {context['extracted_facts']}
💬 最近对话: {context['conversation_history']}
"""),
*state["messages"]
]
}

这是注入逻辑的核心,它把整合好的记忆,以 SystemMessage(系统提示词)的形式,拼接到对话的最前面:

  1. 先插入一段结构化的系统提示,告诉大模型:
    • 有哪些可用的记忆
    • 分别是什么类型(用户画像 / 相关事实 / 最近对话)
    • 具体内容是什么
  2. 再把原有的对话历史 *state["messages"] 拼接在后面,保证用户的提问不会被覆盖。

生产级记忆体系设计

整个分层记忆体系的核心设计规范,它给每一层记忆都定好了数据类型、存储策略、检索方式、更新频率,解决了 “每一层该怎么管” 的问题:

层级 数据类型 存储策略 检索方式 更新频率
静态事实层 用户画像、偏好、固定信息 永久存储,手动更新 精确匹配 低频
提取事实层 从对话中提取的关键信息 长期存储,自动压缩 语义搜索 中频
动态对话层 原始历史对话 短期缓存,滚动更新 时间 / 数量 / 混合 高频

关键设计逻辑

  1. 存储策略分层
    • 静态层:比如用户的姓名、过敏史,几乎不变,永久存 + 人工维护,避免脏数据。
    • 提取层:比如用户的行程、偏好变化,长期存但可以自动合并 / 压缩,避免数据爆炸。
    • 动态层:比如最近的聊天记录,短期缓存 + 定期清理,只保留最近有效数据。
  2. 检索方式匹配场景
    • 静态层:用精确匹配(比如按user_id直接查),效率高、无歧义。
    • 提取层:用语义搜索(向量召回),能找到和当前问题相关的历史事实。
    • 动态层:用时间 / 数量限制(比如最近 7 天 / 10 轮对话),快速裁剪上下文,避免 token 超限。
  3. 更新频率适配变化速度
    • 静态层:低频更新,保证数据稳定。
    • 提取层:中频更新,每次对话后异步提取关键信息。
    • 动态层:高频更新,每轮对话都会追加新消息。

压缩策略

这段代码是动态对话层的核心策略,解决了 “怎么把无限变长的对话历史,塞进有限的上下文窗口” 的问题:

1
2
3
4
5
6
7
# 对话历史压缩策略
compression_strategies = {
"时间窗口": "最近7天对话",
"数量限制": "最近20轮对话",
"Token限制": "最近4000 tokens",
"混合策略": "7天内 + 关键对话提取"
}
  • 时间窗口:只保留最近 N 天的对话,过期的直接归档 / 删除。
  • 数量限制:只保留最近 N 轮对话,更早的直接截断。
  • Token 限制:按对话的总 token 数裁剪,比如最多保留 4000 tokens 的对话,超出的从前往后删。
  • 混合策略:生产环境最常用的方案,比如 “保留 7 天内的对话,同时只保留其中的关键信息”,兼顾时效性和信息密度。

提取事实层工作流

flowchart LR
    A["原始对话
我计划下周去北京旅游"] --> B["关键信息提取
{行程: 北京旅游, 时间: 下周}"] B --> C["结构化存储
向量存储"] C --> D["语义检索"]
  1. 原始对话:用户的自然语言输入,比如 “我计划下周去北京旅游”。
  2. 关键信息提取:通过大模型把自然语言转成结构化数据,比如 {"行程": "北京旅游", "时间": "下周"}
  3. 结构化存储:把提取出的事实,加上向量嵌入,存到向量数据库里。
  4. 语义检索:当用户后续提问(比如 “我之前说过什么时候去北京?”),可以通过向量相似度召回对应的事实。

三层记忆整体架构流程

flowchart TD
    subgraph 三层记忆体系
        A[静态事实层
用户画像/固定偏好] B[提取事实层
对话关键信息] C[动态对话层
原始聊天记录] end subgraph 数据处理规则 A1[精确匹配查询
低频更新/永久存储] B1[LLM提取+向量存储
中频更新/语义检索] C1[多策略压缩裁剪
高频更新/短期缓存] end A --> A1 B --> B1 C --> C1 A1 & B1 & C1 --> D[记忆合并组装] D --> E[注入系统提示词] E --> F[Agent 执行业务逻辑]

框架内置与生产级中间件方案对比

特性维度 框架内置方案 生产级中间件方案
代码复杂度 8 行配置,N 处问题 模块化,职责清晰
记忆体系 单一扁平结构 三层智能架构
检索能力 简单关键词匹配 多维度智能检索
压缩策略 时间 / 数量 / Token 混合
扩展性 修改需侵入 Agent 插件化,即插即用

1. 代码复杂度

  • 框架内置方案:表面看只有几行配置,极其简单,但背后全是隐藏的坑(就是第一张图里的 8 大生产级问题),上线后每一个问题都要你自己填,越改越乱。
  • 生产级方案:模块化拆分,记忆存储、召回、压缩、注入各自独立,每个模块只做一件事,出问题可以单独排查和优化,维护成本低得多。

2. 记忆体系

  • 框架内置方案:所有记忆(用户信息、对话历史、关键事实)都塞在同一个扁平的存储里,没有分层,没有隔离,找东西全靠模糊匹配,效率极低。
  • 生产级方案:用了三层架构(静态事实层 / 提取事实层 / 动态对话层),不同类型的记忆用不同的方式管理,各司其职,互不干扰。

3. 检索能力

  • 框架内置方案:只能做简单的关键词匹配或基础语义搜索,无法区分记忆的时效性、重要性、用户归属,很容易出现 “串记忆”“错用记忆” 的问题。
  • 生产级方案:多维度智能检索,比如按用户 ID + 语义相似度 + 时间窗口 + 重要性权重组合召回,精准度高很多。

4. 压缩策略

  • 框架内置方案:完全没有对话压缩机制,对话历史会无限膨胀,很快就会超出模型的上下文长度,导致 Agent 无法正常工作。
  • 生产级方案:支持时间、数量、Token 三重压缩,还能组合使用,比如 “保留 7 天内的对话,且 Token 不超过 4000”,从根源上控制上下文长度。

5. 扩展性

  • 框架内置方案:记忆逻辑和 Agent 主逻辑强耦合,想改存储、加权限、加监控都要直接改 Agent 的核心代码,越改越难维护。
  • 生产级方案:插件化设计,记忆模块和 Agent 主逻辑解耦,想换向量库、加新的压缩策略、改记忆注入方式,都可以即插即用,不用动 Agent 本身的代码。

智能分层核心的优势

1.智能分层

这是整个方案的基础,把 Agent 的记忆按稳定性和用途分成了三类:

  • 静态事实:用户画像、长期偏好(比如姓名、过敏史、固定习惯),特点是稳定、不常变,用精确匹配直接查询。
  • 提取事实:结构化关键信息(比如用户说过的行程、偏好变化),特点是半结构化、有价值,用语义搜索召回。
  • 动态对话:原始上下文(最近的聊天记录),特点是高频、易过期,用时间 / 数量 / Token 限制裁剪。

2.灵活压缩策略

1
2
3
4
5
6
7
8
# 混合维度压缩
compressed_history = compress_conversation(
strategy="hybrid", # 混合策略
max_messages=10, # 数量维度:只保留最近10轮对话
max_tokens=2000, # Token维度:对话总长度不超过2000 tokens
time_window="7d", # 时间维度:只保留7天内的对话
keep_important=True # 保留关键对话:重要消息不被压缩
)
  • 混合策略(hybrid):同时从「数量、Token、时间」三个维度限制对话长度,比单一维度更精准。

  • keep_important=True:即使超出限制,也会保留关键对话(比如用户的核心需求、重要指令),避免压缩丢失关键信息。

3.生产就绪特性

特性 作用 解决的痛点
✅ 自动清理过期记忆 基于时间、大小自动清理 / 归档记忆,避免数据无限膨胀 数据清理机制空白
✅ 细粒度权限控制 精确控制记忆数据的访问权限,比如用户只能看自己的记忆 权限控制真空
✅ 完整监控指标 跟踪记忆的使用情况、性能瓶颈(比如召回耗时、命中率) 监控能力为零
✅ 可视化调试工具 可视化查看记忆的召回、压缩、使用链路,方便排查问题 调试工具匮乏

避坑指南级最佳实践

这些是很多新手做 Agent 记忆时最容易踩的坑,也是官方方案的核心问题:

陷阱 问题本质 后果
❌ 单一扁平记忆结构 所有记忆(用户信息、对话历史、关键事实)混在一起,没有分层和隔离 无法满足复杂业务需求,比如用户画像、多会话隔离、多维度检索,很容易出现串记忆、错用记忆的问题
❌ 无限制存储历史对话 不做任何裁剪和压缩,对话历史无限增长 模型上下文窗口很快被撑满,导致性能下降、API 调用成本飙升,甚至直接报错
❌ 忽略记忆压缩和提取 直接把所有对话喂给模型,不做关键信息提取和压缩 既浪费了大量 Token 和计算资源,又会让模型被冗余信息干扰,回答质量下降

这四条是生产级 Agent 记忆模块的核心设计原则,和我们之前讲的三层架构、压缩策略完全对应:

  1. ✅ 实现三层记忆架构:静态事实、提取事实、动态对话

    把记忆按「稳定性 + 用途」分层管理:

    • 静态事实:用户画像、长期偏好(永久存储,低频更新)

    • 提取事实:从对话中提取的结构化关键信息(长期存储,中频更新)

    • 动态对话:最近的聊天记录(短期缓存,高频更新)

      这是解决 “单一扁平结构” 问题的根本方案。

  2. ✅ 智能提取关键事实:从对话中提取结构化信息

    不要直接把原始对话当记忆,而是通过 LLM 把对话里的关键信息(比如用户的行程、偏好、需求)提取成结构化数据,再存入向量库。这样后续检索时,能更精准地召回相关信息,减少冗余。

  3. ✅ 混合维度压缩历史:结合时间、数量、Token 维度

    对话历史必须做压缩,而且要用「混合策略」,比如:

    • 时间:只保留最近 7 天的对话

    • 数量:只保留最近 10 轮对话

    • Token:对话总长度不超过 2000 tokens

      还可以加上 “保留关键对话” 的开关,避免压缩时丢失重要信息。

  4. ✅ 定期评估记忆价值:优化记忆存储和检索策略

    记忆不是存进去就完事了,要定期评估:

    • 哪些记忆被高频使用?哪些几乎没被召回过?

    • 哪些记忆已经过时,需要归档或清理?

    • 检索策略的召回率、准确率够不够?

      然后根据评估结果优化存储和检索策略,避免数据膨胀和性能下降。