这篇文章是《从零开发一个 Coding Agent》系列第四篇。聚焦于如何在大模型事件流中使用状态机进行顺序校验，解决使用者常见的事件顺序错误和协议不一致问题。

使用者痛点这方面。事件顺序错误导致 Agent 崩溃

当 Provider 推送的事件缺少必要的 starttext_start 或出现顺序混乱时Agent 会因未处理的非法事件而抛错，导致整个对话流程中断。TypeScript 只能检查单个对象字段，却无法记住前面已收到哪些事件。无法捕捉“内容块开始后直接出现结束”或“工具调用参数不完整”等细节错误。

EventStream 的局限性

EventStream 仅负责数据运输，不关注事件出现的合法性。至于例如，

{ type: "text_delta",contentIndex:。delta: "你好" }

如果它前面没有出现 start 和 text_start它在协议上仍然是非法事件。其实，为此，需要一个有记忆的校验器——状态机。

校验流程示意

Provider -> EventStream.push -> StreamEventValidator.accept -> Agent
合法事件进入队列；非法事件抛出 StreamSequenceError,并让生产者、异步迭代器和 result 同时得到同一错误。

定义状态机与阶段错误

状态机主要概念：

• IDLE： 尚未收到 start
• DYNAMIC： 已开始。可以接收内容块或终止事件
• TERMINAL： 已收到 done/Error”，不能再接收任何事件。
• CURRENT BLOCK： 正在拼接的内容块
• NEXT INDEX： 下一个内容块必须使用该索引且连续递增。话说回来，
• Deltas 与 End 的一致性：
• Deltas 必须属于当前活动块；End 内容必须等于所有 Deltas 的累积结果；完整结束后才能开始下一个索引。

实现文件结构

import type { AssistantContent,StreamEvent } from "../types.ts";type StreamPhase = "idle" | "streaming" | "terminal";type ActiveBlock =
| { kind: "text";contentIndex: number;value: string }
| { kind: "thinking";话说回来，contentIndex: number;value: string }
| {
再看kind，"tool_call";contentIndex: number;id: string，怎么说呢，name: string;argumentsJson: string;老实说，sawArgumentsDelta: boolean;}，interface SequenceState {
再看phase，StreamPhase;nextContentIndex: number;按理说，activeBlock?: ActiveBlock;completedContent: AssistantContent;}
...

可识别错误类型：StreamSequenceError

export class StreamSequenceError extends Error {
readonly eventType: StreamEvent;readonly phase: StreamPhase;constructor(eventType: StreamEvent。phase:
StreamPhase,detail:string){
super;this.name = "StreamSequenceError";不过，this.eventType = eventType;this.phase = phase;}
export interface StreamEventValidator{
accept这方面，void
}

再看辅助函数。统一报错与索引检查

function reject:never{
throw new StreamSequenceError;}
function assertStartIndex:void{
if{
reject(event,state。`expected contentIndex ${state.nextContentIndex},received ${received}`);}
}
...

JSON 数据校验工具函数

type JsonValue =
null|boolean|number|string|JsonValue|{
:JsonValue
};function isJsonValue:value is JsonValue{...}
function isJsonObject:value is Record{...}
function jsonDeepEqual:boolean{...}
...

创建状态机实例：createStreamEventValidator

export function createStreamEventValidator:StreamEventValidator{
const state:{phase:"idle"。nextContentIndex:number,completedContent:Array}={
再看phase，"idle",nextContentIndex:,completedContent:
};return {
accept{
if{
if reject;state.phase="streaming";return，}
if{
reject;按理说，}
// 全局规则已处理完毕后再进入 switch
switch {
case "start":reject;break，...
}
}
};}

文本 & 思考内容块处理

case 'textdelta': case 'thinkingdelta':{ const expectedKind=event.type==='text_delta'?'text':'thinking';const active=state.activeBlock;if reject，if reject;active.value+=event.delta;return，其实，}

case 'textend': case 'thinkingend':{ const expectedKind=event.type==='text_end'? 'text':'thinking';const active=state.activeBlock;if reject，if reject;if reject，state.completedContent.push( expectedKind==='text'?{type:'text',text: active.value}:{type:'thinking'。thinking: active.value});按理说，state.activeBlock=undefined;state.nextContentIndex++;return，} 主要要点 - 必须先检查类型、索引与完整内容，再写入完成列表并清空活跃块。按理说，- 一个非法 end 不会污染后续状态。

⚠️ 使用者痛点：忘记 content 与 delta 一致性检查导致最终消息不匹配，引发调试困难。

工具调用块处理

case 'toolcallstart':{ if reject;assertStartIndex;老实说，state.activeBlock={ kind这方面，'tool_call',contentIndex:event.contentIndex。id:event.id,name:event.name,argumentsJson:'',sawArgumentsDelta:false};return，}

case 'toolcalldelta':{ const active=state.activeBlock!，if reject;不过，if reject;active.sawArgumentsDelta=true;// 标记已收到 delta active.argumentsJson+=event.argumentsDelta;return，}

case 'toolcallend':{ const active=state.activeBlock!，if reject;if ... // 一样做索引检查

let parsed!话说回来，:Record;if{ try{parsed=JSON.parse;}catch{reject;} if) reject }else{ if) reject parsed=event.toolCall.arguments as any }

// ID & Name 必须保持不变 if reject ... state.completedContent.push({ type这方面。'tool_call',id :active.id,name :active.name,arguments :parsed});state.activeBlock=undefined;state.nextContentIdx++;return，}

⚠️ 使用者痛点：工具参数分段传输导致 JSON 拼接错误；若未正确合并，则会被判为非法，从而导致整个流失效。

终止与异常处理

function getExpectedTerminalContent{…}

case 'done':{ if reject;if reject，if) reject;state.phase='terminal';return，}

case 'error':{ if ... const safePartial=getExpectedTerminalContent;// 若有未结束工具则忽略 if ) reject;老实说，state.phase='terminal';return，} default: reject

⚠️ 使用者痛点：error 时仍携带半截 tool JSON 被误认为有效，导致 downstream 调用失败。

集成到 EventStream 中

import { createStreamEventValidator } from "./validation.ts";

export type AssistantMessageEventStream = EventStream;

export function createAssistantMessageEventStream: AssistantMessageEventStream { const validator=createStreamEventValidator;return new EventStream({ validate{validator.accept;},isTerminal{return e.type==='done'||e.type==='error';话说回来，}，getResult{ if return e.message;throw new Error;}} )，说起来，}

⚠️ 使用者痛点：若将 validator 放在模块顶层。多条并发请求共享同一状态会相互干扰，导致后续流被误拒绝。

公开入口与维护策略

`packages/ai/src/index.ts` 导出以下 API：

ts export type { AssistantMessageEventStream } from "./utils/event-stream.ts";其实，export { createAssistantMessageEventStream } from "./utils/event-stream.ts";

export type { StreamEventValidator } from "./utils/validation.ts";export { createStreamEventValidator,StreamSequenceError } from "./utils/validation.ts";

⚠️ 使用者痛点：内部实现细节如 ActiveBlockSequenceState 等不应暴露，否则升级可能破坏依赖代码。

建立与格式化脚本

powershell

Set-Location D:\pi\di-code

npx --no-install biome check --write packages\ai\src\index.ts packages\ai\src\utils\event-stream.ts packages\ai\src\utils\validation.ts

npm run check && npm run build --workspace @di-code/ai

⚠️ 使用者痛点：格式化工具缺失或版本冲突会导致建立失败，需要统一团队依赖版本。

与接下来计划

* 状态机实现了对整条流阶段、活跃块、下标还有完成内容的一致性校验，解决了使用者常见的 “缺少 start”、“块类型错配”、"索引跳跃"、“工具参数不完整”、还有 “终止后继续推送” 等问题。*​ * 这篇文章仅验证统一协议与 JSON 边界。下一篇将进一步加入针对具体工具 schema 的参数校验。*​ * 所有公共入口已安全导出。可放心升级，而内部实现细节保持私有以免破坏兼容性。*​ # eventstream-validation 分支地址 如果你对 Agent 开发感兴趣，欢迎点赞收藏+关注！话说回来，再看专栏，《从零开发一个 Coding Agent》— 东方小月 — 掘金📌📝.

。

---