第 10 章：评测与成本

出处

来源：evals/ · cost_optimization.ipynb · 状态：已读

一句话

先有能过关的 agent 和一把尺子，再谈省钱。 评测要能便宜地反复跑；成本看的是 pass rate × 单次任务美元，不是 token 单价。换小模型是清单最后一项，因为它直接削智能天花板。

核心概念

Eval 四件套，打分方式决定你能不能常跑

misc/building_evals.ipynb：输入（常是模板变量）→ 模型输出 → golden → 分数。出题是一次性固定成本；打分是每次重跑都要付的成本，所以设计时就该偏向可自动打分。

方式	何时用	代价
代码（精确匹配 / 正则 / 包含）	能把任务收成闭集或强制格式	最快最稳
人	几乎任何开放任务	慢、贵，尽量别依赖
模型 grader	语气、开放 QA 等曾经要人的活	要抽检 grader 本身

经验：任务尽量做成可自动评（改选择题是常见手法）；grader 好不好只能试；宁可题多质量一般，不要题少但每题雕花——分布要贴近真实。Claude 可以帮你出题，但 golden 仍要人盯。

misc/generate_test_cases.ipynb：从带 的模板抽变量，让模型编合成值。有黄金例子就喂进去；可改模型自己的 planning 再 prefill（例如「文档用编号行」）。一半战场是逼真输入，另一半是 golden 输出——后者可让模型先写你再改。prompt cache 让多-shot 变便宜，这才是往 prompt 里堆例子的时机。

tool_evaluation/tool_evaluation.ipynb：多个 agent 独立跑同一份评测文件里的任务，盯的是 工具有没有被正确调用，不是最终散文好不好听。第 3 章的计算器在这里才算被验收。

官方深搜分数是 harness，不是模型玄学

evals/agentic_search/reproduce_agentic_search_benchmarks.ipynb：DeepSearchQA / BrowseComp 的公开分 可以在 Messages API 上复现。第三方测低，常见原因是 harness 配置——短对话无所谓的参数，30+ 工具、几十万 token 时会变成扣分项。

现场 cell 只跑 3 道演示题（数美元、数分钟）；全量 900 题是几百美元 + 数小时。杠杆按 notebook 顺序：

1. PTC：code_execution 为唯一 direct 工具，web_search / web_fetch 从代码里调，response_inclusion: "excluded"，页面正文不进对话。一道题 50+ fetch 时这是最大效率差。
2. Compaction 指令：默认摘要不知道基准里「原题 + 答案格式」必须活下来。不写 custom instructions，压完后模型会请你重述问题 → 零分。低 trigger 时效应最大。
3. thinking: adaptive、effort、budget：和第 4、9 章同一组旋钮。

模型卡：DeepSearchQA 常 不开 compaction（1M 窗够用）；BrowseComp 用 3M budget 开 compaction。notebook 用后者的配置，好让 compaction 真的发生。

数据集：BrowseComp 1266（OpenAI）；DeepSearchQA 900（Kaggle）。harness 期望 problem / answer / answer_type。

成本：Applied AI 的清单，模型选择在最后

cost_optimization/cost_optimization.ipynb 把风尚从 tokenmaxxing 改成 budgetmaxxing：不要靠降级模型或砍用例，改周围架构，走到智能–成本 Pareto。整本约 $40 API，不必从头跑。

清单顺序：

1. 先在强模型上跑通，做 eval，记基线
2. Prompt cache
3. 输入：让模型用工具发现上下文，而不是 12k 手册整本塞 system
4. Agent 环：别让多轮历史线性膨胀（compact / 清工具 / PTC）
5. 收紧输出约束
6. 非交互走 Batch
7. 最后才选更便宜的模型和 effort

度量是 usage → 美元。看 cost per task 不看 cost per token：贵模型若回合少，可能更便宜。案例：Acme 车险理赔 agent，system 里嵌 ~12k underwriting 手册，8 个调查工具 + 4 个终局 verdict。基线 Opus effort=high、无 cache、无上下文管理：10/10，~$0.29/任务。百万件索赔，每省 1 分钱 ≈ 年省 $1 万。质量条近 100%——标错比推理费贵，所以优化时不准掉点。

finetuning/finetuning_on_bedrock.ipynb：Bedrock 上微调 Haiku，JSONL system + messages。这是成本杠杆之一，不是默认路径。先 harness 再权重。

对照 recipe（默认折叠；机制片段摘自官方 notebook，全文在 GitHub）：

misc/building_evals.ipynb — 评测本身要设计，打分必须可重复

GitHub 全文

# Define our input prompt template for the task.
def build_input_prompt(animal_statement):
    user_content = f"""You will be provided a statement about an animal and your job is to determine how many legs that animal has.

    Here is the animal statment.
    <animal_statement>{animal_statement}</animal_statment>

    How many legs does the animal have? Return just the number of legs as an integer and nothing else."""

    messages = [{"role": "user", "content": user_content}]
    return messages

# Get completions for each input.
# Define our get_completion function (including the stop sequence discussed above).
def get_completion(messages):
    response = client.messages.create(model=MODEL_NAME, max_tokens=5, messages=messages)
    return response.content[0].text


# Get completions for each question in the eval.
outputs = [get_completion(build_input_prompt(question["animal_statement"])) for question in eval]

# Let's take a quick look at our outputs
for output, question in zip(outputs, eval, strict=False):
    print(
        f"Animal Statement: {question['animal_statement']}\nGolden Answer: {question['golden_answer']}\nOutput: {output}\n"
    )

# Get completions for each input.
# Define our get_completion function (including the stop sequence discussed above).
def get_completion(messages):
    response = client.messages.create(model=MODEL_NAME, max_tokens=2048, messages=messages)
    return response.content[0].text


# Get completions for each question in the eval.
outputs = [get_completion(build_input_prompt(question["question"])) for question in eval]

# Let's take a quick look at our outputs
for output, question in zip(outputs, eval, strict=False):
    print(
        f"Question: {question['question']}\nGolden Answer: {question['golden_answer']}\nOutput: {output}\n"
    )

misc/generate_test_cases.ipynb — 合成数据填模板，仍要人过眼

GitHub 全文

def get_test_data(prompt_template, examples, custom_planning=None):
    """Generate test data using the Claude API."""
    synth_eval_prompt_ready = format_prompt_template_for_synth_evals(prompt_template, examples)

    messages = [
        {
            "role": "user",
            "content": synth_eval_prompt_ready,
        }
    ]
    if custom_planning:
        messages.append(
            {
                "role": "assistant",
                "content": custom_planning,
            }
        )

    message = (
        CLIENT.messages.create(
            max_tokens=4000,
            messages=messages,
            model=MODEL_NAME,
            temperature=1,
        )
        .content[0]
        .text
    )

    return message

# We'll use this function to sample Claude's response to the filled-in template,
# once we have our example values/test case.


def call_claude_with_template(prompt_template, variables):
    """Call Claude with a filled prompt template."""
    filled_template = prompt_template
    for var, value in variables.items():
        filled_template = filled_template.replace(f"{{{{{var}}}}}", value)

    message = (
        CLIENT.messages.create(
            max_tokens=4000,
            messages=[
                {
                    "role": "user",
                    "content": filled_template,
                }
            ],
            model=MODEL_NAME,
            temperature=0.7,
        )
        .content[0]
        .text
    )

    return message

tool_evaluation/tool_evaluation.ipynb — 评的是工具有没有被正确调用

GitHub 全文

client = Anthropic()
model = "claude-sonnet-4-6"


def agent_loop(prompt: str, tools: list[dict[str, Any]] = None) -> tuple[str, dict[str, Any]]:
    """Simplified agent class for tool evaluation"""
    messages = [{"role": "user", "content": prompt}]

    response = client.messages.create(
        model=model,
        max_tokens=4096,
        system=EVALUATION_PROMPT,
        messages=messages,
        tools=tools,
    )

    messages.append({"role": "assistant", "content": response.content})

    # Track tool calls with timing
    tool_metrics = {}  # {tool_name: {"count": N, "durations": [X1, X2, ...]}}

    def _prepare_tool_result(tool_use_id, tool_result):
        return {
            "role": "user",
            "content": [
                {
                    "type": "tool_result",
                    "tool_use_id": tool_use_id,
                    "content": tool_result,
                }
            ],
        }

    while response.stop_reason == "tool_use":
        tool_use = next(block for block in response.content if block.type == "tool_use")
        tool_name = tool_use.name

        tool_start_ts = time.time()
        try:
            # Note: eval is used here for demonstration purposes to dynamically call tool functions.
            # In production, use a safer dispatch mechanism like a dictionary of functions.
            tool_response = eval(  # noqa: S307
                f"{tool_name}(**tool_use.input)"
            )  # Call the tool function with its input
        except Exception as e:
            tool_response = f"Error executing tool {tool_name}: {str(e)}\n"
            tool_response += traceback.format_exc()
        tool_duration = time.time() - tool_start_ts

        # Update tool metrics
        if tool_name not in tool_metrics:
            tool_metrics[tool_name] = {"count": 0, "durations": []}
        tool_metrics[tool_name]["count"] += 1
        tool_metrics[tool_name]["durations"].append(tool_duration)

        # Prepare tool result and append to messages
        messages.append(_prepare_tool_result(tool_use.id, tool_response))
        response = client.messages.create(
            model=model,
            max_tokens=4096,
            system=EVALUATION_PROMPT,
            messages=messages,
            tools=tools,
        )
        messages.append({"role": "assistant", "content": response.content})

    response = next(
        (block.text for block in response.content if hasattr(block, "text")),
        None,
    )
    return response, tool_metrics

evals/agentic_search/reproduce_agentic_search_benchmarks.ipynb — 官方分数在 Messages API 上怎么复现

GitHub 全文

def sample(question: str, *, max_turns: int = 100) -> dict:
    """Run one question to completion through the agentic search loop."""
    # cache_control on the user message turns on prompt caching for the whole
    # request, so the server-side tool loop caches between iterations.
    messages = [
        {
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": USER_PROMPT_TEMPLATE.format(question=question),
                    "cache_control": {"type": "ephemeral"},
                }
            ],
        }
    ]
    request = {
        "model": MODEL,
        "max_tokens": 64_000,
        "tools": TOOLS,
        "betas": BETAS,
        "thinking": THINKING,
        "output_config": OUTPUT_CONFIG,
        "context_management": CONTEXT_MANAGEMENT,
        "messages": messages,
    }

    usage = {"input_tokens": 0, "output_tokens": 0, "cache_read_input_tokens": 0}
    transcript: list[str] = []
    tool_calls = 0
    start = time.time()

    for turn in range(max_turns):
        # Stream so the per-event read-timeout applies, not a per-response one.
        with client.beta.messages.stream(**request) as stream:
            response = stream.get_final_message()

        usage["input_tokens"] += response.usage.input_tokens
        usage["output_tokens"] += response.usage.output_tokens
        usage["cache_read_input_tokens"] += (
            getattr(response.usage, "cache_read_input_tokens", 0) or 0
        )
        tool_calls += sum(
            1 for b in response.content if getattr(b, "type", "") == "server_tool_use"
        )
        transcript.append(f"--- turn {turn} ---\n" + summarize_response(response))

        # Persist the code-execution container so state survives across turns.
        if getattr(response, "container", None) is not None:
            request["container"] = response.container.id

        # Round-trip: append assistant content, drop pre-compaction history.
        messages.append(
            {
                "role": "assistant",
                "content": [b.model_dump(exclude_none=True) for b in response.content],
            }
        )
        messages[:] = truncate_to_last_compaction(messages)

        if response.stop_reason == "end_turn":
            final_text = "".join(
                b.text for b in response.content if getattr(b, "type", "") == "text"
            )
            return {
                "text": final_text,
                "result": extract_result_tag(final_text),
                "turns": turn + 1,
                "tool_calls": tool_calls,
                "elapsed_s": time.time() - start,
                "usage": usage,
                "transcript": transcript,
            }
        if response.stop_reason not in ("pause_turn", "max_tokens"):
            raise RuntimeError(f"unexpected stop_reason={response.stop_reason}")

    raise RuntimeError(f"did not finish within {max_turns} turns")

cost_optimization/cost_optimization.ipynb — 同一任务扫配置，找 Pareto，不找最聪明

GitHub 全文

claim = EVAL_CLAIMS[0]
followups = [
    "Is this loss covered under the policy?",
    "What payout amount would you recommend?",
]


def converse(label, *, cache=False):
    """Ask each follow-up in a growing conversation. The nonce keeps the prefix
    cold: fresh per run when caching, fresh per call when not (so the uncached
    variant's call 2 can't read call 1's implicit entry either)"""
    run_nonce = f"(cache demo session {time.time_ns()})\n\n"
    messages = [{"role": "user", "content": f"Claim {claim['id']}: {claim['summary']}"}]
    total = 0.0
    for i, q in enumerate(followups, 1):
        prefix = run_nonce if cache else f"(cache demo session {time.time_ns()})\n\n"
        messages.append({"role": "user", "content": q})
        r = client.messages.create(
            model=MODEL,
            max_tokens=512,
            system=prefix + SYSTEM_PROMPT,
            tools=TOOLS,
            tool_choice={"type": "none"},  # keep tools in the (cached) prefix; answer in prose
            messages=messages,
            **({"cache_control": {"type": "ephemeral"}} if cache else {}),
        )
        total += print_usage(f"{label} · claim A · call {i}", r.usage)
        messages.append({"role": "assistant", "content": r.content})
    return total


# BEFORE — no cache_control, every turn re-processes the full manual + tools
reset_usage_header()
t1 = converse("no caching")
print(f"→ total ${t1:.4f}\n")

# AFTER — top-level auto caching
reset_usage_header()
t2 = converse("auto cache", cache=True)
print(f"→ total ${t2:.4f}")
print(f"→ {(1 - t2 / t1):.0%} cheaper vs no caching")

finetuning/finetuning_on_bedrock.ipynb — 微调是成本杠杆之一，不是默认路径

GitHub 全文

import json

sample_dataset = []
dataset_path = "datasets/json_mode_dataset.jsonl"
with open(dataset_path) as f:
    for line in f:
        sample_dataset.append(json.loads(line))

print(json.dumps(sample_dataset[0], indent=2))

关键洞察

没有评测的第 5、8、9 章都只是演示。好评测的特征是：能在每次改 prompt / 换模型 / 开 cache 时自动跑，并且同时报质量和美元。DeepSearch 那份把第 3、4 章的 PTC 和 compaction 从「省钱技巧」升级成「分数的一部分」——harness 就是产品。

成本清单把「换 Haiku」放最后，是因为那是唯一直接砍天花板的杠杆。前面六项（cache、工具发现、压历史、收输出、batch）回收的是 没有换来质量的美元。

个人思考

第 1 章分类从 70% 到 95% 已经是评测驱动；本章把同一纪律扩到 agent 和账单。下一章是运行时看见这些数字：Admin API、session 封顶、地理钉扎。评测是实验室；可观测是值班。

相关

• 00-Cookbooks阅读导读
• 05-Agent工作流
• 09-Managed-Agents
• 11-生产可观测
• Uber：Running a Software Factory Efficiently at Uber Scale