第 1 章讲义 · 05 LLM Gateway（统一模型调用服务）

配套代码：整个 src/anna/server/ + src/anna/providers/，入口 main.py

学完本节，你应该能回答：Gateway 解决什么问题？密钥为什么集中在网关？ 请求/响应各在哪层校验？模板、错误、用量为什么都要"统一"？

目录#

1. FastAPI LLM 服务：统一入口与密钥管理
2. 模型调用接口：封装 OpenAI Compatible
3. Streaming 输出：网关代理逐块转发
4. Structured Output：屏蔽底层差异
5. Pydantic 数据校验：入口与出口
6. Prompt 模板：网关统一管理
7. 错误处理与重试：限流、超时、fallback
8. Token / Cost / Latency 日志

1. FastAPI LLM 服务：统一入口与密钥管理#

Gateway 解决的第一件事：各组件不需要各自维护 API 密钥和厂商配置。 业务代码只调一个 HTTP 入口，密钥、端点、模型清单全部集中在网关的配置层 （src/anna/config.py，环境变量驱动）：

export DASHSCOPE_API_KEY=...            # 网关统一持有
export ANNA_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1

业务组件永远不接触厂商 SDK 和密钥，只发请求：

import httpx
resp = httpx.post(
    "http://127.0.0.1:8000/v1/chat/completions",
    json={"model": "auto", "messages": [{"role": "user", "content": "hi"}]},
)

这带来三个收益：密钥泄露面缩小（只有网关持有）、厂商切换对业务无感、 调用行为（限流/用量/错误）全部可观测。入口是 FastAPI 应用（main.py + create_app），/docs 自带 Swagger 调试。

2. 模型调用接口：封装 OpenAI Compatible#

对外暴露 OpenAI 兼容协议（POST /v1/chat/completions），对内用 ChatProvider 统一接口 + adapter 屏蔽厂商差异（讲义 01 §6）：

业务代码 → POST /v1/chat/completions（OpenAI 格式）
              → ChatProvider.complete/stream（统一接口）
                  → OpenAI 兼容 adapter（透传）
                  → Anthropic adapter（协议转换）
                  → Gemini adapter（协议转换）

GET /v1/models 返回可用模型；model 字段支持 auto / 厂商:模型 / 纯模型名三种路由写法（router.py）。收益：厂商升级协议、新增厂商， 业务代码一行不改。

3. Streaming 输出：网关代理逐块转发#

Gateway 代理流式响应是两段式（讲义 02 §3）：上游→网关用 httpx.Client.stream 逐行读（base.py _stream_sse），网关→客户端用 async StreamingResponse 逐块写（app.py _stream_response）。客户端断开时关闭 上游连接停止生成（讲义 02 §5）。

return StreamingResponse(
    generate(),
    media_type="text/event-stream",
    headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
)

首页对比面板可直接验证：流式"首字节耗时"远小于非流式"总耗时"。

4. Structured Output：屏蔽底层差异#

上层只需传一份 JSON Schema，网关负责：

1. 把 response_format 按厂商转换（OpenAI 透传 / Gemini 映射 / Anthropic 隐藏工具，讲义 04 §4）；
2. 返回后做 Schema 校验、纠错重试、失败降级（讲义 04 §5-7，structured.py）。

result = structured_completion(
    provider, messages, json_schema,
    max_attempts=3,
    fallback=default_card,
)

业务层完全不感知"这个模型是 Gemini 还是 Anthropic"，只拿到校验过的 Python 字典/对象——这就是 Gateway 的"屏蔽差异"价值。

5. Pydantic 数据校验：入口与出口#

请求入口：所有 /v1/* 请求体用 Pydantic 模型声明 （server/schemas.py），FastAPI 自动完成反序列化 + 校验，非法请求直接 422：

class ChatRequest(BaseModel):
    model: str = "auto"
    messages: List[ChatMessage]        # role/content 必填
    temperature: Optional[float] = None
    stream: bool = False
    response_format: Optional[dict] = None

响应出口：普通对话组装成 OpenAI 兼容响应（openai_format.py）； 结构化输出在 structured.py 用 jsonschema 校验后才返回。两层拦截的意义： 入口拦住"请求格式错"，出口拦住"模型输出脏"。

诚实边界：普通 chat 响应的出口目前是"组装"而不是"严格校验"（模型输出 文本本来就是自由格式）；结构化输出才是严格校验路径。

6. Prompt 模板：网关统一管理#

模板库由网关管理（prompts.py + /v1/prompts），上层只传变量：

curl -s http://127.0.0.1:8000/v1/prompts/summarize/render \
  -H 'Content-Type: application/json' \
  -d '{"variables":{"passage":"..."}}'

网关还提供版本历史与回滚（讲义 03 §8）、注入隔离变量（讲义 03 §7）。 收益：Prompt 变更走模板版本而不是改业务代码，可审计、可回滚。

7. 错误处理与重试：限流、超时、fallback#

统一错误处理分三层（讲义 01 §5）：

allowed, retry_after = limiter.check(_client_key(request))
if not allowed:
    raise HTTPException(429, detail=error_payload(...), headers={"Retry-After": ...})

业务代码不需要自己写重试——网关统一做，且错误分类（鉴权/限流/超时/额度）是 可回退与否的判据（classify_provider_error）。

8. Token / Cost / Latency 日志#

每次调用记录一条 UsageRecord（usage.py）：provider/model、输入输出 token、 耗时、估算成本、状态与错误。可选落 JSONL（ANNA_USAGE_LOG），提供聚合与明细：

这是成本治理的数据底座：哪家模型贵、哪个接口慢、错误率多高，全部可查。 首页"用量统计"面板直接展示。

小结#

• Gateway = 统一入口 + 集中密钥 + 屏蔽厂商差异 + 统一可观测；
• 协议统一：对外 OpenAI Compatible，对内 ChatProvider 接口；
• 流式两段打通：上游 client.stream + 网关 StreamingResponse；
• 结构化输出：Schema 传入 + 厂商转换 + 校验/重试/降级全在网关；
• 校验两层：入口 Pydantic 拦请求，出口 Schema 拦模型输出；
• 模板、错误、用量全部"统一"：可复用、可回滚、可治理；
• 每个模块都有测试：tests/ 下 95 个用例覆盖。

动手练习：

1. 在首页分别演示：对话、流式对比、结构化输出、模板渲染、用量统计；
2. 把 ANNA_RATE_LIMIT_RPM 调小，连打几条看 429 + Retry-After；
3. 配一个失效模型观察自动 fallback（或直接用 demo:boom 看错误链路）；
4. 发几条请求后打开用量面板，对照 Token/成本/延迟明细。

附录：讲义内容 × 项目实现对照#

图例：✅ 已实现并有测试 / ⬜ 未实现（讲义中已注明或属后续阶段）/ 📘 通用知识或示例说明。