113 lines
5.6 KiB
Markdown
113 lines
5.6 KiB
Markdown
---
|
||
id: python.fastapi.scenario.streaming-chat-api
|
||
title: 大模型对话 API：鉴权、统一错误、SSE 流式、客户端断开即停止
|
||
type: scenario
|
||
domain: python
|
||
stack: [fastapi, asyncio, nginx]
|
||
tags: [对话 API, SSE, 鉴权, 错误处理, 可测试性]
|
||
level: intermediate
|
||
prerequisites: [python.fastapi.streaming-testing]
|
||
est_hours: 2
|
||
related: [python.fastapi.interview, devops.nginx.scenario.ai-gateway-sse-canary]
|
||
week: [5, 12]
|
||
---
|
||
|
||
# 大模型对话 API：鉴权、统一错误、SSE 流式、客户端断开即停止
|
||
|
||
## 业务背景与问题
|
||
|
||
公司要把大模型能力封装成内部统一的对话服务，给客服系统、OA、知识库等多个业务调用，而不是每个业务各自直连模型服务商。要求：
|
||
|
||
- 各业务系统用自己的 API Key 调用，按租户统计用量；
|
||
- 支持流式（打字机效果）和非流式两种模式；
|
||
- 错误格式统一，调用方能程序化处理；
|
||
- 出了问题能用请求 ID 串起调用方和服务端的日志；
|
||
- 未来可以切换模型服务商而不影响调用方。
|
||
|
||
## 约束
|
||
|
||
- 第一版只有两三个人开发，要求简单可靠。
|
||
- 模型调用按 token 计费，用户关闭页面后不应继续生成。
|
||
- 模型 API Key 只保存在这个服务里，不下发给业务系统。
|
||
|
||
## 方案设计
|
||
|
||
```mermaid
|
||
sequenceDiagram
|
||
participant B as 业务系统
|
||
participant N as Nginx
|
||
participant A as 对话服务（FastAPI）
|
||
participant L as 模型服务商
|
||
B->>N: POST /v1/chat（X-API-Key, X-Request-ID）
|
||
N->>A: 转发（关闭缓冲）
|
||
A->>A: 依赖：校验 API Key → 租户；pydantic 校验参数
|
||
A->>L: 流式请求
|
||
loop 每个 token
|
||
L-->>A: delta
|
||
A-->>B: data: {"delta": ...}
|
||
A->>A: 客户端断开？→ 停止
|
||
end
|
||
A-->>B: data: [DONE]（或 event: error）
|
||
```
|
||
|
||
1. **分层**：路由层（校验、鉴权、响应格式）→ 服务层（组装提示词、计量）→ 模型客户端（`LLMClient` 协议，可替换服务商）。
|
||
2. **鉴权**：`current_tenant` 依赖，API Key → 租户；后续计量、限流都按租户。
|
||
3. **统一错误**：`HTTPException`、`RequestValidationError`、兜底异常都转成 `{code, message, request_id}`；兜底异常只记日志不返回细节。
|
||
4. **请求 ID**：中间件读取或生成 `X-Request-ID`，写进响应头和日志。
|
||
5. **流式**：SSE；中途出错发送 `event: error`；客户端断开立即停止。
|
||
6. **可测试**：模型客户端通过依赖注入，测试中替换为脚本化的假实现。
|
||
|
||
## 关键代码
|
||
|
||
完整代码和测试见 [examples/chat-api](../examples/chat-api/)：
|
||
|
||
```python
|
||
@app.post("/v1/chat", response_model=None)
|
||
async def chat(req: ChatRequest, request: Request,
|
||
tenant: Annotated[str, Depends(current_tenant)],
|
||
llm: Annotated[LLMClient, Depends(get_llm)]):
|
||
rid = request.state.request_id
|
||
if not req.stream:
|
||
content = "".join([t async for t in llm.stream(req.messages, req.temperature)])
|
||
return ChatResponse(request_id=rid, content=content)
|
||
|
||
async def events():
|
||
try:
|
||
async for token in llm.stream(req.messages, req.temperature):
|
||
if await request.is_disconnected(): # 用户走了，停止生成
|
||
return
|
||
yield f"data: {json.dumps({'delta': token}, ensure_ascii=False)}\n\n"
|
||
yield "data: [DONE]\n\n"
|
||
except Exception:
|
||
log.exception("生成失败 request_id=%s", rid)
|
||
yield f"event: error\ndata: {json.dumps({'message': '生成失败', 'request_id': rid}, ensure_ascii=False)}\n\n"
|
||
|
||
return StreamingResponse(events(), media_type="text/event-stream",
|
||
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})
|
||
```
|
||
|
||
测试覆盖：未带 API Key 返回 401 统一格式；参数错误返回 422 并列出 `messages.0.role` 等字段；非流式返回完整内容并透传请求 ID；流式事件顺序正确并以 `[DONE]` 结束；模型中途失败时发送 `error` 事件；内部异常不泄露连接串。
|
||
|
||
## 取舍
|
||
|
||
| 选择 | 优点 | 缺点 |
|
||
| --- | --- | --- |
|
||
| **SSE（本方案）** | 基于 HTTP，简单，代理和网关支持好 | 单向；`EventSource` 不支持 POST，前端要用 fetch 解析 |
|
||
| WebSocket | 双向，适合语音、实时协作 | 连接管理和负载均衡更复杂 |
|
||
| 统一对话服务（本方案） | 密钥集中、可计量、可切换服务商 | 多一跳；需要保证高可用 |
|
||
| 业务直连模型服务商 | 少一跳 | 密钥分散、无法统一计量和治理 |
|
||
| 直接使用 LangChain 等框架封装 | 功能多 | 对一个转发 + 治理的服务来说偏重，版本变化快 |
|
||
|
||
## 易错点
|
||
|
||
- 流式中途异常想改状态码：响应头已发出，只能发错误事件。
|
||
- 不检查断开：页面关了模型还在生成。
|
||
- 统一错误处理器漏掉 `RequestValidationError`：422 的格式和其他错误不一致。
|
||
- 兜底异常返回 `str(e)`：可能包含连接串、内部地址。
|
||
- Nginx 没关缓冲：流式变整段（见 [AI 应用网关场景](../../../devops/nginx/scenarios/ai-gateway-sse-canary.md)）。
|
||
- 只在内存里计量：多 worker / 多实例数据不准，要写数据库或 Redis。
|
||
|
||
## 面试怎么讲
|
||
|
||
> “我做了一个内部统一的对话服务，FastAPI 实现。鉴权用依赖注入把 API Key 映射到租户，便于计量和限流；所有错误转成统一的 code/message/request_id 格式，内部异常只记日志；流式用 SSE，中途出错发 error 事件，客户端断开就停止生成以节省 token；模型客户端抽象成协议，测试里用假实现替换，覆盖了鉴权、校验、流式和异常场景。”
|