跳到主要内容

数据截至 (上游 commit be263ee23085)

集成模式 — 把同一条记忆环塞进 12 个集成包(覆盖 10 个框架)

30 秒导读: Zep 的核心价值是一条与框架无关的"记忆环"(见 01-memory-model.md)。可真实世界里 agent 跑在 ADK、CrewAI、LangGraph、AutoGen…… 十几个互不兼容的框架上。本章讲同一条记忆环怎么反复落进每个框架的"原生"记忆钩子,并把这些落地方式归纳成两种通用形态——(A)每轮自动往 system prompt 前拼一段 Context Block、(B)把 graph.search 暴露成模型自己决定何时调用的工具。最后讲一条贯穿所有集成的横切原则:永不因 Zep 失败而拖垮 agent。

1. 这是什么(零基础也能懂)​

一句话定义: 集成层是一组薄适配器——每个把 Zep 的记忆环翻译成某一个 agent 框架"本来就懂"的记忆/上下文接口,让你不用改 agent 代码就获得长期记忆。

它解决什么问题: 假设你已经用 Google ADK 写好了一个客服 agent,现在想让它"记住"用户上次说过的偏好。你不想重写 agent、也不想手动在每轮对话里塞检索代码。集成层的承诺是:装一个包(pip install zep-adk)、把一个工具挂进 agent、剩下的自动发生。

为什么需要一层"集成",而不是直接调 SDK: Zep 的 SDK 到处都一样(thread.add_messages、thread.get_user_context、graph.search),但每个框架"注入记忆"的位置和形状都不同:

  • ADK 有一个 process_llm_request 钩子(在发给模型前改请求);
  • LangGraph 是一张显式的图,你在节点里自己拼 system message;
  • CrewAI 期望一个带 save() / search() 方法的存储对象。

集成层就是把同一条记忆环,分别塞进这三种(乃至十种)不同形状的"插座"里。

一句话直觉/类比: 记忆环是同一枚灯泡;每个框架是不同国家的插座。集成包就是那一堆转接头——灯泡不变,只换插脚。

12 个包一览(事实源:integrations/CLAUDE.md 的 "Layout" 表)。注意:包数 = 12,框架数 = 10——adk 一个框架就出了 python/typescript/go 三份包,其余 9 个框架各一份:

框架语言分发名import
adkpython / typescript / gozep-adkzep_adk
ag2pythonzep-ag2zep_ag2
autogenpythonzep-autogenzep_autogen
crewaipythonzep-crewaizep_crewai
langgraphpythonzep-langgraphzep_langgraph
livekitpythonzep-livekitzep_livekit
mastratypescript@getzep/zep-mastra@getzep/zep-mastra
ms-agent-frameworkpythonzep-ms-agent-frameworkzep_ms_agent_framework
pydantic-aipythonzep-pydantic-aizep_pydantic_ai
vercel-aitypescript@getzep/zep-vercel-ai@getzep/zep-vercel-ai

本章不逐个罗列全部 12 个包。我们用 ADK / CrewAI / LangGraph 三个 Python 包做代表,把"模式"和"两种形态"讲透。最难的适配——ZepStore(在时序图谱上假扮一个 KV 存储)——单独留给 03-zepstore-hybrid-delegate.md。

2. 顶层全景(它大概怎么转)​

怎么读下面这张图: 中间竖线左边是不变的记忆环(所有集成共享),右边是每个框架各自的插座;两种形态 A / B 是灯泡通电的两种方式。

              不变的记忆环(见 01 章)                每个框架的"原生"钩子
        ┌──────────────────────────────┐      ┌────────────────────────────┐
        │  create user → create thread │      │  ADK: process_llm_request  │
        │  → add_messages → 取上下文    │◀────▶│  LangGraph: 图节点里手写    │
        │                              │      │  CrewAI: storage.save/search│
        │  Zep SDK: thread.* / graph.* │      └────────────────────────────┘
        └──────────────┬───────────────┘
                       │ 集成层把环"接"进钩子,分两种形态:
          ┌────────────┴─────────────┐
          ▼                          ▼
  ┌───────────────────┐    ┌────────────────────────┐
  │ 形态 A 自动注入     │    │ 形态 B 按需搜索工具       │
  │ 每轮: 取 Context   │    │ 把 graph.search 变成一个 │
  │ Block, 拼到 system │    │ tool, 由"模型"决定何时调  │
  │ prompt 最前面      │    │ (参数可在构造时 pin 死)   │
  │ ——你不写调用       │    │ ——模型主动写调用          │
  └───────────────────┘    └────────────────────────┘

两种形态一句话职责:

形态谁触发检索落点代表实现
A 自动上下文注入框架(每轮固定发生)system prompt 前缀ZepContextTool、build_system_message、ZepStorage.search
B 模型可调用工具模型(自己决定)一次 function call 的返回值ZepGraphSearchTool、create_graph_search_tool、ZepSearchTool

主线走一遍(高层,不进代码):

  1. 一次用户消息进来。
  2. 形态 A 无条件触发:把消息写进 Zep,同时取回一段"关于这个用户的 Context Block",塞到即将发给模型的 system 指令最前面。
  3. 模型带着这段记忆生成回复。若模型觉得"我还需要查点更具体的东西",它可以主动调用形态 B 的搜索工具,拿到几条事实再继续。
  4. 助手回复被写回 Zep,喂给下一轮。

两种形态不是二选一:典型 agent 会 A 打底(保证每轮都有基本记忆)、B 兜底(让模型能主动深挖)。

3. 目录布局与命名推导规则​

在读代码前,先建立"东西放在哪、叫什么"的地图。这层的组织本身就是一种设计。

framework-first,再 language。 目录先按框架分,框架下再按语言分(事实源:integrations/CLAUDE.md 的 "Layout: platform-first" 节):

integrations/
  <framework>/          # adk, crewai, langgraph, ...
    <language>/         # python | typescript | go
      src/zep_<framework>/
        __init__.py     # 包入口 + __version__ + 依赖自检
        <core>.py       # 核心适配(context / tools / memory)
        exceptions.py

命名从"框架目录名"这一个 key 机械推导,好让 CI 不必维护映射表:

已知推导规则例子
框架目录 key——pydantic-ai
Python import 名zep_ + key(连字符→下划线)zep_pydantic_ai
PyPI 分发名zep- + keyzep-pydantic-ai
Python 包路径integrations/<key>/pythonintegrations/pydantic-ai/python

这条规则的价值在 CI:加一个新 Python 包,只需在 paths-filter 里加一行,路径与 import 名都能从 key 自动算出(事实源:integrations/CLAUDE.md 的 "Naming convention" 段)。

每个包内部结构统一: __init__.py 里都有一段依赖自检——先尝试 import 框架本体,失败就抛一个友好的 ZepDependencyError 告诉你该装什么。例如 ADK 的入口在导入前先探 google.adk.tools.base_tool,见 integrations/adk/python/src/zep_adk/__init__.py:65-96;CrewAI 只硬依赖 crewai.tools(因为 1.x 删了旧的 Storage 基类),见 integrations/crewai/python/src/zep_crewai/__init__.py:51-90。

4. 形态 A:自动上下文注入(每轮拼 Context Block)​

它要解决的小问题: 让 agent"无脑地"每轮都带上关于用户的长期记忆,开发者一行检索代码都不用写。

思路/直觉: 找到每个框架里"消息即将发给模型、但还没发"的那个瞬间,在那一刻:①把这轮用户消息写进 Zep;②取回一段 prompt-ready 的 Context Block;③拼到 system 指令最前面。三步的位置因框架而异,动作完全一样。

4.1 ADK:劫持 process_llm_request 钩子​

ADK 提供了一个官方钩子 process_llm_request——ADK 自带的 PreloadMemoryTool 也用它。Zep 的做法是同一个钩子:定义一个 BaseTool 子类,但它从不被模型调用,只在每次 LLM 请求前修改请求。

真实实现:ZepContextTool.process_llm_request 在 integrations/adk/python/src/zep_adk/context_tool.py:308-324 覆写该钩子。核心步骤都在这个方法里:抽取用户文本 → 从会话 state 解析身份 → 同轮去重 → 写消息+取上下文 → 注入(建资源已改为带外 ensure_user/ensure_thread,见下)。

注入那一步是关键——把上下文包进一对 <ZEP_CONTEXT> 标签,再用 ADK 的 llm_request.append_instructions() 追加进系统指令,见 context_tool.py:406-408:

# 示意,贴近源码。重点看:上下文被包进标签,再 append 进 system 指令
if context_text:
    instruction = (
        "The following context is retrieved from Zep's long-term memory ...\n\n"
        "<ZEP_CONTEXT>\n"
        f"{context_text}\n"
        "</ZEP_CONTEXT>"
    )
    llm_request.append_instructions([instruction])

身份不写死、从会话 state 运行时解析,所以一个 ZepContextTool 实例能被所有用户/会话共享。解析顺序在 ZepContextTool._resolve_identity(context_tool.py:228-293):zep_user_id 优先,回退到 ADK 会话的 user_id;线程同理。

两个易被忽略的坑,Zep 都处理了:

  • 同轮重复触发。 在一轮里如果发生工具调用循环,process_llm_request 会用同一个 user_content Python 对象触发多次。Zep 用 id(user_content) 做"同轮守卫",跳过重复而不误伤后续轮里同样的文本(那会是新对象),见 context_tool.py:355-359。
  • 助手回复由另一个钩子写回。 用户侧消息由 ZepContextTool 写,助手侧由 create_after_model_callback 返回的 after_model_callback 写(integrations/adk/python/src/zep_adk/callbacks.py:30)。这个回调特意跳过含 function_call 的中间响应,只落一条干净的最终助手消息,见 callbacks.py:74-76。

默认单次往返 vs 自定义并行: 默认走 thread.add_messages(return_context=True)——一次 API 调用同时"写消息"和"取上下文"(context_tool.py:382-387)。若传了自定义 context_builder,则改用 asyncio.gather 并行跑"写消息"和"建上下文",降延迟,见 ZepContextTool._persist_and_build_context(context_tool.py:430-485)。

4.2 LangGraph:图节点里显式拼 system message​

LangGraph 没有隐藏钩子——它是一张显式的图,你在节点里自己写。所以 Zep 提供的是辅助函数,让你在 agent 节点顶部一行拿到 system message。

推荐用法(integrations/langgraph/python/src/zep_langgraph/context.py:290 的 build_system_message):

# 示意,贴近源码文档示例。重点看:节点顶部取上下文,prepend 到消息列表
async def agent_node(state):
    system = await build_system_message(
        zep_client,
        thread_id=state["thread_id"],
        base_instructions="You are a helpful assistant.",
    )
    messages = [system, *state["messages"]]
    response = await llm.ainvoke(messages)
    ...

底层就是 thread.get_user_context——注意它从整张用户图组装 Context Block,thread 只用来"界定此刻什么相关",见 get_zep_context(context.py:118)对 zep_client.thread.get_user_context 的封装(context.py:173)。拼装逻辑在 format_context_block(context.py:258),默认用 DEFAULT_CONTEXT_TEMPLATE(context.py:49)把上下文包进 <MEMORY> 标签。

同步/异步双份 API。 LangGraph 节点可能是 sync 也可能是 async,所以每个 helper 都成对出现:get_zep_context / get_zep_context_sync、build_system_message / build_system_message_sync(context.py:191、192)。这是本层一个反复出现的模式。

4.3 CrewAI:实现存储的 search() 契约​

CrewAI 期望一个带 save() / search() 的存储对象。Zep 的 ZepStorage(integrations/crewai/python/src/zep_crewai/memory.py:20)就实现这个契约,把"取上下文"藏在 search() 后面。

它的 search() 值得一看:并行取两个来源再合并——线程上下文(thread.get_user_context)+ 针对用户图的 graph.search(scope="edges"),用 ThreadPoolExecutor(max_workers=2) 同时发,见 memory.py:104-170。

CrewAI 1.x 删掉了旧的 Storage 基类和 ExternalMemory 包装器,所以这些类现在是独立、鸭子类型的适配器(只靠 save/search/reset 三个方法),不再继承任何 CrewAI 基类,见 memory.py:20-26 的类注释。

4.4 三者对照:同一动作,三个落点​

维度ADKLangGraphCrewAI
落点process_llm_request 钩子图节点里手写调用Storage.search() 契约
谁写消息工具 + after-model 回调persist_messages helperZepStorage.save()
取上下文 APIadd_messages(return_context=True)thread.get_user_context线程上下文 + graph.search 合并
身份来源会话 state(运行时)调用方传 thread_id构造时传 user_id/thread_id
共享单实例✅(身份运行时解析)✅(纯函数)❌(每对 user/thread 一个)

共同的暗线: 无论落点在哪,取回的都是同一种东西——一段从整张用户图组装的 prompt-ready 字符串。差异只在"框架允许我在哪一刻把它塞进 prompt"。

5. 形态 B:模型可调用的按需搜索工具​

它要解决的小问题: 形态 A 每轮拼的是"泛泛相关"的上下文;但模型有时需要主动、精确地查一件具体事("用户上次提到的那个项目截止日是哪天?")。这时应该把 graph.search 交给模型,让它自己决定何时搜、搜什么。

思路/直觉: 把 graph.search 包成一个标准工具,放进模型的工具列表。模型看到工具的 schema、自行决定调用、拿到格式化后的文本结果继续推理。

5.1 参数 pin 机制(ADK 的 ZepGraphSearchTool)​

graph.search 有一大堆参数(scope、reranker、limit、mmr_lambda……)。不该把它们全暴露给模型——多数场景你想把 scope 锁成 edges、reranker 锁成 rrf,只让模型填 query。

ADK 的方案叫 pin(钉死):构造时以关键字传入的参数,会从模型看到的 schema 里删掉并锁成固定值。所有可调参数的元定义在 _SEARCH_PARAMS(integrations/adk/python/src/zep_adk/graph_search_tool.py:42)。

_build_declaration 在生成模型可见的函数声明时,跳过所有已 pin 的参数,见 graph_search_tool.py:260-289:

# 示意,贴近源码。重点看:pin 掉的参数不出现在模型 schema 里
for param_name, param_def in _SEARCH_PARAMS.items():
    if param_name in self._pinned:
        continue                     # 已 pin → 对模型隐藏
    properties[param_name] = types.Schema(...)   # 未 pin → 暴露给模型

执行时参数按优先级合并:pinned > 模型提供 > 默认值,见 run_async(graph_search_tool.py:316-324)。

两条巧妙的边界规则:

  • pin 成 None = "隐藏但不发送"。 用来从 schema 里抹掉一个可选参数,又不真的把它传给 SDK(比如 reranker 不是 mmr 时把 mmr_lambda=None)。构造时会拒绝把必填参数 pin 成 None,见 graph_search_tool.py:237-244。
  • user_id 不许 pin。 它必须在运行时从会话 state 解析,构造时显式禁止,见 graph_search_tool.py:225-230。

5.2 目标解析:个人图 vs 共享图​

同一个搜索工具,可能指向两种目标之一:

        构造时给了 graph_id 吗?
              │
      ┌───────┴────────┐
     是                否
      │                │
      ▼                ▼
 搜"共享图"        运行时从会话 state
 (所有用户共用,   解析 user_id, 搜
  如文档知识库)    "该用户的个人图"

ADK 版的解析逻辑在 run_async 开头(graph_search_tool.py:305-311):有 graph_id 就用它;否则调 _resolve_user_id(graph_search_tool.py:349-360)从 state 取 zep_user_id、回退到 ADK 会话 user_id。

LangGraph 版把选择做成互斥且必须恰好给一个:_resolve_target 用 bool(user_id) == bool(graph_id) 判定"两个都给或都不给"就报错,见 integrations/langgraph/python/src/zep_langgraph/tools.py:191-198。这体现了两个框架的哲学差异:ADK 图省事(运行时兜底),LangGraph 求显式(构造时锁死)。

5.3 结果格式化:图 → 模型能读的文本​

搜索结果是结构化的 GraphSearchResults,得压成模型能直接读的紧凑文本。格式随 scope 变(graph_search_tool.py:122-165 scope_results_to_texts、tools.py:54-81):

scope取什么每行渲染成
edgesedge.fact- <事实>
nodesnode.name + node.summary- <名>: <摘要>
episodesepisode.content- <原始内容>
auto预组装的 result.context直接返回整段

5.4 LangGraph 版:同一形态,不同底座​

LangGraph 的 create_graph_search_tool(tools.py:94)做的是同一件事,但用 LangChain 的 StructuredTool.from_function 产出工具(tools.py:156),参数在构造时闭包捕获(不像 ADK 那样跑一套 pin/schema 生成)。暴露给模型的 schema 只有一个 query 字段(_GraphSearchInput,tools.py:46-51)。同样成对提供 sync 版 create_graph_search_tool_sync(tools.py:164)。

CrewAI 的 ZepSearchTool(integrations/crewai/python/src/zep_crewai/tools.py:309)则相反:它把 scope/limit 也暴露给模型(SearchMemoryInput,tools.py:18-26),并支持 scope="all" 一次搜三种范围再拼。这说明同一形态在不同框架里"给模型多少自由度"是可调的——从"只给 query"到"连 scope 都交给模型"。

5.5 两种形态并存:CrewAI 的写侧路由​

顺带一提写侧:CrewAI 的 ZepStorage.save() 用 metadata.type 做路由——message 走 thread.add_messages(进对话历史),json/text 走 graph.add(进知识图谱),见 memory.py:54-102。而 ZepGraphStorage(integrations/crewai/python/src/zep_crewai/graph_storage.py:18)专管非用户维度的共享图,其 search() 复用 search_graph_and_compose_context——并行搜 edges/nodes/episodes 三种 scope、再用 compose_context_string 拼成一段,见 integrations/crewai/python/src/zep_crewai/utils.py:82-162。

6. 横切原则:永不因 Zep 失败而崩溃​

这是贯穿所有 12 个集成的一条铁律(integrations/CLAUDE.md 明写:"Log Zep errors; never crash the agent on a Zep failure")。记忆是增强,不是依赖——Zep 挂了,agent 应该继续跑,只是"这轮没记忆"。

降级点在每一个可能触网的地方,策略统一:catch → log → 返回中性值。

位置失败时的中性返回引用
ADK 取上下文return(这轮不注入)context_tool.py:395-400
ADK 身份解析失败warning + return(跳过本轮持久化)context_tool.py:340-348
ADK 资源不存在warning 提示先跑 ensure_user/ensure_thread,本轮照常跳过注入context_tool.py:414-424
LangGraph 取上下文返回 None(节点照常跑)context.py:177-183
LangGraph 搜索工具返回错误字符串给模型tools.py:151-153
CrewAI 搜索返回空列表graph_storage.py:171-173

注意搜索工具的降级尤其巧妙:它不抛异常,而是把 f"Memory search failed: {exc}" 作为工具返回值交给模型(tools.py:153、graph_search_tool.py:339-341)。模型看到"搜索失败"这句话,自然会绕过去继续答——agent 循环一次都不会断。

建资源改成了显式的带外步骤。 旧版 _ensure_resources 在工具内部惰性建用户/线程;HEAD 把这步整个移出 ZepContextTool——工具从不建资源,要求先调 provisioning.ensure_user/ensure_thread(create-then-catch-conflict:409/"already exists" 视为成功,判定在 _is_already_exists_conflict,provisioning.py:36-54;context_tool.py:360-362 的注释写明该约束)。这样重复启动、并发首轮都不会误伤,身份供给也和多轮注入解耦。

7. 巧妙之处(可借鉴的技术)​

  • 用 id() 做"同轮去重"。 靠比较 Python 对象身份而非内容,精准跳过工具循环里的重复触发,又不误伤后续轮里同样的用户文本,context_tool.py:355-359。
  • 写读合一 vs 并行二选一。 默认 add_messages(return_context=True) 一次往返;需要自定义上下文时才 asyncio.gather 并行,把"简单场景省调用、复杂场景省延迟"都占了,context_tool.py:372-380、430-485。
  • pin-to-None 语义。 同一个"隐藏参数"动作,既能锁定值也能"隐藏且不发送",一个机制覆盖两种意图,graph_search_tool.py:237-244、248-256。
  • 失败即工具返回值。 把 Zep 错误转成给模型读的字符串而非异常,让降级对 agent 循环完全透明,tools.py:151-153。
  • 协议限制显式成常量。 LangGraph 把 Zep 的硬限制写成 MAX_MESSAGE_CHARS = 4096、MAX_MESSAGES_PER_CALL = 30,超长自动截断、超量自动分批,而不是撞 400,integrations/langgraph/python/src/zep_langgraph/persistence.py:44-48、145-164。
  • 命名机械可推导。 从一个框架 key 算出 import 名、PyPI 名、包路径,让加新集成近乎零配置(integrations/CLAUDE.md)。

8. 边界与局限(诚实)​

  • 摄取是异步的,没有读己所写。 刚 add 的一条事实不会在同一轮立刻可检索,集成层显式提醒按"最终可用"设计,persistence.py:24-27。所以形态 A 在当轮注入的是"截至上一轮"的记忆。
  • 形态 A 每轮多一次(或并行一次)网络往返,是延迟成本;高频场景要权衡。
  • CrewAI 存储不能跨用户共享单实例——user_id/thread_id 绑在构造时;ADK/LangGraph 才是运行时解析。
  • ZepStore 不在本章。 它在 KV 存储契约上假扮 Zep,是本层最难、语义最拧巴的适配,单独见 03-zepstore-hybrid-delegate.md。
  • 本章只举 ADK/CrewAI/LangGraph 三个 Python 代表;其余 9 个包(含 3 个 TypeScript、1 个 Go)遵循同样两种形态,不逐一展开。

9. 代码地图(导航索引)​

主题文件路径符号名
形态 A:ADK 自动注入钩子integrations/adk/python/src/zep_adk/context_tool.pyZepContextTool.process_llm_request
身份运行时解析integrations/adk/python/src/zep_adk/context_tool.pyZepContextTool._resolve_identity
惰性建资源(409 容错)integrations/adk/python/src/zep_adk/context_tool.pyZepContextTool._ensure_resources
写读合一 vs 并行integrations/adk/python/src/zep_adk/context_tool.pyZepContextTool._persist_and_build_context
助手回复写回(跳过 tool 调用)integrations/adk/python/src/zep_adk/callbacks.pycreate_after_model_callback
形态 B:参数 pin + schema 生成integrations/adk/python/src/zep_adk/graph_search_tool.pyZepGraphSearchTool._build_declaration
参数合并 + 目标解析integrations/adk/python/src/zep_adk/graph_search_tool.pyZepGraphSearchTool.run_async
结果格式化integrations/adk/python/src/zep_adk/graph_search_tool.pyZepGraphSearchTool._format_results
形态 A:LangGraph 节点助手integrations/langgraph/python/src/zep_langgraph/context.pybuild_system_message / get_zep_context
形态 B:LangGraph 搜索工具integrations/langgraph/python/src/zep_langgraph/tools.pycreate_graph_search_tool / _resolve_target
协议限制常量 + 分批integrations/langgraph/python/src/zep_langgraph/persistence.pyMAX_MESSAGE_CHARS / _chunk_messages
形态 A:CrewAI 存储契约integrations/crewai/python/src/zep_crewai/memory.pyZepStorage.save / ZepStorage.search
共享图存储 + 并行三 scopeintegrations/crewai/python/src/zep_crewai/utils.pysearch_graph_and_compose_context
形态 B:CrewAI 搜索工具integrations/crewai/python/src/zep_crewai/tools.pyZepSearchTool
布局与命名推导(事实旁证)integrations/CLAUDE.md—

同组其它章: index.md(总览与阅读地图) · 01-memory-model.md(通用记忆环:所有集成的地基) · 03-zepstore-hybrid-delegate.md(最难的适配) · 04-mcp-server.md(Go 版只读 MCP) · 05-evaluation.md(评测)