深度研究代理食谱
本食谱演示了如何使用 OpenAI 深度研究 API 和 OpenAI 代理 SDK 构建代理研究工作流。这是基础食谱的延续，如果您尚未熟悉该内容，请考虑这样做。
您将学习如何编排单代理和多代理管道、丰富用户查询以最大化输出质量、流式传输研究进度、集成网络搜索和MCP 用于内部文件搜索，以及构建强大的研究应用程序。
对于需要规划、综合、工具使用或多步推理的任务，请考虑使用深度研究代理。请勿将深度研究用于琐碎的事实查找、简单的问答或短格式聊天，标准的 openai.responsesAPI 会更快、更便宜。
先决条件
- OpenAI API 密钥（在您的环境中设置为 OPENAI_API_KEY）
- Agents SDK 和 OpenAI Python SDK
设置
安装依赖项
%pip install --upgrade "openai>=1.88" "openai-agents>=0.0.19"
导入库并配置客户端
零数据保留
我们通过下面的 os.environ 设置禁用数据保留。这允许企业在零数据保留环境中运行深度研究。如果数据保留不是您的约束条件，那么请考虑将其保持启用状态，以便为您的代理工作流以及与其他平台工具（如评估和微调）的深度集成实现自动化可追溯性。
import os
from agents import Agent, Runner, WebSearchTool, RunConfig, set_default_openai_client, HostedMCPTool
from typing import List, Dict, Optional
from pydantic import BaseModel
from openai import AsyncOpenAI
# 使用环境变量作为 API 密钥并设置长超时
client = AsyncOpenAI(api_key="", timeout=600.0)
set_default_openai_client(client)
os.environ["OPENAI_AGENTS_DISABLE_TRACING"] = "1" # 为零数据保留 (ZDR) 组织禁用跟踪
基本深度研究代理
基本研究代理使用 o4-mini-deep-research-alpha 模型执行深度研究。它具有对公共互联网的原生 WebSearch 访问权限，并直接将研究结果流式传输回笔记本。在这种情况下，我们使用的是 o4-mini-deep-research-alpha 模型，因为它比完整的 o3 深度研究模型更快，并且具有可接受的智能。
学习目标：
完成此操作后，您可以运行单代理研究任务并流式传输其进度。
# 定义研究代理
research_agent = Agent(
name="Research Agent",
model="o4-mini-deep-research-2025-06-26",
tools=[WebSearchTool()],
instructions="您根据用户的问题进行深入的实证研究。"
)
# 异步函数，用于运行研究并打印流式传输的进度
async def basic_research(query):
print(f"正在研究：{query}")
result_stream = Runner.run_streamed(
research_agent,
query
)
async for ev in result_stream.stream_events():
if ev.type == "agent_updated_stream_event":
print(f"\n--- 切换到代理：{ev.new_agent.name} ---")
print(f"\n--- 正在研究 ---")
elif (
ev.type == "raw_response_event"
and hasattr(ev.data, "item")
and hasattr(ev.data.item, "action")
):
action = ev.data.item.action or {}
if action.get("type") == "search":
print(f"[网络搜索] query={action.get('query')!r}")
# 流式传输已完成 → final_output 现在已填充
return result_stream.final_output
# 运行研究并打印结果
result = await basic_research("研究司美格鲁肽对全球医疗保健系统的经济影响。")
print(result)
具有澄清功能的多代理研究
多代理深度研究
考虑如何进一步改进“深度研究”产生的研究质量。在这种情况下，我们在将其提交给深度研究代理之前，利用多代理架构来丰富用户查询的提示，并提供有关最终研究报告的预期内容。
子代理提示丰富
支持代理提示专门设计用于通过为用户初始查询提供结构和严谨性来提高最终研究输出的质量。
# ─────────────────────────────────────────────────────────────
# 提示
# ─────────────────────────────────────────────────────────────
CLARIFYING_AGENT_PROMPT = """
如果用户没有明确要求研究（不太可能），请询问他们想让您进行什么研究。
指南：
1. **保持简洁，同时收集所有必要信息** 提出 2-3 个澄清问题以收集更多研究背景信息。
- 确保以简洁、结构良好的方式收集执行研究任务所需的所有信息。如果适合清晰度，请使用项目符号或编号列表。不要询问不必要的信息，或用户已提供的信息。
2. **保持友好和非居高临下的语气**
- 例如，不要说“我需要关于 Y 的更多细节”，而要说“您能否分享更多关于 Y 的细节？”
3. **遵守安全指南**
"""
RESEARCH_INSTRUCTION_AGENT_PROMPT = """
根据以下指南，获取用户的查询，并将其重写为详细的研究说明。仅输出研究说明，其他任何内容都不输出。转移给研究代理。
指南：
1. **最大化特异性和细节**
- 包括所有已知的用户偏好，并明确列出要考虑的关键属性或维度。
- 将用户的所有详细信息包含在扩展提示中至关重要。
2. **填补未说明但必需的维度，作为开放式问题**
- 如果某些属性对于有意义的输出至关重要，但用户尚未提供，请明确说明它们是开放式的，或默认为“无特定限制”。
3. **避免不必要的假设**
- 如果用户未提供特定详细信息，请勿自行捏造。
- 相反，请说明规格的缺失，并指导深度研究模型将其视为灵活或接受所有可能的选项。
4. **使用第一人称**
- 从用户的角度来构建请求。
5. **表格**
- 如果您确定包含表格有助于说明、组织或增强深度研究输出中的信息，则必须明确要求深度研究模型提供它们。
示例：
- 产品比较（消费者）：比较不同智能手机型号时，请求一个表格，并排列每个型号的功能、价格和消费者评分。
- 项目跟踪（工作）：概述项目可交付成果时，创建一个表格，显示任务、截止日期、负责的团队成员和状态更新。
- 预算规划（消费者）：创建个人或家庭预算时，请求一个表格，详细说明收入来源、月度支出和储蓄目标。
竞争对手分析（工作）：评估竞争对手产品时，请求一个表格，其中包含关键指标，例如市场份额、定价和主要差异化因素。
6. **标题和格式**
- 您应在提示中包含预期的输出格式。
- 如果用户请求的内容最好以结构化格式（例如报告、计划等）返回，请要求深度研究模型“以适当的标题和格式作为报告，确保清晰度和结构。”
7. **语言**
- 如果用户输入不是英语，请告知模型使用该语言进行回复，除非用户查询明确要求使用其他语言进行回复。
8. **来源**
- 如果应优先考虑特定来源，请在提示中指定它们。
- 优先内部知识。每个文件只检索一次。
- 对于产品和旅行研究，优先链接到官方或主要网站（例如，官方品牌网站、制造商页面或信誉良好的电子商务平台，如亚马逊以获取用户评论），而不是聚合网站或 SEO 密集型博客。
- 对于学术或科学查询，优先链接到原始论文或官方期刊出版物，而不是调查论文或二次摘要。
- 如果查询是特定语言，请优先使用以该语言发布的来源。
重要提示：确保此函数的完整负载是有效的 JSON
重要提示：在提示中指定所需的输出语言
"""
四代理深度研究管道
-
分类代理 - 检查用户查询 - 如果缺少上下文，则路由到澄清代理；否则路由到指令代理
-
澄清代理 - 提出后续问题 - 等待用户（或模拟）答案
-
指令构建代理 - 将丰富后的输入转换为精确的研究简报
-
研究代理 (
o3-deep-research) - 使用WebSearchTool执行网络规模的实证研究 - 使用 MCP 对内部知识库执行搜索，如果存在相关片段，代理会将这些相关片段纳入其参考资料中。 - 流式传输中间事件以提高透明度 - 输出最终研究工件（我们稍后会解析）
有关 MCP 服务器如何构建的更多见解。请参阅此资源。
# ─────────────────────────────────────────────────────────────
# 结构化输出（仅对澄清代理需要）
# ─────────────────────────────────────────────────────────────
class Clarifications(BaseModel):
questions: List[str]
# ─────────────────────────────────────────────────────────────
# 代理
# ─────────────────────────────────────────────────────────────
research_agent = Agent(
name="Research Agent",
model="o3-deep-research-2025-06-26",
instructions="根据用户的说明进行深入的实证研究。",
tools=[WebSearchTool(),
HostedMCPTool(
tool_config={
"type": "mcp",
"server_label": "file_search",
"server_url": "https://<url>/sse",
"require_approval": "never",
}
)
]
)
instruction_agent = Agent(
name="Research Instruction Agent",
model="gpt-4o-mini",
instructions=RESEARCH_INSTRUCTION_AGENT_PROMPT,
handoffs=[research_agent],
)
clarifying_agent = Agent(
name="Clarifying Questions Agent",
model="gpt-4o-mini",
instructions=CLARIFYING_AGENT_PROMPT,
output_type=Clarifications,
handoffs=[instruction_agent],
)
triage_agent = Agent(
name="Triage Agent",
instructions=(
"决定是否需要澄清。\n"
"• 如果是 → 调用 transfer_to_clarifying_questions_agent\n"
"• 如果否 → 调用 transfer_to_research_instruction_agent\n"
"仅返回一个函数调用。"
),
handoffs=[clarifying_agent, instruction_agent],
)
# ─────────────────────────────────────────────────────────────
# 自动澄清助手
# ─────────────────────────────────────────────────────────────
async def basic_research(
query: str,
mock_answers: Optional[Dict[str, str]] = None,
verbose: bool = False,
):
stream = Runner.run_streamed(
triage_agent,
query,
run_config=RunConfig(tracing_disabled=True),
)
async for ev in stream.stream_events():
if isinstance(getattr(ev, "item", None), Clarifications):
reply = []
for q in ev.item.questions:
ans = (mock_answers or {}).get(q, "无偏好。")
reply.append(f"**{q}**\n{ans}")
stream.send_user_message("\n\n".join(reply))
continue
if verbose:
print(ev)
#return stream.final_output
return stream
# ─────────────────────────────────────────────────────────────
# 示例运行
# ─────────────────────────────────────────────────────────────
result = await basic_research(
"研究司美格鲁肽对全球医疗保健系统的经济影响。",
mock_answers={}, # 或提供模拟答案
)
代理交互流程
尽管代理 SDK 跟踪原生提供了这些信息，但您可能希望打印人类可读的高级代理交互流程以及工具调用。运行 print_agent_interaction 以获取简化的可读代理步骤序列，包括：代理名称、事件类型（移交、工具调用、消息输出）、简化的工具调用信息（工具名称和参数）。
import json
def parse_agent_interaction_flow(stream):
print("=== 代理交互流程 ===")
count = 1
for item in stream.new_items:
# 代理名称，如果缺失则回退
agent_name = getattr(item.agent, "name", "Unknown Agent") if hasattr(item, "agent") else "Unknown Agent"
if item.type == "handoff_call_item":
func_name = getattr(item.raw_item, "name", "Unknown Function")
print(f"{count}. [{agent_name}] → Handoff Call: {func_name}")
count += 1
elif item.type == "handoff_output_item":
print(f"{count}. [{agent_name}] → Handoff Output")
count += 1
elif item.type == "mcp_list_tools_item":
print(f"{count}. [{agent_name}] → mcp_list_tools_item")
count += 1
elif item.type == "reasoning_item":
print(f"{count}. [{agent_name}] → Reasoning step")
count += 1
elif item.type == "tool_call_item":
tool_name = getattr(item.raw_item, "name", None)
# 如果 tool_name 缺失或为空，则跳过工具调用
if not isinstance(tool_name, str) or not tool_name.strip():
continue # 静默跳过
tool_name = tool_name.strip()
args = getattr(item.raw_item, "arguments", None)
args_str = ""
if args:
try:
parsed_args = json.loads(args)
if parsed_args:
args_str = json.dumps(parsed_args)
except Exception:
if args.strip() and args.strip() != "{}":
args_str = args.strip()
args_display = f" with args {args_str}" if args_str else ""
print(f"{count}. [{agent_name}] → Tool Call: {tool_name}{args_display}")
count += 1
elif item.type == "message_output_item":
print(f"{count}. [{agent_name}] → Message Output")
count += 1
else:
print(f"{count}. [{agent_name}] → {item.type}")
count += 1
# 用法示例：
parse_agent_interaction_flow(result)
引文
以下是一个 Python 代码片段，用于提取和打印与最终输出相关的 URL 引文：
def print_final_output_citations(stream, preceding_chars=50):
# 反向迭代 new_items 以查找最后一个 message_output_item(s)
for item in reversed(stream.new_items):
if item.type == "message_output_item":
for content in getattr(item.raw_item, 'content', []):
if not hasattr(content, 'annotations') or not hasattr(content, 'text'):
continue
text = content.text
for ann in content.annotations:
if getattr(ann, 'type', None) == 'url_citation':
title = getattr(ann, 'title', '<no title>')
url = getattr(ann, 'url', '<no url>')
start = getattr(ann, 'start_index', None)
end = getattr(ann, 'end_index', None)
if start is not None and end is not None and isinstance(text, str):
# 安全地计算前面的片段开始索引
pre_start = max(0, start - preceding_chars)
preceding_text = text[pre_start:start].replace('\n', ' ').strip()
excerpt = text[start:end].replace('\n', ' ').strip()
print("# --------")
print("# MCP CITATION SAMPLE:")
print(f"# Title: {title}")
print(f"# URL: {url}")
print(f"# Location: chars {start}–{end}")
print(f"# Preceding: '{preceding_text}'")
print(f"# Excerpt: '{excerpt}'\n")
else:
# 如果没有索引可用，则回退
print(f"- {title}: {url}")
break
# 用法
print_final_output_citations(result)
## 深度研究研究报告
print(result.final_output)
结论
通过本笔记本中的模式，您现在拥有了使用 OpenAI 深度研究代理构建可扩展、生产就绪的研究工作流的基础。这些示例不仅演示了如何编排多代理管道和流式传输研究进度，还演示了如何集成网络搜索和 MCP 以访问外部知识。
通过利用代理工作流，您可以超越简单的问答，应对需要规划、综合和工具使用的复杂、多步骤研究任务。模块化的多代理设计：分类、澄清、指令和研究代理使您能够将这些管道适应广泛的领域和用例，从医疗保健和金融到技术尽职调查和市场分析。
随着深度研究 API 和代理 SDK 的不断发展，这些模式将帮助您保持在自动化、数据驱动研究的前沿。无论您是构建内部知识工具、自动化竞争情报，还是支持专家分析师，这些工作流都提供了强大、可扩展的起点。
祝您研究愉快！