廉价模型优先、失败时升级：构建可审计的双层 LLM 流水线

Full translation

完整中文译文

我在每个发布周期都会看到同一个模式：一款新的性价比模型发布， discourse 就炸开了锅，48 小时内我的信息流里有一半人在宣称它是万金油替代品。这个说法甚至可能是对的。但问题是，发这些言论的人没有一个能告诉你：它是否适合你的具体工作负载。而大多数时候，这才是唯一重要的问题。

一个更健康的心智模型是：不要把模型选择当作一次性的购物决策，而是把它当作一种运行时策略。默认把工作路由到便宜选项，用非模型的方式检查输出，只有当检查失败时才为重型选项付费。下面是这个策略的一个可工作实现，以及一套测量规范，把这个想法从一个直觉变成一个可审计的系统。

两条通道，不是两个选择

我的配置故意做得很无聊：

通道 A（优先尝试）：当前低成本或免费选项。我有意不在本文中指定具体的 checkpoint——模型名称的衰减比下面的代码还快，你应该持续对当月新发布的模型重新做基准测试。

通道 B（兜底）：更贵、能力更强的模型，只有在通道 A 的输出未通过客观检查时才调用。

成本方面有一点说明：我使用 MonkeyCode 迭代这个管道，写稿时它提供免费模型访问和免费服务器选项，所以实验不会产生账单。披露：本文是作为 MonkeyCode 产品推广的一部分准备的。这个管道本身是提供商无关的——任何支持 OpenAI 兼容 chat API 的服务都可以直接插入，所以把端点当作配置，而不是承诺。

核心规则：升级需要一个非模型裁判

最重要的设计决策：永远不要让模型判断自己的输出（或者同类的输出）是否足够好。LLM 报告的置信度与正确性无关。取而代之的是，只有当外部的、确定性的检查失败时才升级——测试套件、解析器、schema 验证器、与预期输出的 diff。

下面是一个紧凑的实现。标准库加 requests，完全可重运行，每次路由决策都会写入 JSONL 审计日志：

# lane_router.py
import hashlib, json, subprocess, time
from dataclasses import dataclass, asdict

import requests

@dataclass
class RouteRecord:
    job_id: str
    lane: str            # "A" or "B"
    fell_back: bool
    check_ok: bool
    seconds: float
    prompt_fingerprint: str

def chat(endpoint: str, model: str, prompt: str) -> str:
    resp = requests.post(
        f"{endpoint}/v1/chat/completions",
        json={
            "model": model,
            "messages": [{"role": "user", "content": prompt}],
            "temperature": 0.2,
        },
        timeout=120,
    )
    resp.raise_for_status()
    return resp.json()["choices"][0]["message"]["content"]

def objective_check(job: dict, candidate: str) -> bool:
    """Deterministic validation only. No LLM-as-judge allowed here."""
    mode = job["check"]
    if mode == "pytest":
        path = job["write_to"]
        with open(path, "w") as fh:
            fh.write(strip_fences(candidate))
        run = subprocess.run(job["command"], shell=True,
                             capture_output=True, timeout=90)
        return run.returncode == 0
    if mode == "valid_json":
        try:
            parsed = json.loads(strip_fences(candidate))
        except json.JSONDecodeError:
            return False
        required = job.get("required_keys", [])
        return all(k in parsed for k in required)
    if mode == "regex":
        import re
        return re.fullmatch(job["pattern"], candidate.strip()) is not None
    raise ValueError(f"unsupported check: {mode}")

def strip_fences(text: str) -> str:
    """Pull code out of a markdown fence if present; else return as-is."""
    if "```" not in text:
        return text
    block = text.split("```")[1]
    lines = block.splitlines()
    if lines and lines[0].strip().isalpha():  # language tag line
        lines = lines[1:]
    return "\n".join(lines)

def route(job: dict, lanes: list[dict], log_path: str = "routes.jsonl") -> RouteRecord:
    fp = hashlib.sha256(job["prompt"].encode()).hexdigest()[:12]
    fell_back = False
    for idx, lane in enumerate(lanes):
        start = time.time()
        candidate = chat(lane["endpoint"], lane["model"], job["prompt"])
        elapsed = round(time.time() - start, 2)
        passed = objective_check(job, candidate)
        if passed or idx == len(lanes) - 1:
            record = RouteRecord(
                job_id=job["id"],
                lane=lane["label"],
                fell_back=fell_back,
                check_ok=passed,
                seconds=elapsed,
                prompt_fingerprint=fp,
            )
            with open(log_path, "a") as fh:
                fh.write(json.dumps(asdict(record)) + "\n")
            return record
        fell_back = True

lanes = [
    {"label": "A", "endpoint": "https://lane-a-endpoint", "model": "current-budget-model"},
    {"label": "B", "endpoint": "https://lane-b-endpoint", "model": "premium-model"},
]

job = {
    "id": "csv-to-json-migration-014",
    "prompt": (
        "Convert the transformation in migrate.py so it emits newline-delimited JSON. "
        "Return only the complete updated file inside a code fence."
    ),
    "check": "pytest",
    "write_to": "migrate.py",
    "command": "python -m pytest tests/test_migrate.py -q",
}

print(route(job, lanes))

审计日志才是真正的交付物

路由代码大概是一个周末的工作量。真正产生复利价值的是 routes.jsonl。经过几周的真实流量后，你可以计算出那些原本纯属猜测的东西：

每个任务家族的回退率。把记录按任务类型分桶，然后看 fell_back。如果数据格式化任务在通道 A 上 92% 的情况都能通过，那么通道 A 在那里是一个合理默认值。如果多文件重构任务有 60% 的情况会失败，那通道 A 在这类任务上就是虚假经济——你在为一次注定失败的首次尝试买单，同时还在大多数调用上增加了额外延迟。

每个成功任务的实际成本。一次成功的通道 A 调用花 $X，优于一次通道 B 调用花 $5X。一次失败后触发通道 B 的通道 A 调用花 $6X。按类别做算术；答案在每个类别之间经常是不同的——这正是路由而非全局选一个模型的核心意义。

新版本的回归测试。当下一个 checkpoint 发布时，把通道 A 指向它，然后重放你记录的任务语料（保留 prompts，或者在日志旁边保存一份脱敏语料）。对比你自有任务在不同模型版本上的回退率，胜过任何公开排行榜。

我坚持的操作规范

裁判必须是确定性的，这是底线。一旦你加上"让第二个 LLM 来评分"，你就用额外的延迟重建了置信度问题。测试、解析器、schema、正则、精确匹配——不要任何会幻觉的东西。

给 prompts 做指纹而不是必须存储它们。SHA-256 前缀让你可以去重和关联，而不必在日志中持久化潜在敏感内容。

最多两条通道。三级级联读起来像聪明的工程，实际表现却像一个延迟生成器加上更大的调试面。如果通道 B 检查失败，那是人的问题，不是第三个模型的问题。

这个方案的局限性

没有 oracle 的任务。这个方案的有效性取决于检查有多强。有测试的代码、结构化提取、格式转换——很适合。有开放性文本、架构判断、"为高管总结这个"——没有确定性的裁判，靠感觉路由那些任务恰恰是建设这个管道要消除的猜测。

同步、延迟敏感路径。一次失败的通道 A 尝试加通道 B 重试可能大致翻倍墙上时钟时间。把路由器放在后台 worker、批处理任务和异步管道中，直到你有 p99 数据证明可以这样做。

假设免费一直免费。免费模型访问和免费服务器选项是关于现在的声明，不是合约。上面端点即配置的设计是你的保险单；不要把提供商的慷慨硬编码到你的架构里。

小样本量。15 个路由任务什么也证明不了。在一个任务家族有大约 50+ 条记录之前，我不信任回退率数字，而且即使有我也看分布，不只看均值。

如果你的大部分工作负载是无法验证的生成，或者你运行的每件事都有严格的延迟预算，老实说——跳过路由器。在这种情况下直接选强模型是更简单也更正确的工程决策。

一个具体的起点

拉取你最近大约 50 条真实 prompts，把它们分为"有确定性检查"和"没有确定性检查"两类，然后把有检查的子集通过双通道设置跑一周。如果你想让测量阶段零成本，MonkeyCode 的免费模型访问和免费服务器选项可以作为通道 A 和 host，在收集数据的同时——而且由于日志格式是提供商无关的，当你把通道指向别处时，你学到的东西都可以迁移。

"便宜模型够不够用？"这个问题是有答案的，它就在你的 prompt 历史里。测量它；不要把决策外包给发布周的舆论。