# 大模型API网关的核心不是转发，是调度：EWMA动态打分算法的工程实现

![封面](https://developer.qcloudimg.com/http-save/yehe-12517049/5372d8bf65d9c0a7939986d02b7504b0.jpg)>本文所有代码可直接运行，实验数据基于真实压测，不涉及任何厂商赞助。

## 0. 一个让我熬夜到三点的生产事故

上周四凌晨两点，我接到告警：线上对话服务响应超时率从 0.1% 飙升到 17%。
排查了一个小时，定位到原因：我们接入的某家模型上游通道，因为机房网络波动，所有请求的首token延迟从正常的 200ms 暴涨到 8~15 秒，但连接并没有断开、请求也没有返回错误——换句话说，通道**"没死"，只是"变得极慢"**。
而我们的负载均衡用的是业界最常用的**最小连接数算法**：它看到这条通道的连接数和其他通道差不多（因为请求都还挂着没返回），就持续地把新请求往这条慢通道上送。最终结果是雪崩：大量请求堆积，用户看到的就是一直转圈。
**这次事故让我意识到一个反直觉的事实：大模型API网关的调度问题，和传统微服务的负载均衡，根本不是同一个问题。** 把Nginx的轮询/最小连接数/加权轮询直接搬过来用，在大模型场景下是会出事故的。
本文我会从0到1讲清楚：为什么传统算法全部失效，以及我们最终落地的**改造版EWMA（指数加权移动平均）动态打分调度算法**——它上线后，我们的P99延迟下降了62%，超时率从17%降到0.03%，而代码量只有不到200行。
## 1. 为什么传统负载均衡算法在大模型场景下全部失效

传统负载均衡算法，本质上在回答一个问题：**"下一个请求发给谁？"** 但它们隐含了几个前提假设，而这些假设在大模型API场景下全部不成立。

| | | |
|:-:|:-:|:-:|
|算法|核心假设|大模型场景为何失效|
|**轮询（Round Robin）**|所有节点能力相同、响应速度一致|不同上游通道延迟差距可达5~10倍，轮询等于把一部分请求送给慢通道|
|**随机（Random）**|样本量够大时近似均匀|大模型单次请求耗时长，样本量根本摊不平|
|**加权轮询**|节点能力差异可以用静态权重描述|通道质量是分钟级波动的，静态权重追不上实时变化|
|**最小连接数**|连接数越少 = 节点越空闲|大模型连接数不反映真实负载——慢通道连接数不多但每个请求都卡10秒；流式连接无法用连接数衡量|
|**IP Hash**|同一客户端路由到同一节点|大模型API无状态需求，粘性会话毫无意义|
更致命的是大模型API请求的**三大特殊性**，是传统HTTP场景没有的：
**特殊性1：请求耗时长，且波动极大。** 普通HTTP接口响应时间通常在毫秒级，P50和P99差距在3~5倍以内；而大模型API首token延迟（TTFT）从200ms到15秒都正常，生成速度波动也很大。你不能用"最近一次请求快慢"判断通道健康。
**特殊性2：流式输出（SSE）存在"半失败"状态。** 传统HTTP要么成功要么失败；但SSE流式输出可能出现：首token返回成功（HTTP 200），生成到第300个token时断流。请求已经计费，但用户没拿到完整回答——传统健康检查根本检测不到。
**特殊性3：多维度决策，不止看速度。** 选通道不仅要看延迟，还要看当前并发负载、历史成功率、报价（不同通道价格差可达10倍）、是否支持当前上下文长度。这是多目标优化问题，不是单一指标能决定的。
![传统vs大模型](https://developer.qcloudimg.com/http-save/yehe-12517049/3b67423ce96c52b36363283e6cc3bf65.png)图1：传统HTTP负载均衡 vs 大模型API场景对比
## 2. EWMA算法为什么适合这个场景，以及我们做了哪些改造

EWMA（Exponentially Weighted Moving Average，指数加权移动平均）最早在Nginx商业版里用于负载均衡，后来被gRPC采纳为默认的xDS负载均衡策略之一。核心思想很简单：
**给每个后端节点维护一个"响应延迟的指数加权移动平均值"，永远选这个值最小的节点。**
和简单平均不同，EWMA对最近的观测值赋予更高权重，公式是：
```
latency_ewma = latency_ewma * β + current_latency * (1 - β)
```其中 β 是衰减因子（通常取0.9左右），越近的延迟数据权重越高，越早的数据按指数衰减。
**但原生EWMA直接套用到大模型场景有三个问题，我们做了对应改造：**
### 改造1：从单一延迟指标 → 多维度综合健康分

只看延迟不够。一个通道可能很快但成功率只有80%，另一个稍慢但成功率99.9%，后者更值得信任。我们把通道健康分定义为四个维度加权：
```
score = w1 * latency_score   // 延迟得分（越快越好）
      + w2 * success_score   // 成功率得分（越稳越好）
      + w3 * load_score      // 当前负载得分（越闲越好）
      + w4 * cost_score      // 价格得分（越便宜越好）
```每个维度归一化到0~1区间，权重可配置。最终永远选score最高的通道。
### 改造2：加入熔断器，处理"僵尸通道"

EWMA是平滑的，如果一个通道突然挂掉，历史得分会慢慢降下来，但在降下来之前还会接收到新请求。我们加了熔断器：连续失败N次后通道进入"打开"状态，一段时间内完全不分配请求；半开探测期放少量流量试探，恢复正常才完全关闭。
### 改造3：针对SSE流式场景，拆分TTFT和TPS

传统EWMA测的是请求从发起到收到完整响应的时间，但SSE流式请求这个时间和输出长度有关，没法直接比。我们拆成两个独立指标：
- **TTFT（Time To First Token）**：从发请求到收到第一个token的时间，反映通道响应速度
- **TPS（Tokens Per Second）**：生成速率，反映通道吞吐量

最终延迟分 = α × TTFT得分 + (1-α) × TPS得分。
## 3. 完整实现代码（可直接运行）

下面是调度器的核心代码，不到200行，可以直接集成到任何Python的API网关里：
```
import time
import math
import random
import threading
from dataclasses import dataclass, field
from typing import List, Optional, Dict
from enum import Enum


class CircuitState(Enum):
    """熔断器状态"""
    CLOSED = "closed"        # 正常，允许流量
    OPEN = "open"            # 熔断，拒绝流量
    HALF_OPEN = "half_open"  # 半开，试探性放少量流量


@dataclass
class ChannelMetrics:
    """单条上游通道的实时指标"""
    url: str
    price_per_1k_tokens: float       # 每千token报价（元）
    max_concurrent: int = 50         # 最大并发数

    # EWMA指标
    ewma_ttft: float = 1.0           # 首token延迟EWMA（秒），初始1秒
    ewma_tps: float = 20.0           # 生成速率EWMA（tokens/s），初始20tps
    success_rate: float = 1.0        # 成功率EWMA

    # 实时状态
    current_concurrent: int = 0
    circuit_state: CircuitState = CircuitState.CLOSED
    consecutive_failures: int = 0
    circuit_open_time: float = 0
    half_open_success: int = 0

    beta: float = 0.9
    failure_threshold: int = 5
    circuit_open_duration: float = 30
    half_open_threshold: int = 3


class SmartScheduler:
    """大模型API智能调度器"""

    def __init__(
        self,
        weight_latency: float = 0.4,
        weight_success: float = 0.3,
        weight_load: float = 0.15,
        weight_cost: float = 0.15,
    ):
        self.channels: List[ChannelMetrics] = []
        self.lock = threading.Lock()
        self.w_latency = weight_latency
        self.w_success = weight_success
        self.w_load = weight_load
        self.w_cost = weight_cost

    def add_channel(self, channel: ChannelMetrics):
        """注册一条上游通道"""
        with self.lock:
            self.channels.append(channel)

    def _normalize(self, value, min_val, max_val, reverse=False):
        """归一化到[0,1]，reverse=True表示值越小越好"""
        if max_val == min_val:
            return 1.0
        normalized = (value - min_val) / (max_val - min_val)
        return 1.0 - normalized if reverse else normalized

    def _compute_score(self, ch, all_channels):
        """计算综合得分，越高越好"""
        avail = [c for c in all_channels if c.circuit_state != CircuitState.OPEN]
        ttfts = [c.ewma_ttft for c in avail]
        prices = [c.price_per_1k_tokens for c in all_channels]

        latency_score = self._normalize(
            ch.ewma_ttft, min(ttfts), max(ttfts), reverse=True
        ) if ttfts else 0.5
        success_score = ch.success_rate
        load_score = max(0, min(1,
            1.0 - ch.current_concurrent / ch.max_concurrent
        )) if ch.max_concurrent > 0 else 0.5
        cost_score = self._normalize(
            ch.price_per_1k_tokens, min(prices), max(prices), reverse=True
        ) if prices else 0.5

        return (
            self.w_latency * latency_score
            + self.w_success * success_score
            + self.w_load * load_score
            + self.w_cost * cost_score
        )

    def _can_route(self, ch, now):
        if ch.circuit_state == CircuitState.CLOSED:
            return ch.current_concurrent < ch.max_concurrent
        elif ch.circuit_state == CircuitState.HALF_OPEN:
            return True
        else:  # OPEN
            if now - ch.circuit_open_time >= ch.circuit_open_duration:
                ch.circuit_state = CircuitState.HALF_OPEN
                ch.half_open_success = 0
                return True
            return False

    def select_channel(self):
        """选择最优通道（核心调度入口）"""
        now = time.time()
        with self.lock:
            available = [ch for ch in self.channels if self._can_route(ch, now)]
            if not available:
                return None
            scored = [(self._compute_score(ch, available), ch) for ch in available]
            scored.sort(key=lambda x: x[0], reverse=True)
            # epsilon-greedy：10%概率随机选top-3之一，避免赢者通吃
            if random.random() < 0.1 and len(scored) > 1:
                return scored[random.randint(0, min(2, len(scored)-1))][1]
            return scored[0][1]

    def record_success(self, ch, ttft, tps):
        with self.lock:
            ch.current_concurrent -= 1
            ch.ewma_ttft = ch.beta * ch.ewma_ttft + (1-ch.beta) * ttft
            ch.ewma_tps = ch.beta * ch.ewma_tps + (1-ch.beta) * tps
            ch.success_rate = ch.beta * ch.success_rate + (1-ch.beta) * 1.0
            ch.consecutive_failures = 0
            if ch.circuit_state == CircuitState.HALF_OPEN:
                ch.half_open_success += 1
                if ch.half_open_success >= ch.half_open_threshold:
                    ch.circuit_state = CircuitState.CLOSED

    def record_failure(self, ch):
        with self.lock:
            ch.current_concurrent -= 1
            ch.success_rate = ch.beta * ch.success_rate + (1-ch.beta) * 0.0
            ch.consecutive_failures += 1
            if ch.consecutive_failures >= ch.failure_threshold:
                ch.circuit_state = CircuitState.OPEN
                ch.circuit_open_time = time.time()
            if ch.circuit_state == CircuitState.HALF_OPEN:
                ch.circuit_state = CircuitState.OPEN
                ch.circuit_open_time = time.time()

    def increment_concurrent(self, ch):
        with self.lock:
            ch.current_concurrent += 1
```调度器的使用方式：
```
# 初始化调度器
scheduler = SmartScheduler(
    weight_latency=0.4,
    weight_success=0.3,
    weight_load=0.15,
    weight_cost=0.15,
)

# 注册上游通道（实际从配置或API动态获取）
scheduler.add_channel(ChannelMetrics(
    url="https://api.example-a.com/v1/chat/completions",
    price_per_1k_tokens=0.020,
))
scheduler.add_channel(ChannelMetrics(
    url="https://api.example-b.com/v1/chat/completions",
    price_per_1k_tokens=0.012,
))

# 每次请求前选通道
channel = scheduler.select_channel()
if channel is None:
    raise Exception("no available channel")
scheduler.increment_concurrent(channel)

try:
    ttft, tps = send_request(channel.url, messages)
    scheduler.record_success(channel, ttft, tps)
except Exception as e:
    scheduler.record_failure(channel)
```## 4. 熔断器状态机

除了EWMA打分，熔断器是调度器的另一道关键防线。它的状态转换逻辑如下：
![熔断器状态机](https://developer.qcloudimg.com/http-save/yehe-12517049/08b2054e9d803c3aa6638ff9ef77bec3.png)图2：熔断器三态转换状态机
- **CLOSED（关闭）**：正常状态，所有请求放行；连续失败达阈值后切换到OPEN
- **OPEN（打开）**：熔断状态，拒绝所有流量，持续一段时间（我们设30秒）
- **HALF_OPEN（半开）**：熔断时间到后进入，试探性放少量请求；连续成功N次回到CLOSED，失败一次立刻回到OPEN

三态机的核心价值：**给故障通道一段"冷静期"，而不是让它一有好转就立刻接流量**。实际生产中上游故障往往是间歇性的，30秒冷静期足够让大多数网络抖动恢复。
## 5. 压测对比：三种算法在故障场景下的真实表现

我们在压测环境模拟了一个真实场景：3条上游通道，正常TTFT都在200~400ms；压测到第30秒时，通道B发生网络抖动，TTFT从300ms恶化到10秒，但请求不完全失败（成功率从99.9%降到70%），压测持续120秒。
分别用三种算法调度，结果如下：
![压测对比](https://developer.qcloudimg.com/http-save/yehe-12517049/31335c5074a34098cc898db6cc79ad77.png)图3：三种调度算法在故障场景下的压测对比

| | | | |
|:-:|:-:|:-:|:-:|
|指标|轮询调度|最小连接数|EWMA智能调度|
|平均TTFT|3.4s|2.8s|**0.42s**|
|P99 TTFT|12.1s|9.7s|**1.8s**|
|请求成功率|82.3%|86.1%|**99.2%**|
|路由到慢通道比例|33%|28%|**<1%**|
|平均token单价|¥0.018|¥0.018|**¥0.013**|
几个关键发现：
1. **轮询在故障场景下几乎是灾难**——固定按比例分配，1/3的请求送给慢通道；
2. **最小连接数比轮询略好**，但慢通道的连接也在"卡住"状态，无法有效识别；
3. **EWMA智能调度故障后约15秒感知到延迟变化**，自动切走流量，路由到慢通道比例不到1%——且因为考虑价格维度，平均成本也更低。

## 6. 工程落地踩过的5个坑

**坑1：beta值的调优是门艺术。** beta=0.9意味着最近10次请求权重占约65%，太灵敏会频繁切换通道，太迟钝则故障反应慢。我们按通道类型差异化：稳定大厂通道beta=0.95（更平滑），第三方聚合通道beta=0.8（更灵敏）。
**坑2：epsilon-greedy的必要性。** 纯选最高分导致"赢者通吃"——一条通道得分高就把所有流量给它，其他通道没探测数据，它出问题就没有备选。我们加了10%随机探测流量，始终保持对次优通道的最新数据。
**坑3：SSE断流的判定标准。** 什么叫"失败"？HTTP 5xx肯定是，但SSE中途断流怎么算？我们定义：收到首token后输出token数少于预期70%记为失败，预期长度由prompt预估。
**坑4：并发计数必须精确。** `increment_concurrent` 和 `record_success/failure` 必须严格配对，否则并发计数泄漏，通道会被误判"满载"。建议用contextmanager保证。
**坑5：冷启动问题。** 新通道没有历史数据，EWMA初始值设太高没人选、太低易打挂。我们做法是新通道初始得分设为当前平均分，前10次请求用线性权重预热，等积累足够数据再纳入正常调度。
## 7. 写在最后

大模型API网关这个领域，大多数开源项目和商业产品都把精力放在了"协议转换"和"Key管理"上，但在我看来，**真正决定用户体验的核心是调度算法——用户感知到的"快"和"稳"，几乎全是调度层决定的。**
EWMA不是什么新算法，它在负载均衡领域已存在多年。但把它从传统HTTP场景迁移到大模型SSE流式场景，需要对指标定义、熔断机制、多目标权重做大量改造。这篇文章里的代码我们已经在线上跑了几个月，P99延迟和成功率的改善是实打实的。
如果你正在做大模型API相关的基础设施，希望这篇文章能帮你少走一些弯路。代码已整理成可运行的demo，有问题欢迎评论区交流。
**参考说明：**文中调度器的生产版本已在我们团队内部API网关中稳定运行数月，对接方式与OpenAI SDK完全兼容。完整代码和压测脚本已整理好，有需要的同学可以在我个人主页简介处获取，欢迎一起交流讨论。

上周四凌晨两点，我接到告警：线上对话服务响应超时率从 0.1% 飙升到 17%。排查了一个小时，定位到原因：我们接入的某家模型上游通道，因为机房网络波动，所有请求的首token延迟从正常的 200ms 暴涨到 8~15 秒，但连接并没有断开、请求也没有返回错误——换句话说，通道"没死"，只是"变得极慢"。而我们的负载均衡用的是业界最常用的最小连接数算法：它看到这条通道的连接数和其他通道差不多（因为请求都还挂着没返回），就持续地把新请求往这条慢通道上送。最终结果是雪崩：大量请求堆积，用户看到的就是一直转圈。这次事故让我意识到一个反直觉的事实：大模型API网关的调度问题，和传统微服务的负载均衡，根本不是同一个问题。 把Nginx的轮询/最小连接数/加权轮询直接搬过来用，在大模型场景下是会出事故的。本文我会从0到1讲清楚：为什么传统算法全部失效，以及我们最终落地的改造版EWMA（指数加权移动平均）动态打分调度算法——它上线后，我们的P99延迟下降了62%，超时率从17%降到0.03%，而代码量只有不到200行。

大模型API网关的核心不是转发，是调度：EWMA动态打分算法的工程实现

传统轮询、加权轮询、最少连接数这些经典负载均衡算法，在大模型API场景下几乎全部失效：首token延迟差异可达10倍、长连接SSE易断流、单通道故障会引发雪崩。本文从一次凌晨两点超时率飙升至17%的生产事故出发，拆解EWMA指数加权移动平均动态打分算法的工程改造方案，给出不到200行可直接运行的Python调度器完整实现（含多维度质量打分、熔断器三态状态机、epsilon-greedy防赢者通吃。

人工智能

架构设计

后端

大模型API网关调度算法EWMA动态打分改造详解：针对传统负载均衡在LLM场景的失效问题，提出四维健康分评估（延迟/成功率/负载/价格），结合熔断器与SSE流式指标拆分，实现P99延迟下降62%、超时率降至0.03%的工程实战方案。

负载均衡

代码解释

路由

Python

Nginx

腾讯云双十一大促

tione

cngw

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

社区新版编辑器体验调研

大模型API网关的核心不是转发，是调度：EWMA动态打分算法的工程实现-腾讯云开发者社区-腾讯云

大模型API网关的核心不是转发，是调度：EWMA动态打分算法的工程实现

大模型API网关的核心不是转发，是调度：EWMA动态打分算法的工程实现

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐