LangChain DeepAgents stream / astream 调用参数说明

先说结论

deepagents.create_deep_agent() 返回的是一个 LangGraph 的 CompiledStateGraph，所以它的流式接口本质上就是底层 LangGraph 图对象的：

• agent.stream(...)
• agent.astream(...)

也就是说，DeepAgents 没有单独发明一套新的 stream 参数体系，而是在 LangGraph 标准流式接口上，额外强调了“子代理流式输出”这个场景。

完整签名

根据 LangGraph Python Reference，stream / astream 的签名如下：

stream(
    input: InputT | Command | None,
    config: RunnableConfig | None = None,
    *,
    context: ContextT | None = None,
    stream_mode: StreamMode | Sequence[StreamMode] | None = None,
    print_mode: StreamMode | Sequence[StreamMode] = (),
    output_keys: str | Sequence[str] | None = None,
    interrupt_before: All | Sequence[str] | None = None,
    interrupt_after: All | Sequence[str] | None = None,
    durability: Durability | None = None,
    subgraphs: bool = False,
    debug: bool | None = None,
    **kwargs,
) -> Iterator[dict[str, Any] | Any]

astream(
    input: InputT | Command | None,
    config: RunnableConfig | None = None,
    *,
    context: ContextT | None = None,
    stream_mode: StreamMode | Sequence[StreamMode] | None = None,
    print_mode: StreamMode | Sequence[StreamMode] = (),
    output_keys: str | Sequence[str] | None = None,
    interrupt_before: All | Sequence[str] | None = None,
    interrupt_after: All | Sequence[str] | None = None,
    durability: Durability | None = None,
    subgraphs: bool = False,
    debug: bool | None = None,
    **kwargs,
) -> AsyncIterator[dict[str, Any] | Any]

astream(...) 和 stream(...) 参数完全一致，只是一个返回同步迭代器，一个返回异步迭代器。

参数逐项说明

input

本次运行的输入。

在 DeepAgents 里最常见的是：

{
    "messages": [
        {"role": "user", "content": "帮我研究一下 LangGraph"}
    ]
}

也可以传：

• Command(...)：用于中断后恢复执行
• None：某些恢复场景下可用

config

运行配置，通常放线程 ID、可配置项等。

最常见写法：

config = {"configurable": {"thread_id": "thread-123"}}

这个参数在以下场景尤其重要：

• 需要 checkpoint 持久化
• 需要 human-in-the-loop 中断恢复
• 想把多次调用绑定到同一个会话线程

context

静态上下文对象。官方说明是 “The static context to use for the run”。

这个参数一般不是 DeepAgents 入门时最常用的；更多是你在 LangGraph 层自己扩展上下文 schema 时使用。

stream_mode

最重要的参数。决定你“流出来的到底是什么”。

可以传单个模式，也可以传模式列表。

支持的模式有：

• "values"：每一步后输出完整 state
• "updates"：每一步后只输出增量更新
• "messages"：输出 LLM token 流和元数据
• "custom"：输出你在节点/工具里主动写出的自定义流事件
• "checkpoints"：输出 checkpoint 事件，需要 checkpointer
• "tasks"：输出任务开始/结束事件，需要 checkpointer
• "debug"：输出最完整的调试信息

1. values

每一步都给你完整状态快照。

适合：

• 想直接观察整个 state 如何演化
• 调试 graph 状态

缺点是输出通常更重。

2. updates

每一步只给变化部分。

适合：

• 看节点执行进度
• 看子代理阶段性结果
• 前端做状态刷新

在 DeepAgents 里，这通常是看主 agent / subagent 步骤推进时最实用的模式。

3. messages

输出 (LLM token, metadata) 形式的消息流。

适合：

• 实时打印模型回复
• 前端 token-by-token 渲染
• 区分主 agent 和 subagent 的 token 来源

DeepAgents 里如果你想看主代理和子代理的 token 输出，这是最关键的模式之一。

4. custom

输出你在节点或工具内部通过 get_stream_writer() 主动写出来的数据。

适合：

• 自定义进度条
• 工具内部状态上报
• 长任务里的阶段信号

5. checkpoints

输出 checkpoint 事件，格式和 get_state() 类似。

前提：

• 图必须配置 checkpointer

适合：

• 持久化调试
• 观察每次落盘状态

6. tasks

输出任务开始/结束事件，以及结果和错误。

前提：

• 图必须配置 checkpointer

适合：

• 看 node/task 生命周期
• 做执行监控

7. debug

最重但最全的流式输出。

官方说明是它会结合 checkpoints、tasks 以及额外元数据。

适合：

• 深度调试
• 排查为什么 graph 行为不符合预期

print_mode

接受和 stream_mode 一样的值，但只打印到控制台，不改变函数本身返回的数据。

适合：

• 本地调试时顺手看输出
• 不想改主逻辑，只想额外打印某些流式事件

output_keys

指定只流式输出哪些 key。

官方说明是：默认会输出所有非 context channel。

适合：

• state 很大，只关心部分字段
• 想减少流式输出体积

interrupt_before

在指定节点执行前中断。

适合：

• 静态断点调试
• 某些需要人工审批的固定节点前暂停

注意：

• 官方更推荐 human-in-the-loop 场景使用动态 interrupt()，而不是过度依赖静态断点

interrupt_after

在指定节点执行后中断。

用途和 interrupt_before 类似，只是触发点不同。

durability

控制 graph 执行时状态持久化的时机。默认是 "async"。

可选值：

• "sync"：下一步开始前同步持久化
• "async"：下一步执行时异步持久化
• "exit"：只在 graph 退出时持久化

一般理解：

• sync：更稳，但可能更慢
• async：默认折中方案
• exit：最省过程持久化，但中途可靠性最低

subgraphs

是否把子图里的事件也一起流出来。默认 False。

这个参数对 DeepAgents 非常关键，因为 DeepAgents 的子代理本质上就是通过子图/子执行链路暴露流式事件。

如果你想看到 subagent 的事件，通常要：

subgraphs=True

开启后：

• 主 agent 事件的 ns 通常是空元组 ()
• 子代理事件的 ns 会带类似 ("tools:<task_id>",) 这样的命名空间

debug

额外调试开关。底层参数里存在，但实际日常最常用的还是直接用 stream_mode="debug"。

一般场景下不需要主动设置。

**kwargs

保留参数，底层用于兼容旧参数或其他扩展传递。

日常使用里通常不用主动传。

DeepAgents 场景下最常用的调用组合

1. 看步骤推进

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "研究一下 LangGraph"}]},
    stream_mode="updates",
    subgraphs=True,
    version="v2",
):
    ...

适合：

• 看主 agent 和 subagent 每一步在做什么

2. 看 token 实时输出

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "研究一下 LangGraph"}]},
    stream_mode="messages",
    subgraphs=True,
    version="v2",
):
    ...

适合：

• 实时展示模型输出
• 区分主代理和子代理的 token 来源

3. 同时看步骤、token、自定义事件

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "分析客户满意度趋势"}]},
    stream_mode=["updates", "messages", "custom"],
    subgraphs=True,
    version="v2",
):
    ...

适合：

• 做前端可视化
• 做复杂执行监控

version="v2" 为什么几乎总要加

虽然 version 不在上面 reference 的主签名里展开说明，但官方 streaming 文档明确推荐在流式调用时传：

version="v2"

原因是：

• v2 的返回结构统一
• 每个 chunk 都是固定结构
• 更适合 DeepAgents 的 subgraph/subagent 流式处理

v2 下，每个 chunk 都长这样：

{
    "type": "updates" | "messages" | "custom" | "values" | "checkpoints" | "tasks" | "debug",
    "ns": (),
    "data": ...
}

其中：

• type：当前 chunk 属于哪种流模式
• ns：命名空间，用来标识是主 agent 还是哪个 subagent
• data：真正的数据载荷

对 DeepAgents 来说，version="v2" + subgraphs=True 几乎是最推荐的组合。

DeepAgents 里最值得记住的 5 个参数

如果你只想先记最核心的，优先记这 5 个：

• input：本次输入，通常是 {"messages": [...]}
• stream_mode：你要看什么流
• subgraphs：要不要把子代理事件一起流出来
• config：线程 ID、恢复执行等配置
• version="v2"：统一返回格式，官方推荐

一个最小实用示例

from deepagents import create_deep_agent

agent = create_deep_agent(
    system_prompt="You are a helpful research assistant.",
    subagents=[
        {
            "name": "researcher",
            "description": "Researches a topic in depth",
            "system_prompt": "You are a thorough researcher.",
        }
    ],
)

for chunk in agent.stream(
    {
        "messages": [
            {"role": "user", "content": "Research quantum computing advances"}
        ]
    },
    stream_mode=["updates", "messages"],
    subgraphs=True,
    version="v2",
):
    if chunk["type"] == "updates":
        print("UPDATE", chunk["ns"], chunk["data"])
    elif chunk["type"] == "messages":
        token, metadata = chunk["data"]
        if token.content:
            print("TOKEN", chunk["ns"], token.content)

参考资料

• Deep Agents Streaming
• LangGraph Streaming
• LangGraph Interrupts
• LangGraph Reference: Pregel.stream / astream