《AI Agent 工程实战》3.2 流式接口：SSE 推送事件

3.2 流式接口：SSE 推送事件

3.1 的 GET /runs/{id} 只能回答「跑完了没有」——用户提问后盯着转圈，直到整份报告生成完才看到第一行字。真实的研究助理不是这样工作的：它一边检索、一边调用工具、一边吐字，过程本身就是体验。这一节把 run 的执行过程接到 HTTP 上，用 SSE 把每个事件实时推给客户端。

本节把 ResearchCopilot 推进到「过程可见」：新增 researchcopilot/stream.py，用 StreamingResponse 以 text/event-stream 推送事件，事件复用卷一第 7 章的 token / tool_call / tool_result / step / final，并用 httpx 真消费一条完整流。

3.2.1 为什么 run 要流式返回

一次研究 run 动辄十几秒到几分钟：多跳检索、若干次工具调用、最后生成带引用的长报告。如果只给一个「一次性返回」的端点，客户端在整个过程中拿不到任何反馈，体验上等同于卡死。流式返回把这段时间切成许多小事件，每发生一件事就推一条，客户端立刻能渲染「正在检索语料…」「正在调用 search_corpus…」「正在生成…」。用户看到的延迟从「整个 run 的耗时」降到「第一个事件的耗时」。和 3.1 的轮询对比一下，差别很直观：

对 Agent 来说还有一层额外收益：工具调用是可见的。当客户端收到 tool_call 事件，就能把「Agent 正在查 Postgres」明确显示出来。这种可解释性在调试和信任建立上都值钱——用户知道 Agent 不是凭空编的，而是真的去查了。

3.2.2 事件 envelope：三行定契约

SSE 的帧由若干 字段: 值 行组成，帧与帧之间用一个空行分隔。我们把最常用的三个字段固定成 envelope：

三个字段各司其职：id 让客户端记住「我看到第几条了」，第 9.3 节的断线续传就靠它；event 决定这条消息怎么处理；data 是数据本身。载荷用 JSON 而不是裸文本是关键取舍——JSON 序列化后天然不含裸换行，而裸换行会被 SSE 当成帧边界，把一条消息劈成两帧。用 JSON，等于免费买到「任意内容都不会破坏帧结构」。

一条完整的帧，在传输层就长这样（注意结尾那个空行）：

id: 5
event: token
data: {"text": "pgvector"}

id、event、data 各占一行，最后用空行收尾——客户端 aiter_lines() 依次读到的正是这几行。

3.2.3 复用卷一的事件模型

事件类型不重新发明，直接对齐卷一第 7 章 agent.stream() 吐出的六种事件，再加上 run 自身的两个生命周期事件：

这张表就是运行时契约——3.1.6 说过 OpenAPI 描述不了它，只能靠文档钉住。前端（第 9 章）要按这张表写 switch (event)，服务端要按这张表发；任何一边加字段，另一边都得同步。把卷一的 Event 原样映射过来，而不是为 HTTP 重新设计一套，好处是同一套事件模型既服务于进程内调用，也服务于网络传输，引擎与服务层之间不需要翻译层。

3.2.4 把事件编成 SSE 帧

编码函数只有三行拼接，但有个必须记住的细节——帧尾两个换行：

import json


def sse_frame(seq: int, event: str, data: dict) -> str:
    payload = json.dumps(data, ensure_ascii=False)
    return f"id: {seq}\nevent: {event}\ndata: {payload}\n\n"

• 结尾必须是 \n\n：SSE 靠空行切帧，少一个换行，两帧就粘成一条，解析器读到的数据全乱。
• ensure_ascii=False：否则中文会被转成 \uXXXX 转义，合法但没法读；设成 False 后 data: 里是原样中文，调试肉眼可看。
• id 用递增序号：重连时客户端把最后收到的 id 放进 Last-Event-ID 头，服务端据此续传（3.2.8 与第 9.3 节）。

SSE 帧还有两个可选字段偶尔会用到：retry: 告诉客户端重连前等多少毫秒（不写就由浏览器定），: 开头的注释行（如 :\n\n）不含数据、只当心跳用。生产环境里客户端与服务端之间常隔着代理，长连接若长时间无数据会被掐断，靠定期发一个注释帧就能保活——第 9.3 节讲断线重连时会把心跳与 Last-Event-ID 一起展开。

3.2.5 端点：StreamingResponse 直接吃生成器

流式端点只有几行，核心是 StreamingResponse 接受一个可迭代对象，产一片发一片，而不是等全部算完：

from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel


class RunCreate(BaseModel):
    input: str


app = FastAPI(title="ResearchCopilot stream", version="0.3.0")


@app.post("/runs/{rid}/stream")
def stream_run(rid: str, body: RunCreate) -> StreamingResponse:
    if not rid.startswith("r_"):
        raise HTTPException(status_code=404, detail="run not found")
    return StreamingResponse(event_stream(rid), media_type="text/event-stream")

media_type="text/event-stream" 是 SSE 的约定类型，浏览器见到它就交给 EventSource 自动解析。event_stream(rid) 是个生成器，把 3.2.3 那张表逐条 yield 出来；真实实现里它消费的是 minagent 的 agent.stream()，演示里用一段脚本化事件代替，保证输出确定、可断言。

顺带说明一个实现细节：这里的 event_stream 是同步生成器，Starlette 会用线程池把它包起来异步消费（iterate_in_threadpool），所以它不会阻塞事件循环。要做成纯异步，把它写成 async def + async for 即可。

3.2.6 真跑：用 httpx 消费 SSE 流

消费端用 httpx 的异步客户端 + ASGITransport（进程内打 app，语义与真 HTTP 一致），关键 API 是 client.stream(...) 加 aiter_lines()——边收边读，不等整个响应结束：

import asyncio
import httpx


async def consume() -> None:
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        async with client.stream("POST", "/runs/r_d98c8f3c/stream",
                                 json={"input": "pgvector 召回率"}) as resp:
            print("status       :", resp.status_code)
            print("content-type :", resp.headers["content-type"])
            async for line in resp.aiter_lines():
                print(line)  # 空行 = 一帧结束


asyncio.run(consume())

用 httpx 0.28.1 打出来的真实事件流（原样，空行即帧边界）：

status       : 200
content-type : text/event-stream; charset=utf-8

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}

读这条流：run.started 先报到；接着 step + tool_call + tool_result 是一轮工具调用的完整轨迹（模型要查 search_corpus，工具返回「命中 3 条」）；中间四条 token 是打字机效果（pgvector / 在 / 低维 / 场景更省）；最后一个 step 与 final 收敛；run.succeeded 收尾并带上步数。每一帧都以 id / event / data 三行加空行结束——这正是 3.2.4 那个编码函数的落地。整条流里没有任何一处「等全部攒好」，首字节延迟被压到「第一个事件产生」为止。

再点两处细节。其一，data: 里的中文没有被转义成 \uXXXX，说明 ensure_ascii=False 生效了——如果哪天发现日志里全是 \u，先回来检查这个参数。其二，id 是连续递增的整数，客户端只要记下最后收到的 id，重连时就能让服务端从断点续发，不必从头再跑一遍。

3.2.7 SSE 与 WebSocket 的取舍

两者都能「服务器推消息」，但定位不同。Agent 场景九成时间只需要单向推送，这决定了默认选 SSE：

选型规则一句话：只要「客户端只是听」，就用 SSE。ResearchCopilot 的 run 输出正是这种形态——客户端发一次 POST /runs/{id}/stream，之后全是听。等到真需要「运行中让用户插话、实时打断工具」这种双向交互时（第 8 章的取消/干预），再考虑 WebSocket 或另开一个上行端点，不必为少数场景把整条链路升级成双向协议。

最后提醒一个容易忽略的点：SSE 走的是普通 HTTP，每一次重连都是一个全新请求。如果鉴权用的是短时效 token（第 4 章），要确保重连时能带上有效凭证，否则长连接一断就再也续不上，用户只能刷新页面从头开始。

3.2.8 真实环境 / 真实模型下会怎样

真实模型下，这条流的形状完全不变——帧还是那三行，事件还是那几类，前端还是那样 switch。变的只有两个地方：一是 token 会真的一个一个来，帧与帧之间可能有几百毫秒间隔；二是 run 的耗时更长，断线概率更高。因此生产化要补三样：

• 心跳：长时间没有数据时，定期发一个注释帧 :\n\n，防止中间的代理或负载均衡把「长时间无数据」的连接掐掉。
• 断线续传：客户端重连时带 Last-Event-ID，服务端从该序号之后继续发。这要求事件先落库或进 Redis（第 5、6 章），不能只活在内存里。
• 断线取消：客户端断开时服务端要能感知并停止生成，否则 Agent 还在烧 token 却没人收（第 8 章的超时取消）。

本节验收清单：

• 响应头是 text/event-stream
• 每帧由 id / event / data 三行组成，帧间有空行
• event 类型与 3.2.3 的表一一对应，无自造类型
• data 是单行合法 JSON，中文不被转义
• 消费端能边收边处理，不等响应结束

3.2.9 小结

• SSE 用普通 HTTP 做服务器单向推送，帧由 字段: 值 行 + 空行构成。
• envelope 固定为 id / event / data：id 管续传，event 管分发，data 是单行 JSON。
• 事件类型直接复用卷一第 7 章的模型，加 run.started / run.succeeded 两个生命周期事件。
• StreamingResponse 吃生成器，产一片发一片，全程无「攒好再发」。
• 单向推送默认选 SSE；只有需要客户端频繁上行时才上 WebSocket。

到这里 run 的过程能实时看见了。但客户端重试、分页、错误处理还没有统一说法——下一节 3.3 幂等、分页与统一错误契约 把这些「接口的边角」一次补齐。想回看资源建模，翻 3.1 REST 设计：session / run / message 资源建模 。

阅读导航：上一节：3.1 REST 设计：session / run / message 资源建模 · 下一节：3.3 幂等、分页与统一错误契约 。