用 Python 实现 LLM 请求分级与多引擎自动降级

调用大模型时，单点故障几乎不可避免：某个模型返回 429 限流、某个服务 503 过载、某个请求迟迟不返回，或者某个账号的额度已经用光。如果代码里只写死一个模型，这些情况都会直接变成用户侧的错误。这篇教程要解决的问题是：用纯标准库在 Python 里搭一个请求路由器，把请求按难度分成两档，每档配一条引擎链，失败时按错误类型决定是换下一个引擎还是立刻停下，并记录是谁最终回答了请求。

适合的读者是正在把 LLM 接入自己服务、已经遇到限流或超时、希望在不引入额外依赖的前提下加一层容错的 Python 开发者。文中所有引擎名（large-a、large-b、small-a、local）都是表示角色的占位名，示例用伪引擎实现，不需要 API 密钥，离线即可复现。涉及具体模型的价格、配额与可用性，请以各服务商官网当前信息为准。

准备工作

这套实现刻意只依赖 Python 标准库，因此前置条件很少，但有几件事需要先想清楚。

运行环境

• Python 3.10 以上即可，示例代码用到了 str | None 这样的联合类型标注和 match 之外的海象运算符 :=，3.8 以上基本可用，但建议用较新的版本。
• 只需要 re、time、concurrent.futures、json、random 这些标准库模块。
• 不需要 API 密钥。示例中的引擎是伪引擎，用随机数模拟失败。

需要先确定的四件事

1. 分档规则：哪些请求算 hard，哪些算 easy。本文用关键词加长度、代码块、文件数等结构特征来判断。
2. 每档的引擎链：hard 档和 easy 档各自按什么顺序尝试哪些引擎。常见做法是 hard 档从大模型开始，easy 档从小模型开始，两条链的末尾都放一个本地执行作为兜底。
3. 错误分类：哪些错误应该换引擎，哪些应该直接抛出。这个决定直接关系到故障会不会被掩盖。
4. 冷却策略：额度耗尽的引擎要休息多久，链上最后一个引擎是否允许跳过。

把真实引擎包装成可替换的函数

路由器只关心「给一段 prompt，返回一段文本，失败时抛特定异常」。因此真实引擎的接入方式是写一个薄包装：

def call_large_a(prompt: str) -> str:
    resp = client_a.messages.create(
        model="your-model-name",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}],
    )
    return resp.content[0].text

包装函数内部负责把 HTTP 状态码和错误码翻译成本文定义的异常类型，其余逻辑全部交给路由器。这样替换引擎时不需要改动路由代码。

操作步骤

第一步：定义分档结果与判定函数

先定义承载判定结果的数据结构，再写判定逻辑。判定只用正则和几个简单条件，不调用模型，因此没有额外延迟和成本。

from dataclasses import dataclass, field
import re

@dataclass
class Decision:
    tier: str
    reasons: list[str] = field(default_factory=list)

HARD_WORDS = re.compile(
    r"设计|調査|比较|原因|为什么|判断|评审|重构|调试|优化|方案|架构|排查"
)
EASY_WORDS = re.compile(
    r"摘要|总结|翻译|格式化|提取|列成要点|改写|错别字|润色"
)

def classify(prompt: str, n_files: int = 0) -> Decision:
    hard = []
    if len(prompt) > 2000:
        hard.append(f"长输入({len(prompt)} 字)")
    if prompt.count("```") >= 2:
        hard.append("含代码块")
    if n_files >= 2:
        hard.append(f"多文件({n_files})")
    if m := HARD_WORDS.search(prompt):
        hard.append(f"难词: {m.group(0)}")
    if hard:
        return Decision("hard", hard)
    if m := EASY_WORDS.search(prompt):
        return Decision("easy", [f"易词: {m.group(0)}"])
    return Decision("hard", ["无法判定→hard"])

规则按以下顺序生效：

1. 出现长输入、代码块、多文件、难词中的任意一项，判为 hard。
2. 以上都没有，但出现易词（摘要、翻译等），判为 easy。
3. 两类词都没命中，判为 hard。

为什么拿不准时要倒向 hard：两类误判的代价不对称。把真正困难的请求误判成 easy，会交给能力不足的引擎，产出质量事故；把简单请求误判成 hard，只是多花一点成本。所以默认值应该选代价小的那一侧。

在一组 40 条自建测试样本上（easy 20 条、hard 20 条），这个判定器的准确率是 0.9：easy 判对 16 条、判成 hard 4 条，hard 判对 20 条、判成 easy 0 条。4 条误判全部落在安全侧，分别是「回复一句谢谢」「告诉我今天的日程」「想三个问候语」——都没命中关键词，走「无法判定→hard」；以及「总结这个函数」——因为带代码块被判成 hard。这组样本是自建的，真实请求的分布会不一样，上线前应该用自己的日志重新评估。

第二步：把错误分成「换引擎」和「立刻停」两类

这是整套设计里最关键的一步。给每个异常类带上两个属性：fallback 表示是否换下一个引擎，cooldown 表示该引擎需要休息多少秒。

class EngineError(Exception):
    fallback = True
    cooldown = 0.0

class RateLimited(EngineError):
    pass  # 429，临时限流

class Overloaded(EngineError):
    pass  # 503 / 529，过载

class EngineTimeout(EngineError):
    pass  # 响应超时

class QuotaExceeded(EngineError):
    cooldown = 600.0  # 额度用尽，短期内不会恢复

class InvalidRequest(EngineError):
    fallback = False  # 400，请求本身有错

class AuthError(EngineError):
    fallback = False  # 401 / 403，配置有错，不要掩盖

然后写一个把 HTTP 状态码和错误码映射成异常的函数：

QUOTA_CODES = {
    "insufficient_quota",
    "credit_balance_exhausted",
    "organization_spend_limit_exceeded",
    "project_spend_limit_exceeded",
    "organization_usage_limit_exceeded",
}

def map_http_error(status: int, code: str = "",
                   retry_after: str | None = None) -> EngineError:
    if status == 429:
        if code in QUOTA_CODES:
            return QuotaExceeded(code)  # 同样是 429，但等多久都不会恢复
        return RateLimited(f"429 retry-after={retry_after}")
    if status in (500, 502, 503, 529):
        return Overloaded(str(status))
    if status == 504:
        return EngineTimeout("504")
    if status in (401, 403):
        return AuthError(str(status))
    return InvalidRequest(f"{status} {code}")  # 400 / 404 / 413 / 422 等

映射结果可以整理成一张表，方便对照：

status	code	异常类	fallback	cooldown
429	-	RateLimited	True	0
429	organization_spend_limit_exceeded	QuotaExceeded	True	600
503	server_is_overloaded	Overloaded	True	0
529	overloaded_error	Overloaded	True	0
504	timeout_error	EngineTimeout	True	0
400	invalid_request_error	InvalidRequest	False	0
413	request_too_large	InvalidRequest	False	0
401	authentication_error	AuthError	False	0

几个判断依据值得展开说明。

同样是 429，有的等多久都不会恢复。 429 既可能是限流，也可能是额度、支出上限、用量上限被用光。后者重试没有意义，必须换引擎并让该引擎休息。区分手段是看错误码：额度类错误有专门的 code，例如 credit_balance_exhausted、organization_spend_limit_exceeded、project_spend_limit_exceeded、organization_usage_limit_exceeded。有些服务商把额度信息放在 error.type 而不是 error.code 里，所以生产环境应该同时检查两者，任一命中就按额度耗尽处理。本文示例为了简洁只看 code，实际接入时请补上 type 判断。

400 / 401 / 403 不换引擎。 请求本身写错了，换多少个引擎都会以同样方式失败，只会把失败次数乘以引擎数量；配置错了（密钥无效、权限不足）更需要暴露出来修，而不是被下一个引擎的成功掩盖。这两类都应该立刻中断并抛出。

402 的处理。 部分服务商有专门的计费错误（如 billing_error），表示账单或支付信息有问题。这类错误在同一个账号下换模型也会同样失败，所以本文的设计是直接停止。是否要转到另一份合约下的引擎，应该由运维方显式决定，不要默认隐藏。

500 的取舍。 官方文档通常建议对 500 做指数退避后重试同一个引擎，官方 SDK 也默认重试 5xx。本文的设计是等待时间优先，直接换下一个引擎。两种做法都合理，取决于你的截止时间有多长：如果请求可以等几秒，先在同引擎上重试一次往往更省；如果延迟敏感，直接换引擎更快。

已知局限。 有些服务商在组织或工作区的自定义支出上限被触发时返回的是 400，而不是 429。上面的映射会把 400 判成 InvalidRequest 并停止，这其实是误判。另外，支出上限触发的 429 在 error.type 上可能与普通限流完全相同，只看 code 无法区分。要处理这两种情况，映射函数必须同时读取 error.type 和 message。

第三步：给每次调用加超时

不能让一个不响应的引擎拖住整条链。用 ThreadPoolExecutor 提交调用，再用 future.result(timeout) 设期限：

import concurrent.futures as cf

def _call(self, name: str, prompt: str) -> str:
    fut = self.pool.submit(self.engines[name], prompt)
    try:
        return fut.result(timeout=self.timeout)
    except cf.TimeoutError:
        fut.cancel()  # 正在执行的线程无法取消，这里只是不再等待
        raise EngineTimeout(f"{self.timeout}s")

必须清楚这套机制的边界：cancel() 对已经在运行的调用返回 False，线程会在后台继续跑到结束；ThreadPoolExecutor 在解释器退出时会 join 所有线程，所以一个卡死的调用可能让进程迟迟退不出去。换句话说，这里做到的是「不再等待」，不是「真正中止」。

由此有三点必须配套：

• HTTP 客户端本身也要设置连接超时和读取超时，不要只依赖外层超时。
• 官方 SDK 通常默认带重试（例如对连接错误、限流、5xx 做指数退避重试，并遵循 retry-after）。链路的截止时间和 SDK 的重试会叠加，应该先把 SDK 的重试次数和超时定下来，避免双重等待。
• 如果引擎背后会调用写操作类工具（发邮件、下单），超时后那个调用可能仍在后台完成，此时再换引擎会导致同一操作执行两次。写操作必须做成幂等的。

第四步：实现按链顺序尝试的路由器

路由器持有每档的引擎链、超时时间、冷却记录和线程池。run 方法从链头开始逐个尝试：

TIERS = {
    "hard": ["large-a", "large-b", "local"],
    "easy": ["small-a", "large-b", "local"],
}

def run(self, prompt: str, n_files: int = 0, request_id: str = "-") -> Result:
    d = classify(prompt, n_files)
    res = Result(False, d.tier, None, None)
    res.reasons = d.reasons
    chain = self.tiers[d.tier]
    for i, name in enumerate(chain):
        now = self.clock()
        last = i == len(chain) - 1
        cooling = self.cool_until.get(name, 0) > now
        if self.use_cooldown and cooling and not (last and self.never_skip_last):
            res.attempts.append({"engine": name, "outcome": "skip(cooldown)"})
            continue
        t0 = time.perf_counter()
        try:
            text = self._call(name, prompt)
        except EngineError as e:
            ms = round((time.perf_counter() - t0) * 1000)
            res.attempts.append({"engine": name, "outcome": type(e).__name__, "ms": ms})
            if e.cooldown:
                self.cool_until[name] = now + e.cooldown
            if not e.fallback:
                res.error = type(e).__name__
                break
            continue
        ms = round((time.perf_counter() - t0) * 1000)
        res.attempts.append({"engine": name, "outcome": "ok", "ms": ms})
        res.ok, res.answered_by, res.text = True, name, text
        break
    else:
        res.error = "AllEnginesFailed"
    return res

三个要点：

1. fallback=False 的错误直接 break，当场停止，不继续往下试。
2. 带 cooldown 的错误会记录该引擎的恢复时刻，后续请求在冷却期内跳过它。
3. for ... else 处理「全部失败」的情况，统一记为 AllEnginesFailed。

链的排列本身是策略问题。示例里 hard 档是「大模型 A → 大模型 B → 本地执行」，easy 档是「小模型 A → 大模型 B → 本地执行」，两条链共享后两个引擎。共享的好处是省配置，坏处是共享引擎一旦进入冷却，两条链会同时受影响——这一点在第六步会看到具体后果。

第五步：把每次请求记成一行 JSON

日志要能回答三个问题：为什么进了这一档、在哪一步失败、最后是谁答的。一行 JSON 就够：

{"id": "req-1", "tier": "hard", "reasons": ["难词:设计"],
 "attempts": [
   {"engine": "large-a", "outcome": "RateLimited", "ms": 0},
   {"engine": "large-b", "outcome": "EngineTimeout", "ms": 200},
   {"engine": "local", "outcome": "ok", "ms": 0}],
 "answered_by": "local", "error": null}

reasons 记录分档依据，attempts 记录每个引擎的结果和耗时，answered_by 记录最终应答者，error 在失败时给出异常类名。有了这四段信息，一次异常请求的完整路径可以在一行里追完，不需要翻多份日志。

一个完整示例

下面用伪引擎把整条链路跑通。large-a 永远返回 429，large-b 每次睡 1 秒（超时设为 0.2 秒），local 和 small-a 正常返回。整个过程不需要网络。

import json, random, time
import concurrent.futures as cf

# ---- 伪引擎 ----
def fake_large_a(prompt: str) -> str:
    raise RateLimited("429")

def fake_large_b(prompt: str) -> str:
    time.sleep(1.0)
    return "large-b answer"

def fake_small_a(prompt: str) -> str:
    if "INVALID" in prompt:
        raise InvalidRequest("400 invalid_request_error")
    return "small-a answer"

def fake_local(prompt: str) -> str:
    return "local answer"

class Router:
    def __init__(self):
        self.engines = {
            "large-a": fake_large_a,
            "large-b": fake_large_b,
            "small-a": fake_small_a,
            "local": fake_local,
        }
        self.tiers = TIERS
        self.timeout = 0.2
        self.pool = cf.ThreadPoolExecutor(max_workers=4)
        self.cool_until = {}
        self.use_cooldown = True
        self.never_skip_last = True
        self.clock = time.monotonic

    # _call 与 run 见上文

router = Router()
requests = [
    ("req-1", "评审一下这个设计的风险点"),
    ("req-2", "总结这段文字"),
    ("req-3", "你好"),
    ("req-4", "总结这段 INVALID 文字"),
]
for rid, prompt in requests:
    r = router.run(prompt, request_id=rid)
    print(json.dumps({
        "id": rid, "tier": r.tier, "reasons": r.reasons,
        "attempts": r.attempts,
        "answered_by": r.answered_by, "error": r.error,
    }, ensure_ascii=False))
router.pool.shutdown(wait=False)

预期输出（耗时字段会略有浮动）：

{"id": "req-1", "tier": "hard", "reasons": ["难词:设计"], "attempts": [{"engine": "large-a", "outcome": "RateLimited", "ms": 0}, {"engine": "large-b", "outcome": "EngineTimeout", "ms": 200}, {"engine": "local", "outcome": "ok", "ms": 0}], "answered_by": "local", "error": null}
{"id": "req-2", "tier": "easy", "reasons": ["易词:总结"], "attempts": [{"engine": "small-a", "outcome": "ok", "ms": 0}], "answered_by": "small-a", "error": null}
{"id": "req-3", "tier": "hard", "reasons": ["无法判定→hard"], "attempts": [{"engine": "large-a", "outcome": "RateLimited", "ms": 0}, {"engine": "large-b", "outcome": "EngineTimeout", "ms": 200}, {"engine": "local", "outcome": "ok", "ms": 0}], "answered_by": "local", "error": null}
{"id": "req-4", "tier": "easy", "reasons": ["易词:总结"], "attempts": [{"engine": "small-a", "outcome": "InvalidRequest", "ms": 0}], "answered_by": null, "error": "InvalidRequest"}

逐条对照：

• req-1 因为「设计」命中难词进入 hard 档。large-a 返回 429，large-b 在 200ms 处超时，第三个引擎 local 应答。
• req-2 是摘要类请求，进入 easy 档，链头的 small-a 一次成功。
• req-3 是「你好」，两类词都没命中，按默认规则进入 hard 档，走了和 req-1 相同的路径。
• req-4 是带 INVALID 的摘要请求，small-a 抛出 InvalidRequest，因为 fallback=False，链路当场停止，answered_by 为 null，error 为 InvalidRequest。

注意 req-1 里 large-b 的 200ms 超时：那个线程其实还在后台睡满 1 秒，脚本退出时会等它结束。这正是第三步提到的「只是不再等待」。

用伪引擎评估引擎数量

链上放几个引擎合适？可以用伪引擎做一轮量化。设定每个引擎独立地以失败率 p 失败，把链截取前 k 个引擎比较成功率。每组 2000 个请求，其中混入约 2% 的非法请求（含 INVALID），失败类型按 429 占 50%、503 占 20%、超时占 20%（睡 0.1 秒、截止 0.03 秒）、额度耗尽占 10% 分布，冷却功能关闭。

p	k=1 实测 / 理论	k=2 实测 / 理论	k=3 实测 / 理论
0.05	95.02% / 95%	99.75% / 99.75%	100% / 99.99%
0.1	89.94% / 90%	98.68% / 99%	99.95% / 99.9%
0.2	79.47% / 80%	96.44% / 96%	98.93% / 99.2%
0.3	68.65% / 70%	90.80% / 91%	96.75% / 97.3%
0.5	51.58% / 50%	75.81% / 75%	86.18% / 87.5%

实测值与独立失败假设下的理论值 1−p^k 基本吻合。p=0.2、k=3 时实测 98.93% 比理论 99.2% 低 0.27 个百分点，换算成件数是理论失败约 16 件、实测 21 件，落在二项分布的随机波动范围内。结论很直观：失败率 0.2 时，单引擎成功率约 79%，三引擎链能到约 99%。

代价是切换次数和延迟。p 从 0.05 升到 0.5 时，k=3 的每请求平均切换次数从 0.056 升到 0.728，k=2 从 0.048 升到 0.505；k=3 时由最后一个引擎应答的请求数从 8 件升到 214 件。平均耗时在 k=3、p=0.05 时约 0.4ms，p=0.5 时约 5ms——这是伪引擎，时间几乎都花在等超时上，真实环境会完全不同。

另一个值得记下的数字：在全部 18 组条件（6 个 p 值 × 3 个 k）里，非法请求被转发到下一个引擎的次数是 0。这说明 fallback=False 的短路逻辑按预期生效了。

用伪引擎评估冷却策略

额度耗尽比临时限流更麻烦，因为不休息就会一直撞墙。用虚拟时钟做一轮模拟：每个请求推进 1 秒，large-a 在被调用 300 次后必定返回 QuotaExceeded，所有引擎都带 5% 的随机失败率，非法请求同样混入。

策略	large-a 被调用次数	额度耗尽后的无效调用	成功率	实际发生的切换
不冷却	937	637	99.8%	740
冷却 600 秒，最后一个也跳过	302	2	70.4%	65
冷却 600 秒，最后一个必试	302	2	97.3%	65
冷却 60 秒，最后一个必试	322	22	99.4%	120

不冷却时，虽然成功率还有 99.8%，但额度耗尽之后仍然白白调用了 large-a 637 次。加上冷却后，无效调用降到 2 次或 22 次。

这里有一个容易踩的坑： 冷却 600 秒并且连最后一个引擎也跳过时，成功率掉到了 70.4%。原因是模拟中后两个共享引擎（large-b、local）也会随机遇到额度错误，一旦它们进入 600 秒冷却，整条链就长时间无引擎可用。解决办法就是 never_skip_last：链上最后一个引擎即使处于冷却期也强制尝试一次。加上这条之后成功率回到 97.3%；再把冷却时间从 600 秒缩短到 60 秒，成功率升到 99.4%，无效调用 22 次。

需要说明这个模拟的简化之处：它把共享引擎的额度错误当成「随机发生一次」，而真实环境里额度耗尽会一直持续到上限解除。因此真实场景中冷却时间的选择更依赖你对额度恢复周期的判断，建议结合服务商的实际配额规则来定。

注意事项

• 超时不等于中止。 future.cancel() 无法停止正在运行的线程，被超时的调用会在后台继续执行，并可能拖慢进程退出。HTTP 层必须另外设置连接与读取超时。
• SDK 自带重试会与链路超时叠加。 官方 SDK 通常默认对连接错误、限流和 5xx 做指数退避重试，并遵循 retry-after。接入前先确定 SDK 的重试次数和超时，避免两层等待相乘。
• 写操作必须幂等。 超时后换引擎时，前一个引擎可能已经完成了写操作，重复执行会造成副作用。
• 429 需要区分类型。 额度、支出上限、用量上限类错误重试无效，必须换引擎并冷却；普通限流则不需要冷却。区分依据是错误码，部分服务商把额度信息放在 error.type 而非 error.code，两者都要检查。
• 400 不一定是请求错误。 有些服务商在自定义支出上限被触发时返回 400，按本文映射会被判成 InvalidRequest 并停止。需要同时读取 error.type 和 message 才能识别。
• 支出上限触发的 429 可能无法与普通限流区分。 两者的 error.type 可能相同，只看 code 的简化映射会漏判。
• 最后一个引擎强制尝试是必要的。 共享引擎进入长冷却时，如果连最后一个也跳过，整条链会无引擎可用，成功率显著下降。
• 分档准确率依赖样本。 文中 0.9 的准确率来自 40 条自建样本，真实请求分布不同，结果会变。上线前应使用自己的历史请求重新评估关键词表。
• 本文不涉及的范围。 同一引擎内的重试与退避细节、具体模型的选择与定价、常驻进程的重启、工具调用的幂等化实现，都需要另行设计。
• 价格、配额与模型可用性以服务商官网当前信息为准。 本文出现的引擎名均为角色占位，不对应任何具体产品或价格。