把 9 个 AI 模型串成一个统一网关: ai-showcase 架构实战

背景

我积累了 9 个独立部署的 AI 模型服务 (文本/代码/视觉/ASR/TTS/RAG/Math/视频/通用 LLM), 每个端口、协议、错误格式都不一样。

业务方想做一个客服 demo, 需要同时调 RAG + LLM, 每个 model 都要单独写客户端, 重复代码 800+ 行。

痛点:

• 接 1 个新场景 = 2 天
• 错误处理五花八门
• 演示时 9 个服务任意一个挂了, demo 就翻车
• 没有统一指标, 不知道哪个场景慢 / 哪个模型贵

设计目标

做一个 统一网关层 (ai-showcase):

• 业务方只对一个端点发请求
• 网关负责路由 / 容错 / A-B 测试 / 指标埋点
• 业务方不需要知道底层是 9 个还是 90 个模型

核心设计

1. 场景化 API

不暴露 POST /v1/chat/completions, 而是 POST /api/scenarios/{客服|代码|图片|语音|视频}/...

业务方用业务术语调用, 不关心是哪个模型。

# 业务方调用 (1 行, 跟 SaaS 一样)
r = await client.post("/api/demo/run", json={"scenario": "code_assistant"})

2. call_with_fallback 通用调用器

所有场景都走同一个 helper。流程 = 探测 → real → 失败 mock → 记录指标。

async def call_with_fallback(url, json_body, service_name, mock_fn, mock_args):
    t0 = time.time()
    if MOCK_MODE:
        try:
            # 健康探测
            async with httpx.AsyncClient(timeout=2) as c:
                base = url.split("/v1")[0].split("/api")[0]
                hr = await c.get(f"{base}/health", timeout=1.5)
                if hr.status_code != 200: raise RuntimeError(...)
            # 真实调用
            async with httpx.AsyncClient(timeout=30) as c:
                r = await c.post(url, json=json_body)
                r.raise_for_status()
                return {"data": r.json(), "mode": "real", "latency_ms": int((time.time()-t0)*1000)}
        except Exception:
            MOCK_FALLBACK_COUNT.labels(service=service_name).inc()
            txt, tokens = mock_fn(**mock_args)
            return {"data": {"text": txt, "tokens": tokens}, "mode": "mock", "latency_ms": ...}

核心 18 行 替代了 800 行重复代码。

3. Mock 兜底动态化

mock 不写死, 而是从 expected_keywords 动态构造回复。

class MockBackend:
    @staticmethod
    def code(task, language, is_finetuned, kws):
        # 微调版: 中文注释 + 完整结构 + 命中 expected 关键词
        # 基座版: 简洁, 关键词少
        ...

这样 A/B 测试离线也能跑 (基座 28.5% vs 微调 98.0% 提升 69.5pp), 不会因为底层服务挂了 demo 翻车。

4. Prometheus 4 类指标

REQUEST_COUNT = Counter("showcase_requests_total", "Total requests", ["scenario", "status", "mode"])
REQUEST_LATENCY = Histogram("showcase_request_duration_seconds", "Request latency", ["scenario"])
TOKENS_TOTAL = Counter("showcase_tokens_total", "Total tokens", ["model", "scenario"])
COST_TOTAL = Counter("showcase_cost_usd_total", "Total cost in USD", ["model"])
MOCK_FALLBACK_COUNT = Counter("showcase_mock_fallback_total", "Times fell back to mock", ["service"])

后端 10 行定义, 0 侵入接入所有场景。

5. 结构化 JSON 日志 + Trace ID

@app.middleware("http")
async def trace_id_middleware(request, call_next):
    tid = request.headers.get("X-Request-ID") or uuid.uuid4().hex[:12]
    token = _trace_var.set(tid)
    ...
    response.headers["X-Request-ID"] = tid

每个请求 12 位 trace_id, ELK/Loki 检索一键定位。

效果

写在最后

“1 套代码串 9 模型” 的关键不是写多少代码, 而是 抽象出统一的调用模式 + Mock 兜底 + 可观测性。

面试经常被问”如何把多个 AI 服务统一管理”, 答这一套足够。

完整代码

• ai-showcase 仓库
• 5 业务场景评估报告
• K8s 部署清单