企业把 AI 接进内部系统之后很快会遇到同一个现象所有请求走同一个模型、同一套参数改个标题和生成一份长报告排在同一个队列里。前者本该毫秒级返回却要等到后者算完。这篇是一份完整的实战教程带你写出一套按任务分级的路由层先量出请求分布再按任务显式指定模型和参数用统一客户端收敛重试与降级最后验证效果。每段代码都可以单独运行文中的打印输出也是实测结果。阅读前提会 Python用过任意一家大模型的 HTTP 接口。环境准备全文共用这套依赖和环境变量。凭证一律从环境读取不写进源码pip install requests2.32.3# Windows PowerShell $env:AI_API_BASEhttps://api.example.com/v1 $env:AI_API_KEYsk-xxxxxx $env:AI_MODEL_FASTlight-model-name $env:AI_MODEL_MAINmain-model-name分级路由要求同一套代码能在多个模型之间切换所以接入层必须提供多模型的统一入口——否则每换一个模型就得写一套客户端路由层无从下手。接入前先在平台文档里核对这一项。国内常见平台如 jiekou.vip提供多模型统一入口模型名作为参数传入即可本文代码就是按这种形式写的。步骤一量出请求的真实分布分级规则必须来自实测不能靠感觉。先写一个脚本统计线上输入的长度分布# profile_requests.py import json import statistics from pathlib import Path def pct(values, p): 线性插值分位数避免为一个数引入额外依赖。 if not values: return 0.0 s sorted(values) if len(s) 1: return float(s[0]) k (len(s) - 1) * p / 100.0 lo int(k) hi min(lo 1, len(s) - 1) return s[lo] (s[hi] - s[lo]) * (k - lo) def main(pathsamples.jsonl): lengths, by_task [], {} for line in Path(path).read_text(encodingutf-8).splitlines(): if not line.strip(): continue rec json.loads(line) n len(rec[input]) lengths.append(n) by_task.setdefault(rec.get(task, unknown), []).append(n) print(f样本数 {len(lengths)}) print(f中位 {statistics.median(lengths):.0f} fP90 {pct(lengths, 90):.0f} P99 {pct(lengths, 99):.0f}\n) print(f{任务:18}{条数:6}{中位:8}{P90:8}) for task, vals in sorted(by_task.items(), keylambda kv: -len(kv[1])): print(f{task:18}{len(vals):6}{statistics.median(vals):8.0f}{pct(vals, 90):8.0f}) if __name__ __main__: main()samples.jsonl每行一条记录形如{task: classify, input: ...}。实测输出样本数 4820 中位 168 P90 2140 P99 7630 任务 条数 中位 P90 classify 1904 96 142 shorten_title 1372 121 190 kb_answer 981 2860 7410 weekly_draft 563 1980 4120前两类任务占了 68%中位输入不到 130 字符后两类才是长输入。这张表就是下一步分级规则的依据。步骤二写一个显式的分级路由关键设计决定由调用方显式声明任务类型不靠输入长度自动猜。自动判定在分类任务偶尔带上长上下文时会误判行为不可预测。# router.py import os MODEL_FAST os.environ[AI_MODEL_FAST] MODEL_MAIN os.environ[AI_MODEL_MAIN] # 任务 - (模型, 最大输出, 输入截断上限) TASK_PROFILE { classify: (MODEL_FAST, 32, 1200), shorten_title: (MODEL_FAST, 64, 1200), kb_answer: (MODEL_MAIN, 1024, 12000), weekly_draft: (MODEL_MAIN, 2048, 8000), } DEFAULT_PROFILE (MODEL_MAIN, 1024, 8000) def resolve(task, text): 返回该任务应使用的模型、输出上限以及截断后的输入。 model, max_out, limit TASK_PROFILE.get(task, DEFAULT_PROFILE) return { model: model, max_tokens: max_out, input: text[:limit], truncated: len(text) limit, }三个设计点max_tokens按任务收紧。分类任务只需一个标签给 32 足够。统一给 2048 时模型常在标签后追加一段解释还得写正则清理。收紧后返回结构稳定得多。截断上限按任务分开。分类任务给 1200 就够超过这个长度的分类输入基本是上游传错了内容——我们靠这条规则发现了两处调用方 bug。truncated标志必须带出来。默默截断会导致一类问答答案总是不完整且极难定位。步骤三统一客户端收敛重试与降级分级后出现新问题轻量模型偶发不可用时请求直接失败。把重试和降级收进一个客户端只在这里实现一次# client.py import os import time import uuid import logging import requests API_BASE os.environ[AI_API_BASE].rstrip(/) API_KEY os.environ[AI_API_KEY] log logging.getLogger(ai) RETRYABLE_STATUS {429, 500, 502, 503, 504} class AIClient: def __init__(self, timeout30, max_retry3): self.timeout timeout self.max_retry max_retry self.session requests.Session() self.session.trust_env False def chat(self, plan, task, line, user, fallback_modelNone): plan 来自 router.resolve()主模型耗尽后降级到 fallback_model。 trace_id uuid.uuid4().hex[:12] models [plan[model]] if fallback_model and fallback_model ! plan[model]: models.append(fallback_model) last_err None for model in models: for attempt in range(self.max_retry): started time.monotonic() try: resp self.session.post( f{API_BASE}/chat/completions, headers{Authorization: fBearer {API_KEY}}, json{ model: model, max_tokens: plan[max_tokens], messages: [{role: user, content: plan[input]}], }, timeoutself.timeout, ) elapsed int((time.monotonic() - started) * 1000) if resp.status_code 429: log.warning(throttled trace%s task%s attempt%d, trace_id, task, attempt) last_err http_429 time.sleep(min(2 ** attempt, 8)) continue if resp.status_code in RETRYABLE_STATUS: last_err fhttp_{resp.status_code} time.sleep(min(2 ** attempt, 8)) continue resp.raise_for_status() text resp.json()[choices][0][message][content] log.info( ok trace%s task%s line%s user%s model%s ms%d chars_in%d chars_out%d truncated%s retry%d degraded%s, trace_id, task, line, user, model, elapsed, len(plan[input]), len(text), plan[truncated], attempt, model ! plan[model], ) return text except requests.Timeout: last_err timeout time.sleep(min(2 ** attempt, 8)) except requests.RequestException as exc: last_err type(exc).__name__ time.sleep(min(2 ** attempt, 8)) log.warning(model_exhausted trace%s task%s model%s err%s, trace_id, task, model, last_err) log.error(fail trace%s task%s line%s err%s, trace_id, task, line, last_err) raise RuntimeError(fall models failed: {last_err} (trace{trace_id}))日志里的task/line/user三个字段是整套改造的地基。只在调用点多传两个参数但没有它们你分不清一次超时属于哪条业务线的哪个任务。degraded标志同样重要降级成功的请求在调用方看来是正常的不记录的话主模型持续不稳定这件事可以一周无人察觉。429单独记一行throttled别和 500 混在一起——一个是自己请求太密一个是上游故障混记会把限流误判成服务不稳。步骤四验证分级的实际效果改完必须能量出差别。对同一批输入分别按「全走主模型」和「按任务分级」跑一遍# compare.py import statistics import time from client import AIClient from router import resolve, MODEL_MAIN CASES [ (classify, 用户反馈登录后头像不显示), (classify, 请问发票什么时候能开), (shorten_title, 关于近期系统维护安排的通知说明请各团队注意查看), (kb_answer, 根据以下资料回答问题。 资料段落。 * 200), ] def run(plans, client, label): costs [] for task, plan in plans: started time.monotonic() try: client.chat(plan, tasktask, linebench, userbench) except RuntimeError: continue costs.append((time.monotonic() - started) * 1000) if not costs: print(f{label}: 全部失败) return print(f{label}: n{len(costs)} 中位 {statistics.median(costs):.0f}ms f最大 {max(costs):.0f}ms) def main(): client AIClient() graded [(t, resolve(t, s)) for t, s in CASES] baseline [] for t, s in CASES: p resolve(t, s) p[model] MODEL_MAIN # 强制全部走主模型 p[max_tokens] 2048 baseline.append((t, p)) run(baseline, client, 全走主模型) run(graded, client, 按任务分级) if __name__ __main__: main()实测输出全走主模型: n4 中位 3820ms 最大 9210ms 按任务分级: n4 中位 1140ms 最大 8760ms长任务几乎没变本来就该走主模型短任务中位耗时降到三分之一以下。占七成请求的那两类任务变快了这正是分级要拿到的结果。步骤五按业务线聚合定期回看前面记的字段到这一步才产生价值# aggregate.py import re import sys from collections import defaultdict PAT re.compile( r(?Plvlok|fail) trace(?Ptrace\w) task(?Ptask\S) line(?Pline\S) r(?: user(?Puser\S) model(?Pmodel\S) ms(?Pms\d))? ) def main(path): agg defaultdict(lambda: {n: 0, fail: 0, ms: [], users: set(), deg: 0}) for line in open(path, encodingutf-8): m PAT.search(line) if not m: continue rec agg[(m.group(line), m.group(task))] rec[n] 1 if m.group(lvl) fail: rec[fail] 1 continue if m.group(ms): rec[ms].append(int(m.group(ms))) if m.group(user): rec[users].add(m.group(user)) if degradedTrue in line: rec[deg] 1 print(f{业务线:12}{任务:16}{请求:7}{失败率:8} f{中位ms:8}{人数:6}{降级:6}) for (bl, task), r in sorted(agg.items(), keylambda kv: -kv[1][n]): med sorted(r[ms])[len(r[ms]) // 2] if r[ms] else 0 rate r[fail] / r[n] * 100 print(f{bl:12}{task:16}{r[n]:7}{rate:7.1f}% f{med:8}{len(r[users]):6}{r[deg]:6}) if __name__ __main__: main(sys.argv[1] if len(sys.argv) 1 else ai.log)一周的输出业务线 任务 请求 失败率 中位ms 人数 降级 ticket classify 19240 0.3% 92 41 0 ticket shorten_title 13810 0.4% 118 41 0 kb kb_answer 9760 1.1% 2840 77 214 report weekly_draft 2130 0.6% 3610 23 0kb_answer那 214 次降级是这张表第一次跑出来时最有价值的信息主模型在某几个时段不稳定被降级兜住了调用方毫无感知。不记degraded我们大概会以为一切正常。小结完整顺序量分布 → 按任务显式分级 → 统一客户端收敛重试降级 → 用同一批输入验证 → 聚合表定期回看。五步里最容易被跳过的是第一步和第四步但缺了它们分级就成了既无依据也无法验证的改动。如果只能先做一件选步骤三里的task/line/user三个日志字段——改动最小却是后面所有判断的前提。下一篇会写多团队共用同一个 AI 接入口时的凭证隔离与配额控制。