第 22 章：大数据上的 AI Map Reduce——为什么便宜 100 倍

当判断的规模从几百条涨到几百万条，"用大模型逐条判"这条路会先在账单上撞墙。决策层的账本和生成模型完全不同：它不逐 token 生成，只在封闭选项上出一张概率分布，于是同一个任务可以铺成一条分块、并行、聚合的流水线。本章讲清这个 Map Reduce 模式，以及它为什么能把成本压低一两个数量级。

一笔账：贵在哪里

用生成模型判一条记录"是否相关"，你付两笔钱：把 state 塞进上下文的输入 token，以及模型吐出来的输出 token。判别任务的答案本来只有一个词，可生成模型很少只说这一个词——它习惯先写"根据分析，这条记录……"，再给结论。你为那些不需要的文字付了费，还得写解析器把它们剥掉，而解析器又要处理"模型没按格式输出"的兜底。

把这件事放到 2,300 篇论文的规模上，问题就变成了：你按"生成一段文字"的价格，买了 2,300 次"选一个标签"的服务。

决策层不这么算账。它输出类型化决策，不生成 token 序列，只在给定的选项上出一张概率分布。社区的 jev-fanout-bench 测过这个成本结构：在 jev-1.13-20260917 上跑 2,976 次调用，每次请求大致是 261 个固定输入 token，而且每次请求的隐含成本不随问题数量变化——问一个问题还是八个问题，请求的固定开销几乎一样；把八个问题打成一批（batch）时，输入 token 的中位节省是 76–86%。

这就是决策层在规模化场景里的第一层优势：成本不再和"生成了多少字"挂钩，而是和"判了多少次"挂钩，而"判多少次"又能靠批处理大幅摊薄。

这个结构差异换算成真实账单，也出现在更复杂的场景里。Halv 报告说，把 Jev 放进它的 agent 循环后，在 SWE-rebench 的 Astra 配对集上把 AI agent 的成本降低了 57.1%。注意这不是把任务换成更简单的判定，而是在一个原本就存在的大模型 agent 里，用决策层替换掉一部分"每次判断都问一次大模型"的开销——省下来的钱，来自少生成的那些文字和多摊薄的固定开销。

批处理：把固定开销摊薄

既然单次请求的开销近乎固定，让它一次干更多活就是最直接的省钱手段。jev-fanout-bench 已经把这条曲线测出来了：把多个问题放进一次请求时，八个问题相比逐问题单独调用，输入 token 的中位节省是 76–86%，而且批量与单条的答案差异，和重复请求之间的噪声是同一量级——也就是说，批处理几乎不牺牲答案质量。

def ask_many(rows, question, choices=None, batch_size=8):
    """把一批 state 放进一次请求，而不是一条条调用。"""
    results = []
    for chunk in chunks(rows, batch_size):
        d = client.systemone(
            state={"items": [r["text"] for r in chunk]},   # state 本身可以是列表
            questions={f"q{i}": question for i in range(len(chunk))},
            instructions="对每个 item 独立回答对应的问题，不要相互参考。",
        )
        results.extend(align(chunk, d))
    return results

需要提醒的是：批处理摊薄的是固定输入开销，并不是所有成本都随批大小线性下降——问题越多，state 也越大，请求本身也会变慢。所以批量大小是个要按 state 长度和延迟预算实测的参数，不是越大越好。fanout 基准里的"八个问题"是一个实测出来的甜点，不是上限。

Map Reduce 的三步

大规模判定的骨架和经典的 Map Reduce 同构：

• 分块（Split）：把大输入切成一批可独立判定的单元。按文档切、按行切、按段落切都行，关键是每一块要小到能塞进一次请求的 state。
• 映射（Map）：对每一块发起一次类型化判定，拿回带概率的决策。
• 聚合（Reduce）：把所有局部决策汇总成整体结论——投票、加权，或按层级逐层归并。

flowchart LR
    D["大输入
（数千～数百万条）"] --> S["分块 Split"]
    S --> M["Map：逐块类型化判定"]
    M -->|"单块失败"| RT["重试 / 降级"]
    RT --> M
    M --> A["Reduce：聚合"]
    A --> V["投票 / 加权 / 层级归并"]
    V --> R["整体决策"]

和经典 Map Reduce 一样，"映射"这一步天然可并行：每一块互不依赖，可以并发、可以分批、可以失败重试。这正是决策层能压住成本的结构性原因——判定被拆成了可并行的小请求，而不是串行的长生成。

一次判定有多快

并行之后能不能真的省时间，取决于单次判定有多快。DuckDB 的 duckdb-jev 扩展把 Jev 判定直接作用在结构化的 SQL 行上，对一个 1,000 行的 Choice 分类任务，实测 1,943 rows/s；另一个社区 DuckDB 扩展的实测则是 1,000 行约 10 秒。数量级都落在"每秒上千次判定"这一档。

把这些数字放在一起，规模化判定的成本区间大致是：单纯的批量判别比生成式替代方案便宜 10× 到 26×（jevextract 在 Gemini 3.5 Flash 上测到 10–26× 更低成本；4,034 star 的 agent 工具项目 treg 在生产里报告约 10× 更便宜、18× 更快）；当替换的对象是一个前沿模型的逐条推理时，差距能拉到两个数量级（第 23 章里 Vals AI 的独立评测报告约 1/500 的成本）。标题里的"100 倍"不是精确常数，而是这个区间的中位叙事——判定的成本结构本身决定了它便宜，省多少取决于你替换掉的是什么。

主线案例：2,300 篇论文，83 秒，14 美分

一个公开的生产用例给出了这条流水线跑起来的真实手感：把约 2,300 篇 AI 研究论文做整理归类，全程约 83 秒、总计 $0.14。折算下来，每篇论文约 0.036 秒、约 $0.00006——这个价位在逐条调用生成模型的预算里基本不可能出现。

用 Map Reduce 的三步拆开看，它是这样跑的：按论文分块，每篇发起一次类型化判定（一个 Choice 定主题、一个 Noul 判相关性），最后在代码里按主题归并、按置信度分流。

from typesafe import TypesafeClient

client = TypesafeClient()

TAXONOMY = ["检索增强", "对齐与安全", "多模态", "推理与规划", "训练与微调", "评测"]

def classify_paper(paper: dict) -> dict:
    d = client.systemone(
        state=f"标题：{paper['title']}\n摘要：{paper['abstract']}",
        questions={
            "topic": {
                "type": "choice",
                "question": "这篇论文的研究主题属于哪一类？",
                "choices": TAXONOMY,
            },
            "relevant": {
                "type": "noul",
                "question": "这篇论文是否与 AI Agent 的研究方向直接相关？",
            },
        },
        instructions="你是 AI 研究综述的整理助手，只看标题与摘要判断，不要脑补。",
    )
    return {
        "id": paper["id"],
        "topic": d["topic"].choice,
        "topic_p": max(d["topic"].probabilities.values()),
        "relevant": d["relevant"].value,
        "relevant_p": d["relevant"].probability,
    }

一次调用拿两个判断（主题 + 相关性），各自带概率。真正让成本落下来的，是把上面这个函数批量化：不要把 2,300 次调用一条条串起来，而是按块并发。

from concurrent.futures import ThreadPoolExecutor

def run_pipeline(papers: list[dict], workers: int = 32) -> list[dict]:
    with ThreadPoolExecutor(max_workers=workers) as pool:
        return list(pool.map(classify_paper, papers))

# 2,300 篇论文 → 一批批并发判定 → 得到带概率的行
rows = run_pipeline(papers)

83 秒这个数字，本质上就是"每篇 36 毫秒的判定 × 并发"摊出来的。

聚合策略：投票、加权、层级

Map 完成之后，Reduce 才决定整体结论。三种常用策略，选哪种取决于你的判定单元和整体目标的关系。

投票（voting）——同一问题在多个块上重复判定，按多数或按概率求和定论。适合"整体属性由各块联合决定"的任务，比如判断一份长文档的整体语气。

加权（weighted）——不是每块都一样重要，用权重把概率加权后再汇总。

def weighted_verdict(rows: list[dict], weights: dict[str, float]) -> float:
    """把逐块的 P(true) 按块权重加权，得到整体的可信度。"""
    num = sum(r["relevant_p"] * weights[r["id"]] for r in rows)
    den = sum(weights[r["id"]] for r in rows)
    return num / den

层级归并（hierarchical）——先把小块聚成中块，再把中块聚成大块，每层都做一次判定。适合"章节—文档—语料"这种多级结构，也是论文整理案例里按主题归并的隐含做法。

from collections import defaultdict

def reduce_papers(rows: list[dict]) -> dict[str, list[str]]:
    buckets: dict[str, list[str]] = defaultdict(list)
    for r in rows:
        if not r["relevant"]:
            continue
        # 置信度太低的不硬塞：单独放一边，交人工或做二次判定
        if r["topic_p"] < 0.6:
            buckets["待复核"].append(r["id"])
            continue
        buckets[r["topic"]].append(r["id"])
    return buckets

注意这里的两条纪律：低置信不硬判（概率低于阈值就进"待复核"，而不是赌一个标签），以及聚合在代码里做（模型只填每一块的槽位，归并逻辑由你的代码百分之百确定）。

失败与重试：把判定当成不可靠的网络调用

百万级判定里，一定会有失败的请求。把判定看作一次网络调用：可能超时、可能限流、可能瞬时错误。正确姿势是幂等重试 + 有限降级，而不是让整条流水线挂掉。

import time

class TransientError(Exception):
    """超时、429、5xx 这类可重试的错误。"""

def classify_with_retry(paper: dict, attempts: int = 3) -> dict:
    for i in range(attempts):
        try:
            return classify_paper(paper)          # 幂等：同一 state 重复判定安全
        except TransientError:
            time.sleep(0.2 * (2 ** i))            # 指数退避
    # 重试用尽：给出显式的"未知"，不编造结论
    return {"id": paper["id"], "topic": "未知", "topic_p": 0.0,
            "relevant": None, "relevant_p": 0.0}

两个要点。其一，判定是幂等的：同一段 state 重复送进去，choice 一致、概率只有细微浮动，所以重试不会污染结果。其二，耗尽重试后返回显式的"未知"，而不是一个看起来像结论的东西——下游能据此把它挑出来、补判或剔除。

两个可复用的骨架：GroundingJev 与 jevextract

Map Reduce 的价值在于骨架可以跨任务复用。数据标注与抽取这两类最"吃规模"的任务，各有现成的参照。

GroundingJev 是视觉标注里的一个漂亮例子。它是一个受 Jev 启发的 Qwen3.5-0.8B 模型，把一张图和一句指代表达映射成一个前向传播里并行吐出的四个边界框坐标，相比它的自回归基座模型报告了 8.61× 的推理加速。它印证的是同一件事：把"逐步生成"换成"一次并行判定"，速度就上来了。

jevextract 则把 Map Reduce 用在了信息抽取上。它的做法是：代码先提出候选 span（带精确 offset），Jev 对每个 span 回答一个 Choice（属于哪个 schema 类，或不属于任何类），外加每个句级类一个 Noul，保留超过逐类阈值的答案并标记边缘案例。它发布的基准显示，成本比 LangExtract 跑 Gemini 3.5 Flash 低 10–26×，但 F1 也更低（在双语 jx-bench 上 84.2 对 88.5）。这个取舍值得记住——决策层买到的是规模和成本，不是无条件更高的精度；抽取任务尤其如此。

同类的骨架还有几个可以参考：jev-research-pipeline 每天抓取时按（论文，研究问题）跑 Noul 闸门与 Score 维度，超过代码侧阈值才交给 Qwen 写笔记，离线测试重放录制好的 cassette；jev-align 用 Choice/Score/Boolean 逐行评估 CSV、Parquet、JSONL，把模糊样本与审计样本送人工，再用被人接受的人工标签通过 GEPA 优化判定定义；jlink 用 Noul 成对判定做记录链接；jevgrep 对日志流（含 tail -f）逐行一个 Noul，按概率阈值打印。它们都在做同一件事：把"逐条语义判定"做成一条可并行、可重试、可聚合的管道。

医疗领域还有一个把规模与验证结合的例子：jev-medhallu-benchmark 在 Stanford MedHELM 的 MedHallu（1,000 条测试项）上，用单个 Noul 判断一段回答是否歪曲了它的 PubMed 摘要。Jev 得分 92.9%（四个快速 LLM 是 92.4–95.1%），中位延迟 204 毫秒、每 1,000 次检查 $0.03；让 Jev 去裁定它至少 90% 有把握的那 37% 条目，在保持各 LLM 准确率的同时把 LLM 调用量减少了 37%。这正是 Map Reduce 与下一章要讲的验证结合的样子。

小结

• 生成模型的成本随"写多少字"上涨，判别任务的成本只随"判多少次"上涨；决策层不逐 token 生成，这是它在规模上便宜的根本原因。
• jev-fanout-bench 测得每次请求约 261 个固定输入 token、成本不随问题数变化，八个问题打一批可省 76–86% 输入 token。
• Map Reduce 三步：分块、映射（逐块类型化判定）、聚合（投票 / 加权 / 层级归并）；映射天然可并行，这是速度与成本的来源。
• 主线案例：2,300 篇论文整理，83 秒、$0.14；实测降本从约 10×（treg）到 26×（jevextract）再到约 500×（Vals AI）不等，取决于替换对象。
• 把判定当作幂等的网络调用：限次重试、指数退避，耗尽后返回显式的"未知"，绝不编造结论。
• 规模不等于精度：jevextract 省了 10–26× 成本，F1 也从 88.5 降到 84.2；决策层买到的是规模与成本，不是无条件更高的准确率。

批量判定解决的是"判得多、判得省"，但判得对不对、稳不稳，是另一件事。下一章讲通用验证——如何把"验证"抽象成可复用的能力，以及怎么验证决策层自己。