Dev Spec v2 · ← 返回文档首页
02 模型网关（core/gateway）
1. 目的
| 问题 | 网关怎么解决 |
|---|---|
| 国内只能用已备案的模型 | 按区域路由 + 白名单硬校验 |
| 模型会停用、涨价、限流（例如 Gemini 2.5 Flash Image 已宣布 2026-10-02 停用） | 每个任务配置首选和备选模型，出错自动切换；改模型只改配置 |
| 需要知道每单、每个用户花了多少 AI 成本 | 每次调用计量并记到订单或用户上 |
| 防止费用失控 | 每个产品每天有预算上限 |
2. 接口
class Gateway:
def llm(self, task: str, messages: list[dict], *, schema: dict | None = None,
max_tokens: int = 1024, ctx: CallCtx) -> LLMResult: ...
def image_edit(self, task: str, images: list[bytes], prompt: str, *, n: int = 1,
size: str = "1024x1024", ctx: CallCtx) -> list[ImageResult]: ...
def image_generate(self, task: str, prompt: str, *, refs: list[bytes] = (), n: int = 1,
size: str = "1024x1024", ctx: CallCtx) -> list[ImageResult]: ...
def upscale(self, task: str, image: bytes, *, scale: int, ctx: CallCtx) -> ImageResult: ...
def tts(self, task: str, text: str, *, voice: str, fmt: str = "mp3", ctx: CallCtx) -> AudioResult: ...
def stt(self, task: str, audio: bytes, *, lang: str, word_timestamps: bool = True,
ctx: CallCtx) -> Transcript: ...
def vision_judge(self, task: str, images: list[bytes], rubric: str, *,
schema: dict, ctx: CallCtx) -> dict: ...
def pron_score(self, task: str, audio: bytes, text: str, *, lang: str, ctx: CallCtx) -> PronScore: ...
@dataclass
class CallCtx:
region: Literal["global", "cn"]
product: str
ref_kind: str | None = None # "order" | "song" | "episode" | "user"
ref_id: str | None = None
user_id: str | None = None
priority: Literal["paid", "free", "batch"] = "free"
schema用于要求结构化 JSON 输出；网关负责校验，不合格自动重试 1 次pron_score（发音评分）可以由语音识别厂商提供，也可以自部署；同样走网关- 发音评分的供应商风险：支持日语发音评分的厂商较少，国内是否有已备案、支持英语评分的服务也要确认。第 11 周前必须确认至少一家可用；同时实现备选算法：用语音识别的逐词置信度 + 识别结果与目标句的音素 / 假名编辑距离，算出 0~100 分（精度较低，只作兜底）
3. 配置
providers: # 每个厂商一个适配器
gemini: {kind: google, key_env: GEMINI_API_KEY}
fal: {kind: fal, key_env: FAL_KEY}
volc: {kind: volcengine_ark, key_env: VOLC_ARK_KEY}
elevenlabs:{kind: elevenlabs, key_env: ELEVEN_KEY}
local_asr: {kind: self_hosted, url_env: ASR_URL} # 自部署的 faster-whisper / FunASR
models: # 模型 = 厂商 + 模型名 + 单价
gemini-image-latest: {provider: gemini, name: "<以厂商文档为准>", price: {per_image: 0.039}}
volc-seedream: {provider: volc, name: "<以厂商文档为准>", price: {per_image_cny: 0.22}}
...
tasks:
pix.pet_portrait:
global: [gemini-image-latest, seedream-global]
cn: [volc-seedream]
timeout_s: 60
pix.judge:
global: [gemini-flash-vision]
cn: [doubao-vision]
snd.lyrics_asr:
global: [local_asr_whisper]
cn: [local_asr_funasr]
tt.script:
global: [strong-llm]
cn: [doubao-pro]
tt.checkpoint_reply:
global: [small-llm]
cn: [doubao-lite]
cn_allowlist: # 已备案 / 已登记的模型，同时用于前端公示
- {model: volc-seedream, display_name: "...", filing_no: "..."}
- {model: doubao-pro, display_name: "...", filing_no: "..."}
budgets: # 每天上限（美元）
pixorder: 50
sonicflow: 20
talktale: 40
模型名称、单价是示例。上线前按厂商当期文档填写；单价变动时更新配置，日报里的成本会自动跟着变。
4. 行为规则
| 规则 | 说明 |
|---|---|
| 区域白名单 | region=cn 时，候选模型必须在 cn_allowlist 中；否则抛异常（不是警告），并有单元测试覆盖 |
| 自部署模型 | 自部署的开源模型（例如语音识别）在国内使用时，同样要确认是否属于需要备案的"对外提供生成服务"。语音识别本身不生成内容；修音、合成语音需要按 04 文档 Q7 评估 |
| 顺序切换 | 首选模型超时、返回 5xx、限流或输出校验失败 → 换下一个；全部失败 → 抛出 GatewayUnavailable，由产品侧决定是退回额度还是进入异常队列 |
| 熔断 | 某个模型 5 分钟内错误率 > 20% → 熔断 10 分钟，并告警 |
| 优先级 | 预算接近上限时，free 和 batch 请求先被拒绝，paid 请求最后才受影响 |
| 缓存 | 以 (task, 模型, 输入哈希) 为键；01 的内容生产、02 的 SEO 样片大量受益；用户隐私数据（照片、录音）不缓存 |
| 数据 | 发送给厂商的数据遵守各厂商的"不用于训练"选项；用户照片和录音不写入日志 |
5. 计量表
CREATE TABLE core.ai_calls (
id BIGSERIAL PRIMARY KEY,
at TIMESTAMPTZ NOT NULL DEFAULT now(),
region TEXT NOT NULL, product TEXT NOT NULL, task TEXT NOT NULL,
model TEXT NOT NULL, provider TEXT NOT NULL,
units JSONB NOT NULL, -- {"input_tokens":..., "output_tokens":...} / {"images":4} / {"seconds":31}
cost_usd NUMERIC(10,5) NOT NULL,
latency_ms INT, ok BOOLEAN NOT NULL, error TEXT,
ref_kind TEXT, ref_id TEXT, user_id TEXT, fallback_from TEXT
);
CREATE INDEX ON core.ai_calls (product, at);
CREATE INDEX ON core.ai_calls (ref_kind, ref_id);
每个产品的"每单 / 每首 / 每集 AI 成本"都从这张表汇总，这是 PRD 成本指标的唯一数据来源。
6. 测试
- 每个厂商适配器用录制好的响应做单元测试
cn白名单的硬校验有专门的测试（尝试调用白名单之外的模型必须失败）- 每周运行一次模型回归测试：对每个任务用固定的输入跑一遍，比较质量分（视觉打分、文本评测）和成本，发现退化就告警