9.1 事件协议：token/tool/step 统一 schema，TS 与 Python 共享类型
3.2 把 ResearchCopilot 的 run 变成了 SSE 流，帧里 event: 给出事件名、data: 给出单行 JSON。但那张「事件名 → 载荷字段」的映射表此刻只活在文档里：Python 侧发出去的是裸 dict，TS 侧还没人消费。一旦两边各凭记忆写字段，第一个拼错的 run_id 就会静默失配——这是最难查的一类 bug。本节把那张表升级成一份语言无关的 schema，并让两种语言各自把它落成编译期能检查的类型。
本节把 ResearchCopilot 推进到「事件契约显式化」：给 3.2 的七类事件定义带
type判别字段的 schema，Python 用 pydantic 判别联合、TypeScript 用判别联合类型各实现一份，并用同一份 JSON 样本在两侧真跑解析。
9.1.1 判别字段：为什么是 type
一份「多形态」的 schema 有几种组织方式：所有字段都可选（text? / content? / name? 全摆在一个大对象里）、用嵌套的 payload 分桶、或者引入一个判别字段。前两种都会让「拿到事件后不知道该读哪些字段」的问题残留——你无法在类型层面排除「一个 token 事件却带 result」这种非法组合。
判别字段（discriminated union / tagged union）解决的就是这个：每个变体先声明自己是谁（type），再声明这个身份专属的字段。有了它，类型检查器能保证「先判 type，再访问字段」的顺序，访问错变体的字段会直接编译失败。3.2 的 event: 名正好就是天然的判别值，我们把它收进 JSON 载荷，得到七类事件：
type | 载荷字段 | 来源 |
|---|---|---|
run.started | run_id | run 生命周期 |
step | index、kind（tool_call / final） | 循环步骤 |
tool_call | id、name、arguments | 模型侧 |
tool_result | id、name、result | 循环步骤 |
token | text | 模型侧 |
final | content | 循环收敛 |
run.succeeded | run_id、steps | run 生命周期 |
这张表就是本节要「共享」的东西。它不含任何语言特性，只描述「哪种事件有哪些字段」——这正是它能在两种语言里各落一份类型的前提。
9.1.2 TypeScript 侧：判别联合类型
TS 里最贴切的表达就是字面量类型做判别字段的联合。把七类事件写成一个 AgentEvent 联合，每个成员用 type: "..." 钉死身份：
type AgentEvent =
| { type: "run.started"; run_id: string }
| { type: "step"; index: number; kind: "tool_call" | "final" }
| { type: "tool_call"; id: string; name: string; arguments: Record<string, unknown> }
| { type: "tool_result"; id: string; name: string; result: string }
| { type: "token"; text: string }
| { type: "final"; content: string }
| { type: "run.succeeded"; run_id: string; steps: number };
interface SseFrame { id: number; event: string; data: string; }
这段只用了可擦除语法（无 enum、无参数属性、无 namespace），node events.ts 能零构建直跑。好处立刻兑现：写 switch (ev.type) 时，TS 会在每个分支里自动收窄 ev 的类型，case "token" 里访问 ev.text 合法，访问 ev.content 直接报错。下面的 describe 还刻意不写 default——一旦将来加了第八种事件而这里没补分支，编译器会因为「返回类型可能为 undefined」而报警，等于免费买到一张「新增事件必须处理」的清单。
9.1.3 解码：从 SSE 帧到判别联合
网络上来的是 SseFrame（event 名 + data JSON 字符串），要先解成 AgentEvent。解码器按 event 名分发，把 data 里的字段摊进对象：
function decode(frame: SseFrame): AgentEvent {
const raw = JSON.parse(frame.data) as Record<string, unknown>;
switch (frame.event) {
case "run.started":
return { type: "run.started", run_id: String(raw.run_id) };
case "step":
return { type: "step", index: Number(raw.index), kind: raw.kind as "tool_call" | "final" };
case "tool_call":
return { type: "tool_call", id: String(raw.id), name: String(raw.name),
arguments: raw.arguments as Record<string, unknown> };
case "tool_result":
return { type: "tool_result", id: String(raw.id), name: String(raw.name),
result: String(raw.result) };
case "token":
return { type: "token", text: String(raw.text) };
case "final":
return { type: "final", content: String(raw.content) };
case "run.succeeded":
return { type: "run.succeeded", run_id: String(raw.run_id), steps: Number(raw.steps) };
default:
throw new Error(`未知事件类型: ${frame.event}`);
}
}
function describe(ev: AgentEvent): string {
switch (ev.type) {
case "run.started": return `run.started run_id=${ev.run_id}`;
case "step": return `step index=${ev.index} kind=${ev.kind}`;
case "tool_call": return `tool_call ${ev.name}(${JSON.stringify(ev.arguments)})`;
case "tool_result": return `tool_result ${ev.name} -> ${ev.result}`;
case "token": return `token ${JSON.stringify(ev.text)}`;
case "final": return `final ${JSON.stringify(ev.content)}`;
case "run.succeeded":return `run.succeeded steps=${ev.steps}`;
}
}
decode 是信任边界：它把「网络来的 unknown」转成「编译期可信的 AgentEvent」，所以这里必须显式构造每个字段，不能图省事写 { type: frame.event, ...raw } as AgentEvent——那样 type 只是运行时字符串，编译期收窄全失效。边界上的一次显式转换，换来下游全程的类型安全，这笔账值得。
9.1.4 真跑：同一份 JSON，TS 侧解析
把 3.2.6 那段真实输出里的 11 帧原样抄成数据，喂给 decode 并打印判别结果：
const FRAMES: SseFrame[] = [
{ id: 1, event: "run.started", data: '{"run_id": "r_d98c8f3c"}' },
{ id: 2, event: "step", data: '{"index": 0, "kind": "tool_call"}' },
{ id: 3, event: "tool_call", data: '{"id": "c1", "name": "search_corpus", "arguments": {"q": "pgvector 召回率"}}' },
{ id: 4, event: "tool_result", data: '{"id": "c1", "name": "search_corpus", "result": "命中 3 条"}' },
{ id: 5, event: "token", data: '{"text": "pgvector"}' },
{ id: 6, event: "token", data: '{"text": " 在"}' },
{ id: 7, event: "token", data: '{"text": "低维"}' },
{ id: 8, event: "token", data: '{"text": "场景更省"}' },
{ id: 9, event: "step", data: '{"index": 1, "kind": "final"}' },
{ id: 10, event: "final", data: '{"content": "pgvector 在低维场景更省。"}' },
{ id: 11, event: "run.succeeded", data: '{"run_id": "r_d98c8f3c", "steps": 2}' },
];
let tokens = 0;
for (const f of FRAMES) {
const ev = decode(f);
if (ev.type === "token") tokens += 1;
console.log(`#${String(f.id).padStart(2)} ${describe(ev)}`);
}
console.log(`--- 事件总数: ${FRAMES.length}, token 事件: ${tokens}`);
本机实测（Node 26.5.0，node events.ts 直跑）：
# 1 run.started run_id=r_d98c8f3c
# 2 step index=0 kind=tool_call
# 3 tool_call search_corpus({"q":"pgvector 召回率"})
# 4 tool_result search_corpus -> 命中 3 条
# 5 token "pgvector"
# 6 token " 在"
# 7 token "低维"
# 8 token "场景更省"
# 9 step index=1 kind=final
#10 final "pgvector 在低维场景更省。"
#11 run.succeeded steps=2
--- 事件总数: 11, token 事件: 4
11 帧逐条判别成功，4 个 token 事件计数与 3.2 那条流完全吻合。tsc --noEmit 也通过（tsconfig.json 开着 erasableSyntaxOnly），说明类型检查与可擦除运行两件事同时成立。
9.1.5 Python 侧：pydantic 判别联合
同一份 schema 在 Python 侧用 pydantic 的 discriminator 落地。每个变体是一个 BaseModel，type 用 Literal 钉死，最后用 Field(discriminator="type") 合成一个联合：
from typing import Annotated, Literal, Union
from pydantic import BaseModel, Field, TypeAdapter
class Token(BaseModel):
type: Literal["token"]
text: str
class ToolCall(BaseModel):
type: Literal["tool_call"]
id: str
name: str
arguments: dict
class Final(BaseModel):
type: Literal["final"]
content: str
# ……其余四类同构，略……
AgentEvent = Annotated[
Union[RunStarted, Step, ToolCall, ToolResult, Token, Final, RunSucceeded],
Field(discriminator="type"),
]
adapter = TypeAdapter(AgentEvent)
pydantic 2.14.0 的 discriminator 是真判别：它只看 type 的值就直接挑中对应模型，不做「逐个尝试、谁匹配算谁」的暴力回溯，因此校验快且报错准。用同一份 JSON 样本跑：
samples = [
{"type": "run.started", "run_id": "r_d98c8f3c"},
{"type": "step", "index": 0, "kind": "tool_call"},
{"type": "tool_call", "id": "c1", "name": "search_corpus", "arguments": {"q": "pgvector 召回率"}},
{"type": "tool_result", "id": "c1", "name": "search_corpus", "result": "命中 3 条"},
{"type": "token", "text": "pgvector"},
{"type": "final", "content": "pgvector 在低维场景更省。"},
{"type": "run.succeeded", "run_id": "r_d98c8f3c", "steps": 2},
]
for s in samples:
ev = adapter.validate_python(s)
print(f"{ev.type:14} -> {type(ev).__name__}")
本机实测（Python 3.14.6，pydantic 2.14.0）：
run.started -> RunStarted
step -> Step
tool_call -> ToolCall
tool_result -> ToolResult
token -> Token
final -> Final
run.succeeded -> RunSucceeded
--- 判别成功: 7 条
两侧对同一份样本给出了同一套判别结论——这正是「语言无关 schema」的含义：字段名、字段类型、判别值完全一致，只是表达方式各自贴合本语言的习惯（TS 用字面量联合，Python 用 Literal + discriminator）。
9.1.6 共享类型的三种做法
既然 schema 只有一份，怎么让两端的类型定义也不漂移？三种主流做法，代价各不相同：
| 做法 | 机制 | 优点 | 代价 |
|---|---|---|---|
| 手写双份 | TS type + Python Literal 各写一份 | 零工具链、可读性最好、类型是编译期类型 | 两端会漂移，加字段要改两处、漏一处不报错 |
| 从 JSON Schema 生成 | 单一 JSON Schema 源，生成两端类型 | 单一事实源，改一处两端同步 | 要引生成工具链；生成代码可读性差、难以手改 |
| 共享契约文件 | 一份 .json 事件契约，两侧启动时加载校验 | 无代码生成，运行时也校验 | 拿到的是运行时校验，不是编译期类型；IDE 不提示 |
选哪条取决于团队规模和事件数量。七种事件、两端、长期维护——本卷选手写双份：它的漂移风险可以用一个廉价手段压住，就是本节这份「同一份 JSON 样本在两侧都断言」的测试。样本一旦在某一端解不出来，CI 立刻红，漂移就暴露了。等事件类型涨到几十种、或引入第三种消费语言时，再切到「JSON Schema 生成」更划算。
9.1.7 真实环境 / 真实模型下会怎样
真实模型下 schema 的形状不变——type 还是那七个，字段还是那些。变的是演化压力：真实迭代里几乎一定会加事件（比如 citation、run.failed）。判别联合的纪律此时就是护栏：
- 加字段要向后兼容。新增可选字段（如给
final加citations）两端都得容忍缺失；decode里用默认值兜底，别让旧客户端因为多了个字段就崩。 - 加事件要两端同步。新增
type时，TS 的describe缺分支会被编译器逮住，Python 的discriminator会因「没有匹配的模型」抛校验错——两边都在第一时间报警，这是判别联合最值钱的地方。 - 判别值即契约。
type的字符串一旦发布就不能改名（改名等于破坏兼容），要演进只能新增类型、旧类型保留。这条和数据库迁移同一个道理。
本节验收清单：
- 七类事件都有
type判别字段，且取值与 3.2.3 的表一致 - TS 侧是判别联合，
switch无default也能穷尽 - Python 侧用
Field(discriminator="type")，不是暴力Union - 同一份 JSON 样本在两端都能解析，且结论一致
- TS 代码只用可擦除语法，
node x.ts与tsc --noEmit都过
9.1.8 小结
- 事件 schema = 一个
type判别字段 + 各变体专属字段，3.2 的event:名就是天然判别值。 - TS 用字面量联合表达判别联合，
switch自动收窄、漏分支会编译报错。 - Python 用 pydantic
Field(discriminator="type")，真判别、快且报错准。 decode是信任边界，必须显式构造字段，不能靠as强转糊过去。- 共享类型三选一：手写双份（本卷选它）/ JSON Schema 生成 / 共享契约文件，各自代价见 9.1.6 表。
协议与类型立住了，但事件最终要变成用户看得见的东西。下一节 9.2 前端渲染：TS 最小聊天 UI 逐步显示 就用这份判别联合写一个纯函数式渲染器。想回看事件是怎么被编成 SSE 帧推出来的，翻 3.2 流式接口：SSE 推送事件 。
阅读导航：上一节：8.3 取消、超时与泄漏：慢工具注入验证止损 · 下一节：9.2 前端渲染：TS 最小聊天 UI 逐步显示 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。