前沿预览 EP.14

Streaming — 实时观察主 Agent、子 Agent 与工具调用

在 GitHub 编辑此页

研究助手已经跑了两分钟。后台日志显示它调用了搜索工具，也确实启动了 researcher；浏览器里的用户却只看到一个转圈图标。等到最终答案出现时，用户不知道它是在工作，还是已经卡住了。

模型仍然可能给出一份好答案。真正让用户不安的是，应用把两分钟的运行过程压扁成了一个最终值。本章沿着这个故障改造“黑盒研究助手”：先看见委派，再看见子 Agent 的消息和工具调用，随后修复事件顺序，最后处理旧代码里的底层协议。

示例以 Deep Agents v0.6 引入的 Typed Projection API 为主线。新应用优先使用 agent.stream_events(..., version="v3")；agent.stream(..., version="v2") 放在后半章，专门解释 LangGraph 的协议格式、namespace 和 custom updates。两者解决的问题不同，代码也不要混在同一个循环里。

1. 一次“看起来卡住”的研究请求

先看这个应用的原始调用。它没有错，甚至很适合脚本：

result = agent.invoke({"messages": [{"role": "user", "content": prompt}]})
print(result["messages"][-1].content)

对应到页面，运行期间只有这样一条状态：

用户：研究近期的 Agent Streaming 模式
系统：正在生成……

问题出在产品体验。invoke() 只有在整个运行结束后才返回，所以页面无法回答这些最基本的问题：

同一个研究请求在 invoke 与 Streaming 中的体验对比：invoke 只显示运行中和最终答案，Streaming 则逐步展开 coordinator、researcher、搜索工具和消息状态

我们先不急着设计漂亮的进度条。第一步只做一件事：把当前运行中“谁在工作”显示出来。

2. 先让案例稳定复现

如果完全依赖模型自由发挥，模型有时会直接回答问题，不调用 task，Streaming 页面也就没有稳定的子 Agent 可以展示。为了排查显示链路，我们先把研究请求固定委派给 researcher。

import os

from deepagents import create_deep_agent
from langchain_openai import ChatOpenAI


model = ChatOpenAI(
    model=os.environ.get("MODEL_NAME", "zai-org/GLM-5.2"),
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ.get("OPENAI_BASE_URL", "https://api.siliconflow.cn/v1"),
)

agent = create_deep_agent(
    model=model,
    system_prompt=(
        "You are a coordinator. Delegate every research request to the "
        "researcher subagent. Do not research the topic yourself. "
        "After the subagent returns, summarize its result in two sentences."
    ),
    subagents=[
        {
            "name": "researcher",
            "description": "Researches a topic and returns a concise summary.",
            "system_prompt": (
                "Research the topic, use available tools when useful, "
                "and return a concise evidence-aware summary."
            ),
        }
    ],
)

request = {
    "messages": [
        {"role": "user", "content": "Research recent Agent streaming patterns"}
    ]
}

运行到这里，我们已经有了一个可观察的树：顶层是 coordinator，下面是一次 researcher 委派，研究过程中还可能出现工具调用。模板中的搜索工具和模型必须支持 Tool Calling，否则后面只能看到 Agent 状态，看不到工具事件。

3. 第一个修复：先显示“研究助手已启动”

3.1 不要先从 graph node 猜产品状态

最底层的 LangGraph 会产生很多节点名称，例如 model_request、tools。它们对调试有用，却不适合作为产品界面。用户关心的是“研究助手”，不是某个内部节点。

Deep Agents 为委派提供了更直接的视图：stream.subagents。它的每一个 handle 对应一次 task 委派，并带有用户真正需要的名称、路径和生命周期状态。

stream = agent.stream_events(request, version="v3")

for subagent in stream.subagents:
    print(f"[{subagent.name}] {subagent.status}")
    print("path:", subagent.path)

    try:
        print("output:", subagent.output)
    except Exception as exc:
        print(f"failed: {exc}")

handle 刚出现时，官方 status 是 started；页面可以把它渲染成 researcher · running。如果运行成功，output 是子 Agent 的最终状态；如果它失败或被中断，读取最终输出时可能抛出异常，这个异常应成为卡片上的错误，而不是被吞掉。

3.2 拆解 subagent handle

这个 handle 既有身份字段，也有可以继续消费的 projection：

字段或 projection含义读取时机页面怎么用
namecoordinator 调用 task 时选择的 subagent_typehandle 出现时显示 researcher 等角色名称；不作为唯一键
path这次委派在 Agent 树中的 namespace 路径handle 出现时作为当前运行内的卡片路由键，区分多个同名子 Agent
statusstarted、completed、failed、interrupted 等生命周期状态收到 handle 及其后续状态时更新状态徽标；终态仍要结合 output 或异常判断
messages该子 Agent 发出的消息 projection需要文本过程时写入当前 path 对应的消息列表
tool_calls该子 Agent 发起的工具调用 projection需要工具细节时在当前卡片下建立工具行
values该子 Agent 的状态值 projection需要状态快照时调试或状态面板使用，不等同于文本消息
subagents该子 Agent 继续发起的嵌套委派存在多级委派时递归建立子卡片
output该委派的最终状态或完成信号需要等待终态时完成后保存最终结果；失败时捕获异常

name 的来源值得单独记住：它不是内部 graph node 名，也不是一次运行的唯一 ID。它来自 coordinator 为 task 选择的 subagent_type。同一个 researcher 可以被调用多次，所以 UI 应在当前运行内按 path 存储状态，用 name 显示标签。

status 是生命周期，不是质量结论：

状态表示什么应用动作
started委派已经被发现并开始运行创建或更新卡片为 running
completed子 Agent 已结束且可读取最终输出读取 output，保存结果并结束计时
failed子 Agent 以错误结束保存异常，卡片进入 failed
interrupted运行在中断点暂停保留现有状态，等待恢复或取消

completed 也不代表研究内容一定正确，只代表这次委派正常结束。业务验收仍要由测试、Rubric 或人工审核完成。

3.3 path、namespace 和 ns 是什么关系

这三个名字表达的是同一个概念：从根 Agent 走到当前执行位置的 namespace 路径。它们出现在不同 API 层，形状和作用域并不相同：

API字段Python 形状表示什么
v3 Typed Projectionsubagent.pathtuple[str, ...]当前子 Agent 委派的根路径
v3 raw protocolevent["params"]["namespace"]list[str]产生当前 raw event 的完整路径
v2 Streamingchunk["ns"]tuple[str, ...]产生当前 StreamPart 的完整路径
模板应用协议path 或 namespaceJSON list[str]adapter 序列化后的子 Agent 路径或 raw 事件路径

路径中的每一段采用 <node_name>:<runtime_or_task_id> 形式。Deep Agents 通过 task 工具启动一级子 Agent 时，subagent.path 通常只有一个 tools:<id> 段。这里的 tools 是承载委派的图节点，<id> 用于区分这一次执行；它不是 researcher 这样的显示名。

根 Agent 的 v2 ns              ()
根 Agent 的 v3 raw namespace   []
一级子 Agent 的 path            ("tools:abc123",)
子 Agent 根级事件的 v2 ns       ("tools:abc123",)
子 Agent 内部模型事件的 v2 ns   ("tools:abc123", "model_request:def456")
同一内部事件的 v3 namespace     ["tools:abc123", "model_request:def456"]

所以，子 Agent 根级事件的 namespace 在 tuple/list 归一化后通常等于 subagent.path；更深层的模型或工具事件会在后面追加路径段。这时 subagent.path 是完整 namespace 的前缀：

def belongs_to_subagent(namespace: list[str] | tuple[str, ...], path: tuple[str, ...]) -> bool:
    return tuple(namespace[: len(path)]) == path

同名 researcher 可以被调用多次，每次的 <id> 不同。因此 UI 在当前运行内用完整 path 作为卡片键，用 name 显示角色；不要用 name 代替路径，也不要要求子 Agent 内的每个事件 namespace 都与根 path 完全相等。

path 的唯一性边界也是当前运行。若要把多个运行写入数据库或 Trace，应用仍需先建立自己的请求记录，再把 path 作为该记录下的分支键；这不会让某个应用请求 ID 变成框架的流事件字段。

3.4 Projection 是按需打开的

刚才我们只关心“启动、结束、失败”，所以没有订阅子 Agent 的全部消息。v3 的 projection 是惰性的：访问 subagent.messages 或 subagent.tool_calls 时，才打开对应的细流。

这对生产页面很实用。只展示状态和耗时时，不需要消费每一条消息和工具增量；需要详细过程的界面，再打开相应 projection。没有必要为每个内部事件都建立一份 UI 状态。

父级 stream 也提供相同方向的 projection：

父级 projection范围
stream.messagescoordinator 的消息
stream.tool_callscoordinator 发起的工具调用
stream.values顶层运行的状态值
stream.subagentscoordinator 发起的委派
stream.output顶层运行的最终输出

v3 Typed Projection 的层级与作用域：顶层 stream 提供 messages、tool_calls、values、subagents 和 output；subagent handle 保存 name、path、status，并按需打开自己的同类 projections

这里最容易犯的错是范围混淆。stream.messages 不会替你合并所有子 Agent 的消息；subagent.messages 也不会包含 coordinator 的最终总结。它们是两条独立 projection。

values 需要再多解释一句：它不是另一种消息流，也没有 Deep Agents 统一规定的业务字段。它发出的是当前 Agent state 的快照，字段取决于应用和中间件。例如启用了 Todo 或自定义状态时，快照可能包含消息列表、todos 或其他运行状态。把它用于调试和状态面板比较合适；主对话文本仍从 messages 读取，工具结果仍从 tool_calls 读取。

stream = agent.stream_events(request, version="v3")

for value in stream.values:
    print("state snapshot:", value)

不要把每个 value 当成一个可直接追加的增量。它更像“这一时刻的状态快照”：页面只需要当前状态时，按 Agent 路径覆盖最新快照即可；需要持久化和重放时，再使用 raw protocol event 的 seq 排序。timestamp 适合展示时间，不保证严格顺序。

4. 第二个修复：让用户知道它在查什么

状态卡片解决了“是不是卡住”的问题，但用户很快会追问：“它到底在做什么？”现在把 coordinator 和 researcher 的消息都接出来。

stream = agent.stream_events(request, version="v3")

for message in stream.messages:
    print("[coordinator]", message.text)

for subagent in stream.subagents:
    for message in subagent.messages:
        print(f"[{subagent.name}]", message.text)

4.1 message 应该读什么

Typed Projection 已经把底层 content-block 事件整理成 message handle。本章示例只需要一个字段：

字段同步代码异步代码含义
textmessage.textawait message.text当前消息可显示的文本内容

同步和异步写法的差别不是语法装饰。异步 handle 的文本可能仍在到达，读取时需要 await。如果要保留逐个 text-delta 的精确顺序，不应从 message.text 反推，而要使用第 7 节的 raw events。

消息对象解决的是“显示什么”；消息属于谁由你消费的 projection 决定。这里要区分框架对象与应用对象：官方 message handle 没有 source 或 kind 字段，deepagents/streaming 模板的 event_adapter.py 才把它转换成下面的 SSE 事件：

{
    "kind": "message",
    "source": "coordinator",
    "path": [],
    "text": message.text,
    "final": False,
}

其中 kind 是模板 adapter 写入的事件判别字段，source 是模板路由根据当前消费的 projection 显式传入的标签；两者都不是 Deep Agents 或 LangGraph 自动附带的协议字段。子 Agent 消息使用 source="subagent" 并把 subagent.path 写入 path。不要只把文本字符串推给前端，否则两个来源交错时无法还原卡片归属。

这段代码能帮助我们确认上下文隔离是否真的发生：coordinator 负责下达任务和汇总，researcher 在自己的上下文中完成研究。主 Agent 不需要接收每一次搜索结果，只需要接收子 Agent 最后的摘要。

但这段代码还不能直接放进实时页面。它先把 coordinator 的 iterator 消费完，再去消费 subagents。如果 researcher 在后台已经输出了很多内容，页面看到的顺序就会被重新排列：主 Agent 的话全部出现在前面，子 Agent 的话全部出现在后面。

第一次接入后，页面可能会变成这样：

[coordinator] 正在委派研究任务
[coordinator] 这是最终总结……
[researcher] 正在比较不同的 Streaming 接口
[researcher] 已找到相关资料……

researcher 明明先完成研究，却被排在最终总结之后。第 6 节会修复这个顺序问题；在那之前，还要把卡片里缺失的工具活动补上。

5. 第三个修复：工具调用也要能被看见

研究卡片里只有文字仍然不够。用户看到“正在研究”，却不知道它是在等网络、调用搜索，还是工具已经报错。工具调用也按 Agent 层级提供 projection：

stream = agent.stream_events(request, version="v3")

for call in stream.tool_calls:
    print("[coordinator tool]", call.tool_name, call.input)
    print("completed:", call.completed, "error:", call.error)

for subagent in stream.subagents:
    for call in subagent.tool_calls:
        print(f"[{subagent.name} tool]", call.tool_name, call.input)

        for delta in call.output_deltas:
            print(delta, end="", flush=True)

        if call.completed and call.error is None:
            print("\nresult:", call.output)
        elif call.error is not None:
            print("\nerror:", call.error)

5.1 拆解 tool_call

一个 tool-call handle 同时承载调用身份、进行中的增量和终态：

字段或 projection含义使用规则
tool_name模型选择的工具名用于显示和分类；不要把未知工具名直接当成可信操作
input工具输入参数可能包含敏感数据，写日志和推送前先做脱敏
output_deltas工具执行过程中产生的增量输出按到达顺序追加，不要把单个 delta 当成完整结果
completed调用是否进入完成状态True 只表示调用结束，仍要检查 error
output成功调用的最终输出在 completed 且 error is None 后读取
error调用错误非空时进入失败分支，不要继续把 output 当成功结果

因此，工具状态不能只写成 if call.completed: success。可靠判断至少分成下面三支：

if not call.completed:
    status = "running"
elif call.error is not None:
    status = "failed"
else:
    status = "completed"

output_deltas 适合终端或实时日志区，output 适合完成后的结果摘要。两者同时保存时要避免重复：不要先把所有 delta 拼成全文，又把相同的最终 output 再追加一次。

这里有三个容易混淆的时刻：参数或工具输出还在增量到达；调用已经完成；调用完成但带有错误。页面可以把它们映射成“运行中”“已完成”和“失败”，不要把错误调用渲染成一行空白结果。

到这一步，用户看到的不再是一张只会闪烁的卡片：

researcher · running
  search · running   query="Deep Agents event streaming"
  search · completed 5 results
  正在整理 v3 projection 与 v2 protocol 的差异……

如果研究 Agent 还会委派下一层 Agent，投影可以继续向下递归：

stream = agent.stream_events(request, version="v3")

for subagent in stream.subagents:
    print(f"subagent {subagent.name}: {subagent.status}")

    for tool_call in subagent.tool_calls:
        print(f"{tool_call.tool_name}({tool_call.input})")
        for delta in tool_call.output_deltas:
            print(delta, end="", flush=True)

    for nested in subagent.subagents:
        print(f"nested subagent {nested.name}: {nested.status}")

递归时要用 path 作为当前运行内的 UI 唯一键。同名 researcher 可能来自不同委派分支，单独用 name 会把两张卡片的状态写到一起。

6. 顺序乱了：两种方式修复实时消费

6.1 异步服务：并发消费

回到刚才的页面 bug。coordinator 和 researcher 的事件会交错到达，实时 UI 不能把两个 iterator 排队处理。异步服务应同时消费它们：

import asyncio


async def stream_live():
    stream = await agent.astream_events(request, version="v3")

    async def consume_coordinator():
        async for message in stream.messages:
            print("[coordinator]", await message.text)

    async def consume_subagents():
        async for subagent in stream.subagents:
            async for message in subagent.messages:
                print(f"[{subagent.name}]", await message.text)

    await asyncio.gather(consume_coordinator(), consume_subagents())


asyncio.run(stream_live())

asyncio.gather 解决的是阻塞问题：一个投影等待网络时，另一个仍然可以把事件送到页面。它本身不会返回一个已合并、带全局序号的 iterator；如果要审计“哪个 token 先到”，还需要读取 raw protocol event 的 seq。

6.2 同步程序：使用 interleave

如果当前是同步命令行程序，不必为了展示进度重写成异步。v3 提供了 interleave：

stream = agent.stream_events(request, version="v3")

for name, item in stream.interleave("messages", "subagents"):
    if name == "messages":
        print("[coordinator]", item.text)
    else:
        for message in item.messages:
            print(f"[{item.name}]", message.text)

interleave 会把显式选择的 projections 按严格到达顺序合并，每次返回 (name, item)。name 是你传入的 projection 名，决定 item 的类型：

nameitem下一步
messagescoordinator 的 message handle读取 item.text
subagents一次委派的 subagent handle读取 item.name/path/status，再按需消费子 projection

它只合并你显式传入的 projection。上例没有传 tool_calls，所以 coordinator 的工具调用不会凭空出现在循环中；子 Agent 工具也仍需从 item.tool_calls 读取。

它适合快速做一个同步展示。如果要递归合并工具调用和嵌套子 Agent，仍然建议在应用层写一个事件 adapter，而不是让每个组件都理解 iterator 的细节。

coordinator 与 researcher 的事件会交错到达：串行消费把两个来源分组后造成顺序失真，并发消费使用 asyncio.gather 或 interleave 将事件按到达过程送入页面事件流

7. 页面开始工作后，才需要精确顺序

大多数产品只需要“主对话”“researcher 卡片”和“工具行”三个区域。它们有了自己的来源和路径，页面就能正确更新。只有在调试丢事件、重放运行或做审计时，才值得保留所有层级的精确到达顺序。

这时可以读取 v3 的 raw protocol events：

stream = agent.stream_events(request, version="v3")

for event in stream:
    if not isinstance(event, dict):
        continue
    if event.get("method") != "messages":
        continue

    params = event.get("params") or {}
    data = params.get("data")
    if not isinstance(data, (list, tuple)) or not data:
        continue

    payload = data[0]
    if not isinstance(payload, dict):
        continue
    if payload.get("event") != "content-block-delta":
        continue

    block = payload.get("delta") or {}
    if block.get("type") != "text-delta":
        continue

    namespace = params["namespace"]
    source = "subagent" if namespace else "coordinator"
    print(f"#{event['seq']} [{source}] {block['text']}", end="", flush=True)

raw event 的字段属于协议层。上例中的 source 只是根据 namespace 计算出的本地显示标签，并不是 raw event 自带字段。建议集中写一个 adapter，负责校验版本、读取 seq 与 namespace，再转换成应用自己的事件格式；页面只消费转换后的对象。这样协议升级时只改 adapter 和测试，不必逐个修改组件。

7.1 raw event 的字段

上面的过滤器只取了文本增量，实际 adapter 至少应该理解这些字段：

路径官方形状含义与处理建议
event["seq"]int同一次运行内严格递增；排序和查漏使用它，而不是时间戳
event["method"]str协议方法，例如 messages；先按方法分派，再读取 payload
event["params"]ProtocolEventParams事件参数包；先校验对象形状
params["namespace"]list[str]产生事件的图层级路径；空列表表示根层，非空列表用于子图路由
params["timestamp"]int事件时间；可用于展示，但时钟可能漂移，不用于严格排序
params["data"]Any数据形状由 method 决定；只有确认是 messages 后，才能按 content-block 结构解析
data[0]["event"]str在本节 messages 示例中，content-block-delta 表示增量块
data[0]["delta"]dict在本节示例中继续检查 delta["type"] == "text-delta" 后读取文本

namespace 不是子 Agent 的显示名。v3 的 subagent.name 来自 subagent_type，raw event 的 namespace 是当前事件的完整执行路径，每一段采用 <name>:<runtime_id> 形式。它与 subagent.path 使用同一套 namespace 语义：根级事件归一化后通常相等，子 Agent 内部事件则以 subagent.path 为前缀。前端 adapter 可以同时保存 name、path 和 raw namespace，分别用于显示、卡片路由和协议审计。

8. 旧代码为什么还在处理 type/ns/data

团队接手一个已有 LangGraph 服务时，常会看到这样的循环：

for chunk in agent.stream(
    request,
    stream_mode=["updates", "messages", "custom"],
    subgraphs=True,
    version="v2",
):
    print(chunk["type"])
    print(chunk["ns"])
    print(chunk["data"])

这段循环直接消费底层图执行协议，不使用 v3 的 typed projections。v2 的每个 StreamPart 都有 type、ns、data：

字段形状解读
type字符串当前 chunk 属于 updates、messages 或 custom 哪一种模式
nstuple[str, ...]() 表示主 Agent；非空 tuple 是产生当前 chunk 的完整 namespace
data随 type 变化updates 常是节点状态字典，messages 常是消息与 metadata，custom 是工具写入的自定义 payload

v3 与 v2 Streaming 的观察层级对比：v3 Typed Projection 面向产品角色提供 message、subagent 和 tool_call 字段；v2 StreamPart 面向图执行提供 type、ns、data；两者通过应用事件 Adapter 转换为统一页面事件

()                              -> main agent
("tools:abc123",)              -> task 工具启动的子 Agent
("tools:abc123", "model_request:def456") -> 子 Agent 内部模型节点

每一段都由节点名和本次执行 ID 组成。下面的写法可以按官方示例识别 tools: 段，并把子 Agent 更新路由到对应卡片：

for chunk in agent.stream(
    request,
    stream_mode="updates",
    subgraphs=True,
    version="v2",
):
    if chunk["type"] != "updates":
        continue

    ns = chunk["ns"]
    task_segment = next(
        (segment for segment in ns if segment.startswith("tools:")),
        None,
    )

    if task_segment is None:
        print("Main agent:", chunk["data"])
    else:
        task_id = task_segment.split(":", 1)[1]
        print(f"Subagent {task_id}:", chunk["data"])

这个 task_id 来自 namespace 段，适合作为本次执行的路由标识。若存在嵌套子 Agent，路径里可能出现多个 tools: 段；这时应保留完整 ns 或按已知 subagent.path 做前缀匹配，不能只取第一个 ID 当作全局唯一键。

stream_mode 决定 data 的形状：updates 适合看节点状态变化，messages 适合 token 和工具消息，custom 适合应用自定义进度。subgraphs=True 才会让子图事件出现在同一条流里：

typechunk["data"] 常见形状需要继续解读的字段
updates{node_name: state_update}node_name 是图节点名；state_update 的 schema 由节点决定
messages(token, metadata)token.content 是文本；工具调用还要看 token.tool_call_chunks，来源要结合 chunk["ns"]
custom工具通过 writer 写入的任意对象只按应用自定义 schema 解析，框架不会替你验证

例如 v2 的 messages 分支，不能只打印整个 tuple：

from langchain.messages import AIMessageChunk, ToolMessage


for chunk in agent.stream(
    request,
    stream_mode="messages",
    subgraphs=True,
    version="v2",
):
    if chunk["type"] != "messages":
        continue

    token, metadata = chunk["data"]
    source = "subagent" if chunk["ns"] else "main"

    if isinstance(token, AIMessageChunk) and token.tool_call_chunks:
        for tool_chunk in token.tool_call_chunks:
            print(source, "tool:", tool_chunk.get("name"), tool_chunk.get("args"))
    elif isinstance(token, ToolMessage):
        print(source, "tool result:", token.name, token.content)
    elif token.content:
        print(source, token.content, end="", flush=True)

这里的 metadata 也不要丢掉。它通常携带模型调用、节点或运行标签，调试时可以帮助定位来源；但产品 UI 仍应优先使用 namespace 和 adapter 生成的稳定 path，不要把某个 metadata 私有键当成长期 API。

for chunk in agent.stream(
    request,
    stream_mode="updates",
    subgraphs=True,
    version="v2",
):
    if chunk["type"] != "updates":
        continue

    source = "subagent" if chunk["ns"] else "main"
    print(f"[{source}]", chunk["data"])

不要把 data 当成固定 schema。先判断 chunk["type"]，再按对应模式解析。尤其是 messages 的 payload 不能直接当成 v3 的 message.text；v3 已经替你做了投影，v2 仍要求应用处理模式和来源。

这套格式适合两类场景：已有应用迁移时不想一次重写事件路由；调试时必须知道某条更新来自哪一个图节点。新页面仍建议用 v3 的 subagents，因为它直接表达“委派给哪个产品角色”，不用让 UI 猜 namespace。

9. 需要自定义进度时，先定义自己的事件

研究工具可能还想报告“已找到 3 个来源”“正在合并摘要”。这类信息不是内部 node 状态，应该由工具显式发出：

from langchain.tools import tool
from langgraph.config import get_stream_writer


@tool
def analyze_data(topic: str) -> str:
    """Analyze a topic and report structured progress."""
    writer = get_stream_writer()
    writer({"status": "starting", "topic": topic, "progress": 0})
    # 执行实际分析
    writer({"status": "complete", "topic": topic, "progress": 100})
    return f"Analysis complete: {topic}"

在 v2 中，这些信号从 custom 分支的 chunk["data"] 读取；如果应用已经采用 v3，可以按自己的业务 schema 扩展 adapter。当前 deepagents/streaming 模板没有定义 kind="progress"，不要把示例里的 progress payload 误认为模板或框架的固定事件。也不要让前端组件一半读取 v3 对象、一半判断 v2 的字符串字段。

本例的 custom payload 是应用自己定义的，不是框架固定字段：

字段本例含义约束建议
status当前业务阶段使用有限枚举，例如 starting/analyzing/complete
topic本次分析对象只在必要事件中发送，避免每个增量重复大字段
progress进度百分比约定 0-100，且只表示业务估算，不冒充真实完成比例

既然 schema 由应用掌握，就应该为它加版本和验证。生产事件可增加 schema_version，消费者遇到未知版本时记录并降级，而不是猜测字段含义。

10. 从“能看到”到“能交付”

到这里，页面终于能把一次研究请求说清楚：

coordinator message  -> 主对话
subagent started     -> researcher 卡片进入 running
subagent message     -> 写入 researcher 卡片
tool call            -> 卡片里的工具行，显示参数和结果
subagent completed   -> 卡片收起，保留摘要和状态
final output         -> 主对话里的最终答案
error/interrupted    -> 对应层级的错误或中断提示

现在不需要再凭空设计一套内部 schema。专用的 deepagents/streaming 应用模板 已经在 event_adapter.py 中把 typed projections 转换成前端消费的 SSE 事件：

模板 helperkind主要字段框架数据来源
message_eventmessagesource、path、text、finalcoordinator 或 subagent 的 messages projection
subagent_eventsubagentphase、name、path、statussubagent.name/path/status
tool_eventtool_callphase、source、path、tool_name、input、delta、output、errortool_calls 及 output_deltas
values_eventvaluessnapshotvalues projection
output_eventoutputphase、output、error顶层 output projection 或读取异常
raw_eventrawsequence、method、namespace、dataraw protocol 的 seq/method/params

raw_event 将官方 seq 重命名为 sequence，并保留 method、params.namespace 与 params.data；当前模板没有把 params.timestamp 转发给浏览器。

这些字段分属三层，不能混为“框架返回字段”：

字段精确来源用途
kind模板 event_adapter.py 中每个 helper 写入的字面量让前端分派应用事件；不是 Deep Agents 或 LangGraph 字段
source模板 routes.py 在消费 coordinator 或 subagent projection 时显式传入决定写入主对话还是子卡片；不是 raw protocol 字段
pathcoordinator 使用 []；子 Agent 的官方 tuple subagent.path 由模板转成 JSON list路由 message、tool 和 subagent 卡片
phase模板路由根据消费步骤写入 started/delta/completed/failed/in_progress表达模板事件处理阶段；不要与官方 subagent.status 混用
status官方 subagent.status，由模板原样转为字符串表达 started/completed/failed/interrupted 等委派生命周期
delta官方 tool-call handle 的 output_deltas只出现在模板的 tool_call 事件中，不是所有事件的通用 payload
sequenceraw protocol 顶层 seq，由模板重命名保留同一次运行内的严格事件顺序
namespaceraw protocol 的 params.namespace当前 raw 事件的完整路径；与 subagent.path 同源，但可能包含更深层路径段

这里没有 run_id。三份官方 Event Streaming 文档没有把它定义成 projection 字段或 ProtocolEvent 字段，专用模板也没有发出它。模板请求只有可选的 thread_id：浏览器用它保持后续请求的会话连续性，后端把它传给 checkpointer；它不是单条流事件的 ID，也不能改名成 run_id 后声称来自框架。

页面不需要展示 model_request 这种内部节点名，但 adapter 要理解路径前缀关系。模板事件里的 path 标识子 Agent 卡片根，raw 事件里的 namespace 标识当前事件位置；二者使用同一套 namespace 语义，却因作用域和 JSON 形状不同而分别保留。name 只负责用户看到的角色标签。

四个生产问题

页面上线前还要补四个运行边界：

  1. Streaming 不等于并行。是否并行取决于编排方式，不能从“事件交错”反推执行模型。
  2. 客户端断开后怎么办？长任务需要明确选择超时、取消，还是转入后台继续执行。
  3. 慢客户端怎么办？服务端要限制队列大小，或丢弃能够重建的增量，不能无限积压。
  4. 刷新页面后还能不能看见刚才的过程？如果需要重放，就把标准化事件写入 Trace、数据库或对象存储；内存 iterator 不是持久化日志。

工具错误、子 Agent 失败和中断状态都要原样保留。只有把失败也做成事件，用户才知道“研究没有完成”和“页面没有刷新”是两回事。

11. 一次最小实验：运行专用 Streaming 应用

agentseek-templates PR #20 已经提供完整的 deepagents/streaming 模板，不需要再从 deepagents/research 手工改造：

agentseek create deepagents/streaming --checkout main --no-input
cd deepagents_streaming
agentseek task --list

按生成项目 README 配置 .env，再执行 task --list 展示的依赖安装任务。完成后运行：

agentseek doctor
agentseek dev

这个模板直接包含 coordinator、固定委派的 researcher、本地工具、Event Streaming v3 后端、自定义 SSE route 和时间线前端。页面会同时展示 coordinator/subagent messages、subagent lifecycle、tool input/deltas/output、state snapshots、final output 和 raw protocol events。

实验时输入一个普通问题，然后对照两个文件核验来源：

  1. 在 routes.py 中找到 run.messages、run.subagents、run.tool_calls、run.values、run.output 和 raw run 的并发消费。
  2. 在 event_adapter.py 中找到 message_event、subagent_event、tool_event、values_event、output_event 和 raw_event。
  3. 比较 raw 事件的 seq/method/params.namespace/params.timestamp/params.data 与浏览器事件的 sequence/kind/source/path/phase，确认哪些来自协议、哪些由应用适配器生成。
  4. 连续提问两次，确认浏览器复用 thread_id；重启开发后端后，当前模板的内存会话会重新开始。

验收不看某一段固定文本，因为模型输出会变化。只检查五件事：请求确实委派给了 researcher，子 Agent 有独立状态，工具调用没有被吞掉，raw 事件保留 seq，最终答案仍由 coordinator 汇总。

本章小结

这次改造从一个具体故障开始：最终答案能返回，但用户看不见中间过程。解决它的顺序也很重要：

官方参考

课件下载

Lec 18: Streaming — 实时观察智能体和工具

相关资源

Lec 18: Streaming — 实时观察智能体和工具 B 站视频讲解 小红书图文