wsk/zhuangang

0 Fork 0

Files

T

Shikui Wang 7f52513ffc 阶段3（上）：Python/asyncio/FastAPI 专题的课程、企业场景、面试题和可运行示例

2026-10-23 15:44:13 +08:00

5.6 KiB

Raw Permalink Blame History

id, title, type, domain, stack, tags, level, prerequisites, est_hours, related, week

title

type

domain

stack

大模型对话 API：鉴权、统一错误、SSE 流式、客户端断开即停止

业务背景与问题

公司要把大模型能力封装成内部统一的对话服务，给客服系统、OA、知识库等多个业务调用，而不是每个业务各自直连模型服务商。要求：

各业务系统用自己的 API Key 调用，按租户统计用量；
支持流式（打字机效果）和非流式两种模式；
错误格式统一，调用方能程序化处理；
出了问题能用请求 ID 串起调用方和服务端的日志；
未来可以切换模型服务商而不影响调用方。

约束

第一版只有两三个人开发，要求简单可靠。
模型调用按 token 计费，用户关闭页面后不应继续生成。
模型 API Key 只保存在这个服务里，不下发给业务系统。

方案设计

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）

分层：路由层（校验、鉴权、响应格式）→ 服务层（组装提示词、计量）→ 模型客户端（LLMClient 协议，可替换服务商）。
鉴权：current_tenant 依赖，API Key → 租户；后续计量、限流都按租户。
统一错误：HTTPException、RequestValidationError、兜底异常都转成 {code, message, request_id}；兜底异常只记日志不返回细节。
请求 ID：中间件读取或生成 X-Request-ID，写进响应头和日志。
流式：SSE；中途出错发送 event: error；客户端断开立即停止。
可测试：模型客户端通过依赖注入，测试中替换为脚本化的假实现。

关键代码

完整代码和测试见 examples/chat-api：

@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 应用网关场景）。
只在内存里计量：多 worker / 多实例数据不准，要写数据库或 Redis。

面试怎么讲

“我做了一个内部统一的对话服务，FastAPI 实现。鉴权用依赖注入把 API Key 映射到租户，便于计量和限流；所有错误转成统一的 code/message/request_id 格式，内部异常只记日志；流式用 SSE，中途出错发 error 事件，客户端断开就停止生成以节省 token；模型客户端抽象成协议，测试里用假实现替换，覆盖了鉴权、校验、流式和异常场景。”

5.6 KiB Raw Permalink Blame History Unescape Escape