通宵批处理拆给便宜模型和强模型：十行代码能写出的分流规则

为什么要把批处理拆开

通宵批处理通常包含大量条目，但并非所有条目都需要同等强度的模型。把简单条目交给便宜模型、难条目交给强模型，可以在保证质量的同时控制成本。本文给出一套可落地的分流规则和十行左右的代码示例，并说明如何在同一把 API key 上做成本对比。

分流的核心思路

分流的本质是给每个条目打一个“难度分”，然后按阈值路由。难度分可以来自：

• 规则判断：文本长度、是否包含多步推理关键词、是否需要结构化输出、是否涉及代码或数学。
• 小模型预判：用便宜模型先给条目打一个 0-1 的难度分或分类标签。
• 历史反馈：之前失败或重试过的条目自动标记为“难”。

关键是把“难”的定义收敛成可执行的信号，而不是凭感觉。

可落地的分流规则

以下规则可以单独或组合使用，按优先级从高到低：

• 长度规则：超过一定字符数的条目直接走强模型，因为长上下文更容易出错。
• 关键词规则：包含“证明”“推导”“多步”“比较方案”“代码审查”等词的条目走强模型。
• 输出结构规则：要求严格 JSON、表格或多字段对齐的条目，优先走强模型或先让强模型生成模板。
• 预判规则：用便宜模型对条目做二分类（简单/复杂），置信度低于阈值时升级到强模型。
• 重试规则：便宜模型输出未通过校验（格式错误、空结果、明显矛盾）时，自动重试一次强模型。

这些规则不需要复杂基础设施，一个函数就能实现。

十行代码示例

下面是一个最小分流器，输入条目文本，输出应使用的模型名。实际使用时把模型名替换为聚合站支持的标识即可。

def route(item, cheap="deepseek-chat", strong="claude-sonnet"):
    text = item.get("text", "")
    if len(text) > 2000:
        return strong
    hard_words = ["证明", "推导", "多步", "比较方案", "代码审查"]
    if any(w in text for w in hard_words):
        return strong
    if item.get("require_json") and item.get("fields", 0) > 5:
        return strong
    return cheap

这段代码只做规则路由。如果要加入预判，可以在 return cheap 之前调用一次便宜模型做分类，置信度低时返回 strong。

同一把 key 上的成本对比

聚合站的一个 API key 可以调用多个模型，因此成本对比可以在同一个账户内完成。对比时关注两个量：

• 每条目的实际扣费：按官方价乘以倍率计算。用户侧统一按官方价 ×1.3 扣费，所以模型之间的相对成本等于官方价之比。
• 分流后的加权成本：假设简单条目占比 p，难条目占比 1-p，则加权成本约为 p × 便宜模型官方价 ×1.3 + (1-p) × 强模型官方价 ×1.3。

举例来说，如果便宜模型官方价是强模型的十分之一，且 80% 条目走便宜模型，那么加权成本大约是全部走强模型的 0.8×0.1 + 0.2×1 = 0.28 倍。这个比例取决于实际占比和模型价差，需要用自己的数据测算。

如果同时贡献 key，贡献侧按官方价 ×1.1（优质 ×1.2）返 USDC，这部分收益与消费侧分开计算，不影响上面的分流逻辑。

实施建议

• 先跑一批历史数据，统计简单/难条目的比例和便宜模型的失败率，再定阈值。
• 把分流决策和实际扣费记录到日志里，便于事后核对加权成本。
• 对失败条目保留原始输入和模型输出，方便调整规则。
• 不要一开始就追求精细分类，先用长度和关键词两条规则跑通，再逐步加入预判和重试。

小结

按难度分流不需要复杂架构：几条规则加一个路由函数就能落地。配合同一把 key 的多模型调用，可以在一个账户内完成成本对比和效果验证。关键是先用数据确定“难”的信号，再逐步迭代规则。