第 14 章：Streaming — 实时观察主 Agent、子 Agent 与工具调用

研究助手已经跑了两分钟。后台日志显示它调用了搜索工具，也确实启动了 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() 只有在整个运行结束后才返回，所以页面无法回答这些最基本的问题：

• coordinator 有没有真的把任务交给 researcher？
• 子 Agent 是正在搜索，还是已经失败？
• 工具参数还在生成，还是工具已经返回错误？
• 最终答案出现前，究竟发生了哪些中间步骤？

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

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：

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

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

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

3.3 path、namespace 和 ns 是什么关系

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

路径中的每一段采用 <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：

这里最容易犯的错是范围混淆。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。本章示例只需要一个字段：

同步和异步写法的差别不是语法装饰。异步 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 同时承载调用身份、进行中的增量和终态：

因此，工具状态不能只写成 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 的类型：

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

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

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 至少应该理解这些字段：

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：

()                              -> 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 才会让子图事件出现在同一条流里：

例如 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 是应用自己定义的，不是框架固定字段：

既然 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 事件：

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

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

这里没有 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 汇总。

本章小结

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

• 用 stream.subagents 先显示产品层的委派状态；
• 按需打开 .messages、.tool_calls、.subagents 和 .output；
• 用异步并发消费或 interleave 保留实时更新；
• 只有需要审计时，才读取 raw events 和 namespace；
• 新应用默认使用 v3，v2 留给底层调试、custom updates 和迁移；
• kind/source/phase 属于模板应用协议，seq/method/params 才属于 raw protocol；
• Streaming 不负责持久化、取消、超时和背压，生产边界要由应用补齐。

官方参考

• Deep Agents Event Streaming
• Deep Agents Streaming：v2 Namespaces
• LangChain Event Streaming
• LangGraph Event Streaming
• LangGraph Streaming：v2 StreamPart 与 ns
• AgentSeek deepagents/streaming 模板
• AgentSeek Templates PR #20