Deep Agent 多轮上下文解决方案(实战指南)

适用对象: 正在或即将集成 deepagents 框架的开发者阅读时长: 10 分钟实操时长: 30 分钟

本指南讲解 deepagents / LangGraph 的会话上下文(对话记忆)机制,并以糖豆问数项目为例,给出从"上下文失效"到"完整可用"的完整修复路径。

0. TL;DR(三句话版)

python

# 1. checkpointer 必须是进程级单例
from langgraph.checkpoint.memory import InMemorySaver
CHECKPOINTER = InMemorySaver()

# 2. 每次 stream / invoke 必须传 thread_id
config = {"configurable": {"thread_id": f"{user_id}_{conv_id}"}}
result = await agent.ainvoke(messages, config=config)

# 3. agent 实例本身在进程内复用(不要每次请求重建)

如果只能记住三件事,就这三件。

1. 背景:DeepAgent 的状态机制

1.1 两层概念

概念	作用	默认行为
`thread_id`	会话隔离键(一个 thread = 一个独立会话)	不传时所有调用共享一个匿名 thread,历史互相覆盖
`checkpointer`	LangGraph 的状态持久化器,按 `thread_id` 索引快照	不传时完全无状态,每次调用都从头跑

1.2 数据流

agent.ainvoke(messages, config={"configurable": {"thread_id": "U1_C1"}})
    │
    ├─ ① checkpointer.load("U1_C1")
    │     └─ 拿到上次的 state(messages, scratchpad, 自定义字段)
    │
    ├─ ② state.messages += messages   # 追加新消息
    │
    ├─ ③ LLM + tools 推理
    │
    └─ ④ checkpointer.save("U1_C1", new_state)   # 保存新快照

下一轮 ainvoke 传同一个 thread_id,会从 ① 重新加载,LLM 看到完整历史。

1.3 为什么"看起来没生效"?

大多数开发者遇到的坑:

现象	原因
第二次对话记不住上文	checkpointer 没设 → 完全无状态
服务重启后上下文全丢	checkpointer 是 `InMemorySaver`
多 worker 时上下文断断续续	每个 worker 独立的 saver
多个用户上下文串台	没传 `thread_id` 或拼错
agent 每次请求都很慢	agent 实例在每次请求重新构建(掩盖了 checkpointer 问题)

2. 实施步骤

Step 1:确认 deepagents 版本

bash

pip show deepagents

当前糖豆项目用 >=0.2.x。本文档所有 API 在该版本上验证过。

Step 2:加 checkpointer 单例

新建 / 修改 app/agent/agent.py:

python

from langgraph.checkpoint.memory import InMemorySaver

# 关键:模块级单例,所有 agent 共用
_CHECKPOINTER = InMemorySaver()


def get_checkpointer():
    """给 chat_service 调用的 accessor。"""
    return _CHECKPOINTER

⚠️ 不要在函数内部 InMemorySaver(),那每次都是新对象。

Step 3:`create_deep_agent` 接入

python

from deepagents import create_deep_agent

def create_deep_agent_instance(provider, base_dir, checkpointer=None):
    ...
    agent = create_deep_agent(
        model=...,
        system_prompt=...,
        tools=...,
        skills=...,
        memory=...,
        backend=...,
        checkpointer=checkpointer or _CHECKPOINTER,  # ← 关键
        debug=True,
    )
    return agent

Step 4:`stream_config` 必传 `thread_id`

python

thread_id = f"{user_id}_{conversation_id}"
stream_config = {
    "recursion_limit": 25,
    "configurable": {"thread_id": thread_id},
}

agent.stream(
    {"messages": [{"role": "user", "content": msg}]},
    config=stream_config,
    stream_mode="updates",
)

Step 5:agent 实例进程级复用(可选但强烈推荐)

python

class ChatService:
    def __init__(self):
        self._checkpointer = get_checkpointer()
        self._agents: dict[str, Any] = {}
        self._agent_lock = threading.Lock()

    def _get_agent(self, provider: str):
        """按 provider 懒加载;同 provider 后续请求直接复用。"""
        agent = self._agents.get(provider)
        if agent is not None:
            return agent
        with self._agent_lock:  # double-check lock
            agent = self._agents.get(provider)
            if agent is None:
                agent = create_deep_agent_instance(
                    provider, self.base_dir, checkpointer=self._checkpointer,
                )
                self._agents[provider] = agent
        return agent

3. 验证 Checklist

复制下面这段到团队 wiki,逐项打勾:

markdown

- [ ] checkpointer 是**进程级单例**,不是函数内 `new`
- [ ] 每个 `ainvoke/stream` 都传了 `configurable.thread_id`
- [ ] `thread_id` 拼接规则:同一会话稳定,不同会话唯一
- [ ] (可选)agent 实例缓存,避免每次重建
- [ ] (生产)checkpointer 换成 `PostgresSaver` / `RedisSaver`
- [ ] (生产)如用多 worker,所有 worker 共享同一数据库

快速验证脚本

python

# debug_checkpointer.py
import asyncio
from app.agent.agent import get_checkpointer, create_deep_agent_instance

cp = get_checkpointer()
print(f"[CHECK] checkpointer id={id(cp)}")

agent = create_deep_agent_instance(provider='ollama', checkpointer=cp)
print(f"[CHECK] agent id={id(agent)}")

async def main():
    cfg = {"configurable": {"thread_id": "test_thread"}}
    r1 = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "你好,我叫张三"}]},
        config=cfg,
    )
    r2 = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "我叫什么?"}]},
        config=cfg,
    )
    # r2 的 messages 里应当包含 "我叫张三" 的历史

asyncio.run(main())

4. 常见陷阱与对策

4.1 `thread_id` 撞车

python

# ❌ 危险
thread_id = f"{user_id}_{conversation_id}"
# alice_smith + 42 = "alice_smith_42"
# alice + smith_42 = "alice_smith_42"  # 撞!

# ✅ 加类型前缀
thread_id = f"u{user_id}_c{conversation_id}"

4.2 `InMemorySaver` vs 持久化

方案	重启丢历史	跨 worker 共享	性能
`InMemorySaver`	✅ 丢	❌ 不共享	最快
`SqliteSaver`	❌ 不丢	❌ 文件锁,单进程	快
`PostgresSaver`	❌ 不丢	✅ 共享	中
`RedisSaver`	❌ 不丢	✅ 共享	快

生产推荐:PostgresSaver(已有 PG)或 RedisSaver(已有 Redis)。

4.3 多 worker 部署的 sticky session

如果短期切不到 PostgresSaver,可用 sticky session 让同一用户请求落到同一 worker:

nginx

upstream backend {
    ip_hash;  # 按 IP 哈希
    server worker1:8000;
    server worker2:8000;
    server worker3:8000;
}

这样 worker1 上的会话历史对 worker2 不可见,但因为 sticky,用户永远只走 worker1,历史能保留。

4.4 多 provider 不共享 thread_id

如果项目支持多个 LLM provider(ollama + minimax),不要让不同 provider 共享同一 thread_id:

python

# ❌ 错
thread_id = f"{user_id}_{conv_id}"
agent_ollama.stream(..., config={"configurable": {"thread_id": t}})
agent_minimax.stream(..., config={"configurable": {"thread_id": t}})  # 共享 state,但 agent 不同!

# ✅ 对
thread_id = f"{provider}_{user_id}_{conv_id}"

原因是:LangGraph 的 state 包含消息历史,不同 agent 看到的字段格式可能不同(不同 LLM 的 tool calling 协议不一样),混用会让对方解析失败。

4.5 `interrupt_on` 配合 checkpointer

checkpointer 是 interrupt_on(人机协作审批)的前置条件:

python

agent = create_deep_agent(
    ...,
    checkpointer=cp,
    interrupt_on={"sql_db_query": True},  # 执行 SQL 前暂停
)

# 首次调用 → agent 暂停,返回 __interrupt__
result = await agent.ainvoke(messages, config)

# 人工审核后,resume
from langgraph.types import Command
result = await agent.ainvoke(Command(resume=True), config)

没有 checkpointer,interrupt_on 不会生效。

5. 进阶:`configurable` 的其他字段

configurable 不止 thread_id,还能塞自定义数据:

python

stream_config = {
    "recursion_limit": 25,
    "configurable": {
        "thread_id": thread_id,
        "user_id": user_id,
        "checkpoint_id": "...",    # 时间旅行:从指定 checkpoint 重放
        "checkpoint_ns": "...",   # checkpoint 命名空间
    },
}

checkpoint_id 用法:

python

# 从历史第3 个 checkpoint 重放(比如某个 tool 调用结果异常,想跳过)
config = {"configurable": {"thread_id": t, "checkpoint_id": "1ef4f..."}}
result = await agent.ainvoke(None, config)  # 传 None 重放

6. 与 LangChain `Memory` 的区别

LangChain 旧版有 ConversationBufferMemory / ConversationSummaryMemory,但和 LangGraph checkpointer 不是一回事:

	LangChain Memory	LangGraph Checkpointer
状态范围	只持久化 messages	持久化整个 graph state(含工具结果、自定义字段)
适用范围	老式 Chain / AgentExecutor	LangGraph / DeepAgent
多轮控制	自动	通过 `thread_id` 显式
时间旅行	不支持	✅ 支持

结论:用 deepagents 就不要用 LangChain Memory,直接用 LangGraph checkpointer。

7. 与 `agent.memory=` 参数的区别

deepagents 的 create_deep_agent(memory=[file.md]) 是把 .md 文件当作长期知识塞进 system prompt(类似 RAG 文档):

python

agent = create_deep_agent(
    memory=['/path/to/AGENTS.md'],  # 每次调用都把这文件内容拼进 system message
)

与 checkpointer 的区别:

	`memory=`	`checkpointer=`
内容	静态知识(规则、约定)	动态会话历史
持久	文件系统	saver(内存/DB)
是否需要 thread_id	❌	✅

两者并存不冲突,本项目两个都启用。

8. 性能与监控

关键指标

python

import time
start = time.perf_counter()
result = await agent.ainvoke(messages, config)
logger.info(f"agent took {time.perf_counter() - start:.2f}s, "
            f"thread_id={config['configurable']['thread_id']}, "
            f"msgs_in={len(messages['messages'])}, "
            f"msgs_out={len(result['messages'])}")

告警

单次 ainvoke > 30s → LLM 调用慢
msgs_out - msgs_in 不增长 → agent 没生成回复
msgs_out 暴涨(>20)→ agent 死循环,被 recursion_limit 截断

9. 调试 Checklist

如果上下文不生效,按这个顺序排查:

id(checkpointer) 两次请求相同?——不同 → 单例没生效
thread_id 两次请求相同?——不同 → 拼接逻辑有问题
configurable.thread_id 真的传进去了?——打印 stream_config 看
第二次 ainvoke 后 state 真的更新了?——checkpointer.get(thread_id) 应该返回非空 tuple
agent 是同一个?——不同 agent 看到的状态格式可能不同

10. 完整示例(糖豆项目风格)

python

# app/services/chat_service.py
from app.agent.agent import create_deep_agent_instance, get_checkpointer


class ChatService:
    def __init__(self):
        self._checkpointer = get_checkpointer()  # 单例
        self._agents: dict[str, Any] = {}        # agent 单例缓存
        self._agent_lock = threading.Lock()

    def _get_agent(self, provider: str):
        agent = self._agents.get(provider)
        if agent is not None:
            return agent
        with self._agent_lock:
            agent = self._agents.get(provider)
            if agent is None:
                agent = create_deep_agent_instance(
                    provider, self.base_dir, checkpointer=self._checkpointer,
                )
                self._agents[provider] = agent
        return agent

    async def _run(self, message, conversation_id, user_id, provider, db):
        thread_id = f"u{user_id}_c{conversation_id}"
        agent = self._get_agent(provider or load_config().llm_provider)

        # 这里 stream_config 必传 thread_id
        stream_config = {
            "recursion_limit": 25,
            "configurable": {"thread_id": thread_id},
        }

        async for chunk in agent.stream(
            {"messages": [{"role": "user", "content": message}]},
            config=stream_config,
            stream_mode="updates",
        ):
            ...

11. 参考资料

LangGraph Persistence 官方文档
LangGraph Thread 概念
DeepAgents 源码
本项目修复详情:[context-fix.md](file:///d:/1sqyai/tangdou/backend/docs/context-fix.md)

12. 一句话总结

checkpointer 进程级单例 + thread_id 每次传 + agent 实例复用 = 多轮上下文生效。

Deep Agent 多轮上下文解决方案(实战指南) ​

0. TL;DR(三句话版) ​

1. 背景:DeepAgent 的状态机制 ​

1.1 两层概念 ​

1.2 数据流 ​

1.3 为什么"看起来没生效"? ​

2. 实施步骤 ​

Step 1:确认 deepagents 版本 ​

Step 2:加 checkpointer 单例 ​

Step 3:create_deep_agent 接入 ​

Step 4:stream_config 必传 thread_id ​

Step 5:agent 实例进程级复用(可选但强烈推荐) ​

3. 验证 Checklist ​

快速验证脚本 ​

4. 常见陷阱与对策 ​

4.1 thread_id 撞车 ​

4.2 InMemorySaver vs 持久化 ​

4.3 多 worker 部署的 sticky session ​

4.4 多 provider 不共享 thread_id ​

4.5 interrupt_on 配合 checkpointer ​

5. 进阶:configurable 的其他字段 ​

6. 与 LangChain Memory 的区别 ​

7. 与 agent.memory= 参数的区别 ​

8. 性能与监控 ​

关键指标 ​

告警 ​

9. 调试 Checklist ​

10. 完整示例(糖豆项目风格) ​

11. 参考资料 ​

12. 一句话总结 ​

Deep Agent 多轮上下文解决方案(实战指南)

0. TL;DR(三句话版)

1. 背景:DeepAgent 的状态机制

1.1 两层概念

1.2 数据流

1.3 为什么"看起来没生效"?

2. 实施步骤

Step 1:确认 deepagents 版本

Step 2:加 checkpointer 单例

Step 3:`create_deep_agent` 接入

Step 4:`stream_config` 必传 `thread_id`

Step 5:agent 实例进程级复用(可选但强烈推荐)

3. 验证 Checklist

快速验证脚本

4. 常见陷阱与对策

4.1 `thread_id` 撞车

4.2 `InMemorySaver` vs 持久化

4.3 多 worker 部署的 sticky session

4.4 多 provider 不共享 thread_id

4.5 `interrupt_on` 配合 checkpointer

5. 进阶:`configurable` 的其他字段

6. 与 LangChain `Memory` 的区别

7. 与 `agent.memory=` 参数的区别

8. 性能与监控

关键指标

告警

9. 调试 Checklist

10. 完整示例(糖豆项目风格)

11. 参考资料

12. 一句话总结