在 Messages API 中复现 Claude 的智能体搜索基准分数

本教程面向有编程基础的开发者，详细讲解如何在公共 Messages API 上复现 Claude 的智能体搜索基准分数（如 DeepSearchQA、BrowseComp）。文章强调测试框架配置对长周期智能体任务的关键影响，并逐步介绍程序化工具调用、思考与预算控制、上下文压缩、往返循环等核心环节。通过实际代码示例，读者将学会构建一个可复现的智能体搜索循环，理解每个配置选择背后的原理，并能轻松适配其他基准测试。教程还提供了常见错误排查和生产环境优化建议，帮助开发者避免常见陷阱，提升评估准确性。

技术背景

随着 AI 智能体在复杂任务中的广泛应用，准确评估其性能变得至关重要。Anthropic 发布的智能体搜索基准分数（如 DeepSearchQA、BrowseComp）为衡量智能体在深度研究任务中的能力提供了标准。然而，第三方评估分数往往偏低，主要原因在于测试框架配置不当。许多 API 参数在短对话中无关紧要，但在智能体运行数十次工具调用、处理数十万 token 时，却成为决定成败的关键。本教程旨在揭示这些关键配置，帮助开发者复现官方分数，并理解其背后的工程原理。

核心原理

核心原理在于通过程序化工具调用（PTC）和上下文压缩来优化长周期智能体任务的效率与准确性。PTC 允许模型在代码执行沙箱内调用 web_search 和 web_fetch，仅将摘要返回对话，避免大量页面内容淹没上下文。上下文压缩通过自定义指令确保在压缩后保留原始问题和答案格式要求，防止智能体在长任务中迷失方向。此外，通过设置思考模式、努力级别和任务预算，控制模型的推理深度和输出长度，使其在有限资源下高效完成任务。

架构分析

系统架构围绕一个客户端驱动的循环构建：客户端发送请求，服务器端执行工具调用，返回 pause_turn 或 end_turn 状态。客户端负责维护消息历史、处理压缩、持久化代码执行容器，并循环调用直到 end_turn。关键组件包括：工具定义（code_execution、web_search、web_fetch）、请求参数（thinking、output_config、context_management）、消息往返逻辑（truncate_to_last_compaction）、以及评分器（模型即评判者的 F1 评分）。设计决策包括使用流式传输以应对长响应、设置高重试次数以应对限流、以及固定评分模型以保证可比性。

完整实战教程

复现 Claude 的智能体搜索基准分数

Claude 的已发布智能体搜索分数（DeepSearchQA、BrowseComp）可以在公共 Messages API 上复现。关键在于测试框架配置：一些对短对话无关紧要的 API 参数，在智能体运行 30 多次工具调用、处理数十万 token 时变得至关重要。

第三方评估分数偏低的一个常见原因是测试框架配置。本教程将逐一介绍每个参数并解释其重要性。

成本与运行时间：本笔记本中的实时单元格运行 3 个演示问题（大约 5-10 分钟，API 花费几美元）。复现完整的 900 个问题基准测试需要几百美元和几个小时（在中等并发下）；本笔记本不会重新执行完整运行。

学完本教程后，你将能够：

先决条件

所需知识

所需工具

推荐

设置

%%capture
%pip install -U "anthropic>=0.111.0" pandas python-dotenv

import time

import anthropic
import pandas as pd
from dotenv import load_dotenv
from utils.agentic_search import (
    SAMPLE_QUESTIONS,
    USER_PROMPT_TEMPLATE,
    extract_result_tag,
    grade_response,
    summarize_response,
)

load_dotenv()

MODEL = "claude-sonnet-5"
# Hold the grader model fixed so scores are comparable across the model under test.
GRADER_MODEL = "claude-opus-4-6"

# With server-side tools, a single streamed call can run for many minutes of
# in-sandbox tool execution before the first token arrives. We raise the read
# timeout to 60 minutes; with streaming this is a per-event ceiling, not a
# whole-response one, so it only matters for the gap before the first event.
client = anthropic.Anthropic(
    max_retries=20,
    timeout=anthropic.Timeout(5.0, read=3600.0, write=600.0, pool=600.0),
)

任务

以下是 DeepSearchQA 格式的三个演示问题。每个问题都可以回答，但需要经过多轮搜索、阅读和交叉引用之后才能得出答案：

for q in SAMPLE_QUESTIONS:
    print(f"[{q['example_id']}] ({q['answer_type']})")
    print(f"  Q: {q['problem']}")
    print(f"  A: {q['answer']}\n")

我们将每个问题包装在一个简短的提示中，要求Claude自由规划，并将其最终答案放在<result>标签中，以便评分器可以干净地提取它：

print(USER_PROMPT_TEMPLATE.format(question=SAMPLE_QUESTIONS[0]["problem"]))

模型需要规划、搜索网络、获取并阅读页面、运行计算，并持续这样做直到它确信，然后输出一个我们可以评分的干净最终答案。这句话中的几乎每个部分都是天真的测试框架会悄悄失分的地方。我们将逐步构建这个循环。

上述三个问题是为本手册编写的（并非来自 DeepSearchQA-900），以保持实时单元快速且廉价。对于真正的基准测试，请参阅下面的*数据集*。

数据集

| 基准 | 规模 | 获取方式 |

|---|---|---|

| BrowseComp | 1,266 个问题 | Wei 等人，2025 (OpenAI) |

| DeepSearchQA | 900 个问题 | Kaggle Benchmarks (论文) |

测试框架期望每一行都有 problem、answer 和 answer_type 字段；加载时请相应映射源模式。

1. 工具：程序化工具调用

最大的效率杠杆是 程序化工具调用（PTC）：不是让 Claude 直接调用 web_search 和 web_fetch 并通过 API 往返传递每个结果，而是让 Claude 在 code_execution 沙箱中编写 Python，并从该代码内部调用搜索/获取。只有代码打印的摘要返回对话。

对于深度研究，这至关重要。一个简单的问题可能触发 50 多次获取。没有 PTC，每个页面正文都会进入上下文；有了 PTC，模型在沙箱中阅读它们，只呈现所需内容。

我们通过将 code_execution 列为唯一可直接调用的工具，并将 web_search/web_fetch 标记为可从代码执行中调用（其结果从响应中排除）来实现这一点：

TOOLS = [
    {
        "type": "code_execution_20260521",
        "name": "code_execution",
    },
    {
        "type": "web_search_20260318",
        "name": "web_search",
        "max_uses": 10_000,
        "allowed_callers": ["code_execution_20260521"],
        "response_inclusion": "excluded",
    },
    {
        "type": "web_fetch_20260318",
        "name": "web_fetch",
        "max_uses": 10_000,
        "max_content_tokens": 1_000_000,
        "allowed_callers": ["code_execution_20260521"],
        "response_inclusion": "excluded",
    },
]

BETAS = [
    "compact-2026-01-12",
    "task-budgets-2026-03-13",
]

有两个值得指出的细节：

2. 思考、努力、预算与压缩

四个请求级参数告诉 Claude *要付出多大努力* 以及 *如何管理自身的上下文*：

THINKING = {"type": "adaptive"}

OUTPUT_CONFIG = {
    "effort": "max",
    "task_budget": {"type": "tokens", "total": 3_000_000},
}

# Match the system-card configuration. To reproduce a different model's score,
# use the effort tier and compaction trigger from that model's model card.
COMPACTION_TRIGGER = 200_000

COMPACT_INSTRUCTIONS = (
    "Your summary MUST begin by restating the user's ORIGINAL QUESTION "
    "verbatim and in full, wrapped in <original_question> and "
    "</original_question> tags. Then summarize the research progress so "
    "far: key sources found, data extracted, partial answers, and what "
    "remains to be done. Do NOT call any tools. Your summary MUST also "
    "include this instruction verbatim: 'Provide your final answer "
    "wrapped in <result> and </result> tags.'"
)

CONTEXT_MANAGEMENT = {
    "edits": [
        {
            "type": "compact_20260112",
            "trigger": {"type": "input_tokens", "value": COMPACTION_TRIGGER},
            "instructions": COMPACT_INSTRUCTIONS,
        }
    ]
}

为什么 COMPACT_INSTRUCTIONS 如此重要

当对话达到触发阈值时，API 会对其进行压缩：一次独立的模型调用会总结历史记录，然后代理从总结继续。默认的总结提示是任务无关的；它不知道*对于基准测试*，原始问题和答案格式指令是绝对必须保留的两件事。自定义 instructions 就是告诉它的方式。

没有它们，压缩后的代理只有*它发现了什么*的总结，而没有*它被问了什么*，在长问题上它经常要求用户重述问题，这得零分。

这种影响随着压缩触发的频率而扩大：在低触发阈值下最大（单个问题可能被压缩多次），在本文使用的 200K 触发阈值下较小，但无论如何，这是测试框架中最便宜的保险。

注意：最新的 Claude 模型卡报告 DeepSearchQA *没有*压缩（模型有 100 万 token 的上下文窗口，因此 100 万预算的任务适合一个窗口），而 BrowseComp 报告*有*压缩，总预算为 300 万。我们在这里使用 300 万预算的 BrowseComp 配置，因此压缩是有意义的；在 DeepSearchQA 上，大多数问题在 100 万以下就能完成，永远不会触发压缩。

其他三个参数：

| 参数 | 原因 |

|---|---|

| thinking: adaptive | 让 Claude 在工具调用结果令人惊讶时思考，否则跳过思考 |

| effort | 模型卡上的数字是在模型最高努力级别下报告的 |

| task_budget.total | 告诉 Claude 它在所有轮次和压缩中的累计输出预算，以便它可以调整节奏而不是过早放弃 |

3. 往返循环

使用服务端工具时，每次 API 调用都可以在内部运行多次工具迭代，然后返回 stop_reason="pause_turn" 以设置检查点。我们在客户端的工作只是：获取响应内容，将其作为助手轮次追加，然后再次调用。我们一直继续，直到 stop_reason="end_turn"。

一个辅助函数可以保持请求体在多次轮次中整洁：

def truncate_to_last_compaction(messages: list[dict]) -> list[dict]:
    """Drop history that precedes the most recent compaction block.

    The server already discarded that history when it compacted; resending it is
    pure overhead and on long runs can exceed the 32 MB request-body limit.
    `messages[0]` (the original user prompt) is always kept so the request still
    starts with a user turn.
    """
    for mi in range(len(messages) - 1, 0, -1):
        content = messages[mi].get("content")
        if not isinstance(content, list):
            continue
        for bi in range(len(content) - 1, -1, -1):
            blk = content[bi]
            if isinstance(blk, dict) and blk.get("type") == "compaction":
                trimmed = {**messages[mi], "content": content[bi:]}
                return [messages[0], trimmed, *messages[mi + 1 :]]
    return messages

现在来看循环本身。它是一个单一的函数，虽然很长，但每一行都在做真实基准测试运行所需的事情：

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")

重试

SDK 会在连接错误、408、409、429 和 ≥500 响应时自动重试，采用指数退避，并遵循 API 在限流响应中发送的 Retry-After 头。默认重试次数为 2；对于完整的基准测试运行，我们在客户端（见上文 *设置*）设置了 max_retries=20，以便运行中途的短暂过载不会将问题得分为零。

发现静默工具限流：服务器工具限流与模型令牌限制是分开的。耗尽它们不会引发 API 错误；too_many_requests 会作为工具结果错误出现在响应中，Claude 通常会在沙箱内重试。如果完整基准测试运行得分意外偏低且没有异常，请检查 result["transcript"] 中是否有打印 rate_limit_error 或 too_many_requests 的代码执行单元格。客户端的解决方法是降低并发（或联系 Anthropic 支持以提高工具限流）；客户端重试无济于事，因为 API 调用返回了 200。
同样，将每个问题的 sample() 调用包装在 try/except 中，以便单个 stop_reason="refusal" 或不可恢复的错误将该问题得分为零，而不是中止整个运行。

4. 端到端运行一个问题

让我们运行第一个问题。预计此单元格需要一两分钟：在单个 stream() 调用背后，Claude 正在多次编写和执行搜索代码。

result = sample(SAMPLE_QUESTIONS[0]["problem"])

print(f"tool_calls={result['tool_calls']}  elapsed={result['elapsed_s']:.0f}s")
print(
    f"tokens: in={result['usage']['input_tokens']:,}  out={result['usage']['output_tokens']:,}  "
    f"cache_read={result['usage']['cache_read_input_tokens']:,}"
)
print()
print("Final <result>:", result["result"])

转录内容展示了工作的形态。每一轮都是在沙箱内执行一批工具调用后的一个 pause_turn 检查点：

for line in result["transcript"]:
    print(line + "\n")

5. 评分

基准测试评分使用模型即评判者的 F1 评分器。该评分器会查看问题、标准答案以及提取出的 <result> 文本，然后报告哪些标准答案项被找到，以及出现了哪些不在标准答案集中的额外项。据此我们计算精确率、召回率和 F1 值。

有两件事需要正确处理：

评分提示词如下：

~~~text

你的任务是评估给定响应是否得出了正确答案。

问题：<question>{question}</question>

正确答案（类型：{answer_type}）：<correct_answer>{answer}</correct_answer>

待评估响应：<response>{response}</response>

对于每个预期答案项，指出它是否出现在响应中。

然后列出响应中不在正确答案列表中的任何答案。

措辞不需要完全匹配。

请以以下精确的 XML 格式回复：

<evaluation>

<explanation>一句话</explanation>

<correctness_details>

<item answer="expected_item_1" correct="true|false"/>

</correctness_details>

<excessive_answers>

<item>额外项（如果有）</item>

</excessive_answers>

</evaluation>

~~~

解析以及精确率/召回率/F1 的计算位于 utils/agentic_search.py 中；这里我们只需调用它：

grade = grade_response(client, GRADER_MODEL, SAMPLE_QUESTIONS[0], result["text"])
print(f"answer    : {grade['extracted_answer']}")
print(f"precision : {grade['precision']:.2f}")
print(f"recall    : {grade['recall']:.2f}")
print(f"F1        : {grade['f1']:.2f}")

6. 运行演示集

现在，将这三个问题及其使用情况和 F1 分数放在一个对比表中。（要运行完整的 900 题基准测试，请将 SAMPLE_QUESTIONS 替换为完整数据集并添加并发；请参阅*后续步骤*。）

rows = []
for q in SAMPLE_QUESTIONS:
    r = sample(q["problem"])
    g = grade_response(client, GRADER_MODEL, q, r["text"])
    rows.append(
        {
            "id": q["example_id"],
            "tool_calls": r["tool_calls"],
            "in_tokens": r["usage"]["input_tokens"],
            "out_tokens": r["usage"]["output_tokens"],
            "f1": g["f1"],
            "answer": g["extracted_answer"][:60],
        }
    )

df = pd.DataFrame(rows)
print(df.to_string(index=False))
print(f"\nMean F1: {df['f1'].mean():.2f}")
# in_tokens is the uncached billed input; with caching on, most of the actual
# context is cache reads (tracked separately in usage.cache_read_input_tokens).
print(f"Total tokens: {df['in_tokens'].sum():,} in / {df['out_tokens'].sum():,} out")

7. 扩展到完整基准测试

使用本笔记本的配置在完整的 DeepSearchQA-900 和 BrowseComp-1266 数据集上运行，可以复现已发布的模型卡分数。有关各模型已发布的智能体搜索分数，请参阅各模型的 Claude 模型卡。

8. 适配 BrowseComp

BrowseComp 的问题是单答案（“哪部 1995 年的电影……”）而非集合型，因此唯一的测试框架改动在于评分器：

from utils.agentic_search import grade_browsecomp

# Grading: a single-letter judge (A=correct, B=incorrect, C=abstain).
# accuracy = fraction of "A" verdicts.
bc_grade = grade_browsecomp(client, GRADER_MODEL, SAMPLE_QUESTIONS[0], result["text"])
print(bc_grade)

工具、测试版和 sample() 循环保持不变。如果你想严格复现，请查看模型卡，了解 BrowseComp 报告时使用的确切努力级别和压缩触发条件；§2 中的值是很好的默认值。

这就是重点：一旦循环正确，更换基准测试就只是数据集、几行配置和一个评分器的问题。

回顾

你构建了一个智能体搜索循环，它：

后续步骤

常见问题

常见错误包括：1) 未设置自定义压缩指令，导致压缩后丢失原始问题，智能体要求重述问题而得分零；2) 未使用程序化工具调用，导致上下文被大量页面内容淹没，超出模型上下文窗口；3) 未正确往返 pause_turn 响应，导致状态丢失；4) 未持久化代码执行容器，导致跨轮次状态丢失；5) 评分时使用被测模型而非固定评分模型，导致分数不可比。排查方法：检查转录中是否有 rate_limit_error 或 too_many_requests，降低并发或联系支持；确保 try/except 包裹 sample() 调用，避免单点失败中止整个运行。

企业应用扩展

生产环境优化包括：1) 使用并发执行（如 ThreadPoolExecutor）加速完整基准测试，注意客户端线程安全；2) 调整并发数以平衡速度和限流，SDK 内置重试机制可处理 429/529 错误；3) 考虑多智能体编排模式，可能以更低延迟获得更高准确率；4) 适配新基准时，只需更换数据集、配置和评分器，循环逻辑保持不变；5) 参考模型卡片获取每个基准的精确配置（如努力级别、压缩触发条件），确保复现准确性。

总结

本教程面向有编程基础的开发者，详细讲解如何在公共 Messages API 上复现 Claude 的智能体搜索基准分数（如 DeepSearchQA、BrowseComp）。文章强调测试框架配置对长周期智能体任务的关键影响，并逐步介绍程序化工具调用、思考与预算控制、上下文压缩、往返循环等核心环节。通过实际代码示例，读者将学会构建一个可复现的智能体搜索循环，理解每个配置选择背后的原理，并能轻松适配其他基准测试。教程还提供了常见错误排查和生产环境优化建议，帮助开发者避免常见陷阱，提升评估准确性。

内容来源与验证

继续系统学习

站内还有更多完整实战教程

加入会员，解锁全部教程、完整代码与持续更新内容。

加入会员
查看全部模型评估与微调教程