合集

Claude Agent SDK 实战

已更新 27 章
阅读进度 89%

第 23 章

研究 Agent:从“搜一下”到可复现报告

研究系统最难的不是“找到”,是“证明你找到的东西值得信”。

一个研究 Agent 的 demo,通常很惊艳:给它一个问题,它搜五个网页,写出一篇像模像样的报告。

上线后,问题变成:

  • 同一个问题跑两次,结论不一样,来源也不一样;
  • 十个子 Agent 同时抓网页,被目标站点限流;
  • 报告里的 URL 有一半打不开;
  • 两个来源矛盾,模型取了个“平均值”;
  • 成本超了,因为每个子 Agent 都把整页 HTML 塞进上下文。

这些都不是“模型不会研究”,是 Harness 没控制好“来源、上下文、预算、验证”。


23.1 研究系统的四层

text
POST /research
   │
   ▼
[Planner] ──► [Subquestion Store]
   │
   ▼
[Fan-out] ──► [Researcher × N] ──► [Source Registry]
   │                                  │
   │                                  ▼
   │                             [Fetch Cache]
   ▼
[Claim Extractor] ──► [Citation Verifier] ──► [Report Store]

四层:

  • Planner:把大问题拆成 3–5 个子问题;
  • Researcher:每个子问题一个 Agent,只回“结论 + 来源 + 置信度”;
  • Source Registry:所有 URL/抓取结果/摘要,去重、缓存、可追溯;
  • Verifier:URL 可访问、日期在范围、结论有来源;没有的标“未验证”。

SDK 负责 Researcher 的循环;Planner/Registry/Verifier 是 Harness 的外围。


23.2 Harness 控制点

六问本项目的实现对应章节
看见什么提纲 + 来源摘要 + 引用;不把全文塞进主上下文第 6、7 章
能做什么WebSearch/WebFetch,只读;不给 Bash/Write第 4、11 章
允许做什么域名白名单/黑名单、网络出口、子 Agent 工具面第 5、12、21 章
什么时候停子问题数、来源数、轮数、预算第 18 章
失败怎么办抓取重试/缓存/降级;部分报告第 20 章
做完留下什么来源表、claim 表、报告、引用校验第 7、19 章

研究 Agent 的“看见什么”尤其重要:它决定你是不是把 20 页网页塞进了主上下文。


23.3 看见什么:提纲先于搜索

不要让模型“先搜了再说”。先让它产出提纲:

python
@dataclass
class Subquestion:
    id: str
    question: str
    sources_needed: int = 2
    status: str = "pending"       # pending/running/done/failed

@dataclass
class ResearchPlan:
    question: str
    subquestions: list[Subquestion]
    max_sources: int = 10

Planner 的输出不是自由文本,是这个结构。主循环拿到结构后,才能:

  • 限制 fan-out 数量;
  • 限制每个子问题的来源数;
  • 在 max_sources 用完后停止;
  • 把“未完成”显式标出来。

这对应第 6 章的“上下文预算”:先决定“要什么”,再决定“读多少”。


23.4 能做什么:只读工具 + 子 Agent

Researcher 的工具面要窄:

python
RESEARCHER = AgentDefinition(
    description="调研员:回答一个子问题,返回结论 + 来源 URL + 置信度。",
    prompt=(
        "只回答交给你的子问题。\n"
        "输出:结论 / 证据(最多 3 条,每条带 URL)/ 冲突 / 置信度。\n"
        "找不到来源就写'未验证';不要编 URL。"
    ),
    tools=["WebSearch", "WebFetch"],
    maxTurns=8,
)

父 Agent:

python
options = ClaudeAgentOptions(
    cwd=str(ROOT),
    agents={"researcher": RESEARCHER},
    tools=["Agent", "WebSearch", "WebFetch"],
    allowed_tools=["Agent", "WebSearch", "WebFetch"],
    permission_mode="dontAsk",
    max_turns=20,
    max_budget_usd=1.0,
)

不给子 Agent Bash/Write:它只负责查,不负责执行/落盘。第 8 章讲过子 Agent 的权限边界;第 21 章讲过“工具就是权限”。


23.5 允许做什么:来源也要有边界

研究 Agent 的“权限”不只是工具,还有网络:

  • 域名白名单:只允许公司知识库/指定站点;
  • 域名黑名单:社交平台/论坛/不可信源;
  • 出口代理:记录每个请求,防 SSRF;
  • 子 Agent 不能访问内网/元数据地址。

这些规则放在工具层,不放在提示词里:

python
def allowed_url(url: str) -> bool:
    host = urlparse(url).hostname or ""
    if host in BLOCKED:
        return False
    if ALLOWLIST and host not in ALLOWLIST:
        return False
    return not host.startswith(("127.", "10.", "192.168.", "169.254."))

模型可以“建议”抓某个 URL;工具决定“能不能抓”。第 5 章的权限管线,在这里换成“网络权限”。


23.6 什么时候停:来源数、轮数、预算

研究任务很容易“越搜越多”。三个刹车:

  • subquestions 数量(3–5);
  • sources_needed(每个 2–3);
  • max_budget_usd / max_turns。

Fan-out 也有限制:

python
async def fan_out(plan: ResearchPlan, limit: int = 4):
    sem = asyncio.Semaphore(limit)
    async def one(q: Subquestion):
        async with sem:
            return await run_researcher(q)
    return await asyncio.gather(*(one(q) for q in plan.subquestions))

第 8 章的并行 demo 是“三个 researcher 同时跑”;生产要加 Semaphore,否则十个子 Agent 会把目标站点打挂。


23.7 失败怎么办:缓存 + 部分报告

抓取失败是常态:超时、403、HTML 解析失败、内容为空。

三个处理:

  • 缓存:URL → 内容哈希/抓取时间/摘要;同一 URL 不重复抓;
  • 重试:指数退避,只对 5xx/超时;403 不重试;
  • 降级:某个子问题失败,报告里写“未完成/未验证”,不要让整个任务失败。
python
@dataclass
class Source:
    url: str
    fetched_at: str
    content_hash: str
    summary: str
    status: str = "ok"        # ok/error/blocked

status 进报告:读者能看到“这条结论基于 2 个来源,其中 1 个抓取失败”。第 20 章的“降级/部分成功”,在这里是产品能力。


23.8 做完留下什么:claim、citation、report

研究系统的“产物”不是一段文本,是三张表:

  • sources:URL、抓取时间、哈希、摘要、状态;
  • claims:结论、来源 ID、置信度、冲突;
  • reports:问题、claim 列表、未验证项、生成时间。

引用校验:

python
def verify_citations(claims: list[Claim], sources: dict[str, Source]) -> list[str]:
    problems = []
    for c in claims:
        if not c.sources:
            problems.append(f"{c.id}: no source")
            continue
        for sid in c.sources:
            s = sources.get(sid)
            if s is None or s.status != "ok":
                problems.append(f"{c.id}: bad source {sid}")
    return problems

它只验“有没有、能不能访问”,不验“结论对不对”。事实正确性要人工抽查/规则比对;别把“引用覆盖”当“事实正确”。

第 19 章的 trace、第 17 章的评测,在这里变成“来源表 + claim 表 + 报告”的可复现数据。


来源质量:不是所有 URL 等权

一条 claim 有三个来源,不代表它更可信:三个都是二手博客,不如一个官方文档。

一个最小评分:

python
SOURCE_RANK = {"official": 3, "repo": 2, "news": 1, "blog": 0}

def source_score(source: Source) -> int:
    return SOURCE_RANK.get(source.kind, 0) + (1 if source.published else 0)

kind 要由抓取层/人工标注,不由模型自己说“这是官方”。published 用来标时间敏感信息;没有日期的来源,置信度要降。

来源质量不直接等于事实正确,但它决定“值不值得花时间验证”。

23.9 评测:引用覆盖与事实抽查

研究 Agent 的评测,不能只看“读起来像不像”。至少四个指标:

  • 引用覆盖:多少 claim 带可访问 URL;
  • 事实正确性:抽查 claim 与原文是否一致;
  • 冲突处理:有矛盾时有没有标出来;
  • 成本/时延:来源数、抓取页数、子 Agent 数、墙钟。

引用覆盖可以先写成一行代码:

python
def citation_coverage(claims: list[dict]) -> float:
    if not claims:
        return 0.0
    return sum(1 for c in claims if c.get("sources")) / len(claims)

它只测“有没有来源”,不测“来源对不对”。事实正确性仍然要抽查 URL、日期和原文。

可复现性也要进评测:同一问题、同一计划、同一来源哈希,应该能重放出同一份 claim 表。否则“上次那个结论”永远找不回来。


23.10 一次实测的教训

旧书 ch36 做过一组 3 次运行的对比,任务/判分和本章不同,只看量级:

  • 把结果压成“标题 + 摘要”后,主上下文峰值从约 10.5k 降到约 1.6k,但完成度从 3/3 掉到 2/3;
  • 并行 fan-out 的墙钟是 8.52s,串行是 6.54s;并行没有赢。

两个教训:

  • 摘要省上下文,但会丢信息;要保留“原文链接 + 关键引用”,不能只留摘要;
  • 并行不是默认更快;子 Agent 有启动/汇总开销,来源少时串行更稳。

这一章没有重跑 5 子问题研究;上面的数字只作量级参考,不是本章成功率。


23.11 下一章:业务 Agent

研究 Agent 控制“信息”,业务 Agent 控制“钱和客户”。

下一章把六问换到客服:只读/写分离、审批、脱敏、恢复、多租户。


本章代码 可运行引用/来源控制点:https://github.com/kevinkelin/claude-agent-sdk-book/blob/master/code/projects/prod/research.py fan-out 参考:https://github.com/kevinkelin/claude-agent-sdk-book/blob/master/code/sdkbook/research_fanout.py 说明:本章是生产架构与关键控制点;没有用当前凭证重跑 5 子问题研究。