1. 本章目标 #
前面八章讲的都是「怎么让智能体干对事」。这一章讲怎么让人看见它在干什么。
这件事的重要性容易被低估。一个 deep agent 跑一次可能要几分钟、调十几次工具、派三个子智能体。如果界面上只有一个转圈的加载动画，用户的体验是「卡住了」而不是「在工作」——哪怕它干得完全正确。
Deep Agents 在 LangGraph 的流式协议上加了一层子智能体投影，让「谁在做什么」这件事变得可查询。这一章把两套 API 讲清楚，并给出选择标准。
学完你应能：
- 知道 Deep Agents 的流式 API 就是 LangGraph 的，没有独立封装（§2）
- 熟练使用低层
stream()的各个stream_mode，并知道为什么必须加subgraphs=True（§3） - 用
stream_events(version="v3")的投影对象按类型消费事件（§4） - 用
stream.subagents给每个子任务一个独立句柄，做出「子任务卡片」式的界面（§5） - 理解
tool_calls投影的懒求值语义（§6） - 用
interleave保证主与子的输出顺序正确（§7） - 在流里处理中断（§8）
- 知道官方前端组件覆盖了哪些场景（§10）
前置依赖： 第 26 章（LangGraph 流式基础）、第 47 章（子智能体——本章一半内容围绕它）、第 48 章（中断，§8 会用到）、第 46 章 §7.4（摘要 token 过滤）。
参考文档：
建议阅读顺序： 如果你只想快速做个能用的界面，看 §3.2 和 §5 就够了。§2.2 那个 API 限制建议先看，能省掉一次困惑。
本章验证环境：deepagents 0.7.12、langgraph 1.x。全部实验零 token——事件流的结构用脚本化假模型看得最清楚。
2. Deep Agents 没有自己的流式 API #
2.1. 返回的就是标准 LangGraph 图 #
先破除一个可能的误解：
"""确认 create_deep_agent 返回什么。"""
agent = create_deep_agent(model=..., backend=StateBackend())
print(type(agent).__name__)
print([n for n in dir(deepagents) if "stream" in n.lower()])CompiledStateGraph
[]deepagents 顶层没有任何 stream 相关的导出，也没有 deepagents.stream 模块。返回的是标准的 CompiledStateGraph。
所以你在第 26 章学的 LangGraph 流式知识全部直接可用。Deep Agents 加的东西只有一个：stream.subagents 这个投影（§5）。
2.2. 同步 stream_events() 只支持 v3 #
这个限制会让人卡一下：
agent.stream_events(inp, cfg, version="v2")NotImplementedError: stream_events(version='v2') is not supported.
Use astream_events() for v1/v2, or stream_events(version='v3') on a supported subclass.规则是：
| 方法 | 支持的 version |
|---|---|
stream_events()（同步） |
只有 v3 |
astream_events()（异步） |
v1 / v2 / v3 |
而 version 的默认值是 v2，所以同步调用必须显式写 version="v3"，否则直接报错。
v3 还是实验性的：
LangChainBetaWarning: The v3 streaming protocol on Pregel is experimental.2.3. 两套 API 怎么选 #
stream() / astream() |
stream_events(version="v3") |
|
|---|---|---|
| 返回 | 原始 chunk 迭代器 | GraphRunStream 投影对象 |
| 抽象层次 | 低（节点、状态更新） | 高（消息、工具调用、子智能体） |
| 子智能体 | 靠 subgraphs=True + 命名空间自己区分 |
stream.subagents 直接给句柄 |
| 稳定性 | 稳定 | beta |
| 适合 | 调试、日志、简单界面 | 面向用户的执行过程面板 |
我的建议：
做日志和调试用
stream(stream_mode="updates")；做用户界面用stream_events(version="v3")。
后者贵在 subagents 投影——自己从原始事件里把子智能体的活动拼出来相当麻烦（§3.3 会看到）。
3. 低层 API：stream() 与 stream_mode #
stream_mode 有七个值：
StreamMode = Literal['values', 'updates', 'checkpoints', 'tasks', 'debug',
'messages', 'custom']常用的是三个。
3.1. updates：谁更新了什么 #
for chunk in agent.stream(inp, cfg, stream_mode="updates"):
for node, upd in chunk.items():
keys = list(upd.keys()) if isinstance(upd, dict) else type(upd).__name__
print(f"节点 {node:34s} 更新 {keys}")节点 PatchToolCallsMiddleware.before_agent 更新 NoneType
节点 model 更新 ['messages']
节点 tools 更新 ['files', 'messages']
节点 model 更新 ['messages']这是调试首选——能直接看到执行路径和每步改了什么。注意 tools 节点同时更新了 files 和 messages，因为文件工具改的是状态里的文件字典。
3.2. messages：token 级输出 #
for msg, meta in agent.stream(inp, cfg, stream_mode="messages"):
ns = meta.get("langgraph_checkpoint_ns", "") or ""
src = "子智能体" if "tools:" in str(ns) else "主智能体"
print(f"[{src}] node={meta.get('langgraph_node')} "
f"{type(msg).__name__} {str(msg.content)[:34]!r}")[主智能体] node=model AIMessage ''
[子智能体] node=tools ToolMessage '子智能体报告：已完成。'
[主智能体] node=model AIMessage '主智能体：全部完成。'这是做打字机效果的 API。每次 yield 一个 (消息块, 元数据) 元组，接真模型时消息块是 token 级的增量。
元数据里有不少有用的字段：
{
"langgraph_node": "model",
"langgraph_step": "2",
"langgraph_checkpoint_ns": "model:8bcf8157-a2f8-3e73-cf0f-aff2e978d7",
"langgraph_path": "('__pregel_pull', 'model')",
"langgraph_triggers": "('branch:to:model',)",
"thread_id": "s6",
"ls_provider": "scen",
"ls_model_type": "chat",
"lc_agent_name": "None"
}第 46 章 §7.4 提过的摘要过滤就是靠这里：摘要产生的 token 带 lc_source == "summarization"，前端应该丢掉它们，否则用户会看到智能体突然开始自言自语地总结。
3.3. subgraphs=True：不加就看不见子智能体内部 #
这是低层 API 最容易漏的一点。 上面 messages 那个输出里，「子智能体」那条其实是 task 工具的返回值，子智能体内部的 write_file 调用完全没出现。
对比一下：
# 不带 subgraphs=True
for chunk in agent.stream(inp, cfg, stream_mode="updates"):
...共 4 个 chunk# 带 subgraphs=True，返回值变成 (命名空间, chunk) 元组
for ns, chunk in agent.stream(inp, cfg, stream_mode="updates", subgraphs=True):
who = "子" if ns else "主"
print(f"[{who}] ns={ns} 节点={list(chunk.keys())}")[主] ns=() 节点=['PatchToolCallsMiddleware.before_agent']
[主] ns=() 节点=['model']
[子] ns=('tools:577381ca-87e6-b25f-8dc2-30016522c6a7',) 节点=['PatchToolCallsMiddleware.before_agent']
[子] ns=('tools:577381ca-87e6-b25f-8dc2-30016522c6a7',) 节点=['model']
[子] ns=('tools:577381ca-87e6-b25f-8dc2-30016522c6a7',) 节点=['tools']
[子] ns=('tools:577381ca-87e6-b25f-8dc2-30016522c6a7',) 节点=['model']
[主] ns=() 节点=['tools']
[主] ns=() 节点=['model']
共 8 个 chunk4 个变 8 个，子智能体内部的四步全出来了。区分办法是：
ns为空元组就是主智能体，非空就是子智能体。
这也解释了为什么 stream.subagents 有价值——用低层 API 你得自己维护「哪个 ns 对应哪个子智能体、它现在什么状态」，而投影 API 直接给你句柄。
3.4. 多模式同时订阅 #
for mode, chunk in agent.stream(inp, cfg, stream_mode=["updates", "messages"]):
if mode == "messages":
msg, meta = chunk
print(f"[{mode:8s}] {type(msg).__name__} {str(msg.content)[:30]!r}")
else:
print(f"[{mode:8s}] 节点 {list(chunk.keys())}")[updates ] 节点 ['PatchToolCallsMiddleware.before_agent']
[messages] AIMessage ''
[updates ] 节点 ['model']
[messages] ToolMessage '子智能体报告：已完成。'
[updates ] 节点 ['tools']
[messages] AIMessage '主智能体：全部完成。'
[updates ] 节点 ['model']传列表时每个 chunk 变成 (模式名, 内容) 元组。这样一次订阅就能同时驱动打字机效果（messages）和进度指示（updates）。
4. 投影 API：GraphRunStream #
stream = agent.stream_events(inp, cfg, version="v3")
print(type(stream).__name__, type(stream).__module__)GraphRunStream langgraph.stream.run_stream它的公开成员：
| 成员 | 类型 | 说明 |
|---|---|---|
messages |
StreamChannel |
主智能体的消息 |
tool_calls |
StreamChannel |
主智能体的工具调用 |
values |
StreamChannel |
状态快照序列 |
subagents |
StreamChannel |
子智能体句柄（Deep Agents 加的） |
subgraphs |
StreamChannel |
子图（图结构层面） |
lifecycle |
StreamChannel |
生命周期事件 |
output |
dict |
最终状态 |
interrupted |
bool |
是否中断了 |
interrupts |
list |
中断详情 |
interleave(*names) |
方法 | 交错消费多个投影 |
abort() |
方法 | 中止运行 |
每个投影都是独立的通道，你只订阅用到的那些。文档说这个投影是「轻量的」——发现子智能体任务是先做的，消息和工具调用流只在你访问句柄上对应属性时才打开。
4.1. values 与 output #
for v in stream.values:
print(f"keys={list(v.keys())} 消息数={len(v.get('messages', []))}")第 1 次 values: keys=['messages', 'files'] 消息数=1
第 2 次 values: keys=['messages', 'files'] 消息数=2
...
第 6 次 values: keys=['messages', 'files'] 消息数=6values 是每步之后的完整状态快照（消息数递增），output 是最终那一份：
最终 output 的 key: ['messages', 'files']
output['files']: ['/main_note.md', '/sub_out.md']注意 output['files'] 里同时有主和子写的文件——这就是第 47 章 §4 那条「文件系统完全共享」的另一个证据。
4.2. lifecycle #
for ev in stream.lifecycle:
print(ev){'event': 'started', 'namespace': ['tools:646b5b83-...'], 'graph_name': 'worker',
'trigger_call_id': '646b5b83-...'}
{'event': 'completed', 'namespace': ['tools:646b5b83-...']}只有子智能体的启停事件。graph_name 是子智能体名字，trigger_call_id 指向触发它的那次 task 调用——做界面时可以用它把子任务卡片挂到对应的工具调用下面。
4.3. subagents 和 subgraphs 的区别 #
subagents:
name='worker' graph_name='worker' trigger_call_id='104dba8a-...'
subgraphs:
1. name='worker' path=('tools:2b761cf2-...',)这个例子里数量相同，但语义不同：
subgraphs是图执行结构，subagents是产品级的任务委派。
官方明确建议：面向用户的界面用 subagents，因为它隐藏了内部图节点，直接暴露「子智能体」这个概念。subgraphs 留给调试。
5. stream.subagents：每个子任务一个句柄 #
这是 Deep Agents 在 LangGraph 之上唯一的增量，也是本章最有价值的部分。
"""每个 task 调用得到一个独立句柄。"""
stream = agent.stream_events(inp, cfg, version="v3")
for sub in stream.subagents:
print(f"句柄: name={sub.name!r} path={sub.path!r} status={sub.status!r}")
for m in sub.messages:
print(f" [{sub.name}] {m.text!r}")
print(f"完成后 status={sub.status!r}")句柄: name='worker' path=('tools:33c7e42a-4190-3b89-3d51-892f3dddd48e',) status='started'
[worker] ''
[worker] '子智能体报告：已完成。'
完成后 status='completed'三个要点：
name就是subagent_type——主智能体调task时选的那个名字。你在SubAgent(name=...)里定义的标签，就是流里用来过滤和路由的标签status是动态的：刚拿到句柄时是'started'，消费完messages后变成'completed'- 句柄本身就是一个完整的流对象
5.1. 句柄的完整字段 #
name path status graph_name trigger_call_id
messages tool_calls values subagents output lifecycle subgraphs
error cause interrupted interrupts
abort() interleave() extensions比顶层多了四个：graph_name、trigger_call_id、error、cause。
status 的取值是 started / completed / failed / interrupted。
5.2. 只跟踪生命周期 #
如果界面只需要显示「哪些子任务开始了、结束了」，不用订阅消息流：
"""轻量的子任务进度跟踪。"""
stream = agent.stream_events(inp, version="v3")
running, completed, failed = 0, 0, 0
for sub in stream.subagents:
running += 1
print(f"{sub.name}: 开始")
try:
_ = sub.output # 访问 output 会等它跑完
running -= 1
completed += 1
print(f"{sub.name}: 完成")
except Exception:
running -= 1
failed += 1
print(f"{sub.name}: 失败")这就够驱动一个「3 个子任务，2 完成 1 进行中」的进度条了。
5.3. 主与子的消息分开渲染 #
"""主智能体的话进主聊天区，子智能体的话进折叠卡片。"""
stream = agent.stream_events(inp, version="v3")
for m in stream.messages:
print("[主]", m.text)
for sub in stream.subagents:
for m in sub.messages:
print(f"[{sub.name}]", m.text)顶层 stream.messages 只有主智能体的消息，不含子智能体的——这跟低层 API 不加 subgraphs=True 的效果一致，但这里是设计好的分离而不是需要绕过的限制。
界面上通常这样处理：主智能体的输出直接显示，子智能体的输出收进一个可展开的卡片。因为子智能体的中间过程对用户来说是「实现细节」，想看的时候能展开就行。
6. tool_calls 投影的懒求值 #
工具调用投影有个容易误判的语义：
for c in stream.tool_calls:
print(f"{c.tool_name}")
print(f" 刚拿到: completed={c.completed}")
deltas = list(c.output_deltas)
print(f" 读 output 后: completed={c.completed} error={c.error}")
print(f" output={str(c.output)[:80]!r}")write_file
刚拿到: completed=False
output_deltas: []
读 output 后: completed=True error=None
output="content='Updated file /main_note.md' name='write_file' tool_call_id='m1'"
tool_call_id='m1'
task
刚拿到: completed=False
读 output 后: completed=True error=None
output="Command(update={'files': {'/main_note.md': {'content': '主的笔记', ...刚迭代到时 completed 是 False。 这不是 bug——迭代器在工具开始执行时就把句柄交给你了，所以你能立刻在界面上显示「正在调用 write_file...」。completed 变 True 要等你访问 output 或消费完 output_deltas。
所以正确的用法是：
"""工具调用的正确渲染顺序。"""
for c in stream.tool_calls:
render_pending(c.tool_name, c.input) # 立刻显示「正在执行」
for delta in c.output_deltas: # 有流式输出就逐块渲染
render_delta(delta)
if c.error is not None: # 访问 output 后状态才确定
render_error(c.error)
else:
render_done(c.output)别在拿到句柄时就判断 c.completed——那时候它必然是 False。
字段清单：tool_name、input、output、output_deltas、completed、error、tool_call_id。
顺便看一眼 task 的 output：它是个 Command(update={'files': ...})，而不是普通的 ToolMessage。子智能体的文件改动是以状态更新的形式回传的——这是第 47 章「文件系统共享」在流层面的实现细节。
7. interleave：保证顺序正确 #
stream.messages 和 stream.subagents 是两个独立通道。分别遍历会得到「先主全部、再子全部」的顺序，跟实际发生的顺序不符。
同步代码用 interleave：
"""按实际发生顺序交错消费多个投影。"""
stream = agent.stream_events(inp, cfg, version="v3")
for name, item in stream.interleave("messages", "subagents", "tool_calls"):
if name == "messages":
print(f"[messages] {item.text!r}")
elif name == "tool_calls":
print(f"[tool_calls] {item.tool_name}")
else:
print(f"[subagents] {item.name} status={item.status}")
for m in item.messages:
print(f" └ {m.text!r}")[messages] ''
[tool_calls] write_file
[messages] ''
[tool_calls] task
[subagents] worker status=started
└ ''
└ '子智能体报告：已完成。'
[messages] '主智能体：全部完成。'这个顺序就是界面上该显示的顺序：主智能体说话 → 调 write_file → 派子智能体 → 子智能体干活 → 主智能体收尾。
异步代码用 asyncio.gather 并发消费：
"""异步并发消费主与子的消息流。"""
import asyncio
stream = await agent.astream_events(inp, version="v3")
async def consume_coordinator():
async for message in stream.messages:
print("[主]", 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())注意异步版里 message.text 是需要 await 的。
7.1. 需要精确到达顺序时 #
interleave 的粒度是「投影项」。要精确到每个 token 的到达顺序，就得读原始协议事件：
"""按 namespace 区分来源，逐 token 处理。"""
stream = agent.stream_events(inp, version="v3")
for event in stream:
if event.get("method") != "messages":
continue
payload = event["params"]["data"][0]
if not isinstance(payload, dict) or payload.get("event") != "content-block-delta":
continue
block = payload.get("delta") or {}
if block.get("type") == "text-delta":
source = "subagent" if event["params"]["namespace"] else "coordinator"
print(f"[{source}] {block['text']}")判断来源的办法和低层 API 一样：namespace 非空就是子智能体。
8. 流里的中断 #
第 48 章的审批在流式场景下也能工作：
stream = agent.stream_events(inp, cfg, version="v3")
for m in stream.messages:
print(f"[message] {m.text!r}")
print(f"interrupted={stream.interrupted}")
for i in stream.interrupts:
print(i.value)[message] ''
interrupted=True
{"action_requests": [{"name": "write_file",
"args": {"file_path": "/main_note.md", "content": "主的笔记"},
"description": "Tool execution requires approval\n\n..."}],
...}流会在中断点停下（messages 只吐了一条），interrupted 变 True，interrupts 里是第 48 章 §3.2 那个结构。
前端的处理流程是：
- 消费流，发现
stream.interrupted为真 - 从
stream.interrupts取出action_requests和review_configs，渲染审批界面 - 用户点了之后调
Command(resume={"decisions": [...]})，开一个新的流继续
关键是第 3 步——恢复是一次新的 stream_events 调用，不是在原来的流上继续。
9. 一个完整的执行面板 #
把前面的东西拼起来：
"""能看清「谁在做什么」的执行过程面板（控制台版）。"""
import warnings
warnings.filterwarnings("ignore") # v3 是 beta，会打警告
def render(agent, inp, cfg):
stream = agent.stream_events(inp, cfg, version="v3")
for name, item in stream.interleave("messages", "tool_calls", "subagents"):
if name == "messages":
if item.text: # 过滤纯工具调用的空消息
print(f"\n💬 {item.text}")
elif name == "tool_calls":
print(f"\n🔧 {item.tool_name}({_brief(item.input)})", end="", flush=True)
for delta in item.output_deltas:
print(delta, end="", flush=True)
if item.error is not None:
print(f" ✗ {item.error}")
else:
print(f" ✓ {_brief(item.output)}")
else: # subagents
print(f"\n┌─ 子任务 [{item.name}] 开始")
for sname, sitem in item.interleave("messages", "tool_calls"):
if sname == "messages":
if sitem.text:
print(f"│ 💬 {sitem.text}")
else:
print(f"│ 🔧 {sitem.tool_name}({_brief(sitem.input)})")
_ = sitem.output # 等它完成
print(f"└─ 子任务 [{item.name}] {item.status}")
if stream.interrupted:
print("\n⏸ 等待审批:")
for i in stream.interrupts:
for req in i.value["action_requests"]:
print(f" {req['name']}: {req['args']}")
return None
print(f"\n✅ 完成。产出文件: {list(stream.output.get('files', {}) or {})}")
return stream.output
def _brief(x, n=60):
s = str(x)
return s if len(s) <= n else s[:n] + "…"这个结构直接对应界面上的三种元素：主聊天气泡、工具调用行、可折叠的子任务卡片。
10. 官方前端组件 #
官方提供了一套 React 组件，覆盖三个场景：
| 组件 | 文档 | 解决什么 |
|---|---|---|
| 任务清单 | todo-list | 渲染 write_todos 的三状态清单（第 47 章 §2） |
| 子智能体流式 | subagent-streaming | 把主消息和子任务卡片分开的 UI 模式 |
| 沙箱视图 | sandbox | 展示沙箱里的文件系统和命令执行（第 43 章 §6） |
这些是 TypeScript / React 的，Python 后端通过 SSE 或 WebSocket 把事件推给它们。三个组件对应的正是本章的三类投影：values 里的 todos、subagents 句柄、output 里的 files。
如果你自己写前端，这三个组件的形态值得参考——它们把「智能体在长时间工作」这件事的可视化做了标准化。
11. 实战约定与坑 #
- Deep Agents 没有自己的流式 API，返回的是标准
CompiledStateGraph。第 26 章的知识全部可用（§2.1） - 同步
stream_events()只支持version="v3"，而默认值是v2——必须显式写，否则报NotImplementedError（§2.2） - v3 协议是 beta，会打
LangChainBetaWarning（§2.2） - 选择标准：日志调试用
stream(stream_mode="updates")，用户界面用stream_events(version="v3")（§2.3） - 低层 API 不加
subgraphs=True就看不到子智能体内部。实测 4 个 chunk 变 8 个（§3.3） ns为空是主智能体、非空是子智能体，低层 API 靠这个区分来源（§3.3）stream_mode传列表时，chunk 变成(模式名, 内容)元组（§3.4）- 过滤摘要 token：
metadata里lc_source == "summarization"的要丢掉，否则用户会看到智能体自言自语（§3.2，第 46 章 §7.4） stream.subagents的name就是subagent_type——SubAgent(name=...)里定义的标签就是流里过滤用的标签（§5）status是动态的：拿到句柄时started，消费完变completed。取值还有failed/interrupted（§5）tool_calls的completed刚拿到必然是False。 别在这时候判断，先渲染「正在执行」，访问output后状态才确定（§6）- 顶层
stream.messages不含子智能体消息，这是设计好的分离（§5.3） - 分别遍历多个投影会丢掉顺序，同步用
interleave、异步用asyncio.gather（§7） - 异步版的
message.text需要 await（§7） - 要精确的 token 到达顺序就读原始事件，靠
namespace判断来源（§7.1） - 中断后恢复是一次新的
stream_events调用，不是在原流上继续（§8） - 界面用
subagents不用subgraphs：前者是产品概念，后者是图结构（§4.3） task的output是Command(update={...})而不是ToolMessage，解析时注意（§6）- 子智能体的输出收进可折叠卡片——中间过程对用户是实现细节，能展开就够（§5.3）
- 只要生命周期就别订阅消息流，投影是按需打开的，不访问就不产生开销（§5.2）
12. 练习 #
对比两套 API。 同一个带子智能体的任务，分别用
stream(stream_mode="updates")和stream_events(version="v3")跑一遍，数一数各自要写多少代码才能区分主与子。踩 version 的坑。 调
stream_events(inp)不传version，确认报NotImplementedError。再用astream_events(inp, version="v2")，确认异步版可以。验证
subgraphs=True。 数一下带与不带的 chunk 数量差，并把每个 chunk 的ns打出来。验证懒求值。 在
for c in stream.tool_calls循环里，先打印c.completed，再访问c.output，再打印一次c.completed。确认它从False变True。顺序问题。 先分别遍历
stream.messages和stream.subagents，记下输出顺序；再用interleave跑一遍，对比差异。做执行面板。 把 §9 那个函数跑起来，接一个真模型和两个子智能体（比如「调研」和「写作」），看输出是否清晰。
流式审批。 给
write_file配interrupt_on，在流里检测stream.interrupted，渲染审批提示，然后用Command(resume=...)开新流恢复。（选做）SSE 服务。 用 FastAPI 把
stream_events的投影包成 SSE 端点，每种投影发一类事件（event: message/event: tool/event: subagent），前端用EventSource消费。
13. 本章小结 #
- Deep Agents 没有独立的流式 API，
create_deep_agent返回标准CompiledStateGraph。它加的唯一东西是stream.subagents投影。 - 同步
stream_events()只支持version="v3"，默认值v2会直接报错；v1/v2 只能用astream_events()。 - 两套 API 的分工：
stream()做日志调试，stream_events(version="v3")做用户界面。 - 低层 API 必须加
subgraphs=True才能看到子智能体内部（4 chunk → 8 chunk），靠ns是否为空区分来源。 GraphRunStream有六个投影通道（messages/tool_calls/values/subagents/subgraphs/lifecycle）加上output/interrupted/interrupts，按需订阅。stream.subagents给每个task调用一个独立句柄，name就是subagent_type，status会从started变到completed。tool_calls是懒求值的：刚拿到completed=False，访问output后才确定。正确做法是先渲染「正在执行」再取结果。- 顶层
messages不含子智能体消息，这是设计好的分离——界面上主消息直接显示、子消息收进折叠卡片。 - 分别遍历投影会丢顺序：同步用
interleave(...)，异步用asyncio.gather，要精确 token 顺序就读原始事件按namespace判断。 subagents是产品概念、subgraphs是图结构，面向用户的界面用前者。- 中断在流里表现为
interrupted=True+interrupts列表，恢复是一次新的流调用。 - 官方三个前端组件（任务清单、子智能体流式、沙箱视图）正好对应本章的三类投影。