Agent事件流可观测：astream_events类型化与SSE

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 事件流本质上是一棵带父子关系的执行树刚开始我以为事件就是一个线性的列表读一条处理一条。后来发现不是它其实是一棵执行树只是被压平成流式序列吐出来。每条事件都带这几个关键字段我把它们的作用列清楚字段类型作用实际使用场景eventstr事件类型如on_chat_model_stream决定这条事件怎么处理namestr触发事件的组件名区分同类型的多个节点比如两个不同的工具run_idstr本次运行的唯一 ID作为节点身份标识做状态索引parent_idslist[str]从根到直接父节点的 ID 链还原调用层级做缩进展示tagslist[str]组件上打的标签按业务标签过滤比如只要某个子链的事件metadatadict元信息含模型名、自定义 metadata埋点、归因、成本统计datadict具体载荷结构随事件类型变化取出 token 块、工具入参、最终输出理解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也硬套事件流结果代码量翻了三倍收益几乎为零。判断标准其实很简单用户的等待时间里有没有值得展示的中间过程。场景是否有值得展示的中间态建议方案单轮问答、纯文本生成没有直接用astream带工具调用的 Agent有工具名和入参是重要信息用astream_eventsRAG 检索问答有命中的文档片段值得展示用astream_events抓检索事件多 Agent 协作 / 多步骤规划强需求步骤本身是产品价值必须用事件流 树形展示后台批处理任务无人在看不需要实时流落日志即可还有一个判断维度是排障需求。即使前端不需要展示中间过程后端也常常需要知道这一次请求到底在哪一步慢了。事件流天然带时间戳每条事件的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_factorylist) tags: list[str] Field(default_factorylist) # 只保留我们关心的元信息避免把整个 metadata 透传给前端 model_name: Optional[str] None ts: float payload: dict[str, Any] Field(default_factorydict)关键点在于payload不直接放原始data而是经过一次裁剪。原始的data里可能带着完整的消息历史、模型返回的原始对象直接序列化给前端会有两个问题体积大一次事件几十 KB、有泄漏风险历史消息里可能有系统提示词。我一般只保留前端真正需要的字段比如流式事件只留text_delta工具开始事件只留tool_name和截断后的入参。裁剪这步看着麻烦但它是把能跑变成能上线的分水岭。我曾经偷懒直接把event整体json.dumps推给前端结果一个稍微长点的对话就把浏览器卡住了因为每条 token 事件都带着完整上下文重传了一遍。2.3 事件到 UI 状态的映射规则类型化之后还要定一套稳定的映射规则把事件翻译成界面动作。这套映射一旦定下来前端就不用关心底层是什么框架、什么事件名协议层完全解耦。我的项目里用的是下面这套事件界面动作备注chain_start根节点显示思考中骨架标记会话轮次开始model_start状态文案切到正在理解问题可选通常很快tool_start插入一条工具卡片显示工具名 入参摘要入参要做脱敏和截断tool_end卡片状态改为完成展示结果摘要和耗时耗时从两条事件的时间差算model_stream追加文本到当前气泡需要按run_id区分气泡model_end气泡结束记录 token 用量从usage_metadata取custom按自定义业务名分发用于进度、阶段提示这里有个细节值得单独说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(modelllm, tools[lookup_order]) async def main(): async for event in agent.astream_events( {messages: [{role: user, content: 帮我查下 A123 这单到哪了}]}, versionv2, ): 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 逐条吐一次回答几十上百条很正常。这里有第一个必须记住的点versionv2这个参数不是可选的装饰。不写它不同版本的行为不一样老版本会走 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, versionv2, 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_factorylist) 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 推给前端后端拼好了事件对象接下来就是推给前端。我的选择是 SSEServer-Sent Events理由很实际它是单向的正好匹配服务端推流给客户端这个场景基于普通 HTTP不需要额外的协议升级浏览器原生有EventSource接入成本极低。服务端用异步框架做流式响应核心是生成器不断yield格式化的数据块import json async def event_source(payload: dict): async for event in agent.astream_events(payload, versionv2): envelope to_envelope(event) # 前面定义的裁剪 类型化 if envelope is None: continue # SSE 要求每条消息以 data: 开头以两个换行结束 yield fid: {next_seq()}\n yield fevent: {envelope.kind}\n yield fdata: {envelope.model_dump_json()}\n\n前端侧用EventSource接但有两个必须注意的点。一是SSE 连接默认会超时重连如果不做处理重连之后用户会看到重复内容。解决办法是给每条消息带id重连时浏览器会把Last-Event-ID发回来服务端据此跳过已发送的部分或者更省事的做法重连就直接重开一轮前端清空当前气泡。二是多字节字符切分问题流式传输时如果按字节切中文字符可能被拦腰截断成乱码所以一定要在服务端保证每个yield都是完整的 UTF-8 字符串别在字节层面做切分。WebSocket 当然也能做如果你的场景需要双向通信比如用户在生成中途打断、补充信息那就该上 WebSocket。但只是展示进度的话SSE 的心智负担和运维成本都低得多。4. 几个必须提前知道的坑4.1 版本参数和事件重复前面提过versionv2这里补一个相关的坑同一个节点可能产生多个看起来重复的事件。比如一个链里套了一个模型你可能会同时收到on_chain_stream和on_chat_model_stream两者携带的文本内容高度相似。这不是 bug是设计使然——链级别的流式事件会把子节点的输出向上冒泡方便你在高层节点统一接收。但如果你不做判断就全部渲染界面上的文字就会变成双份。我的处理原则是只在一层消费流式事件。具体来说要么只监听最细粒度的on_chat_model_stream信息最全、带模型标识要么只监听根链的on_chain_stream最简单绝不同时消费两层。选哪个取决于你要不要区分这一段文本是哪个模型节点产的需要区分就用前者。4.2 同步节点会掐断流式这个坑我踩得最惨。现象是本地跑得好好的流式输出接到某个业务链之后就变成憋很久然后一次性全出来打字机效果完全消失。排查了很久才定位到原因——链中间插入了一个同步的RunnableLambda里面做了一些字符串处理。同步函数在异步事件流里执行时会阻塞整个事件循环导致上游的流式块没法及时吐出来全被攒到函数执行完才一起放出去。判断方法很简单把链上每个节点单独测一遍看哪个节点接上去之后流式就断了。修复方式也直接——把同步函数改成异步或者用框架提供的异步包装把它丢到线程池里执行。这个问题的隐蔽之处在于功能完全正常只是体验退化不做性能对比根本发现不了。4.3 工具内部再调模型事件归属怎么算当你的工具本身也是个 Agent比如一个专门的检索助手它内部也会调用模型。这时候事件流里会出现嵌套的on_chat_model_streamparent_ids里挂着工具节点的run_id。如果不做处理前端会把工具内部的中间思考当成最终回答渲染出来用户看到一堆莫名其妙的推理过程。处理方式有两种一是在消费端根据parent_ids判断层级只渲染深度为特定值的模型事件二是干脆在工具内部调用时给链打上 tag外层用exclude_tags把这些噪音排除掉。我更倾向第二种因为过滤在产生侧就生效省掉了传输和解析的开销。给内部的模型调用加上类似internal的标签外层订阅时排除掉代码干净很多。4.4 常见问题速查表把我在实际项目里遇到的和被问过的问题整理成一张表出问题时可以先对一遍现象大概率原因处理方式data[chunk]取不到没传versionv2显式指定版本参数界面文字重复两遍同时消费了链级和模型级流式事件只在一层消费流式变成一次性输出链中混入同步节点阻塞事件循环改异步或包装到线程池事件量大到拖慢服务没做过滤 全量序列化加include_types 裁剪载荷并发请求内容串台只用event判断没用run_id一律按run_id分桶工具事件丢失工具内部有同步阻塞或异常被吞先单独跑工具验证再看异常日志前端收到乱码流式切分破坏了多字节字符按完整字符串yield重连后内容重复未处理Last-Event-ID加序号或重连清屏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把缓冲区里剩下的部分补发出去补完再切回实时流。这个方案在单机部署下够用多实例的话就得把缓冲区放到共享存储里但那是另一个话题了。这套东西做完之后我最大的感受是事件流真正的价值不在于展示而在于它把一个黑盒变成了可度量、可回放、可归因的系统。前期多花的那点封装成本在第一次线上事故排查时就连本带利收回来了。