数据截至 (上游 commit e42bdbedc27c)
04 · 传输层与跨语言一致性
这一章讲底层:事件怎么在网络上走、为什么有「chunk 简写事件」、以及同一份协议怎么在 TypeScript 和 Python 之间保持一致。
4.1 传输与语义解耦:一条事件流可以走任何管道
AG-UI 刻意把「事件是什么」和「事件怎么传」分开。README 说它「works with any event transport(SSE/WebSockets/webhooks)」。客户端的传输入口是 transformHttpEventStream(packages/client/src/transform/http.ts:13),它先缓冲、读到 HTTP headers 再按 content-type 决定解析器:
// packages/client/src/transform/http.ts:43-51(节选)
if (contentType === proto.AGUI_MEDIA_TYPE) {
parseProtoStream(bufferSubject).subscribe(/* protobuf 二进制 */);
} else {
parseSSEStream(bufferSubject, log).subscribe(/* SSE + JSON */);
}
两种格式:
| 格式 | media type | 解析器 | 何时用 |
|---|---|---|---|
| SSE + JSON(默认) | 非 proto 的任意(通常 text/event-stream) | parseSSEStream + EventSchemas.parse | 绝大多数场景 |
| protobuf(二进制) | application/vnd.ag-ui.event+proto | parseProtoStream | 高吞吐/省带宽 |
AGUI_MEDIA_TYPE 这个常量是二进制协议的「暗号」,TS 与 Python 各定义一份且字面值完全相同(packages/proto/src/index.ts:3、sdks/python/ag_ui/encoder/encoder.py:7)。
4.2 SSE 解析:严格遵循标准,处理三件麻烦事
parseSSEStream(packages/client/src/transform/sse.ts)按 SSE 标准把字节流切成 JSON,处理了三个真实麻烦:
- UTF-8 跨块:用
new TextDecoder("utf-8")的{ stream: true }模式,避免一个多字节字符被切在两个网络块之间时乱码。 - 事件按
\n\n分隔、可跨块:buffer 累积,split(/\n\n/)后把最后一段不完整的留在 buffer 里等下一块。 - 只取
data:行、忽略event:/id:/retry::多行 data 拼接后再JSON.parse。
后端的编码方对称地简单——Python 的 EventEncoder._encode_sse 就一行(sdks/python/ag_ui/encoder/encoder.py:28-37):
# sdks/python/ag_ui/encoder/encoder.py(节选)
def _encode_sse(self, event: BaseEvent) -> str:
return f"data: {event.model_dump_json(by_alias=True, exclude_none=True)}\n\n"
注意 by_alias=True:Python 端字段用 snake/别名,序列化时按协议的 camelCase 别名输出,这样 wire 上和 TS 完全对齐。
4.3 chunk 事件:给后端的「简写」,在客户端就地展开
协议里有三个 *_CHUNK 事件(TEXT_MESSAGE_CHUNK/TOOL_CALL_CHUNK/REASONING_MESSAGE_CHUNK)。它们不是新语义,而是「start+content 合一」的便捷写法——后端不想显式发 START/END 时,可以只发 chunk,客户端的 transformChunks(ch02 工序②)会自动补出标准的 START/CONTENT/END。
它的工作方式是个小状态机,记录当前处于 text/tool/reasoning 哪种「模式」(chunks/transform.ts:59):
- 来一个 chunk,若当前不在对应模式(或 id 变了),先关掉上一个(补一个
*_END),再开一个新的(补一个*_START),然后把 delta 转成*_CONTENT。 - 流结束时
finalize里补上最后一个*_END(chunks/transform.ts:384-387),保证不留未闭合的消息。
收到: TOOL_CALL_CHUNK{id=c1,name=f,delta="{"} TOOL_CALL_CHUNK{delta="}"} <流结束>
展开: TOOL_CALL_START{c1,f} → TOOL_CALL_ARGS{c1,"{"} → TOOL_CALL_ARGS{c1,"}"} → TOOL_CALL_END{c1}
这解释了 ch03 为什么 apply 阶段见到 chunk 就 throw:chunk 必须在 verify/apply 之前被展开,否则下游的状态机和 reducer 不认识它。
4.4 内容协商:后端怎么知道该发 SSE 还是 protobuf
编码侧(@ag-ui/encoder)用标准 HTTP 内容协商。EventEncoder 看请求的 Accept 头,用 preferredMediaTypes(从 negotiator 库改写而来,encoder/src/media-type.ts)判断客户端是否接受 proto:
// packages/encoder/src/encoder.ts:64-67(节选)
const preferred = preferredMediaTypes(acceptHeader, [proto.AGUI_MEDIA_TYPE]);
return preferred.includes(proto.AGUI_MEDIA_TYPE); // 接受就发二进制,否则 SSE
这是教科书式的 HTTP 做法:同一个端点,按 Accept 头决定响应格式,客户端和服务端不用预先约定。
4.5 跨语言一致性:同一份协议,各语言一份同形 schema
AG-UI 的协议本体没有用一个跨语言的 IDL 单一来源(proto 只覆盖二进制传输);事件/类型的「真源」是各语言手写但刻意同形的 schema:
| 关切 | TypeScript(Zod) | Python(Pydantic) |
|---|---|---|
| 事件类型 | packages/core/src/events.ts EventType | sdks/python/ag_ui/core/events.py |
| 数据模型 | packages/core/src/types.ts | sdks/python/ag_ui/core/types.py |
| 能力声明 | packages/core/src/capabilities.ts | sdks/python/ag_ui/core/capabilities.py |
| SSE 编码 | @ag-ui/encoder | ag_ui/encoder/encoder.py |
一致性靠几件事撑住:
- 字段别名对齐:Python 序列化
by_alias=True,wire 上是同一套 camelCase 名。 - 宽松校验:
BaseEventSchema.passthrough()(ch01)让一端新增字段不会让另一端解码失败。 - 兼容补丁:为对方序列化习惯让路,如 TS 接受 Python 输出的
"outcome": null(ch01 §1.5)。 - 客户端中间件做版本翻译(ch02 §2.4):老事件形态在客户端就地升级。
这是一种「靠纪律而非工具保证一致」的策略:好处是各语言能用最地道的 schema 库(Zod/Pydantic)、错误信息友好;代价是新增/修改事件得在多处同步,且像 verifyEvents 这种「规则即代码」的逻辑,严格说每个语言都得各实现一遍。
4.6 能力声明:让客户端「先问后用」
AgentCapabilities(packages/core/src/capabilities.ts:220)是个分类齐整的「我支持什么」清单:transport(streaming/websocket/httpBinary/resumable…)、tools、state(snapshots/deltas)、reasoning(含 encrypted 零数据保留)、humanInTheLoop(含 interrupts/approveWithEdits)、multimodal 等,全部字段可选。AbstractAgent.getCapabilities?()(agent.ts:148)是可选实现——agent 自报能力,客户端据此决定显示哪些 UI(文件上传按钮、批准对话框…)。所有字段「省略 = 未声明(未知)」而非「不支持」(schema 顶部注释明说),这是个谨慎的语义选择。
4.7 边界与局限(诚实)
- 运行时以 TypeScript 为主:
verifyEvents/defaultApplyEvents这类「消费端引擎」目前最完整的是 TS(packages/client);Python 等 SDK 侧重协议类型与编码,消费端引擎的完整度需按具体 SDK 核实,本文未逐一验证非 TS 的客户端折叠逻辑 (inferred)。 - 没有单一 IDL 真源:协议契约靠各语言手写同形 schema + 纪律保持一致(§4.5),不是从一份 schema 自动生成全部语言。
- 顺序规则散落在状态机:合法事件顺序的权威是
verifyEvents代码而非独立规范,跨语言对齐靠各自重写。 - 传输适配仍以 HTTP/SSE 为中心:WebSocket/webhook 在能力枚举里有位置,但本文只核实了 HTTP+SSE 与 protobuf 两条解码路径的真实实现。
4.8 横向对比(同 shelf 兄弟协议)
README 自己点明了三件套的分工——这是理解 AG-UI 定位最好的一句话:
| 协议 | 解决什么 | 一句话 |
|---|---|---|
| MCP | 给 agent 工具 | agent ↔ 工具/数据源 |
| A2A | agent 之间通信 | agent ↔ agent |
| AG-UI | 把 agent 带进用户界面 | agent ↔ 用户/UI |
和 MCP 对比一个具体取舍:MCP 是请求/响应 + 工具发现为主的 JSON-RPC 风格;AG-UI 是单向事件流 + 客户端状态折叠为主——因为 UI 场景的本质是「实时回放一个正在发生的过程」,而不是「调一个函数拿返回值」。
4.9 代码地图
| 主题 | 文件 | 符号名 |
|---|---|---|
| 按 content-type 选解析器 | packages/client/src/transform/http.ts | transformHttpEventStream |
| SSE 解析(UTF-8/跨块/多行 data) | packages/client/src/transform/sse.ts | parseSSEStream |
| protobuf 解析 | packages/client/src/transform/proto.ts | parseProtoStream |
| chunk 事件展开 | packages/client/src/chunks/transform.ts | transformChunks |
| 二进制 media type 常量 | packages/proto/src/index.ts | AGUI_MEDIA_TYPE |
| 内容协商编码器 | packages/encoder/src/encoder.ts | EventEncoder |
| Python SSE 编码 | sdks/python/ag_ui/encoder/encoder.py | EventEncoder._encode_sse |
| 能力声明 | packages/core/src/capabilities.ts | AgentCapabilitiesSchema |