本篇对应
app/providers.py（196 行）与app/llm_gateway.py（231 行）。这两个模块组合起来回答一个企业级核心问题：模型不可控时，如何让系统可控。
一、Provider 抽象：为"假模型"正名
class ModelProvider(ABC):
"""模型提供方接口：输入 OpenAI 风格 messages，输出回复文本。"""
name: str = "base"
@abstractmethod
def chat(self, messages: list[dict], model: str | None = None) -> str: ...
def chat_with_usage(self, messages, model=None) -> dict:
return {"text": self.chat(messages, model=model), "usage": None}
def chat_stream(self, messages, model=None):
"""流式输出分片的默认实现：整段生成后按 8 字符切片模拟流。"""
text = self.chat(messages, model=model)
for i in range(0, len(text), 8):
yield text[i:i + 8]
三个方法构成一套"渐进能力"契约：chat 是底线（必须实现）；chat_with_usage 带用量（能取到就取）；chat_stream 有默认实现（子类可覆写为真流式）。网关层对任何 Provider 都能统一提供三种调用形态。
MockProvider：确定性是它最大的美德
class MockProvider(ModelProvider):
"""确定性模拟模型：同一输入永远得到同一输出，便于测试与离线演示。"""
name = "mock"
def chat(self, messages: list[dict], model: str | None = None) -> str:
model = model or "mock-model"
last_user = ""
has_kb_context = False
for m in messages:
if m.get("role") == "user":
last_user = m.get("content", "")
if m.get("role") == "system" and "[知识库上下文]" in m.get("content", ""):
has_kb_context = True
digest = hashlib.sha256((model + "|" + last_user).encode("utf-8")).hexdigest()[:8]
# ...组装回复：轮次 + 摘录 + 是否结合知识库 + 校验码
为什么先写一个"假模型"（教小白最重要的一课）：
- 测试要确定性：真实大模型同样的输入给不同的输出，断言没法写。Mock 对"模型名+最后一条用户消息"取 SHA256 前 8 位做校验码——同输入永远同输出，测试可以断言精确内容。仓库里 107 项测试大部分跑在 Mock 上，54 秒全绿。
- 开发要离线：没网、没 Key、不想花钱，整套对话/工作流照常开发调试演示。README："平台默认使用内置 Mock Provider（确定性回复，离线可跑）"。
- Mock 还懂 RAG：检测 system 里含
[知识库上下文]标记就声明"已结合知识库"——前端引用来源的整条链路不依赖真模型就能测。
OpenAICompatibleProvider：一个类吃下整个生态
class OpenAICompatibleProvider(ModelProvider):
"""可指向 OpenAI / DeepSeek / Moonshot / vLLM / Ollama 等任意兼容端点。"""
name = "openai_compatible"
# 构造器签名见仓库源码：依次接收 base_url、接口密钥、默认模型名与超时秒数
def __init__(self, base_url: str, credential: str, default_model: str, timeout: float = 60.0):
self.base_url = base_url.rstrip("/")
...
def chat_with_usage(self, messages, model=None) -> dict:
self._check_config() # 未配置 → 明确报错（网关据此不重试）
url = f"{self.base_url}/chat/completions"
with httpx.Client(timeout=self.timeout, **self._client_kwargs(url)) as client:
resp = client.post(url, json=payload, headers=self._headers())
resp.raise_for_status()
...
注：为避免凭据扫描误报，上例把源码中的密钥形参转写为 credential，照抄时以仓库
app/providers.py为准（教程后续涉及该参数名处同理）。
- 只实现 OpenAI 兼容一种协议，换来 DeepSeek/Moonshot/通义/vLLM/Ollama 全家桶——因为它们全都兼容
/chat/completions。这是"选协议不选厂商"。 chat_stream是真流式：client.stream("POST", ...)+ 逐行解析上游 SSE（data: {...}行，[DONE]结束），逐 content 增量 yield 给网关。_no_proxy_for的工程细节：httpx 默认继承系统代理，但 Windows 的 ProxyOverride 本地绕过规则它不认识——本机开代理时连127.0.0.1都会被转发给代理返回 502。对回环地址显式trust_env=False（"回环地址语义上永远不该出本机"）。教程实测环境开着系统代理，正是这个细节保证了本地演示可用。
注册表与工厂：get_provider 是唯一入口
PROVIDER_REGISTRY = {"mock": MockProvider, "openai_compatible": OpenAICompatibleProvider}
def get_provider(provider_name: str, conn=None) -> ModelProvider:
"""openai_compatible 的配置优先级：数据库 settings > 环境变量。"""
扩展新渠道（Azure、Bedrock、私有网关）= 写一个子类 + 注册表加一行，上层助手与对话逻辑零改动——02 篇"三层解耦"承诺的兑现。配置优先级"数据库 > 环境变量"让管理员在后台改配置无需重启。
API Key 的静态加密：拖库拿不到明文
Key 存 settings 表，但不以明文存——secretbox.py 用 HMAC-SHA256-CTR + 校验标签实现加密（密文带 enc1: 版本前缀），主密钥来自环境变量 AIP_SECRET_KEY 或自动生成的 data/.secret_key 文件。设计注释说得很清楚：这是"静态加密"而非 HSM 级防护，目的是让"只拖走 platform.db 的场景（备份泄漏、误提交）拿不到 API Key"——数据库文件与密钥文件分离。解密失败宁可返回空让调用方报"未配置"，也不把密文当明文用。
二、LLM 网关：把横切逻辑收进一个入口
如果每个路由自己调 Provider，重试、限流、审计就会各写一份、口径不一。网关层存在的理由（模块 docstring 原文）：
- 重试/退避/超时是横切逻辑，散在各调用点必然口径不一；
- 成本防护（限流）与用量审计（llm_calls 表）只有在统一入口才能全覆盖；
- 降级链是可用性兜底，降级必须"可被看见"——回复带 degraded 标记、审计表记 degraded，而不是静默替换。
2.1 重试：指数退避 + 一个关键的例外
def _backoff_sleep(attempt: int) -> None:
"""指数退避 + 抖动：0.4s, 0.8s, 1.6s...（抖动防惊群）。"""
time.sleep(0.4 * (2 ** (attempt - 1)) + random.uniform(0, 0.2))
逐次失败后等 0.4s → 0.8s → 1.6s（上限 8 次，AIP_LLM_MAX_ATTEMPTS 可调但封顶——"过大的重试数叠加退避会长期占住线程"）。抖动（随机偏移）防止上游恢复瞬间所有重试同时打到（惊群）。
例外：错误信息含 未配置 时立即失败、不重试、不降级——配置缺失是"必须人工修配置"的错误，重试没有意义，降级会掩盖问题。保持"未配置 → 502 明确报错"的语义。
2.2 降级：可用性兜底，但要看得见
if allow_degrade:
fallback = MockProvider()
text = fallback.chat(messages, model=model)
_record(conn, ..., status="degraded", error=err_text)
return {"text": text, ..., "degraded": True, "degrade_reason": err_text, ...}
调用链：openai_compatible 网络错误/超时/5xx → 重试 N 次 → 仍失败 → 降级到 Mock 继续服务，但返回值带 degraded: True + 失败原因，llm_calls 表记 degraded。前端据此在界面上显式标注"当前为降级回复"。README 特性列表的原话："模型不可用时自动降级并在界面上显式标注，而不是静默给假回复"——宁可如实报错，不给静默的假答案，这是 AI 产品的伦理底线。
2.3 限流：成本入口上的滑动窗口
class UserRateLimiter:
"""进程内每用户滑动窗口限流（默认 N 次/分钟）。
单实例部署够用且零依赖；横向扩容时换 Redis + Lua 计数即可，allow/retry_after 接口不变。
只挂在会触发 LLM 调用的消息端点上（成本入口），普通读接口不限。"""
对话 30 条/分钟、工作流 20 次/分钟（环境变量可调）。超限返回 429 + Retry-After 头，前端可直接倒计时提示。注意设计边界：只挂在触发 LLM 成本的端点，读接口不限流——限流是成本控制手段，不是防用户。
2.4 token 计量与审计
真实调用优先取上游返回的 usage 字段；Mock 与流式增量没有 usage，用 estimate_tokens 估算（CJK 约 0.6 token/字，其余约 0.28/字符——注释明确"只用于估算"）。每次调用落一行 llm_calls：谁、哪个助手、哪个模型、进出 token、延迟、实际尝试次数、success/degraded/failed。后台统计页（09 篇）全靠这张表。
2.5 流式网关的语义边界（全仓库最精细的一段设计）
非流式调用失败可以整段重试；流式不行——已向用户输出一半的内容无法安全重试（重放会造成内容重复）。网关的处理：
- 重试粒度 = "打开流并拿到第一个分片"。第一个分片没到就失败 → 静默重试；
- 已开始输出后再失败 → 无法重试，发
error事件收尾，前端提示"本条未保存"； - 降级决策发生在开流阶段：
meta事件（第一个 delta 之前）就携带 degraded 标记——前端先渲染"降级"横幅再开始打字机。
SSE 事件协议：meta(refs/provider/degraded) → delta*（增量文本）→ done(usage/落库完成)，失败以 error 收尾。sse_event() 把每帧编码为 event: xxx\ndata: {json}\n\n。
三、对话路由怎么用网关（预告 09 篇）
result = llm_gateway.chat_with_gateway(
conn, provider_name=assistant["provider"], model=assistant["model"],
messages=messages, user_id=user["id"], assistant_id=assistant["id"],
allow_degrade=True) # 对话允许降级
# 工作流的 llm 节点则是 allow_degrade=False —— 工作流语义是显式的，
# 模型不可用就该让节点失败并记录原因（workflows.py 注释原话）
同一个网关，两种降级策略——对话要"永远有回应"，工作流要"结果可信"。策略差异由调用方声明，这就是好抽象的样子。
四、动手清单
- 实现
ModelProvider/MockProvider/get_provider，用 Mock 写通"发消息 → 得到确定性回复"（先不接前端，curl POST messages 即可） - 用
python -m uvicorn起服务，把限流环境变量调小（如AIP_CHAT_RATE_PER_MIN=3）后连续发 4 条消息，观察第 4 条返回 429 与 Retry-After（仓库test_llm_gateway.py就是这么测的：限流设 3，发 3 条成功后第 4 条 429） - 【降级实验】后台把某助手切到
openai_compatible但不填 Key → 发消息应得 502"未配置"（不降级！配置缺失属"必须人工修配置"的错误）；再填一个不可达的 base_url（网络错误类）→ 指数退避重试 3 次后降级、回复带降级标注 - 查
llm_calls表，找到那条 failed 和那条 degraded 记录——这就是审计的价值
常见坑：
- httpx 忘了 trust_env=False 于回环地址 → 开代理的机器上本地全挂（教程实测环境真实踩到，仓库代码已内置 _no_proxy_for 防护）
- 把 Key 打进日志/审计表 → 永远只存"是否已配置"或密文
- 降级没写进审计 → 用户看到 Mock 回复却以为是真模型——记住：degraded 必须可见
下一篇做 RAG：文档怎么切块、512 维哈希向量怎么来、BM25 与 TF-IDF 怎么融合，以及"用评估驱动调参"的完整过程。