字节笔记本
2026年10月24日 · 约 12 分钟读完
WebSocket 流式的三层架构
本文是《从 0 写 Agent》系列第 36 章。这个系列基于开源项目 agents-go（OpenAI Agents SDK 的 Go 移植版），用 Go 从零实现一个生产级 Agent 运行时；本章讲「WebSocket 流式：三层架构」。
前端要看到 Agent 实时打字、实时显示工具调用、实时收到审批请求，这需要 WebSocket 流式推送。
agents-server 把"SDK 流式事件 → 前端 UI 更新"拆成三层，连接层、分发层、运行层。这一章拆这个端到端的流式架构。
为什么是三层
SDK 的流式输出（第 6 章）是 Go channel 的事件流。但前端在浏览器里，要通过 WebSocket 接收。中间要做，
- 连接管理，WebSocket 鉴权、并发写安全、断线处理
- 事件分发，不同事件类型路由到不同处理（run 事件、审批事件、取消事件）
- 运行桥接，把 SDK 事件翻译成前端能理解的 protocol 事件
三层各自职责清晰，这是复杂流式系统的必要拆分。
第一层：连接层（server/ws.go）
WebSocket 连接的建立和鉴权：
// server/ws.go（概念）
upgrader := websocket.Upgrader{
CheckOrigin: checkOrigin, // 校验 origin 防 CSRF
}
func HandleWSWithAuth(conn, token) {
// 强制第一条消息是 auth
conn.ReadJSON(&msg)
if msg.Type != "auth" || !constantTimeCompare(msg.Token, token) {
conn.Close()
return
}
conn.WriteJSON({Type: "auth.ok"}) // 鉴权成功
// 进入业务循环
}关键，WebSocket 升级后，HTTP 头鉴权失效，要在 WS 应用层重做 token 鉴权。 第一条消息必须是 auth，constant-time 比对防时序攻击。
WSConn 的写锁
type WSConn struct {
*websocket.Conn
mu sync.Mutex // 并发写锁
}
func (w *WSConn) WriteJSON(v any) error {
w.mu.Lock()
defer w.mu.Unlock()
return w.Conn.WriteJSON(v)
}WebSocket 连接不支持并发写，多个 goroutine 同时 WriteJSON 会损坏帧。mu 保证串行写。这是 WebSocket 编程的必备防护。
第二层：分发层（handler/ws_handler.go）
WSHandler 的 Handle（ws_handler.go:69）是个消息循环：
// ws_handler.go:69（骨架）
func (h *WSHandler) Handle(conn *WSConn) {
for {
var msg protocol.Envelope
conn.ReadJSON(&msg)
switch msg.Type {
case "run.create":
go h.handleRunCreate(msg) // 每个 run 一个 goroutine
case "tool.approve":
h.handleApprove(msg) // HITL 审批
case "tool.reject":
h.handleReject(msg)
case "run.cancel":
h.runner.CancelRun(msg.RunID)
}
}
}四种 client→server 消息，创建 run、审批、拒绝、取消。每种有自己的处理。
wsSink，事件出口
// ws_handler.go:62
wsSink := func(env *protocol.Envelope) {
conn.WriteJSON(env) // 把事件推给前端
}wsSink 是个闭包，把"往这个连接写 JSON"封装成一个 EventSink 函数，传给 bridge 层。bridge 层产生事件时调 sink，事件就到了前端。
pending map，HITL 暂存
// ws_handler.go:31
pending map[string]*pendingRun // runID → 被审批中断的 run 状态当 run 被工具审批中断，SDK 的 RunState 存进 pending map。 用户审批后，从 pending 取出，ResumeStreamed。第 37 章详讲。
第三层：运行层（bridge/runner.go）
RunStreamed（runner.go:53）是真正的流式引擎：
// runner.go:53（骨架）
func (r *Runner) RunStreamed(input, agentConfig, sink) {
// 1. 构建 Agent（从 DB 配置）
agent := r.BuildFullAgent(agentConfig)
// 2. 构建 provider（可能多路由）
provider := r.BuildRouterProvider(agentConfig)
// 3. 起 SDK 流式 Run
sr := agents.RunStreamed(ctx, agent, input, opts)
// 4. 逐事件消费，翻译成 protocol 事件，推给前端
for event, err := range sr.Events() {
r.handleStreamEvent(event, sink) // 翻译+推送
}
}事件翻译
handleStreamEvent（runner.go:361）把 SDK 事件翻译成前端 protocol 事件：
| SDK 事件 | protocol 事件 | 前端怎么显示 |
|---|---|---|
RawResponsesStreamEvent（output_text.delta） | run.step（带 delta） | 打字效果 |
RunItemStreamEvent（tool_called） | run.tool_call | "调用工具 X" |
RunItemStreamEvent（tool_output） | run.tool_result | 显示结果 |
RunItemStreamEvent（handoff） | run.handoff | "切换到 Agent Y" |
AgentUpdatedStreamEvent | run.agent_start | 新 Agent 接管 |
| Interruptions（需审批） | run.tool_call（NeedsApproval:true） | 弹审批框 |
关键，response.output_text.delta 是打字效果的唯一来源（runner.go:416 的 extractDelta）。SDK 透传的原始事件里，只有这个类型是"又吐了一个字"。
额外亮点：自动生成标题
// runner.go:289
func (r *Runner) maybeGenerateTitle(session, firstMessage) {
// 用一个小 agent（title_gen，max_turns=1）给会话起标题
title := agents.Run(ctx, titleAgent, firstMessage, ...)
sink(&protocol.Envelope{Type: "session.title_updated", Title: title})
}首条消息后，自动给会话生成标题（类似 ChatGPT 的"New Chat"变成实际标题）。通过事件推回前端更新侧边栏。
三层协作的完整流程
前端发 run.create
→ 分发层：起 goroutine
→ 运行层：BuildFullAgent → RunStreamed
→ SDK 跑循环，产出事件流
→ 运行层：handleStreamEvent 翻译
→ wsSink → 连接层：WriteJSON（持锁）
→ 前端收到 run.step（打字）/ run.tool_call / ...前端根据 protocol 事件的 type 渲染 UI，收到 run.step 追加文字，收到 run.tool_call 显示工具调用卡片，收到审批请求弹框。
小结
读完这一章，你应该理解 WebSocket 流式的三层：
- 连接层，WS 鉴权（首消息 auth）、并发写锁（mu）、CheckOrigin
- 分发层，消息循环 + 四种消息处理 + pending map（HITL 暂存）+ wsSink 闭包
- 运行层，SDK RunStreamed + 事件翻译（SDK→protocol）+ 自动标题
这是"SDK 事件流 → 浏览器 UI"的完整端到端管道。 下一章讲这条管道里最复杂的部分，HITL 审批。
下一篇
第 37 章，HITL 审批的端到端实现。当 SDK 遇到危险工具暂停，agents-server 怎么把"暂停状态"推给前端，用户审批后怎么恢复？pending map + RunState 序列化 + ResumeStreamed 的完整流程。
本文是《从 0 写 Agent》系列第 36 章，案例代码基于开源项目 agents-go（OpenAI Agents SDK 的 Go 移植版）。全系列目录见总览。
