> **本文是「架构实战·第二季」第六篇**，AI 系列基础篇最后一篇。从这一篇开始，我们进入"用 LLM 一定会踩到的工程问题"——网关层。

---

## 引子：一条粗心配置带来的事故

某电商公司大促第一天，凌晨 3 点，告警响起：

```
[CRITICAL] LLM 服务异常：客服 AI 全部报错
```

值班同事爬上服务器一看——OpenAI 的 API key 写到代码里，**前两天换供应商时没同步更新**，调用 OpenAI 走了已经欠费的账户。

开始补救：手动改配置、写自动化脚本，搞了两小时。

**第二天复盘，老板就一句话**："为什么没有一层'上游挂了自动切下游'的机制？"

这就是 LLM 网关要做的事。

---

## 一、为什么需要 LLM 网关

### 1.1 直接调 LLM API 的风险

假设你有 5 个业务（客服、营销、数据分析、AI 客服、AI 编程助手），都直接调用 LLM API：

```
业务1 ───┐
业务2 ───┤
业务3 ───┼──▶ OpenAI / Claude / ...
业务4 ───┤
业务5 ───┘
```

直接调用的问题：

| 问题 | 表现 |
|------|------|
| 供应商切换 | 改 5 个服务的代码 |
| 限流配置 | 每个服务各写一遍 |
| 内容审查 | 散落各处，难统一 |
| 成本统计 | 各算各账，无法汇总 |
| 监控日志 | 格式不一致 |
| 降级策略 | 出问题各自为政 |

### 1.2 网关介入后的形态

```
业务1 ───┐
业务2 ───┤
业务3 ───┼──▶ [LLM 网关] ──▶ OpenAI / Claude / 自建 ...
业务4 ───┤
业务5 ───┘
```

**LLM 网关的核心价值**：

1. **统一入口**：业务方只关心"调哪个模型"，不关心上游；
2. **路由抽象**：换模型不动业务代码；
3. **统一治理**：限流、成本、安全可集中配置；
4. **可观测**：所有调用在一个地方打点；
5. **降级保护**：上游挂了自动切备用。

---

## 二、LLM 网关的核心能力

### 2.1 能力全景

| 能力 | 关键点 |
|------|--------|
| 协议适配 | 兼容 OpenAI API / Anthropic / 私有协议 |
| 多模型路由 | 按 header / path / 业务方分配 |
| 流量染色 | 给请求打标，方便追踪 |
| A/B 测试 | 按权重分流到不同模型版本 |
| 灰度发布 | 1% → 10% → 50% → 100% |
| 限流 | 按业务方 / 用户 / Token / QPS |
| 熔断 | 上游错误率超阈值自动熔断 |
| 降级 | 主模型失败降级到备用模型 |
| 重试 | 指数退避 + 抖动，避免雪崩 |
| 内容安全 | 输入/输出敏感词过滤 |
| 计费埋点 | 按 token / 业务方分账 |
| 全链路 Trace | 单次请求跨多个上游都能追踪 |
| 缓存 | 完全相同请求复用响应 |
| Prompt 模板托管 | 统一管理 prompt 版本 |
| Function Calling 路由 | 把工具调用路由到具体后端 |

### 2.2 关键技术决策

**协议层：用 OpenAI 兼容协议作为统一规范**

不管你用 OpenAI、Claude、还是自建模型，都让它们"看起来"像 OpenAI API：

```
业务代码 → OpenAI 协议 → 网关 → 各模型供应商
```

代价是 OpenAI 协议不完整，需要在网关做协议转换。**绝大多数 LLM 网关都默认兼容 OpenAI**，因为生态最广。

---

## 三、主流 LLM 网关对比

### 3.1 选型对比

| 网关 | 形态 | 优势 | 劣势 |
|------|------|------|------|
| **LiteLLM** | 开源 Python | 多模型支持最广，路由能力强 | 性能一般 |
| **Portkey** | 商业 SaaS | UI 友好，可观测完善 | 私有化不便 |
| **Cloudflare AI Gateway** | 云服务 | 集成 CDN，价格便宜 | 自定义能力弱 |
| **OpenRouter** | 路由服务商 | 一站式多模型 | 数据出境问题 |
| **Kong AI Gateway** | 商业产品 | 复用 Kong 体系 | 复杂 |
| **自研** | - | 完全可控 | 工作量大 |

**中小团队首选 LiteLLM**——开源、轻量、功能齐全。

### 3.2 LiteLLM 部署

```yaml
# config.yaml
model_list:
  # 主模型：GPT-4o
  - model_name: gpt-4o
    litellm_params:
      model: openai/gpt-4o
      api_key: os.environ/OPENAI_API_KEY

  # 备用 1：Claude
  - model_name: claude-3.5-sonnet
    litellm_params:
      model: anthropic/claude-3-5-sonnet-20241022
      api_key: os.environ/ANTHROPIC_API_KEY

  # 备用 2：自建 Qwen
  - model_name: qwen-72b-local
    litellm_params:
      model: openai/Qwen/Qwen2.5-72B-Instruct
      api_base: http://vllm-server:8000/v1
      api_key: "EMPTY"

  # 别名路由：把 'smart' 路由到 GPT-4o，'fast' 路由到 gpt-4o-mini
  - model_name: smart
    litellm_params:
      model: openai/gpt-4o

  - model_name: fast
    litellm_params:
      model: openai/gpt-4o-mini

# 路由策略
router_settings:
  enable_dynamic_loadbalancing: true
  routing_strategy: simple-shuffle

# 熔断
litellm_settings:
  drop_params: true
  request_timeout: 60
  num_retries: 3
  allowed_fails: 5
  cooldown_time: 30

# 通用
general_settings:
  master_key: os.environ/LITELLM_MASTER_KEY
  database_url: "postgresql://litellm:xxx@postgres/litellm"

# 回调：监控埋点
litellm.callbacks = ["prometheus", "langfuse"]
```

启动：

```bash
docker run -d --name litellm \
  -p 4000:4000 \
  -v $(pwd)/config.yaml:/app/config.yaml \
  -e OPENAI_API_KEY=sk-xxx \
  -e ANTHROPIC_API_KEY=sk-ant-xxx \
  -e LITELLM_MASTER_KEY=sk-litellm-master \
  -e DATABASE_URL=postgresql://litellm:xxx@host.docker.internal:5432/litellm \
  ghcr.io/berriai/litellm:main-stable \
  --config /app/config.yaml --port 4000
```

调用：

```bash
curl http://localhost:4000/v1/chat/completions \
  -H "Authorization: Bearer sk-litellm-master" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "smart",
    "messages": [{"role": "user", "content": "你好"}]
  }'
```

业务代码始终只认 `smart` / `fast` 这样的语义化别名，要切换上游不需要改业务代码，只改网关配置。

---

## 四、降级策略

LLM 服务的稳定性永远是不可信的——上游厂商也会挂，自己的服务更别说了。

### 4.1 降级矩阵

| 主模型 | 失败时降级到 |
|--------|------------|
| GPT-4o | Claude 3.5 → Qwen-72B 自建 |
| Claude 3.5 | GPT-4o → Qwen-72B 自建 |
| 自建 Qwen | GPT-4o-mini（价格兜底） |

**降级原则**：

- 永远保留一个"价格兜底"——任何情况下都要保证业务可用；
- 降级链不要太长，3 级足够；
- 降级时要打标（`degraded: true`），方便事后分析。

### 4.2 LiteLLM 配置降级

```yaml
model_list:
  # 主模型
  - model_name: gpt-4o
    litellm_params:
      model: openai/gpt-4o

  # 降级 1
  - model_name: gpt-4o
    litellm_params:
      model: anthropic/claude-3-5-sonnet
      api_key: os.environ/ANTHROPIC_API_KEY

  # 降级 2
  - model_name: gpt-4o
    litellm_params:
      model: openai/Qwen/Qwen2.5-72B-Instruct
      api_base: http://vllm-fallback:8000/v1
```

LiteLLM 会自动按优先级降级。

### 4.3 自研降级逻辑

对于复杂场景，建议在业务层实现：

```python
class LLMGateway:
    def __init__(self):
        self.fallback_chain = [
            ("gpt-4o", self.call_openai),
            ("claude-3.5-sonnet", self.call_anthropic),
            ("qwen-72b-local", self.call_local),
        ]

    async def call(self, messages, **kwargs):
        last_error = None
        for model_name, fn in self.fallback_chain:
            try:
                response = await fn(messages, **kwargs)
                if model_name != self.fallback_chain[0][0]:
                    logger.warning(f"降级到 {model_name}: {last_error}")
                    metrics.counter("llm.degraded", tags={"model": model_name}).inc()
                return response, model_name
            except Exception as e:
                last_error = e
                metrics.counter("llm.failure", tags={"model": model_name}).inc()
                continue
        raise AllModelsFailed(last_error)
```

---

## 五、限流与成本控制

### 5.1 多维限流

**Token 级限流**（最关键）：

```python
class TokenRateLimiter:
    def __init__(self, tpm: int, window_sec: int = 60):
        self.tpm = tpm  # 每分钟 token 数
        self.window_sec = window_sec
        self.buckets = {}  # {user_id: deque}

    def check(self, user_id: str, tokens: int) -> bool:
        now = time.time()
        bucket = self.buckets.setdefault(user_id, deque())

        # 清掉窗口外的记录
        while bucket and bucket[0][0] < now - self.window_sec:
            bucket.popleft()

        # 当前窗口已用 token
        used = sum(t for _, t in bucket)
        if used + tokens > self.tpm:
            return False

        bucket.append((now, tokens))
        return True
```

**请求级限流**（防止恶意调用）：

```python
# 用 Redis + Lua 脚本实现
def request_rate_limit(user_id: str, qps: int = 10) -> bool:
    lua = """
    local cur = redis.call('INCR', KEYS[1])
    if cur == 1 then
      redis.call('EXPIRE', KEYS[1], 1)
    end
    if cur > tonumber(ARGV[1]) then
      return 0
    end
    return 1
    """
    return redis.eval(lua, 1, f"rl:{user_id}", qps)
```

### 5.2 成本埋点

每次调用都要打点：

```python
def cost_tracking(model: str, input_tokens: int, output_tokens: int,
                  user_id: str, business: str):
    cost = calculate_cost(model, input_tokens, output_tokens)
    metrics.counter("llm.cost", tags={"model": model, "business": business}).inc(cost)
    metrics.histogram("llm.input_tokens", tags={"model": model}).observe(input_tokens)
    metrics.histogram("llm.output_tokens", tags={"model": model}).observe(output_tokens)
    
    # 记明细到日志（用于账单）
    logger.info(json.dumps({
        "ts": now(),
        "user_id": user_id,
        "business": business,
        "model": model,
        "input_tokens": input_tokens,
        "output_tokens": output_tokens,
        "cost_usd": cost,
    }))
```

成本计算参考：

```python
PRICING = {
    # 单位：USD / 1M tokens
    "gpt-4o": {"input": 2.5, "output": 10.0},
    "gpt-4o-mini": {"input": 0.15, "output": 0.6},
    "claude-3.5-sonnet": {"input": 3.0, "output": 15.0},
    "qwen-72b-local": {"input": 0.0, "output": 0.0},  # 自建不计外部成本
}

def calculate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
    p = PRICING.get(model, {"input": 0, "output": 0})
    return (
        input_tokens / 1_000_000 * p["input"]
        + output_tokens / 1_000_000 * p["output"]
    )
```

### 5.3 预算控制

按用户/业务设置日预算，超额自动降级或拒绝：

```python
async def check_budget(user_id: str, estimated_cost: float) -> bool:
    used = await redis.get(f"budget:{user_id}:{today()}")
    used = float(used or 0)
    
    budget = get_user_budget(user_id)
    if used + estimated_cost > budget:
        # 降级到便宜模型
        return False  # 或者返回降级 token
    return True
```

---

## 六、内容安全

### 6.1 双侧过滤

LLM 内容安全必须在**输入侧 + 输出侧**双重过滤：

```python
INPUT_BLOCKLIST = [...]   # 违规关键词
OUTPUT_BLOCKLIST = [...]  # 敏感输出词

def content_filter(text: str, side: str = "input") -> str:
    blacklist = INPUT_BLOCKLIST if side == "input" else OUTPUT_BLOCKLIST
    for block in blacklist:
        if block in text:
            logger.warning(f"内容拦截: side={side}, pattern={block}")
            metrics.counter(f"content.blocked.{side}").inc()
            if side == "input":
                return "抱歉，我只能回答合规问题。"
            else:
                return ""
    return text

async def safe_llm_call(messages):
    # 输入过滤
    for msg in messages:
        msg.content = content_filter(msg.content, "input")
    
    # LLM 调用
    response = await llm.call(messages)
    
    # 输出过滤
    response = content_filter(response, "output")
    return response
```

### 6.2 LLM-as-a-judge 双保险

关键词过滤容易被绕过，对关键场景加 LLM 二次审查：

```python
JUDGE_PROMPT = """
你是内容安全审查员，请判断以下文本是否包含：
- 政治敏感
- 暴力血腥
- 色情低俗
- 歧视性言论
- 违法犯罪引导

【文本】
{text}

请输出 JSON：{"safe": true/false, "reason": "..."}
"""

async def llm_judge(text: str) -> bool:
    result = await small_llm.complete(JUDGE_PROMPT.format(text=text))
    parsed = json.loads(result)
    if not parsed["safe"]:
        logger.warning(f"LLM Judge 拦截: {parsed['reason']}")
    return parsed["safe"]
```

---

## 七、可观测性

### 7.1 必埋的核心指标

```promql
llm_request_total{model, status}        # 请求数（按状态分类）
llm_request_duration_seconds{model}     # 请求延迟 P50/P95/P99
llm_input_tokens{model}                 # 输入 token 分布
llm_output_tokens{model}                # 输出 token 分布
llm_cost_usd{model, business}           # 成本
llm_first_token_latency{model}          # TTFT
llm_degraded_total{from_model, to_model}# 降级次数
llm_rate_limited_total{user_id}         # 限流次数
llm_content_blocked_total{side}         # 内容拦截次数
```

### 7.2 Trace

每次请求都应该生成一个 trace_id，贯穿整个调用链：

```python
import uuid
from opentelemetry import trace

tracer = trace.get_tracer("llm-gateway")

async def call_llm_with_trace(messages, user_id, business):
    span = tracer.start_span("llm_call")
    span.set_attribute("user.id", user_id)
    span.set_attribute("business", business)
    span.set_attribute("model", messages["model"])
    span.set_attribute("trace.id", str(uuid.uuid4()))
    
    try:
        response = await llm.call(messages)
        span.set_attribute("input_tokens", response.usage.prompt_tokens)
        span.set_attribute("output_tokens", response.usage.completion_tokens)
        span.set_attribute("cost.usd", calculate_cost(messages["model"], ...))
        return response
    except Exception as e:
        span.record_exception(e)
        raise
    finally:
        span.end()
```

### 7.3 业务级可观测

不止"调模型好不好"，还要看"业务用得好不好"：

- 用户满意度（点赞/点踩）
- 采纳率（用户复制了 AI 回答的比例）
- 转人工率
- 单次对话平均轮次

这些指标才是 LLM 应用的真正"北极星"。

---

## 八、踩坑提示

### 坑 1：API key 散落各处

**症状**：开发同事离职带走 key，新供应商 key 没同步到所有服务。

**解决**：

1. 密钥全部走统一网关，不要写在业务代码；
2. 网关层用密钥管理服务（Vault / KMS）；
3. 密钥定期轮换，建议 90 天一次。

### 坑 2：超时没分类型

**症状**：所有超时都按 30s，导致慢请求拖死系统。

**解决**：

```python
TIMEOUT_CONFIG = {
    "stream": 120,    # 流式响应可以等久一点
    "batch": 300,     # 批处理不限
    "interactive": 15 # 交互式必须快
}

async def call(messages, call_type="interactive"):
    timeout = TIMEOUT_CONFIG.get(call_type, 30)
    return await asyncio.wait_for(llm.call(messages), timeout=timeout)
```

### 坑 3：流式响应连接超时

**症状**：用 SSE 流式响应，闲置 5 分钟断流，用户看到一半。

**解决**：每 30s 发心跳包（仅 `:` 内容）保活：

```python
async def stream_with_heartbeat(response):
    async for chunk in response:
        yield chunk

    # 每 30s 一个心跳
    while True:
        yield ":\n\n"
        await asyncio.sleep(30)
```

### 坑 4：网关单点

**症状**：LLM 网关自己挂了，所有业务全停。

**解决**：

1. 网关本身水平扩展，多副本 + LB；
2. 关键业务配置"网关直连"备用通道（绕开网关）；
3. 网关监控 + 主备切换预案。

### 坑 5：Prompt 缓存反而降级

**症状**：缓存命中的 prompt 是修改前的版本，但 prompt 已经改过。

**解决**：

```python
def cached_call(prompt_with_version: str):
    # 把 prompt 版本号也作为 cache key 一部分
    cache_key = f"prompt:{prompt_id}:{version}:{hash(prompt)}"
    ...
```

---

## 九、本篇要点回顾

| 能力 | 实现 |
|------|------|
| 多模型路由 | LiteLLM 别名 + alias 系统 |
| 协议统一 | OpenAI 兼容协议 |
| 降级 | 主备链配置 + 自动 failover |
| 限流 | Token / QPS / 业务多维度 |
| 成本 | 每次调用打点 + 日预算控制 |
| 内容安全 | 输入输出双侧 + LLM-as-judge |
| 可观测 | Prometheus 指标 + OpenTelemetry Trace |

**LLM 网关的三个核心原则**：

1. **网关自己做高可用**——网关挂了等于全挂；
2. **降级永远留个兜底**——任何上游都可能挂；
3. **可观测是上线第一天的事**——亡羊补牢为时已晚。

---

## 十、下一篇预告

基础篇 6 篇告一段落，从下一篇开始进入**应用篇**。下一篇 【架构实战】**LLM Agent开发：Function Calling与多步骤推理**：

- Agent 的本质：从 LLM 到自主决策系统；
- Function Calling 的协议与实现细节；
- ReAct、Plan-and-Execute、Reflexion 三大主流框架；
- 工具注册表设计与安全沙箱；
- 实战：从零搭建一个能调 10 个工具的 AI Agent。

LLM Agent 是 AI 应用从"会聊天"升级到"会干活"的关键一步。

---

> **作者**：架构实战 第二季  
> **版本**：v1.0  
> **最后更新**：2026-10  
> **全文约 5500 字**

本文是「架构实战·第二季」第六篇，AI 系列基础篇最后一篇。从这一篇开始，我们进入"用 LLM 一定会踩到的工程问题"——网关层。

【架构实战】LLM API网关：多模型路由与降级策略

研发部经理

大数据

本文深入解析LLM网关的架构设计与工程实践，聚焦统一接入、多模型路由、降级熔断、限流成本控制及内容安全等核心能力。通过LiteLLM开源方案与自研逻辑对比，提供可直接落地的配置示例与踩坑经验，帮助技术团队构建高可用、可观测的AI服务网关层。

2026上云采购 | AI焕新·智启新局

CDN免费试用


文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

【架构实战】LLM API网关：多模型路由与降级策略-腾讯云开发者社区-腾讯云

【架构实战】LLM API网关：多模型路由与降级策略

【架构实战】LLM API网关：多模型路由与降级策略

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐