企业官网建设流程全解析

Event Streaming 这一章我在官方文档上折返了三次才彻底吃透，第一次读只觉得是一堆on_xxx_start和on_xxx_end的事件名堆在一起，第二次动手写才发现事件量比想象中大得多，第三次把它接到前端做实时状态展示时，才算真正明白stream_events这套设计想解决的问题。简单说，它把 Agent 从"一个黑盒函数"变成了"一条可观测的类型化事件流"——模型什么时候开始思考、调用了哪个工具、工具入参是什么、哪一步在等外部接口、哪一段花的时间最长，全都能拿到结构化数据。这对做 AI 应用的人来说不是锦上添花，而是决定产品体验的下限。这篇就把我精读文档加实际落地过程中的理解完整拆开：事件结构怎么读、类型化封装为什么值得单独做一层、过滤参数怎么用才不踩坑、事件树怎么拼、SSE 怎么接、以及我在真实项目里踩过的几个坑。适合已经能跑通一个 Agent、但被"用户盯着空白屏幕等十秒"这个问题困扰的开发者。

1. stream_events 到底在解决什么问题

1.1 astream 的天花板在哪里

刚开始做流式输出的时候，大多数人用的都是astream。它确实好用，一句话就能让模型吐字变成打字机效果，用户感知到的等待时间从"十秒"压缩到"一秒出第一个字"。但只要你的应用里出现了 Agent、出现了工具调用，astream立刻就不够用了。

原因很直接：astream只负责把最终节点的 token 一个块一个块地推出来。一个典型的 Agent 执行链路是"用户提问 → 模型判断需要调工具 → 生成工具入参 → 执行工具（可能是查数据库、调外部接口、跑一段检索）→ 把工具结果塞回上下文 → 模型再次生成 → 最终回答"。在这整条链路里，astream只在最后那一步吐字，前面工具执行的那几秒到十几秒，用户看到的就是一个转圈的加载图标。

我做过一个内部数据查询的 Agent，工具那一步要跑一次跨表聚合，平均 6 到 8 秒。上线第一天就收到反馈："这玩意儿是不是卡死了？"用户不知道它在干什么，就会怀疑、会刷新、会关掉。这不是模型能力问题，是过程不可见的问题。

astream_events解决的正是这个：它不止给你最终输出的 token，而是把链路上每一个组件（模型、工具、检索器、提示词模板、输出解析器、子链）的开始、流式输出、结束都作为事件推出来。你能在工具开始执行的那一刻就告诉用户"正在查询订单数据"，而不是等八秒后给一个结果。

1.2 事件流本质上是一棵带父子关系的执行树

刚开始我以为事件就是一个线性的列表，读一条处理一条。后来发现不是，它其实是一棵执行树，只是被压平成流式序列吐出来。

每条事件都带这几个关键字段，我把它们的作用列清楚：

理解run_id和parent_ids是理解整个事件流的关键。可以把它类比成物流：run_id是每个包裹的单号，parent_ids记录了这个包裹经过了哪几个转运中心。当一条on_tool_start事件的parent_ids里包含某个on_chain_start的run_id时，你就知道这个工具是被那条链调起来的，而不是被另一条并发链调起来的。

这一点在并发场景下尤其重要。用户可能同时发两个请求，或者一个请求里并行跑了三个检索任务，事件流会交错吐出来。如果你只用event字段来判断，一定会串台；必须用parent_ids做归属判断，才能把交错的事件重新归到各自的执行树上。

1.3 什么场景值得上事件流，什么场景纯属过度设计

不是所有项目都需要astream_events。我见过有同学做一个单轮问答的 Demo，也硬套事件流，结果代码量翻了三倍，收益几乎为零。判断标准其实很简单：用户的等待时间里，有没有"值得展示的中间过程"。

还有一个判断维度是排障需求。即使前端不需要展示中间过程，后端也常常需要知道"这一次请求到底在哪一步慢了"。事件流天然带时间戳（每条事件的data里通常有对应的回调时间信息，你也可以在接收端自己打点），把它落盘后就是一份现成的执行轨迹。这种情况下即使不上前端，也值得在服务端跑一遍。

2. 类型化这件事为什么值得单独做

2.1 原始事件是字典，字典在业务代码里会失控

astream_events吐出来的是普通的dict。文档示例里那种写法，处理一两个事件还行，一旦业务复杂起来就会迅速失控。

我第一版的代码大概长这样：一堆if event["event"] == "on_chat_model_stream"，里面再嵌if event["name"] == "xxx"，取值全靠event["data"]["chunk"].content。问题在两周后集中爆发：某次框架小版本升级，某个字段的层级变了一点，代码没报错，但前端显示变成空白，排查了大半天才发现是取值路径失效。更烦的是这类字典访问在编辑器里没有任何提示，字段名拼错了只有运行时才炸，而且往往炸在线上。

所谓类型化事件流，就是在这层原始字典之上加一层自己的模型：把每条事件解析成一个有明确字段、有类型约束的对象，业务代码只和这个对象打交道。好处有三个：字段拼错立刻报错、IDE 能补全、前后端能基于同一个模型约定协议。

2.2 用 Pydantic 定义自己的事件模型

我的做法是先定义一个统一的信封（envelope），再针对高频事件定义具体的载荷模型。这样既保持了通用性，又拿到了类型提示。代码大致如下：

from typing import Any, Literal, Optional from pydantic import BaseModel, Field EventKind = Literal[ "chain_start", "chain_stream", "chain_end", "model_start", "model_stream", "model_end", "tool_start", "tool_stream", "tool_end", "retriever_start", "retriever_end", "custom", ] class EventEnvelope(BaseModel): kind: EventKind name: str run_id: str parent_ids: list[str] = Field(default_factory=list) tags: list[str] = Field(default_factory=list) # 只保留我们关心的元信息，避免把整个 metadata 透传给前端 model_name: Optional[str] = None ts: float payload: dict[str, Any] = Field(default_factory=dict)

关键点在于payload不直接放原始data，而是经过一次裁剪。原始的data里可能带着完整的消息历史、模型返回的原始对象，直接序列化给前端会有两个问题：体积大（一次事件几十 KB）、有泄漏风险（历史消息里可能有系统提示词）。我一般只保留前端真正需要的字段，比如流式事件只留text_delta，工具开始事件只留tool_name和截断后的入参。

裁剪这步看着麻烦，但它是把"能跑"变成"能上线"的分水岭。我曾经偷懒直接把event整体json.dumps推给前端，结果一个稍微长点的对话就把浏览器卡住了，因为每条 token 事件都带着完整上下文重传了一遍。

2.3 事件到 UI 状态的映射规则

类型化之后，还要定一套稳定的映射规则，把事件翻译成界面动作。这套映射一旦定下来，前端就不用关心底层是什么框架、什么事件名，协议层完全解耦。我的项目里用的是下面这套：

这里有个细节值得单独说：model_stream事件在工具调用场景下会出现在不同的位置。模型第一次调用时吐的是工具调用的结构化片段（tool_call_chunks），文本内容往往是空的；工具执行完第二次调用时吐的才是真正的回答。如果前端不加区分地把所有model_stream都追加到同一个气泡里，用户会看到一堆空白或者奇怪的字符。我的处理方式是判断当前流的所属run_id是否已经有对应的气泡，没有就新建，同时检查chunk.content是否为空字符串，为空就跳过。

3. 动手：从零搭一条 Agent 事件流管道

3.1 最小骨架，先跑通再谈优化

不管后面要做多少封装，第一步永远是先让事件流跑起来，看清它到底吐了什么。我建议所有人都先写这段最小代码，然后把事件打印出来扫一眼，这比读十页文档都管用。

import asyncio from langchain_core.tools import tool from langchain.agents import create_agent @tool def lookup_order(order_no: str) -> str: """根据订单号查询订单状态。""" return f"订单 {order_no} 状态：已发货" agent = create_agent(model=llm, tools=[lookup_order]) async def main(): async for event in agent.astream_events( {"messages": [{"role": "user", "content": "帮我查下 A123 这单到哪了"}]}, version="v2", ): kind = event["event"] print(f"[{kind}] name={event['name']} run={event['run_id'][:8]}") asyncio.run(main())

跑完这段，你会看到一串类似on_chain_start、on_chat_model_start、on_chat_model_stream、on_chat_model_end、on_tool_start、on_tool_end、on_chain_end的输出。第一次看可能会觉得事件比想象中多，尤其是on_chat_model_stream会按 token 逐条吐，一次回答几十上百条很正常。

这里有第一个必须记住的点：version="v2"这个参数不是可选的装饰。不写它，不同版本的行为不一样，老版本会走 v1 的实现，事件语义和字段结构都有差异，我见过有人照抄了一份很老的示例代码，结果data里怎么都找不到chunk，折腾了半天才发现是版本参数的问题。写的时候建议直接写在参数里，别指望默认值。

3.2 用过滤参数把事件量砍下来

事件量是这套机制最容易被低估的成本。一次稍复杂的 Agent 调用，产生几百条事件很常见。如果你只是想在界面上显示"正在调用工具"，却把几百条 token 事件全都序列化、全都过一遍网络，那就是纯粹的浪费。

astream_events提供了六个过滤参数，我按使用频率排一下：

• include_types：按事件类型过滤，最常用。比如只要["on_tool_start", "on_tool_end"]。
• include_names：按组件名过滤。比如只要某个特定检索器的结束事件。
• include_tags：按标签过滤。适合你在构建链时给某些节点打了自定义 tag，事后想单独拿出来。
• exclude_types/exclude_names/exclude_tags：反向排除，适合"除了某个噪音节点，其它都要"的场景。

async for event in agent.astream_events( inputs, version="v2", include_types=["on_tool_start", "on_tool_end", "on_chat_model_stream", "on_chat_model_end"], ): ...

关于这几个参数，我踩过一个坑值得提一句：include_names和include_types同时传的时候，过滤是变宽而不是变窄的。直觉上你会觉得两个条件是"与"关系，实际跑下来是"或"——只要匹配其中任一条件，事件就会被放出来。我第一次发现这个问题是因为明明只想看工具事件，结果模型 token 也一起出来了。所以我的习惯是：一次只用一种过滤维度，需要多维度组合就在自己的消费端再筛一遍，别指望框架帮你做交集。

还有一点，过滤是在事件产生侧生效的，被过滤掉的事件不会走你的消费逻辑，所以它确实能省 CPU 和序列化开销，但注意它不影响底层链的实际执行——工具该跑还是跑，只是你看不到事件了。

3.3 把事件拼成树，还原完整调用链路

拿到一堆扁平事件之后，最有价值的加工就是把它拼回树。这一步做完，你就能回答"这次请求里哪个环节最耗时"这类问题。

思路很简单：维护一个run_id -> 节点的字典，遇到_start事件就建节点挂到父节点下面，遇到_end事件就回填耗时和输出。核心代码大概这样：

import time from dataclasses import dataclass, field @dataclass class Node: run_id: str kind: str name: str start_ts: float end_ts: float | None = None children: list["Node"] = field(default_factory=list) summary: str = "" error: str | None = None @property def cost_ms(self) -> float | None: if self.end_ts is None: return None return round((self.end_ts - self.start_ts) * 1000, 1) class TreeBuilder: def __init__(self): self.nodes: dict[str, Node] = {} self.root: Node | None = None def feed(self, event: dict): rid = event["run_id"] parents = event.get("parent_ids") or [] kind = event["event"] if kind.endswith("_start"): node = Node(rid, kind, event["name"], time.time()) self.nodes[rid] = node if parents and parents[-1] in self.nodes: self.nodes[parents[-1]].children.append(node) elif self.root is None: self.root = node elif kind.endswith("_end"): node = self.nodes.get(rid) if node: node.end_ts = time.time() # 回填摘要，注意不同事件类型的 data 结构不同 node.summary = self._summarize(kind, event.get("data", {})) elif kind.endswith("_stream"): pass # 流式事件只用于增量展示，不入树 def _summarize(self, kind: str, data: dict) -> str: if kind == "on_tool_end": return str(data.get("output"))[:120] if kind == "on_chat_model_end": out = data.get("output") text = getattr(out, "content", "") or "" return str(text)[:120] return ""

这段代码有两个细节值得展开。第一，on_*_stream事件我直接跳过不入树，因为它们数量太多且没有结构价值，只用于增量渲染。第二，_summarize里做了getattr兜底，因为不同版本、不同类型的模型节点，data["output"]的类型不完全一致，有的是消息对象，有的是结果包装对象，硬取值容易在某个分支上炸掉。

有了这棵树，前端做缩进展示就非常容易了——递归遍历，按层级加缩进，叶子节点显示耗时。用户看到的是"总耗时 9.2 秒，其中工具 lookup_order 占 7.8 秒"，这比一句"查询中"有用得多。

3.4 接 SSE 推给前端

后端拼好了事件对象，接下来就是推给前端。我的选择是 SSE（Server-Sent Events），理由很实际：它是单向的，正好匹配"服务端推流给客户端"这个场景；基于普通 HTTP，不需要额外的协议升级；浏览器原生有EventSource，接入成本极低。

服务端用异步框架做流式响应，核心是生成器不断yield格式化的数据块：

import json async def event_source(payload: dict): async for event in agent.astream_events(payload, version="v2"): envelope = to_envelope(event) # 前面定义的裁剪 + 类型化 if envelope is None: continue # SSE 要求每条消息以 data: 开头，以两个换行结束 yield f"id: {next_seq()}\n" yield f"event: {envelope.kind}\n" yield f"data: {envelope.model_dump_json()}\n\n"

前端侧用EventSource接，但有两个必须注意的点。一是SSE 连接默认会超时重连，如果不做处理，重连之后用户会看到重复内容。解决办法是给每条消息带id，重连时浏览器会把Last-Event-ID发回来，服务端据此跳过已发送的部分；或者更省事的做法，重连就直接重开一轮，前端清空当前气泡。二是多字节字符切分问题，流式传输时如果按字节切，中文字符可能被拦腰截断成乱码，所以一定要在服务端保证每个yield都是完整的 UTF-8 字符串，别在字节层面做切分。

WebSocket 当然也能做，如果你的场景需要双向通信（比如用户在生成中途打断、补充信息），那就该上 WebSocket。但只是展示进度的话，SSE 的心智负担和运维成本都低得多。

4. 几个必须提前知道的坑

4.1 版本参数和事件重复

前面提过version="v2"，这里补一个相关的坑：同一个节点可能产生多个看起来重复的事件。比如一个链里套了一个模型，你可能会同时收到on_chain_stream和on_chat_model_stream，两者携带的文本内容高度相似。

这不是 bug，是设计使然——链级别的流式事件会把子节点的输出向上冒泡，方便你在高层节点统一接收。但如果你不做判断就全部渲染，界面上的文字就会变成双份。

我的处理原则是：只在一层消费流式事件。具体来说，要么只监听最细粒度的on_chat_model_stream（信息最全、带模型标识），要么只监听根链的on_chain_stream（最简单），绝不同时消费两层。选哪个取决于你要不要区分"这一段文本是哪个模型节点产的"，需要区分就用前者。

4.2 同步节点会掐断流式

这个坑我踩得最惨。现象是：本地跑得好好的流式输出，接到某个业务链之后就变成"憋很久，然后一次性全出来"，打字机效果完全消失。

排查了很久才定位到原因——链中间插入了一个同步的RunnableLambda，里面做了一些字符串处理。同步函数在异步事件流里执行时，会阻塞整个事件循环，导致上游的流式块没法及时吐出来，全被攒到函数执行完才一起放出去。

判断方法很简单：把链上每个节点单独测一遍，看哪个节点接上去之后流式就断了。修复方式也直接——把同步函数改成异步，或者用框架提供的异步包装把它丢到线程池里执行。这个问题的隐蔽之处在于功能完全正常，只是体验退化，不做性能对比根本发现不了。

4.3 工具内部再调模型，事件归属怎么算

当你的工具本身也是个 Agent（比如一个专门的检索助手），它内部也会调用模型。这时候事件流里会出现嵌套的on_chat_model_stream，parent_ids里挂着工具节点的run_id。

如果不做处理，前端会把工具内部的中间思考当成最终回答渲染出来，用户看到一堆莫名其妙的推理过程。处理方式有两种：一是在消费端根据parent_ids判断层级，只渲染深度为特定值的模型事件；二是干脆在工具内部调用时给链打上 tag，外层用exclude_tags把这些噪音排除掉。

我更倾向第二种，因为过滤在产生侧就生效，省掉了传输和解析的开销。给内部的模型调用加上类似internal的标签，外层订阅时排除掉，代码干净很多。

4.4 常见问题速查表

把我在实际项目里遇到的和被问过的问题整理成一张表，出问题时可以先对一遍：

5. 让事件流产生真正的可观测价值

5.1 从事件里算出时延和 token 消耗

事件流最被低估的用途是成本与性能观测。既然每次调用都会吐出完整的事件序列，那顺手把关键指标算出来几乎是零成本的事。

时延方面，_start和_end的时间差就是每个节点的耗时，聚合到根节点就是端到端耗时。我习惯把耗时按类型分组统计，比如"模型调用总耗时""工具执行总耗时""检索总耗时"，这样一眼就能看出瓶颈在哪。很多时候优化方向根本不是换更强的模型，而是把那个跑了六秒的数据库查询加个索引。

token 方面，模型结束事件里通常带着用量信息（新版本的消息对象上有usage_metadata，包含输入、输出、总计）。把这些数字按run_id归集，再和metadata里的模型名对照，就得到了一份按模型维度的消耗报表。

def extract_usage(event: dict) -> dict | None: out = event.get("data", {}).get("output") usage = getattr(out, "usage_metadata", None) if not usage: return None return { "run_id": event["run_id"], "model": event.get("metadata", {}).get("ls_model_name"), "input_tokens": usage.get("input_tokens", 0), "output_tokens": usage.get("output_tokens", 0), }

这里要提醒一句：不同模型服务商返回的用量字段结构不完全一致，有的只给总数不给分项，有的干脆不返回。所以这段代码一定要做空值兜底，别让它成为事件流里唯一一个会抛异常的地方——一旦抛出去，整个异步迭代就中断了，后面的回答用户永远等不到。

5.2 自定义事件与业务埋点

框架内置的事件类型覆盖了常规组件，但业务上总有些状态是它不知道的。比如"正在生成报表第 3 页，共 10 页"，这种进度信息只能自己埋。

框架提供了自定义事件的分发接口，在链里某个节点里主动打一个点，外层就能用对应类型（on_custom_event）收到，name就是你指定的业务名。这个能力在长任务场景下非常实用——用户最怕的不是慢，而是不知道还剩多久。

我的做法是把长流程拆成几个明确的阶段，每个阶段开始时打一个自定义事件，前端拿到之后更新进度条和文案。要注意的是自定义事件里的data是自己随便塞的，前端解析时最好带上版本号或者做宽松校验，避免将来改字段导致线上前端报错。

5.3 落盘回放与断线重连

最后一个我觉得值得做的是事件落盘。把每条精简后的事件连同时间戳写进一张表，好处有三个：线上出问题时可以完整复现当时的执行路径；可以离线分析慢请求的共同特征；可以做"回放"，把某次执行的事件按原速重放给用户看。

落盘的代价不大，因为前面已经裁剪过载荷了，单条事件通常就几百字节到几 KB。如果量特别大，可以只在发生错误、或者耗时超过阈值的请求上落全量，正常请求只落骨架。

断线重连则要结合前面的 SSE 序号来做。我的方案是给每个会话维护一个内存里的事件缓冲区（带最大长度），重连时根据Last-Event-ID把缓冲区里剩下的部分补发出去，补完再切回实时流。这个方案在单机部署下够用，多实例的话就得把缓冲区放到共享存储里，但那是另一个话题了。

这套东西做完之后，我最大的感受是：事件流真正的价值不在于"展示"，而在于它把一个黑盒变成了可度量、可回放、可归因的系统。前期多花的那点封装成本，在第一次线上事故排查时就连本带利收回来了。