事件契约

事件契约是定义 AI Agent 应用中各层之间事件格式、类型和语义的协议规范。它是 React 前端、Go 后端和 Python AI Service 之间的通信契约。

#type / concept #status / evergreen #tech / ai #tech / architecture

[!info] related notes

事件契约

一句话定义

事件契约是定义 AI Agent 应用中各层之间事件格式、类型和语义的协议规范。它确保 Python AI Service 产生的事件能被 Go 后端正确转发、被 React 前端正确消费。

它解决什么问题

三层架构中，事件从 Python → Go → React 逐层传递。如果没有统一的契约：

  • Python 输出 {"type": "text", "content": "..."}, Go 期望 {"event": "message", "data": "..."}
  • 前端不知道有哪些事件类型，怎么渲染
  • 新增事件类型时，三层都要改，没有参照标准
  • 出了问题不知道是哪层的问题

事件契约让三层有统一的”语言”。

核心原理

事件类型体系

事件类型分层:

1. 文本事件
   - text_delta: 文本增量（流式输出）
   - text_done: 文本完成

2. 工具调用事件
   - tool_call_start: 工具调用开始
   - tool_call_delta: 工具参数增量
   - tool_call_result: 工具执行结果

3. 状态事件
   - progress: 进度更新
   - state_patch: 状态补丁
   - interrupt: 需要人类介入

4. 控制事件
   - error: 错误
   - done: 完成
   - heartbeat: 心跳

Event Envelope 结构

interface AgentEvent {
  // 事件类型
  event: "text_delta" | "tool_call_start" | "tool_call_result" |
         "progress" | "interrupt" | "error" | "done" | "heartbeat";

  // 事件数据（类型由 event 字段决定）
  data: TextDeltaData | ToolCallStartData | ...;

  // 元数据
  meta: {
    run_id: string;       // Run 唯一标识
    step_id: string;      // Step 唯一标识
    timestamp: number;    // 事件时间戳
    sequence: number;     // 事件序号（用于排序和去重）
  };
}

各事件类型的 data 结构

// 文本增量
interface TextDeltaData {
  delta: string;           // 增量文本
  accumulated?: string;    // 累积文本（可选）
}

// 工具调用开始
interface ToolCallStartData {
  tool_call_id: string;
  tool_name: string;
  arguments?: object;      // 完整参数（非流式时）
}

// 工具调用结果
interface ToolCallResultData {
  tool_call_id: string;
  tool_name: string;
  result: any;
  error?: string;
  duration_ms: number;
}

// 进度
interface ProgressData {
  message: string;
  percent?: number;
  step: number;
  total_steps?: number;
}

// 中断（需要人类审批）
interface InterruptData {
  interrupt_id: string;
  type: "approval_required" | "info_needed";
  message: string;
  tool_call?: ToolCallStartData;
  options: string[];
}

// 错误
interface ErrorData {
  code: string;
  message: string;
  recoverable: boolean;
}

// 完成
interface DoneData {
  total_tokens: number;
  total_duration_ms: number;
  steps: number;
}

在 React + Go + Python AI Service 架构中的位置

Python AI Service:
  产生 AgentEvent
       │
       ▼
Go 后端 (SSE Gateway):
  验证事件格式 → 附加元数据 → 转发
       │
       ▼
React 前端 (Event Reducer):
  解析事件 → 更新状态 → 渲染 UI

Go 后端的转发逻辑

func sseProxy(w http.ResponseWriter, r *http.Request) {
    // 1. 调用 Python AI Service
    resp, _ := http.Post(aiServiceURL+"/chat", body)
    defer resp.Body.Close()

    // 2. 设置 SSE 响应头
    w.Header().Set("Content-Type", "text/event-stream")
    w.Header().Set("Cache-Control", "no-cache")

    // 3. 逐行转发事件
    scanner := bufio.NewScanner(resp.Body)
    for scanner.Scan() {
        line := scanner.Text()

        // 4. 验证事件格式
        event, err := parseEvent(line)
        if err != nil {
            continue // 跳过无效事件
        }

        // 5. 附加元数据
        event.Meta.SessionID = sessionID
        event.Meta.Timestamp = time.Now()

        // 6. 持久化（可选）
        persistEvent(event)

        // 7. 转发到前端
        fmt.Fprintf(w, "event: %s\ndata: %s\n\n", event.Type, event.JSON())
        w.(http.Flusher).Flush()
    }
}

React 前端的事件消费

// Event Reducer
function eventReducer(state: ChatState, event: AgentEvent): ChatState {
  switch (event.event) {
    case "text_delta":
      return { ...state, text: state.text + event.data.delta };

    case "tool_call_start":
      return {
        ...state,
        toolCalls: [...state.toolCalls, {
          id: event.data.tool_call_id,
          name: event.data.tool_name,
          status: "running",
        }],
      };

    case "tool_call_result":
      return {
        ...state,
        toolCalls: state.toolCalls.map(tc =>
          tc.id === event.data.tool_call_id
            ? { ...tc, status: "done", result: event.data.result }
            : tc
        ),
      };

    case "interrupt":
      return { ...state, interrupt: event.data, status: "waiting_approval" };

    case "error":
      return { ...state, error: event.data, status: "error" };

    case "done":
      return { ...state, status: "done" };

    default:
      return state;
  }
}

常见设计模式

1. 事件版本化

在事件中加入 version 字段，支持协议升级时的向后兼容。

2. 事件序号

每个事件带递增的 sequence，前端可以检测丢包和乱序。

3. 心跳保活

定期发送 heartbeat 事件，防止 SSE 连接被中间代理断开。

4. 事件压缩

对于大量 text_delta 事件，可以批量发送以减少 HTTP 请求次数。

常见坑

  1. 没有统一契约: 三层各自定义事件格式，对接困难
  2. 事件类型不够用: 新增功能时发现现有事件类型无法表达
  3. 不做事件验证: Python 输出了格式错误的事件，Go 转发后前端崩溃
  4. 不处理乱序: 网络抖动导致事件到达顺序与发送顺序不同
  5. 不处理丢包: SSE 断连后丢失了中间事件

和其他概念的关系

  • vs 流式协议: 流式协议是传输层，事件契约是应用层
  • vs SSE: SSE 是传输载体，事件契约是传输内容的定义
  • vs Event Envelope: Envelope 是事件的包装格式
  • vs Event Reducer: Reducer 是前端消费事件的逻辑

总结

事件契约是三层架构的通信标准。没有契约，三层就像说不同语言的人，每次对接都要翻译。

参考资料

创建于 2026/6/30 更新于 2026/7/15