流式LangChain深度智能体：实时计划、工具调用与令牌。
用流式处理替代阻塞式调用，以便用户查看深度智能体的工作过程：三种流式模式、子智能体的命名空间以及过滤与总结令牌。
来源: LangChain Deep Agents: Your AI Agent Is Not Frozen. It Just Cannot Tell You It Is Working. II
几乎没有什么比屏幕冻结更能迅速破坏信任了。当用户无法判断智能体是在思考、卡住了还是已经崩溃时，他们往往会往最坏的情况想。而流式处理则用实时的动态反馈取代了这种沉默：计划、工具调用以及相关文字会随着操作实时显示，让人们清楚知道有事情正在发生。
为何 invoke 会让深度智能体看起来像已崩溃
试着在有人观看的情况下让深度智能体执行实际任务。调用会通过 invoke 发送，之后界面就会静止四十秒。实际上，智能体正在读取文件、构思计划、运行工具并逐步推理。用户看到的只是一个静止的框和旋转图标，这可能意味着它正在努力处理，也可能已经无声崩溃。由于两者看起来毫无区别，人们就会重新加载页面并提交工单，称系统卡住了。
没有任何内容被挂载出来。该智能体根本没有用于自我描述的渠道。invoke函数按设计就是如此——它在后台执行整个循环，最终只返回一个完整的答案，这正适合那些无人监控的脚本或定时任务。但一旦有人类在另一端操作，这种做法就不合适了。
这种解决方案则不会干扰循环流程，它用流式处理替代了阻塞式调用，从而使计划、工具调用及令牌能够实时显示出来。
从invoke到stream
流式处理与invoke密切相关。你无需等待最终状态，只需循环调用agent.stream(...)，每当循环执行某项操作时就会有新数据块到达。还有一个异步版本agent.astream(...)，结构相同，正是Web服务器会使用的形式。
这种思维模型较为简单。深度智能体背后的持久化 LangGraph 会在每一步生成一个事件，而流式处理机制会在这些事件产生时将其传递下去。这并非另一个智能体或不同的循环，只不过是对你已熟悉的那个智能体的一种视角展示而已。
有一个细节必须首先明确，因为所有示例都依赖于它：即通过 version="v2" 启用的 v2 流格式。每个数据块都是一个包含三个键的小字典：
type表示事件的类型。ns即命名空间，用于标识事件的生成者。data则存储事件的实际内容。
无论传输什么类型的数据，这三个键都保持不变，这样就能让代码更易读，避免出现复杂的元组解包结构。
三种模式，三种层次
流并非单一的消防水龙带。stream_mode设置决定了您会接收何种类型的事件，而三种常见的选择对应着三种不同的详细程度。
updates为逐步视图。每当一个图节点处理完成时就会触发一个事件，实际上这意味着模型完成了一次迭代或某个工具运行了一次。这种详细程度适合用于进度记录，既能显示智能体正在执行的任务，又不会让信息过于冗长，是用于展示“智能体当前正在做什么”的简洁信息的合理默认选项。
messages 是逐条消息的视图。模型在生成内容时会逐条输出结果，就像聊天助手打字时的效果一样。当需要让人实时看到智能体的输出内容时，可使用此视图。流式调用也会在这里触发，因此你可以在模型决定执行调用时立即宣布“正在调用 read_file”，而非等到调用完成之后。
values 是快照视图。每执行一步后，你都会收到智能体的完整状态，而不仅仅是发生变化的部分。虽然这种方式的数据量更大且使用频率较低，但当需要在每个时刻都了解完整情况时，比如要实时显示智能体的待办列表或正在处理的文件，它才是最佳选择。
传入模式列表可一次性返回多个模式，这正是精致用户界面通常的运作方式：updates负责驱动进度指示器，而messages则用于显示文本内容。updates生成的简洁进度记录如下所示：
for chunk in agent.stream(
{”messages”: [{”role”: “user”, “content”: “Summarize the repo’s failing tests”}]},
stream_mode=”updates”,
version=”v2”,
):
if chunk[”type”] == “updates”:
for node_name in chunk[”data”]:
print(f”step: {node_name}”)
输出的内容是随着各个图步骤完成而依次显示的列表，每个模型轮次和工具运行都会对应一行。原本看似停滞的界面现在能够明显地向前推进。若将stream_mode设置为"messages"，则会在智能体写入内容的同时直接提供文本，便于立即打印。
区分主智能体与子智能体
深度智能体很少会长时间独自工作。一旦它将子任务分配给子智能体，就会同时有两个或更多智能体产生输出。如果将所有输出汇总到同一条流中，结果就会变得混乱不堪：有协调智能体的句子片段，有研究智能体的几个标记，还有不知从何处传来的工具调用。ns字段可以解决这个问题。当使用subgraphs=True开启子图流式处理时，每个数据块都会携带一个该字段值。
该规则很容易应用。如果命名空间为空，则表示是主代理发出了事件；如果其中包含tools:部分，则表示是某个子代理发出的事件，同时还会标明是哪个子代理以及对应的部分名称。之后可以按来源对输出结果进行排序：协调器的内容放在一处，每个子代理的工作结果分别列出。
for chunk in agent.stream(
{”messages”: [{”role”: “user”, “content”: “Research and summarize quantum computing”}]},
stream_mode=”messages”,
subgraphs=True, # ①
version=”v2”,
):
if chunk[”type”] == “messages”: # ②
token, metadata = chunk[”data”] # ③
is_subagent = any(s.startswith(”tools:”) for s in chunk[”ns”]) # ④
source = “subagent” if is_subagent else “main” # ⑤
if token.content:
print(f”[{source}] {token.content}”, end=”“, flush=True) # ⑥
在那个代码片段中有六个关键细节。首先，subgraphs=True才是赋予ns意义的关键，因为没有它的话，子代理事件就没有任何区分用的命名空间。其次，该分支仅保留messages事件，因为这个循环处理的是令牌而非步骤。第三，每个数据块的负载会拆解为一个令牌及其元数据。第四，命名空间检测才是真正起作用的部分：ns中的任何tools:片段都会将该令牌标记为来自子代理。第五，由此产生的布尔值会被转换成易于理解的源标签。第六，以该标签作为前缀进行输出，就能将混乱的流数据分成两条带有属性的通道。
在打印之前，每个标记都会标注其生成者，因此协调器转交给研究人员的内容会以两个带名称的流的形式呈现。构建完整的网页界面是一项更复杂的任务，但每个数据块的命名空间与React前端所读取的信号是相同的。
非代理发言产生的异常标记
初次进行长时间流式处理时，这一点常常会让人们感到意外。为了管理自身的上下文，当对话内容接近模型的处理极限时，该系统会压缩较旧的消息。这个压缩过程本身也属于模型调用，因此会生成标记，这些标记会与代理的实际回复一同流入messages流中。如果该流直接发送给用户，他们就会看到代理突然开始复述自己的历史记录，这看起来就像是一个漏洞。
幸运的是，这些标记都有对应的标签。每个messages事件都包含元数据，而摘要标记的lc_source值均为"summarization"。只需跳过这些标记，保留其余内容即可。
for chunk in agent.stream(
{”messages”: [{”role”: “user”, “content”: “...”}]},
stream_mode=”messages”,
version=”v2”,
):
token, metadata = chunk[”data”]
if metadata.get(”lc_source”) == “summarization”:
continue
# ... handle real agent tokens here
只需检查一下元数据，就能区分与用户对话的智能体和与自身对话的框架。只有前者会被显示出来。这虽是项简单的工作，却能显著提升整体质量。
监控运行过程与维持对话
流式处理与对话常常被混为一谈，因此需要明确区分。上述内容均涉及对某次运行过程的观察。每次stream调用以及每次invoke调用默认都是单次性的。智能体完成任务后会立即忘记，因此通过新的调用进行后续操作时，不会保留之前对话的任何记忆。
通过多轮对话保持对话的连续性，使得“也修复另一个测试”这一指令能够生效——因为智能体能够记住之前的修复操作——这是一种独立的机制。它依赖于线程和检查指针，以及用于存储持久化图状态的内存层。需要区分的是：流式处理能够提供单次运行过程中的实时可见性，而线程则能确保不同运行之间的连续性。二者相互独立，但通常需要同时使用。
一个展示处理过程的简单终端界面
接下来，将所有组件整合成一个可实际运行的控制台。该控制台应在智能体每前进一步时打印一行输出，并实时显示智能体生成的文本；因此需要同时设置两种模式，并根据type字段的值进行分支处理。
from deepagents import create_deep_agent
agent = create_deep_agent(
model="anthropic:claude-sonnet-4-6",
system_prompt="You are a code-maintenance assistant for a small Python repo.",
) # ①
prompt = "List the files you'd look at first to find a failing test, and explain why."
for chunk in agent.stream(
{"messages": [{"role": "user", "content": prompt}]},
stream_mode=["updates", "messages"], # ②
version="v2",
):
if chunk["type"] == "updates": # ③
for node_name in chunk["data"]:
print(f"\n[step] {node_name}")
elif chunk["type"] == "messages": # ④
token, metadata = chunk["data"]
if metadata.get("lc_source") == "summarization": # ⑤
continue
if token.content:
print(token.content, end="", flush=True) # ⑥
代码虽然简短，但每个部分都有其功能。由于create_deep_agent的构建方式与之前完全相同，因此结构并未改变。一次调用中请求两种模式意味着单个循环即可同时处理步骤指示器和文本流。当数据块为updates事件时，循环会为刚完成处理的节点打印一行[step]；而当其为messages事件时，则会打印生成的文字内容。前一节中的摘要检查功能就位于该分支内部，这样框架的常规处理流程就不会影响到终端输出，且在打印之前会跳过空标记。
其结果为：每次智能体向前移动时都会生成一条[step]记录，同时还会逐个字符输出其文本内容，且已有过滤机制确保不会有任何信息泄露。虽然外观不美观，但内容真实可靠。用户能够看到智能体正在运行，并观察它的思考过程。
较新版本的Deep Agents提供了更高级的API，即在version="v3"下使用的agent.stream_events(...)，以及诸如stream.subagents这样的类型化投影功能。文档会引导新应用使用这些功能，因为每个投影都是独立的迭代器，无需自行处理stream_mode或解码命名空间。不过，首先学习stream_mode方式仍然是值得的，因为这些投影功能正是建立在这些相同事件之上的。在决定使用复杂接口之前，请先查看当前文档。
今日即可实施的检查清单
- 首先找出所有有任务正在等待处理的
invoke，并将它们改为在updates模式下使用agent.stream(...)。有简单的步骤日志总比空白屏幕要好。 - 查看一个真实的数据块。使用
version="v2"运行它，打印输出并读取type、ns和data字段，以便明确其结构。 - 在真实用户看到输出之前，删除那些元数据中显示
lc_source == "summarization"的messages数据块。 - 如果代理将任务分配给子代理，需设置
subgraphs=True，并按命名空间对输出进行排序，每个代理对应一条输出行。 - 在代码仓库中保留类似上述的小型控制台。这样你可以实时观察每个新功能的运行情况，而无需根据上一条消息猜测。
哪些内容发生了变化，哪些没有
最终结果：原本的阻塞式调用变成了实时数据流。stream_mode决定了你能获取的详细程度，ns标识出每个事件的生成者，摘要过滤功能则去除了工具自身的冗余信息。现在应该很清楚，观察运行过程与记住运行结果其实是两回事。
代理及其循环逻辑没有任何变化。唯一的新增内容是一个界面窗口，而这个窗口正是区分人们信赖的工具与他们在恐慌中急于刷新的工具的界限。当代理展示其工作过程时，用户就会认为它确实在正常运行。
因此，下次再有人报告代理“卡住”时，先别用性能分析工具，试着使用实时数据流看看。其实代理一直都在“说话”，它只是需要一个传输渠道而已。
相关阅读
- 你的智能体文件系统即决策：Deep Agents后端详解 —— LangChain Deep Agents后端如何决定哪些内容会被保留：五种内置功能、CompositeBackend路由机制、多用户存储中的命名空间问题以及沙箱模式。
- 托管式Deep Agents测试版：基于目录的LangChain智能体部署方式 —— LangChain的托管式Deep Agents可将项目文件夹转换为托管智能体。各类文件的用途、mda部署的工作原理，以及值得了解的测试版缺陷。