使用 Agents SDK 的 Deep Research API
来源: https://developers.openai.com/cookbook/examples/deep_research_api/introduction_to_deep_research_api_agents 抓取时间: 2026-07-21 16:25:21
本 Cookbook 演示了如何使用 OpenAI Deep Research API 和 OpenAI Agents SDK 构建智能体研究工作流程。这是 基础 Cookbook 的延续，如果您还不熟悉该内容，请考虑先阅读。
您将学习如何编排单智能体和多智能体管道，丰富用户查询以最大化输出质量，流式传输研究进度，集成网络搜索和 用于内部文件搜索的 MCP，以及构建强大的研究应用程序。
考虑将 Deep Research 智能体用于需要规划、综合、工具使用或多步骤推理的任务。不要将 Deep Research 用于简单的事实查找、简单的问答或短格式聊天，普通的 openai.responses API 会更快且更便宜。
先决条件
- OpenAI API 密钥（在您的环境中设置为 OPENAI_API_KEY）
- Agents SDK 和 OpenAI Python SDK
设置
安装依赖项
%pip install --upgrade "openai>=1.88" "openai-agents>=0.0.19"
导入库并配置客户端
零数据保留
我们通过下面的 os.environ 设置禁用数据保留。这允许企业在零数据保留环境中使用 Deep Research 进行操作。如果数据保留对您来说不是一个主动约束，那么考虑保持启用状态，以便您可以对智能体工作流程进行自动追踪，并与评估和微调等其他平台工具深度集成。
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
# Use env var for API key and set a long timeout
client = AsyncOpenAI(api_key="", timeout=600.0)
set_default_openai_client(client)
os.environ["OPENAI_AGENTS_DISABLE_TRACING"] = "1" # Disable tracing for Zero Data Retention (ZDR) Organizations
基础 Deep Research 智能体
基础研究智能体使用 o4-mini-deep-research-alpha 模型执行 Deep Research。它可以对公共互联网进行原生 WebSearch 访问，并将其发现直接流式传输回笔记本。在这种情况下，我们使用 o4-mini-deep-research-alpha 模型，因为它比完整的 o3 deep research 模型更快，且智能水平可接受。
学习目标：
完成本节后，您可以运行单智能体研究任务并流式传输其进度。
# Define the research agent
research_agent = Agent(
name="Research Agent",
model="o4-mini-deep-research-2025-06-26",
tools=[WebSearchTool()],
instructions="You perform deep empirical research based on the user's question."
)
# Async function to run the research and print streaming progress
async def basic_research(query):
print(f"Researching: {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--- switched to agent: {ev.new_agent.name} ---")
print(f"\n--- RESEARCHING ---")
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"[Web search] query={action.get('query')!r}")
# streaming is complete → final_output is now populated
return result_stream.final_output
# Run the research and print the result
result = await basic_research("Research the economic impact of semaglutide on global healthcare systems.")
print(result)
带澄清功能的多智能体研究
多智能体 Deep Research
考虑如何进一步提高"Deep Research"产生的研究质量。在这种情况下，我们利用多智能体架构，在提交给深度研究智能体之前，使用更多关于用户查询的信息以及我们期望在最终研究报告中看到的内容来丰富提示。
子智能体提示丰富
支持智能体提示经过专门设计，通过为用户的初始查询提供结构和严谨性来提高最终研究输出的质量。
# ─────────────────────────────────────────────────────────────
# Prompts
# ─────────────────────────────────────────────────────────────
CLARIFYING_AGENT_PROMPT = """
If the user hasn't specifically asked for research (unlikely), ask them what research they would like you to do.
GUIDELINES:
1. **Be concise while gathering all necessary information** Ask 2–3 clarifying questions to gather more context for research.
- Make sure to gather all the information needed to carry out the research task in a concise, well-structured manner. Use bullet points or numbered lists if appropriate for clarity. Don't ask for unnecessary information, or information that the user has already provided.
2. **Maintain a Friendly and Non-Condescending Tone**
- For example, instead of saying "I need a bit more detail on Y," say, "Could you share more detail on Y?"
3. **Adhere to Safety Guidelines**
"""
RESEARCH_INSTRUCTION_AGENT_PROMPT = """
Based on the following guidelines, take the users query, and rewrite it into detailed research instructions. OUTPUT ONLY THE RESEARCH INSTRUCTIONS, NOTHING ELSE. Transfer to the research agent.
GUIDELINES:
1. **Maximize Specificity and Detail**
- Include all known user preferences and explicitly list key attributes or dimensions to consider.
- It is of utmost importance that all details from the user are included in the expanded prompt.
2. **Fill in Unstated But Necessary Dimensions as Open-Ended**
- If certain attributes are essential for a meaningful output but the user has not provided them, explicitly state that they are open-ended or default to "no specific constraint."
3. **Avoid Unwarranted Assumptions**
- If the user has not provided a particular detail, do not invent one.
- Instead, state the lack of specification and guide the deep research model to treat it as flexible or accept all possible options.
4. **Use the First Person**
- Phrase the request from the perspective of the user.
5. **Tables**
- If you determine that including a table will help illustrate, organize, or enhance the information in your deep research output, you must explicitly request that the deep research model provide them.
Examples:
- Product Comparison (Consumer): When comparing different smartphone models, request a table listing each model's features, price, and consumer ratings side-by-side.
- Project Tracking (Work): When outlining project deliverables, create a table showing tasks, deadlines, responsible team members, and status updates.
- Budget Planning (Consumer): When creating a personal or household budget, request a table detailing income sources, monthly expenses, and savings goals.
Competitor Analysis (Work): When evaluating competitor products, request a table with key metrics—such as market share, pricing, and main differentiators.
6. **Headers and Formatting**
- You should include the expected output format in the prompt.
- If the user is asking for content that would be best returned in a structured format (e.g. a report, plan, etc.), ask the Deep Research model to "Format as a report with the appropriate headers and formatting that ensures clarity and structure."
7. **Language**
- If the user input is in a language other than English, tell the model to respond in this language, unless the user query explicitly asks for the response in a different language.
8. **Sources**
- If specific sources should be prioritized, specify them in the prompt.
- Prioritize Internal Knowledge. Only retrieve a single file once.
- For product and travel research, prefer linking directly to official or primary websites (e.g., official brand sites, manufacturer pages, or reputable e-commerce platforms like Amazon for user reviews) rather than aggregator sites or SEO-heavy blogs.
- For academic or scientific queries, prefer linking directly to the original paper or official journal publication rather than survey papers or secondary summaries.
- If the query is in a specific language, prioritize sources published in that language.
IMPORTANT: Ensure that the complete payload to this function is valid JSON
IMPORTANT: SPECIFY REQUIRED OUTPUT LANGUAGE IN THE PROMPT
"""
四智能体 Deep Research 管道
-
分诊智能体
- 检查用户的查询
- 如果缺少上下文，路由到澄清智能体；否则路由到指令智能体
-
澄清智能体
- 提出后续问题
- 等待用户（或模拟）答案
-
指令构建器智能体
- 将丰富的输入转换为精确的研究简介
-
研究智能体 (
o3-deep-research)- 使用
WebSearchTool执行网络级实证研究 - 使用 MCP 对内部知识库执行搜索，如果有相关文档，智能体会将这些相关片段合并到其参考材料中。
- 流式传输中间事件以保持透明度
- 输出最终研究工件（我们稍后会解析）
- 使用
有关如何构建 MCP 服务器的更多见解，请参见此资源。
# ─────────────────────────────────────────────────────────────
# Structured outputs (needed only for Clarifying agent)
# ─────────────────────────────────────────────────────────────
class Clarifications(BaseModel):
questions: List[str]
# ─────────────────────────────────────────────────────────────
# Agents
# ─────────────────────────────────────────────────────────────
research_agent = Agent(
name="Research Agent",
model="o3-deep-research-2025-06-26",
instructions="Perform deep empirical research based on the user's 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=(
"Decide whether clarifications are required.\n"
"• If yes → call transfer_to_clarifying_questions_agent\n"
"• If no → call transfer_to_research_instruction_agent\n"
"Return exactly ONE function-call."
),
handoffs=[clarifying_agent, instruction_agent],
)
# ─────────────────────────────────────────────────────────────
# Auto-clarify helper
# ─────────────────────────────────────────────────────────────
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, "No preference.")
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
# ─────────────────────────────────────────────────────────────
# Example run
# ─────────────────────────────────────────────────────────────
result = await basic_research(
"Research the economic impact of semaglutide on global healthcare systems.",
mock_answers={}, # or provide canned answers
)
智能体交互流程
尽管通过 Agent SDK 追踪原生提供，但您可能希望打印带有工具调用的人类可读的高级智能体交互流程。运行 print_agent_interaction 以获取智能体步骤的简化可读序列，包括：智能体名称、事件类型（交接、工具调用、消息输出）、简要工具调用信息（工具名称和参数）。
import json
def parse_agent_interaction_flow(stream):
print("=== Agent Interaction Flow ===")
count = 1
for item in stream.new_items:
# Agent name, fallback if missing
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)
# Skip tool call if tool_name is missing or empty
if not isinstance(tool_name, str) or not tool_name.strip():
continue # skip silently
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
# Example usage:
parse_agent_interaction_flow(result)
引用
以下是提取和打印与最终输出相关的 URL 引用的 Python 代码片段：
def print_final_output_citations(stream, preceding_chars=50):
# Iterate over new_items in reverse to find the last 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):
# Calculate preceding snippet start index safely
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:
# fallback if no indices available
print(f"- {title}: {url}")
break
# Usage
print_final_output_citations(result)
## Deep Research Research Report
print(result.final_output)
结论
通过本笔记本中的模式，您现在拥有了使用 OpenAI Deep Research 智能体构建可扩展、生产就绪研究工作流程的基础。这些示例不仅演示了如何编排多智能体管道和流式传输研究进度，还演示了如何集成网络搜索和 MCP 以进行外部知识访问。
通过利用智能体工作流程，您可以超越简单的问答，解决需要规划、综合和工具使用的复杂、多步骤研究任务。模块化的多智能体设计：分诊、澄清、指令和研究智能体使您能够将这些管道适应广泛的领域和用例，从医疗保健和金融到技术尽职调查和市场分析。
随着 Deep Research API 和 Agents SDK 的持续发展，这些模式将帮助您站在自动化、数据支持研究的前沿。无论您是构建内部知识工具、自动化竞争情报，还是支持专家分析师，这些工作流程都提供了一个强大、可扩展的起点。
祝研究愉快！