SDK:一行代码研究 Agent

用最简代码构建研究助手

🌐 查看英文原文 | 源码 Notebook


构建一个单行代码研究 Agent

研究任务消耗大量专家时间:市场分析师手动收集竞争情报,法务团队跟踪法规变化,工程师在文档中调查 bug 报告。核心挑战不在于找到信息,而在于根据刚刚发现的内容知道下一步该搜索什么。

Claude Agent SDK 让你能够构建无需预定义工作流即可自主探索外部系统的 Agent。与遵循固定步骤的传统工作流自动化不同,研究 Agent 会根据发现的内容调整策略——追踪有价值的线索、综合相互冲突的信息来源,并判断何时已经收集了足够的信息来回答问题。

通过本 cookbook,你将能够:

  • 用几行代码构建一个能够自主搜索和综合信息的研究 Agent

这一基础适用于任何所需信息无法预先获取的任务:竞争分析、技术故障排除、投资研究或文献综述。

为什么需要研究 Agent?

研究是一个理想的 Agent 应用场景,原因有二:

  1. 信息不是自包含的。输入问题本身不包含答案。Agent 必须与外部系统(搜索引擎、数据库、API)交互来收集所需信息。
  2. 路径在探索过程中涌现。你无法预先确定工作流。Agent 应该搜索公司财务信息还是监管文件,取决于它对商业模式的发现。最优策略通过调查逐步显现。

在最简单的形式下,研究 Agent 搜索网络并综合发现。下面,我们将用 Claude Agent SDK 内置的网络搜索工具,仅用几行代码来实现这一点。

注意:你也可以查看 Claude Code 内置工具的完整列表

前提条件

在遵循本指南之前,请确保你具备:

所需知识

  • Python 基础知识——熟悉 async/await、函数和基本数据结构
  • 对 Agent 模式有基本了解——如果你刚接触 Agent,我们建议先阅读 构建有效的 Agent

所需工具

推荐: * 熟悉 Claude Agent SDK 的概念 * 了解 LLM 中的工具使用模式

环境配置

首先,安装所需的依赖:

%%capture
%pip install -U claude-agent-sdk python-dotenv

注意:确保你的 .env 文件包含:

ANTHROPIC_API_KEY=your_key_here

加载环境变量并配置客户端:

from dotenv import load_dotenv

load_dotenv()

MODEL = "claude-opus-4-6"

构建你的第一个研究 Agent

让我们从最简单的实现开始:一个能够搜索网络并综合发现的研究 Agent。使用 Claude Agent SDK,这只需要几行代码。

关键是 query() 函数,它创建一个无状态的 Agent 交互。我们将为 Claude 提供一个工具 WebSearch,让它根据我们的研究问题自主决定何时以及如何使用它。

from utils.agent_visualizer import (
    display_agent_response,
    print_activity,
)

from claude_agent_sdk import ClaudeAgentOptions, query

messages = []
async for msg in query(
    prompt="Research the latest trends in AI agents and give me a brief summary and relevant citiations links.",
    options=ClaudeAgentOptions(model=MODEL, allowed_tools=["WebSearch"]),
):
    print_activity(msg)
    messages.append(msg)
🤖 Using: WebSearch()
🤖 Using: WebSearch()
🤖 Using: WebSearch()
✓ Tool completed
✓ Tool completed
✓ Tool completed
🤖 Thinking...
display_agent_response(messages)
<IPython.core.display.HTML object>

这里发生了什么:

  • query() 创建单轮 Agent 交互(无对话记忆)
  • allowed_tools=["WebSearch"] 授予 Claude 搜索网络的权限,无需请求批准
  • Agent 自主决定何时搜索、运行什么查询以及如何综合结果

来自 utils.agent_visualizer 的可视化工具: - print_activity() - 实时显示 Agent 的操作(工具调用、思考过程) - display_agent_response() - 将最终响应渲染为样式化的 HTML 卡片 - visualize_conversation() - 创建完整对话的时间线视图

就这样!仅用几行代码就构建了一个功能完备的研究 Agent。该 Agent 将搜索相关信息,追踪有价值的线索,并提供带有引用的综合摘要。

query() 函数创建一个无状态的 Agent 交互。每次调用都是独立的——没有对话记忆,没有之前查询的上下文。这使它非常适合需要快速答案而无需维护状态的一次性研究任务。

工具权限的工作原理:

allowed_tools=["WebSearch"] 参数授予 Claude 搜索权限而无需请求批准。这对于自主运行至关重要:

  • Allowed tools(允许的工具)- Claude 可以自由使用这些工具(本例中为 WebSearch)
  • Other tools(其他工具)- 可用但使用前需要批准
  • Read-only tools(只读工具)- 如 Read 等工具默认始终允许
  • Disallowed tools(禁用工具)- 将工具添加到 disallowed_tools 以将它们从 Claude 的上下文中完全移除

何时使用无状态查询:

  • 上下文无关的一次性研究问题
  • 并行处理独立的研究任务
  • 每次查询都需要全新上下文的场景

何时不使用无状态查询:

  • 基于之前发现的多轮调查
  • 基于初始结果迭代优化研究
  • 需要持续上下文的复杂分析

让我们使用 visualize_conversation 辅助函数来检查 Agent 实际做了什么:

from utils.agent_visualizer import visualize_conversation

visualize_conversation(messages)
<IPython.core.display.HTML object>

从原型到生产:三个关键改进

我们的单行研究 Agent 可以工作,但有局限性。没有记忆的单次查询无法处理迭代研究(“找到 X,然后根据你的发现分析 Y”)。让我们探讨三种进一步改进实现的方法。

1. 使用 ClaudeSDKClient 实现对话记忆:无状态查询无法基于之前的发现进行构建。如果你问”顶级 AI 创业公司有哪些?“然后问”它们是如何融资的?“,第二个查询没有关于你指的是哪些创业公司的上下文。我们可以使用 ClaudeSDKClient 来跨多次查询维护对话历史。

2. 使用系统提示实现专业化行为:研究领域通常有特定的要求。财务分析需要与技术新闻摘要不同的严谨度。使用系统提示来编码你的研究标准、首选来源、引用格式或输出结构。请参阅我们的 Agent 提示指南获取研究相关的示例。

3. 使用 Read 工具进行多模态研究:真正的研究不仅仅是文本。市场报告有图表,技术文档有图示,竞争分析需要截图对比。启用 Read 工具让 Claude 可以分析图像、PDF 和其他视觉内容。

让我们为研究 Agent 实现这三项改进。

from claude_agent_sdk import ClaudeSDKClient

# System prompt with citation requirements for research quality
RESEARCH_SYSTEM_PROMPT = """You are a research agent specialized in AI.

When providing research findings:
- Always include source URLs as citations
- Format citations as markdown links: [Source Title](URL)
- Group sources in a "Sources:" section at the end of your response"""

messages = []
async with ClaudeSDKClient(
    options=ClaudeAgentOptions(
        model=MODEL,
        cwd="research_agent",
        system_prompt=RESEARCH_SYSTEM_PROMPT,
        allowed_tools=["WebSearch", "Read"],
        max_buffer_size=10 * 1024 * 1024,  # Increase to 10MB for image handling
    )
) as research_agent:
    # First query: Analyze the chart image
    await research_agent.query("Analyze the chart in research_agent/projects_claude.png")
    async for msg in research_agent.receive_response():
        print_activity(msg)
        messages.append(msg)

    # Second query: Use web search to validate/contextualize the chart findings
    await research_agent.query(
        "Based on the chart analysis, search for recent news or data that validates or provides context for these findings. Include source URLs."
    )
    async for msg in research_agent.receive_response():
        print_activity(msg)
        messages.append(msg)
🤖 Using: Read()
✓ Tool completed
🤖 Thinking...
🤖 Using: Glob()
✓ Tool completed
🤖 Thinking...
🤖 Using: Read()
✓ Tool completed
🤖 Thinking...
🤖 Using: WebSearch()
🤖 Using: WebSearch()
🤖 Using: WebSearch()
🤖 Using: WebSearch()
✓ Tool completed
✓ Tool completed
✓ Tool completed
✓ Tool completed
🤖 Thinking...

🔧 处理大型响应和缓冲区限制

处理图像或大数据时,你可能会遇到缓冲区溢出错误:

Fatal error in message reader: Failed to decode JSON: JSON message exceeded maximum buffer size of 1048576 bytes

为什么会出现这种情况: - 默认 max_buffer_size 为 1MB(1,048,576 字节) - 图像在消息中以 base64 编码,显著增加大小 - 磁盘上约 200KB 的图表图像在 base64 编码后变为约 270KB+,外加消息开销

解决方案: 在处理图像或大型工具输出时,将 ClaudeAgentOptions 中的 max_buffer_size 设置为更高的值(例如 10MB)。

最佳实践: - 根据你的用例设置缓冲区大小:典型的多模态工作使用 10MB,大型文档处理使用更大的值 - 考虑是否真的需要传递完整图像——有时描述或较小的缩略图就足够了 - 监控缓冲区错误并相应调整 - 在系统提示中包含引用要求,以确保可验证的研究输出

这里发生了什么:

本示例结合了所有三项改进:对话记忆、引用感知的系统提示和多模态分析。

关键组件:

组件 用途
ClaudeSDKClient 跨多次查询维护对话状态
RESEARCH_SYSTEM_PROMPT 强制引用格式和来源 URL
allowed_tools=["WebSearch", "Read"] 启用网络搜索和图像/文档分析
max_buffer_size=10MB 处理 base64 编码的图像而不会溢出

执行流程:

  1. 第一次查询 - 使用 Read 工具分析图表图像
  2. 第一轮响应循环 - 收集所有消息直到 Agent 完成
  3. 第二次查询 - 搜索网络以验证/补充图表发现的背景信息
  4. 上下文继承 - 第二次查询记住第一次的图表分析

为什么选择 ClaudeSDKClient 而非 query():

async with ClaudeSDKClient() 上下文管理器维护对话状态。每次 receive_response() 调用都基于之前的上下文构建。这与创建独立、无状态会话的 query() 不同。

visualize_conversation(messages)
<IPython.core.display.HTML object>

面向生产环境构建

Jupyter notebook 非常适合学习,但生产系统需要可复用的模块。我们已将研究 Agent 打包到 research_agent/agent.py 中,提供了清晰的接口:

核心函数:

  • print_activity() - 实时显示 Agent 正在做什么(从共享工具导入)
  • get_activity_text() - 提取活动文本用于自定义处理,例如日志或监控
  • send_query() - 研究查询的主入口点,内置活动显示

内置最佳实践:

该模块包含 RESEARCH_SYSTEM_PROMPT,它确保: - 来源 URL 始终作为引用包含在内 - 引用格式化为 markdown 链接以便清晰渲染 - “Sources:”部分将所有引用分组

显示控制:

send_query() 函数有一个 display_result 参数(默认值:True): - display_result=True - 在 Jupyter notebook 中渲染样式化的 HTML 卡片 - display_result=False - 仅返回文本结果用于程序化使用

这个 Agent 现在可以在任何 Python 脚本中使用!

适用于对话上下文无关的独立问题。

该模块自动处理: - 执行期间的活动显示 - 新对话的上下文重置 - 最终响应的样式化 HTML 渲染

from research_agent.agent import send_query

# The module handles activity display, context reset, and result visualization internally
result = await send_query("What is the Claude Code SDK? Only do one websearch and be concise")
🤖 Using: WebSearch()
✓ Tool completed
🤖 Thinking...
<IPython.core.display.HTML object>

现在我们测试一个复用相同对话的多轮对话。

多轮对话可以无缝工作——只需传入 continue_conversation=True:

result1 = await send_query("What is Anthropic? Only do one websearch and be concise")
🤖 Using: WebSearch()
✓ Tool completed
🤖 Thinking...
<IPython.core.display.HTML object>
# Continue the conversation to dig deeper by setting continue_conversation=True
result2 = await send_query(
    "What are some of their products?",
    continue_conversation=True,
)
🤖 Thinking...
<IPython.core.display.HTML object>

总结

你构建了什么

在本 cookbook 中,你构建了三个复杂度递增的研究 Agent:

  • 无状态研究 Agent - 用于独立研究任务的单行查询
  • 有记忆的有状态 Agent - 基于之前发现的多轮调查
  • 生产模块 - 可集成到应用中的可复用研究函数

关键要点

何时使用无状态查询(query()):

  • 独立的研究问题
  • 并行处理不相关的任务
  • 每次都需要全新上下文的场景

何时使用有状态 Agent(ClaudeSDKClient):

  • 基于之前发现的多轮调查
  • 研究的迭代优化
  • 需要持续上下文的复杂分析

当信息不是自包含的且最优工作流在探索过程中涌现时,研究 Agent 表现出色——竞争分析、技术故障排除、文献综述和调查性新闻都符合这一模式。

后续步骤

自主研究的这一基础为你准备好了企业级多 Agent 系统。在下一个 notebook 中,你将学习:

在协调 Agent 下编排专业化子 Agent 通过 hooks 和自定义命令实现治理 为不同利益相关者调整输出风格(高管 vs. 技术团队)

下一步:01_The_chief_of_staff_agent.ipynb - 从单 Agent 到多 Agent 编排。


Tip

中文读者提示:本章节的代码和输出为英文原文。如需理解具体实现细节,可参考上方中文导读和代码注释。如有疑问,欢迎在 GitHub Issues 讨论。