零预算跑通十几个模型能力,这件事我做过不止一次。标题里那个词说得很直白,本质上就是把能零成本调用的模型能力,全部收拢到一个统一入口。dsh-our-free-model 2.0 是我给这套东西起的代号,它不是一个产品,而是一份配置清单加一层薄薄的适配代码:13 个可以零成本起步的模型接入渠道,统一成一套请求格式、一套密钥管理、一套降级逻辑,换渠道只改一行配置,不动业务代码。
先说清楚它解决什么问题。做过批处理的人都有体会:手上有一批文本要清洗、要分类、要打标签,量不大不小,几千到几万条。买商业额度太贵,本地跑大模型又吃显存。这时候你会去找各种免费额度、社区试用、开源自建方案,找着找着就发现——渠道是找到了十几个,但每个渠道的接口长得都不一样,密钥散落在五六个文件里,超时行为各不相同,某个渠道额度用完了你第二天早上才发现,任务已经卡了一整夜。
这套清单干的就是把这件事标准化。适合两类人看:一类是手上有一堆小任务、预算为零的个人开发者;另一类是团队里负责搭内部工具、需要给同事提供一个稳定统一入口的人。不要求你懂模型原理,但需要你会写一点 Python、能看懂 YAML。下面我把设计思路、13 个渠道怎么分类、代码怎么落地、踩过哪些坑,一层层拆开讲。
1. 为什么要把免费能力做成一个入口,而不是十份配置
1.1 零成本接入的真实痛点:不是没渠道,是渠道太散
我最早的做法很土:每个渠道写一个脚本,脚本里硬编码 URL 和密钥。前三个渠道还挺顺,到第五个就开始出问题。最典型的一次是跑一个 4000 条的文本分类任务,前半段用的是渠道 A,跑到 60% 的时候触发限速,脚本没有重试逻辑,直接抛异常退出。我第二天早上看到日志,等于白跑一夜。
渠道散带来的麻烦,具体可以归成四类,这四类问题几乎每个做零成本接入的人都会撞上:
- 鉴权方式不统一。有的走 Header 里的 Bearer,有的走自定义字段,有的还要签名。每换一个渠道,鉴权代码就得重写一遍。
- 请求体结构不统一。同样是对话补全,有的把系统提示词放在消息数组第一条,有的单独开一个字段;有的支持
max_tokens,有的叫max_output_tokens。 - 限速策略不透明。文档上写着每分钟多少次,实际跑起来是按并发数算的,或者干脆是滑动窗口,你根本猜不到下一次被拒是什么时候。
- 额度耗尽的反馈不一致。有的返回 429,有的返回 200 但内容为空,有的直接给你一段道歉文本。第三种最坑,因为它不会报错,脏数据会一路流进你的结果文件。
这四类问题叠加起来,导致一个很尴尬的局面:渠道越多,整体稳定性反而越差。因为任何一个渠道出问题,都会污染整条链路,而你排查的时候还得先定位是哪个渠道。
所以我后来定了个原则:渠道数量可以多,但对外只能暴露一个入口。业务代码永远只认一个函数,剩下的脏活全在适配层里消化。
1.2 统一接入层的设计取舍:适配器模式加配置驱动
方案选型上我试过三条路。
第一条是直接用各家官方 SDK,在业务层里做 if-else 分支。这条路起步最快,但依赖会越装越多,而且不同 SDK 对超时、重试的处理各有一套逻辑,你想统一控制根本做不到。更麻烦的是版本冲突,某个 SDK 依赖的 HTTP 库版本和另一个打架,光是解决依赖就够喝一壶。
第二条是搞一个本地网关,把所有渠道注册进去,业务侧只访问本地端口。这条路适合团队场景,但对个人项目来说太重了,多一个进程就多一个要维护的东西。
第三条是我最后选的:适配器模式加配置驱动。核心就两点。第一,定义一份内部统一请求体,字段尽量少,只保留messages、temperature、max_tokens、stop这四个真正通用的东西。第二,每个渠道对应一个适配器类,负责把内部请求体翻译成该渠道要的格式,再把返回结果翻译回内部统一结构。
为什么选这条路?理由很实在。更换渠道的成本被压到最低。业务代码里写的是router.chat(messages),底层走哪个渠道由配置决定。哪天渠道 A 挂了,我改一行 YAML 换成渠道 B,重启都不用,热加载就能生效。适配器本身也很薄,一个渠道大概三四十行代码,写起来不心疼。
这里有个取舍要说明:统一请求体必然会丢掉一部分渠道特性。比如某些渠道支持函数调用、支持结构化输出、支持多模态输入,这些在统一层里我全砍掉了。原因是这些能力的格式差异太大,硬要抽象会得到一个谁都不好用的四不像。我的做法是留一个extra字段作为逃生舱,渠道专有参数塞进去,适配器自己解析。这样 90% 的常规请求走标准路径,剩下 10% 的特殊需求走逃生舱。
1.3 2.0 版本相对 1.0 到底改了什么
1.0 那版能跑,但写得很糙。渠道切换是硬编码的 if-else,降级逻辑写死在函数里,加一个新渠道要改三个文件。2.0 做了四件事的改造,这四件事也是我认为任何一个多渠道路由层都该具备的。
第一件,配置外置。13 个渠道的元信息全部挪到一份 YAML 里,包括名称、协议类型、基础地址、鉴权方式、限速参数、权重。代码里不再出现任何硬编码地址。
第二件,健康状态内建。每个渠道维护一个运行时状态:最近一次成功时间、连续失败次数、当日剩余额度估算、当前冷却截止时间。路由的时候先按状态过滤,再按权重选择。
第三件,降级策略声明化。配置文件里可以写fallback: [C03, C07, C11]这样的链,主渠道失败自动往下走,走到底还失败才抛给业务层。
第四件,密钥与配置分离。配置里只写环境变量名,真实密钥通过环境变量或者本地密钥文件注入,配置文件本身可以放心提交到仓库。
注意:13 这个数字不是固定目标,它是当时我手上能跑通并且长期可用的渠道数量。渠道会失效,会改规则,所以清单本身要有新陈代谢的机制,这一点在第 5 节会专门讲怎么维护。
2. 13 个渠道的分类逻辑与挑选标准
2.1 按能力形态分成四类:本地权重、社区试用、云厂商赠送、自建推理
13 个渠道如果平铺着列,看完记不住。我按能力形态把它们分成四类,这个分类方式的好处是:同一类渠道的接入代码高度相似,可以复用适配器;出问题的时候排查思路也一致。
第一类,本地权重直跑。就是把开源模型权重下载到本机,用推理框架加载起来,暴露一个本地端口。这类渠道的优势是完全不受外部限制,没有限速、没有额度、断网也能跑;劣势是吃硬件,而且首次加载慢。我这套清单里放了 3 个,分别对应不同参数量级,小参数量的跑得快用来做粗筛,中等参数量的用来做精细分类。
第二类,开源社区的在线试用额度。很多模型社区会给注册用户一定的免费调用额度,用来体验模型能力。这类渠道的特点是模型质量好、协议规范,缺点是额度有限而且通常有并发限制。清单里放了 4 个,主要用来处理那些对质量要求高、但数据量不大的任务。
第三类,云服务商的新用户赠送额度。这类额度的特点是总量大、有效期长,但通常需要绑定支付方式,而且赠送期结束后会自动转成付费。清单里放了 4 个。这类渠道我用来跑大批量的离线任务,因为吞吐高、稳定。
第四类,自建推理服务。找一台闲置的机器,或者用容器化方式部署推理服务,对外暴露一个接口。这类渠道本质上和第一类重合,区别在于它可以被多台机器共享,适合团队场景。清单里放了 2 个。
四类加起来正好 13 个。这个结构的好处是互补:本地渠道保证断网可用,社区渠道保证质量,云厂商渠道保证吞吐,自建渠道保证团队共享。
2.2 一张表看清 13 个渠道的定位与取舍
下面这张表是我实际维护的版本,渠道名用 C01 到 C13 编号代替,协议类型和限制条件写得比较实。
| 编号 | 形态分类 | 协议类型 | 典型用途 | 主要限制 | 建议权重 |
|---|---|---|---|---|---|
| C01 | 本地权重 | 自定义 HTTP | 断网兜底、粗筛 | 首加载慢,显存占用高 | 3 |
| C02 | 本地权重 | 兼容 CC 协议 | 中等批量分类 | 单并发,吞吐一般 | 4 |
| C03 | 本地权重 | 兼容 CC 协议 | 高质量小批量 | 响应慢,不适合流式 | 2 |
| C04 | 社区试用 | 兼容 CC 协议 | 质量敏感任务 | 每分钟请求数限制 | 6 |
| C05 | 社区试用 | 兼容 CC 协议 | 长文本摘要 | 上下文有上限,超长会截断 | 5 |
| C06 | 社区试用 | 自定义 HTTP | 结构化抽取 | 需自行拼装请求体 | 4 |
| C07 | 社区试用 | 兼容 CC 协议 | 通用对话 | 每日额度较小 | 5 |
| C08 | 云厂商赠送 | 兼容 CC 协议 | 大批量离线任务 | 需绑定支付方式 | 8 |
| C09 | 云厂商赠送 | 兼容 CC 协议 | 大批量离线任务 | 赠送额度有有效期 | 8 |
| C10 | 云厂商赠送 | 兼容 CC 协议 | 中等并发在线任务 | 并发数按账号计算 | 6 |
| C11 | 云厂商赠送 | 自定义签名 | 高吞吐批处理 | 签名算法需单独实现 | 7 |
| C12 | 自建推理 | 兼容 CC 协议 | 团队共享 | 需自己维护服务可用性 | 5 |
| C13 | 自建推理 | 兼容 CC 协议 | 内网隔离场景 | 仅内网可达 | 4 |
表里"建议权重"这一列是路由层的选择依据,数值越大越优先被选中。权重不是拍脑袋定的,我的定法是:权重等于该类渠道在我历史任务中的成功率乘以吞吐系数。比如 C08 和 C09 成功率高、吞吐大,权重给到 8;C03 虽然质量好但太慢,权重压到 2,只在明确要求高质量的场景下手动指定。
2.3 挑渠道的三个硬指标:上下文长度、并发限速、协议兼容度
渠道多不等于好用,我筛渠道的时候只看三个硬指标,不满足的直接淘汰,不浪费时间去试。
第一个是上下文长度。这个指标最容易被忽略,但坑最深。有些渠道标称支持很长的上下文,实际跑起来超过某个长度就静默截断,不报错。你拿到一个看起来通顺但漏了关键信息的输出,排查半天才发现是截断。我的做法是在接入时先跑一个探针:构造一段包含特殊标记的长文本,看返回里标记还在不在。这个探针脚本放在第 4 节讲。
第二个是并发限速。文档上写的限速往往是最宽松的那种,实际跑起来更严。而且限速的计量方式差别很大:有的按请求数算,有的按 token 数算,有的是滑动窗口。我的经验是先按文档限速的三分之一跑,稳定了再往上加,一次加 20%,观察有没有 429。这比一上来就压满然后被限速要省事得多。
第三个是协议兼容度。我这套东西的核心假设是大部分渠道能用同一种协议描述。如果某个渠道的协议差异太大,适配器要写上百行,我就会权衡:它的能力是不是别处拿不到?拿得到就砍掉。清单里 C06 和 C11 是仅有的两个自定义协议渠道,保留它们的原因是 C06 的结构化抽取确实好用,C11 的吞吐确实高,值得为它们单独写适配器。
提示:这三个指标建议在接入前花二十分钟测一遍,比接进去之后再返工省太多时间。我当时图快,前三个渠道都是直接接,结果其中一个上下文只有标称的一半,导致一批摘要任务全部重跑。
3. 一键接入的工程实现:从配置文件到路由层
3.1 目录结构与配置格式设计
先看目录结构,这个结构我改过好几版,现在这版是层次最清楚的:
model-router/ ├── config/ │ ├── channels.yaml # 13 个渠道的元信息 │ └── routes.yaml # 业务场景到渠道链的映射 ├── router/ │ ├── __init__.py │ ├── core.py # 路由主逻辑 │ ├── state.py # 渠道健康状态 │ └── adapters/ │ ├── base.py # 适配器基类 │ ├── cc_compat.py # 兼容 CC 协议的通用适配器 │ ├── custom_a.py # C06 专用 │ └── custom_b.py # C11 专用 ├── tools/ │ ├── probe_context.py # 上下文探针 │ └── smoke_test.py # 全渠道冒烟测试 └── .env.examplechannels.yaml的格式设计有个关键决定:配置里绝对不写真实密钥,只写环境变量名。这样配置文件可以进仓库,团队协作的时候也不用互相传密钥。
channels: - id: C04 name: community_trial_a adapter: cc_compat base_url: https://api.example-a.com/v1 auth: type: bearer env: C04_API_KEY model: model-family-a-large limits: rpm: 60 max_context: 32000 max_output: 4096 timeout: connect: 5 read: 60 weight: 6 fallback: [C07, C08] - id: C08 name: cloud_grant_a adapter: cc_compat base_url: https://api.example-b.com/v1 auth: type: bearer env: C08_API_KEY model: model-family-b-chat limits: rpm: 300 max_context: 128000 max_output: 8192 timeout: connect: 5 read: 90 weight: 8 fallback: [C09, C10]注意里面两个细节。fallback是渠道级的兜底链,和业务级的路由链是两回事,前者处理"这个渠道临时挂了",后者处理"这个任务该优先用谁"。timeout分了连接和读取两个值,这一点很关键:连接超时设短(5 秒),能快速判定渠道不可达;读取超时设长(60 到 90 秒),因为大模型出首字确实慢,设太短会误杀。
3.2 适配器与统一请求体
适配器基类定义得很薄,就两个方法,一个负责请求转换,一个负责响应转换。
from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import Any @dataclass class ChatRequest: messages: list[dict] temperature: float = 0.7 max_tokens: int = 1024 stop: list[str] = field(default_factory=list) extra: dict[str, Any] = field(default_factory=dict) @dataclass class ChatResult: text: str channel_id: str model: str prompt_tokens: int = 0 completion_tokens: int = 0 truncated: bool = False class BaseAdapter(ABC): def __init__(self, cfg: dict): self.cfg = cfg @abstractmethod def build_payload(self, req: ChatRequest) -> dict: ... @abstractmethod def parse_response(self, raw: dict) -> ChatResult: ... def build_headers(self) -> dict: import os auth = self.cfg["auth"] key = os.environ.get(auth["env"]) if not key: raise RuntimeError(f"missing env: {auth['env']}") return { "Authorization": f"Bearer {key}", "Content-Type": "application/json", }兼容 CC 协议的渠道,适配器几乎是空实现,直接映射就行。真正需要写代码的是 C06 和 C11。C06 的差异在于它的系统提示词要单独放一个顶层字段,不能进消息数组;C11 的差异在于签名,要对请求体做一次哈希再放进 Header。
class CustomAAdapter(BaseAdapter): """C06:系统提示词独立字段,响应结构也不同。""" def build_payload(self, req: ChatRequest) -> dict: system = "" msgs = [] for m in req.messages: if m["role"] == "system": system = m["content"] else: msgs.append(m) return { "system_prompt": system, "dialog": msgs, "temperature": req.temperature, "max_new_tokens": req.max_tokens, **req.extra, } def parse_response(self, raw: dict) -> ChatResult: text = raw.get("data", {}).get("output_text", "") usage = raw.get("data", {}).get("usage", {}) return ChatResult( text=text, channel_id=self.cfg["id"], model=raw.get("data", {}).get("model_name", ""), prompt_tokens=usage.get("input", 0), completion_tokens=usage.get("output", 0), truncated=raw.get("data", {}).get("finish_reason") == "length", )关于truncated这个字段,我要特别强调。判断输出是否被截断,不能只看文本长不长,要看返回里的结束原因字段。很多渠道在被截断时不会报错,只是把结束原因标成length之类的值。我早期的代码完全没管这个,导致一批长摘要任务的输出全都是半截,还是用户反馈给我的时候才发现。
3.3 失败降级与额度耗尽的兜底策略
路由主逻辑的骨架不复杂,但顺序很讲究。
def chat(req, route_name="default", max_retry=3): chain = build_chain(route_name) # 业务链 + 各级 fallback 展开 errors = [] for cid in chain: if not state.is_available(cid): errors.append(f"{cid}: cooling down") continue for attempt in range(max_retry): try: result = call_channel(cid, req) state.mark_success(cid) if result.truncated: state.mark_soft_fail(cid) return result except RateLimitError: state.mark_rate_limited(cid) break # 限速直接换渠道,不重试 except TransientError as e: state.mark_fail(cid) errors.append(f"{cid}: {e}") sleep(backoff(attempt)) except FatalError as e: state.mark_dead(cid) errors.append(f"{cid}: {e}") break raise AllChannelsFailed(errors)这段代码有三个设计决定值得说。
限速不重试,直接换渠道。这是反直觉的。直觉上碰到限速应该等一会儿再试,但实测下来等待的时间成本远高于换渠道。因为限速窗口通常是分钟级的,你等 30 秒可能还是被拒,不如直接切到备用渠道,任务整体延迟更低。代价是消耗了备用渠道的额度,所以冷却时间要设得合理,我设的是 90 秒,够一个滑动窗口滑过去。
瞬时错误才重试,致命错误直接判死。网络超时、5xx 这类可以重试,重试三次退避递增(1 秒、3 秒、9 秒)。但鉴权失败、模型不存在这类错误重试没有意义,直接标记该渠道不可用,避免每次都浪费一轮重试。
截断算软失败。渠道没报错但输出被截断了,这种情况不把渠道标记为死亡,只降一点权重,因为可能是这次请求的内容太长导致的,换个渠道就正常。
额度耗尽的检测是这套东西里最脏的部分,因为渠道的反馈方式各不相同。我总结出三种信号,同时监控:
- 明确的限流状态码,这个是最好处理的。
- 返回内容里出现余额、配额相关的固定话术,这个需要维护一个关键词表,我目前维护了十几个词。
- 连续 N 次返回空内容,且响应时长明显偏短(因为额度耗尽时服务端不做推理,返回得特别快)。
第三种信号最隐蔽。我在状态管理里加了一个"短响应空内容"计数器,连续三次就认为该渠道额度耗尽,冷却一小时。
3.4 密钥管理:绝不进仓库的三种做法
密钥管理这件事,踩过一次坑就再也不会马虎了。我的一个早期脚本不小心把密钥打印进了日志,虽然只是本地文件,但也吓出一身汗。现在的做法是三层防护。
第一层,配置文件只存环境变量名。前面 YAML 里的env: C04_API_KEY就是这个意思。适配器加载密钥的时候从环境变量读,读不到直接抛异常,绝不使用空值或默认值。
第二层,本地开发用.env文件,并确保它被忽略。.env.example里只放字段名和占位符,真实文件永远在忽略列表里。团队协作的时候大家各自填自己的,谁也不用问谁要。
第三层,日志脱敏。所有出站请求在打日志之前先过一遍脱敏函数,把 Header 里的鉴权字段和 URL 里的查询参数替换成占位符。这个函数只有十几行,但必须有。
import re SENSITIVE = re.compile(r"(sk-|Bearer\s+)[A-Za-z0-9_\-]{8,}") def scrub(text: str) -> str: return SENSITIVE.sub(lambda m: m.group(1) + "***REDACTED***", text)注意:脱敏要在写日志的那一刻做,不要指望事后清理。日志一旦落地就可能被备份、被打包、被转发,事后清理的成本高得多。
4. 实操:从零跑通第一个渠道到十三个全通
4.1 环境准备与最小可运行样例
环境这块不用搞复杂,Python 3.10 以上,装三个包就够:httpx负责 HTTP 请求(比传统库更好控制超时),pyyaml读配置,tenacity可选,不想引入就自己写退避。
python -m venv .venv source .venv/bin/activate pip install httpx pyyaml cp .env.example .env最小可运行样例我建议从本地权重渠道开始,也就是 C01 或 C02。原因是它不依赖任何外部服务,没有鉴权、没有限速,跑通了说明整条链路(配置加载、适配器选择、请求发送、响应解析)都是通的。如果一上来就接外部渠道,出了问题你分不清是自己的代码问题还是渠道的问题。
# smoke_one.py import os, yaml, httpx from dotenv import load_dotenv load_dotenv() cfg = yaml.safe_load(open("config/channels.yaml")) ch = next(c for c in cfg["channels"] if c["id"] == "C02") resp = httpx.post( f"{ch['base_url']}/chat/completions", json={ "model": ch["model"], "messages": [{"role": "user", "content": "用一句话解释什么是适配器模式"}], "temperature": 0.3, "max_tokens": 128, }, headers={"Authorization": f"Bearer {os.environ[ch['auth']['env']]}"}, timeout=httpx.Timeout(connect=5.0, read=60.0), ) print(resp.status_code, resp.json())这段代码跑通之后,把它包进适配器,再接路由层,增量就很小了。
4.2 逐个渠道接入的通用四步法
13 个渠道我用同一套流程接,一步都不跳,这套流程能挡住绝大部分坑。
第一步,探针测上下文。写一个脚本,构造一段 8000 字左右的文本,开头和结尾各埋一个唯一标记,让模型复述结尾的标记。如果它能复述,说明上下文足够;如果复述不出来但返回了正常内容,说明被静默截断了,把配置里的max_context往下调。
def probe_context(cfg, target_marker="MARKER_9F3A"): body = ("这是填充文本。" * 800) + f"请原样输出这个标记:{target_marker}" # ... 发送请求 ... return target_marker in resp_text第二步,探针测限速。并发 3 个请求连发 20 轮,记录 429 的比例。如果 429 超过 10%,把配置里的rpm除以 2 再试。
第三步,验证截断字段。故意把max_tokens设成一个很小的值(比如 16),看返回里结束原因字段是不是标记成截断。这一步是为了确保适配器的parse_response能正确识别截断,别等到生产环境才发现。
第四步,跑一遍真实任务的小样本。从你实际的业务数据里抽 50 条,走完整链路跑一遍,人工检查 10 条结果。这一步能发现前面三步发现不了的问题,比如格式不符合预期、输出带多余前缀、中文编码问题。
这四个步骤走完,一个渠道大概花 15 到 20 分钟。13 个渠道一次性接,一个下午能搞定,看起来慢,但比接完之后反复返工快得多。
4.3 联调验收与压测脚本
全部接完要做一次联调验收,这个环节的目的不是测性能,是确认降级链真的能生效。方法很土但很有效:把主渠道的密钥临时改成一个无效值,跑任务,看它是不是自动切到了 fallback 里的下一个,并且最终的输出质量没有明显下降。
压测脚本我写得很简单,就是多线程跑固定条数的请求,统计几个指标。
| 指标 | 计算方式 | 我的验收阈值 |
|---|---|---|
| 端到端成功率 | 成功条数 / 总条数 | 高于 98% |
| P95 延迟 | 按耗时排序取 95 分位 | 低于 25 秒 |
| 平均降级次数 | 触发 fallback 的总次数 / 总条数 | 低于 0.3 |
| 空内容率 | 返回空字符串的条数 / 总条数 | 低于 0.5% |
| 截断率 | 标记为截断的条数 / 总条数 | 低于 1% |
平均降级次数这个指标容易被忽略,但它很有用。如果这个数持续偏高,说明主渠道的权重设置不合理,或者主渠道本身已经不太行了,该调整配置了。
from concurrent.futures import ThreadPoolExecutor def run_batch(items, workers=8): with ThreadPoolExecutor(max_workers=workers) as pool: results = list(pool.map(handle_one, items)) ok = sum(1 for r in results if r) print(f"success_rate={ok/len(results):.3f}")线程数不要一上来给太大,我从 8 开始,因为大部分渠道的限速在 8 到 10 并发这个量级就开始有感觉了。想压更高并发,正确做法是增加渠道数而不是增加单渠道的并发数,这是这套多渠道路由架构最有价值的地方。
5. 常见问题与排查技巧实录
5.1 高频报错速查表
下面这张表是我这半年攒下来的,覆盖了九成以上的实际报错。
| 现象 | 最可能的原因 | 排查动作 | 处理方式 |
|---|---|---|---|
| 401 / 鉴权失败 | 环境变量没加载或密钥过期 | 打印环境变量名是否存在,不要打印值 | 重新配置密钥,标记渠道状态 |
| 404 找不到模型 | 模型名写错或该模型已下线 | 对照渠道文档核对模型标识 | 更新配置里的模型名 |
| 429 频繁出现 | 实际限速比文档严 | 降到三分之一并发重试 | 下调配置里的 rpm 并拉长冷却 |
| 200 但内容为空 | 额度耗尽或触发内容过滤 | 看响应头是否有配额字段 | 冷却该渠道,走 fallback |
| 响应超时 | 读取超时设得太短 | 统计首字延迟分布 | 读取超时提到 90 秒 |
| 输出半截 | 静默截断 | 检查结束原因字段 | 下调 max_tokens 或换长上下文渠道 |
| 中文乱码 | 编码声明缺失 | 检查响应头 content-type | 强制按 UTF-8 解码 |
| 同一请求结果差异极大 | 温度参数被渠道忽略 | 用温度 0 连发三次对比 | 降低对该渠道的依赖权重 |
表里最后一行的坑我要单独说。有些渠道在温度设成 0 的时候依然不稳定,原因是它的采样实现不同,或者后端有多副本且没做一致性保证。这种情况你没法从代码层面解决,只能在业务层做结果校验,比如结构化抽取的字段缺失就重试,而不是指望渠道给你确定性输出。
5.2 限速与超时的处理经验
限速这块我摸索出一条比较实用的做法:给每个渠道维护一个令牌桶,桶的容量按 rpm 折算成一个更保守的值,比如文档写 60 rpm,我按 40 配桶,留三分之一的余量吸收突发。这样即使在流量高峰也不会撞墙。
令牌桶比简单的定时器好在它能吸收短时突发。比如业务侧突然来了 10 个请求,桶里有 10 个令牌就直接放行,不用排队;而按固定间隔发送的做法会让这 10 个请求排成一条长队,延迟很难看。
超时这块的经验是连接超时和读取超时要分开设,而且读超时要跟输入长度挂钩。固定 60 秒对短请求够用,但对输入几千 token 的请求可能不够。我的做法是基础 60 秒,每 1000 个输入 token 加 10 秒,上限 180 秒。这个公式不精确,但比固定值合理得多。
还有个小技巧:超时之后不要立刻重试同一个渠道。因为超时的原因往往是服务端正在排队处理,立刻重试只会加剧排队。正确的做法是标记一次软失败,换个渠道,把这个渠道放进 30 秒的冷却。
5.3 踩坑记录:三个我实际翻车的地方
第一个坑,把流式响应当成普通响应解析。早期我没注意某个渠道默认开流式,返回的是一段一段的 SSE 格式数据,我按普通 JSON 解析,直接抛异常。后来在适配器里加了个判断:如果响应头的内容类型是事件流,就走逐行解析的路径。这件事的教训是接入任何渠道前先看一眼原始响应,别假设它一定返回完整 JSON。
第二个坑,忽略了赠送额度的有效期。有个渠道我一直在用,用了两个月都正常,某天突然全部失败。排查半天发现是赠送额度到期了,服务端返回的是权限相关的错误码,而不是配额错误。这件事之后我在配置里加了grant_expire字段,手工记录有效期,快到期的时候提前预警。这件事的教训是零成本资源都有到期日,必须在配置里显式记账。
第三个坑,在降级链里放了两个同源的渠道。我以为配置了两级兜底很稳,结果这两个渠道其实共用同一套后端,主渠道挂了备用渠道也一起挂,降级链形同虚设。这件事的教训是降级链里的渠道必须来自不同的能力形态,比如本地权重渠道加社区渠道加云厂商渠道这样的组合,才算真正的冗余。
提示:这三个坑有个共同点——都不是代码写错,而是对渠道的假设错了。所以我在配置文件的注释里专门留了一段,写清楚每个渠道的"我假设它是什么样的",渠道变了先回来改这段注释,提醒自己重新验证。
5.4 清单的长期维护:让渠道有进有出
13 这个数字会变。渠道会关停、会改规则、会收紧免费额度,这是常态,不用纠结。关键是让维护这件事有节奏。
我现在固定每两周做一次全渠道冒烟测试,跑 13 个最小请求,看成功率。连续两次失败的直接从配置里移到channels_retired.yaml,不删,留个记录。同时每周花一点时间看有没有新的渠道可以补进来,补进来的走一遍第 4 节那四步法。
配额的记账也很重要。我在配置里给每个云厂商渠道加了两个字段:quota_total和quota_used,每次请求成功后累加用量,写回本地文件。这个数字不精确,因为我不可能拿到服务端的真实用量,但它能给我一个量级判断——快用完了就提前把权重调低,别等突然失效打乱整个流程。
这套东西跑了大半年,最大的感受是:零成本接入的真正门槛不在技术,而在纪律。适配器写起来都不难,难的是每次都老老实实走那四步验证,难的是把每一次翻车都记进速查表,难的是接受渠道会消失然后主动去维护。我个人在实际操作中的体会是,宁可只维护六个稳定渠道,也别贪多去接二十个半死不活的,因为每一个不稳定的渠道都会变成一条随时会炸的引线,而排查它花的时间,远超过省下来的那点钱。