跳转至内容

通过 SSE 流式传输响应

本演练讲解将追踪一条聊天响应从 FastAPI 路由器到浏览器客户端的完整流程。您将了解精确的 SSE 传输格式、StreamingResponse 的使用方式,以及 Python 技术栈为何有意保持与 .NET API 相同的契约。

API 入口点

app/routers/chat.py 处理 POST /api/chat/stream。它接收一个 ChatRequest,选择请求的模型或默认的 claude-haiku-4.5,并返回 FastAPI 的 StreamingResponse

@router.post("/stream")
async def stream_chat(request: Request, body: ChatRequest) -> StreamingResponse:
return StreamingResponse(
    event_stream(),
    media_type="text/event-stream",
    headers={"Cache-Control": "no-cache", "Connection": "keep-alive"},
)

这些头信息告知中间件和浏览器,此为长时连接流式传输,而非需缓冲至完成的普通 JSON 响应。

数据流格式

对于每个来自 CopilotChatService.chat_stream 的分块,路由器会序列化一个小型 JSON 对象并写入一条 SSE 消息:

yield f"data: {json.dumps({'content': chunk})}\n\n"

当流式传输正常结束时,端点会写入终止符:

yield "data: [DONE]\n\n"

若在流式传输启动后抛出异常,路由器会在同一 SSE 数据通道中写入错误事件:

yield f"data: {json.dumps({'error': str(ex)})}\n\n"

此时已无法可靠地切换为 HTTP 错误状态,因为状态码和响应头已经发送。报告后期故障的唯一有效方式,是将错误写入流式传输载荷。

有意保持 .NET 兼容性

传输契约与 .NET API 一致:使用以 data: {...}\n\n 表示的数据帧,并以 data: [DONE]\n\n 结束。Python 的 json.dumps 与 .NET 的 JsonSerializer 生成的 JSON 空白可能不同,但 content/error 载荷结构和结束标记完全相同。由于 FastAPI 在同一进程中同时提供 API 和 UI,Python 页面使用端口 5070

请求体别名

ChatRequest 使用 Pydantic 的 camelCase 别名生成器:

model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)

prompt: str | None = None
model: str | None = None
system_message: str | None = None

这保持了 Python 代码的惯用性(system_message),同时保留了 .NET 客户端和文档中使用的 HTTP 合约(systemMessage)。

断开连接处理

在生成器内部,路由器会在写入下一帧前检查浏览器是否已关闭:

async for chunk in service.chat_stream(prompt, model, body.system_message):
    if await request.is_disconnected():
        break
    yield f"data: {json.dumps({'content': chunk})}\n\n"

如果标签页关闭或请求被放弃,API 即可停止写入并结束流式传输处理。

缓冲响应与健康检查

该路由器还提供 POST /api/chat 端点以返回缓冲响应,并提供无需 SDK 传输的 GET /api/chat/health 健康检查端点:

response = await _service(request).chat(body.prompt or "", model, body.system_message)
return ChatResponse(content=response, model=model)

实际的健康检查输出:

{"status":"healthy","service":"CopilotChat","availableModels":["claude-haiku-4.5","gpt-4.1","gpt-5","claude-sonnet-4.5","claude-opus-4.5","gemini-2.5-pro"]}

使用 curl 测试

将请求发送到端口 5070。使用 curl -sN,避免 curl 缓冲响应:

$ curl -sN -X POST http://localhost:5070/api/chat/stream \
    -H 'Content-Type: application/json' \
    -d '{"prompt":"Reply with exactly: streaming works","model":"claude-haiku-4.5"}'

data: {"content": "streaming works"}

data: [DONE]

⚠️ 该字段为prompt,而非message。若此处存在未识别的键,则会被忽略,因此此处的拼写错误会静默地发送一个空提示,模型将回复通用问候语而非报错。

助手文本的具体内容取决于模型和账户状态,但帧结构是关键部分。较长的回答仅会在哨兵出现前以更多data:帧的形式到达。

浏览器客户端

静态 UI 在 app/static/app.js 中读取相同的 data: 数据帧。它使用 fetch()res.body.getReader()TextDecoder;完整渲染流程详见 Web 界面