---
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；模型客户端抽象成协议，测试里用假实现替换，覆盖了鉴权、校验、流式和异常场景。”