😸
OpenAI Agents SDK エージェントパターンの比較
OpenAI Agents SDKは、複数のLLMエージェント間でタスクを分担させたり、関数呼び出しとしてエージェント同士を連携させたりできるPython向けライブラリです。本記事では、SDKの基本を紹介した後、サンプルコード: research_botのエージェント定義を少し変更したいくつかのパターンの挙動を比較します。
複数のパターンの挙動を比較することで、OpenAI Agents SDKのふるまいを理解し、実際のプロジェクトで活用できるようになることを目指します。
1. エージェントの基本
まずは、OpenAI Agents SDKの概要を説明します。
OpenAI Agents SDKでは、Agentクラスを用いてエージェントを定義します。主に以下の要素を指定します。
-
name: エージェント名 -
instructions: エージェントの指示 -
tools: 関連する関数や他エージェントをツールとして登録するリスト -
handoffs: タスクを引き継ぐ先のエージェントリスト - その他: GuardrailsやModelなど
例: シンプルな検索エージェント
from openai_agents import Agent, function_tool
# 検索処理を行う関数
@function_tool
async def search_tool(query: str) -> str:
# 実際には外部APIやデータベース検索などを実装
return f"[検索結果] {query} に関する情報"
# Agentにツールとして関数を登録して定義
search_agent = Agent(
name="search_agent",
instructions="検索エージェントです。検索結果を返します。",
tools=[search_tool]
)
エージェントの実行方法
定義したAgentは、runメソッドを呼び出すだけで実行できます。内部ではモデル呼び出しとツール連携がシームレスに行われます。
# 非同期実行例
result = await search_agent.run("OpenAI Agents SDKとは何ですか？")
print(result)
実行結果として、LLMの応答や登録したツールの出力が返されます。
同期的に実行したい場合はrun_syncを、ストリーミング処理にしたい場合はrun_streamedを呼び出します。
tools
-
toolsにはPython関数や、他のAgentインスタンスを登録できます。 - ツールとして呼び出された
Agentや関数は、処理結果を呼び出し元に返します。これにより、ワークフローを連続的に制御できます。
handoffs
-
handoffsには、タスクを引き継ぐ他のAgentインスタンスを登録します。 - handoffsによって、呼び出された
Agentへ制御が完全に移譲され、呼び出し元には戻りません。指定されたエージェントのワークフローが継続されます。
toolsとhandoffsの挙動の違い
| 方式 | 戻り値の扱い | フローの制御 |
|---|---|---|
| tools | 処理結果を呼び出し元に返す | 呼び出し元のエージェントに戻る |
| handoffs | 戻り値を返さず制御を移譲 | 指定先エージェントが継続 |
toolsとhandoffsは、どちらも他のエージェントの処理を呼び出すために使用できます。呼び出し結果をもとに元のエージェントで処理を続けたい場合はtoolsを、処理の主体を完全に別のエージェントに移したい（ルーティングしたい）場合はhandoffsを選択します。
2. AgentSampleによる挙動比較
以下のサンプルコードは、公式のresearch_botを参考にしています。
research_botでは、3つのエージェント「PlannerAgent」「SearchAgent」「WriterAgent」を用意し、Plan->Search->Reportの順で処理を行うようにします。
処理を実行すると、詳細なレポートが生成されます。
この基本的な処理フローを、toolsとhandoffsを使ってどのように実現できるか、そしてその挙動がどう異なるかを検証するために、以下の3つのパターンで比較します。
- すべてのエージェントをtoolsとして
ManagerAgentに登録 - すべてのエージェントをhandoffsとして
ManagerAgentに登録 - データ連携や特定のタスク実行のみを
function_toolとして定義し、ManagerAgentから呼び出す
モデルはすべて、gpt-4oで検証します。
2.1 すべてtoolsとして定義した場合
ここでは、ManagerAgentが全体のプロセスを管理します。planner_agent、search_agent、writer_agentは、それぞれ計画、検索、レポート作成のタスクを担当するツールとしてManagerAgentに登録されます。ManagerAgentは、ユーザーのクエリに基づいてこれらのツールを適切な順序で呼び出し、最終的なレポートを生成します。
# 各Agent定義（省略）
manager_agent = Agent(
name="ManagerAgent",
instructions=MANAGER_PROMPT,
tools=[planner_agent.as_tool(tool_name="planner_agent", tool_description="このエージェントは何を検索するのかを計画します。"), search_agent.as_tool(tool_name="search_agent", tool_description="このエージェントは計画に従って検索を行います。"), writer_agent.as_tool(tool_name="writer_agent", tool_description="このエージェントは検索結果をまとめてレポートを作成します。")]
)
# 実行すると、plan→search→reportが順に呼び出され、最終結果を返す
result = await Runner.run(manager_agent, "ここにクエリを入れる")
print(result.final_output)
コード全文
from __future__ import annotations
import asyncio
from termcolor import colored
from pydantic import BaseModel
from agents import (
Agent,
Runner,
WebSearchTool,
ModelSettings,
gen_trace_id,
trace,
TResponseInputItem,
)
from agents.model_settings import ModelSettings
from openai.types.responses import ResponseTextDeltaEvent
from dotenv import load_dotenv
load_dotenv()
# ---------- Planner Agent ----------
PLANNER_PROMPT = (
"You are a helpful research assistant. Given a query, come up with a set of web searches "
"to perform to best answer the query. Output between 5 and 20 terms to query for."
)
class WebSearchItem(BaseModel):
"""
reason(str): Your reasoning for why this search is important to the query.
query(str): The search term to use for the web search.
"""
reason: str
query: str
class WebSearchPlan(BaseModel):
"""
searches(list[WebSearchItem]): A list of web searches to perform to best answer the query.
"""
searches: list[WebSearchItem]
planner_agent = Agent(
name="PlannerAgent",
instructions=PLANNER_PROMPT,
model="gpt-4o",
output_type=WebSearchPlan,
)
# ---------- Search Agent ----------
SEARCH_PROMPT = (
"You are a research assistant. Given a search term, you search the web for that term and "
"produce a concise summary of the results. The summary must 2-3 paragraphs and less than 300 "
"words. Capture the main points. Write succinctly, no need to have complete sentences or good "
"grammar. This will be consumed by someone synthesizing a report, so its vital you capture the "
"essence and ignore any fluff. Do not include any additional commentary other than the summary "
"itself."
)
search_agent = Agent(
name="Search agent",
instructions=SEARCH_PROMPT,
tools=[WebSearchTool()],
model_settings=ModelSettings(tool_choice="required"),
)
# ---------- Writer Agent ----------
WRITER_PROMPT = (
"You are a senior researcher tasked with writing a cohesive report for a research query. "
"You will be provided with the original query, and some initial research done by a research "
"assistant.\n"
"You should first come up with an outline for the report that describes the structure and "
"flow of the report. Then, generate the report and return that as your final output.\n"
"The final output should be in markdown format, and it should be lengthy and detailed. Aim "
"for 5-10 pages of content, at least 1000 words."
)
class ReportData(BaseModel):
"""
short_summary(str): A short 2-3 sentence summary of the findings.
markdown_report(str): The final report
follow_up_questions(list[str]): Suggested topics to research further
"""
short_summary: str
markdown_report: str
follow_up_questions: list[str]
writer_agent = Agent(
name="WriterAgent",
instructions=WRITER_PROMPT,
model="gpt-4o",
output_type=ReportData,
)
# ---------- Manager Agent ----------
MANAGER_PROMPT = (
"You are a senior project manager."
"To complete the task, you need to use handoff to delegate next step of the task."
)
class OutputType(BaseModel):
"""
message(str): The message to display to the user.
is_finished(bool): Whether the task is finished.
"""
message: str
is_finished: bool
manager_agent = Agent[OutputType](
name="ManagerAgent",
instructions=MANAGER_PROMPT,
tools=[planner_agent.as_tool(tool_name="planner_agent", tool_description="このエージェントは何を検索するのかを計画します。"), search_agent.as_tool(tool_name="search_agent", tool_description="このエージェントは計画に従って検索を行います。"), writer_agent.as_tool(tool_name="writer_agent", tool_description="このエージェントは検索結果をまとめてレポートを作成します。")],
output_type=OutputType,
model="gpt-4o",
)
# ---------- Research Bot ----------
class ResearchManager:
def __init__(self):
pass
async def run(self, query: str) -> None:
trace_id = gen_trace_id()
with trace("Research trace", trace_id=trace_id):
inputs: list[TResponseInputItem] = [{"role": "user", "content": f"Query: {query}"}]
while True:
result = Runner.run_streamed(
manager_agent,
inputs,
)
result_stream = result.stream_events()
async for evt in result_stream:
# tool_called イベントを見て検索進行度らしきものを表示
# We'll ignore the raw responses event deltas
if evt.type == "raw_response_event":
if isinstance(evt.data, ResponseTextDeltaEvent):
print(evt.data.delta, end="")
continue
# When the agent updates, print that
elif evt.type == "agent_updated_stream_event":
print(colored(f"Agent updated: {evt.new_agent.name}", "green"))
print("-"*50)
continue
# When items are generated, print them
elif evt.type == "run_item_stream_event":
if evt.item.type == "tool_call_item":
print(colored(f"Tool was called: {evt.item.raw_item}", "blue"))
print("-"*50)
elif evt.item.type == "tool_call_output_item":
print(colored(f"Tool output: {evt.item.output}", "blue"))
print("-"*50)
elif evt.item.type == "message_output_item":
print(colored("output", "blue"))
print("-"*50)
# print(evt.item.output)
else:
pass # Ignore other event types
result_output = result.final_output_as(OutputType)
print("="*50)
print(result_output)
if result_output.is_finished:
break
inputs = result.to_input_list()
report = result_output.message
print(report)
with open("report_tool_simple.md", "w", encoding="utf-8") as f:
f.write(report)
async def main() -> None:
query = "2025年4月26日（土）の東京の天気について、日本語でレポートをまとめてください。"
await ResearchManager().run(query)
if __name__ == "__main__":
asyncio.run(main())
結果
Sequentialな処理を定義しなくてもplanner_agentが呼び出された後、planner_agentで計画した複数の検索をsearch_agentで処理、その後、writer_agentで作成したレポートをManagerAgentがまとめて返すという順番で処理が行われました。
ManagerAgentでは、writer_agentで作成された「short_summary」と「follow_up_questions」の情報は失われてしまっていました。WriterAgentで生成されていた情報をすべて利用せず、マネージャーのさじ加減で情報が失われてしまうようです。正確な情報を返したい場合は、Toolsとしてエージェントを実行するのは不向きかもしれません。
4.2 すべてhandoffsとして定義した場合
次にHandoffsとして定義する場合を調べてみます。Handoffsで定義した場合は、ルーティング後、呼び出し元に制御が戻らないため、ループ処理などを用いてマネージャーエージェントに処理を戻す必要があります。
main_agent = Agent(
name="ManagerAgent",
instructions=MANAGER_PROMPT,
handoffs=[plan_agent, search_agent, report_agent]
)
query = "ここにクエリを入れる"
inputs: list[TResponseInputItem] = [{"role": "user", "content": f"Query: {query}"}]
while True:
result = await Runner.run(main_agent, inputs)
inputs = result.to_input_list() # 会話履歴を取得
if result.last_agent.name == "ReportAgent":
print(result.final_output)
break
コード全文
from __future__ import annotations
import asyncio
from pydantic import BaseModel
from agents import (
Agent,
Runner,
WebSearchTool,
ModelSettings,
gen_trace_id,
trace,
TResponseInputItem,
)
from agents.model_settings import ModelSettings
from openai.types.responses import ResponseTextDeltaEvent
from termcolor import colored
from dotenv import load_dotenv
load_dotenv()
# ---------- Planner Agent ----------
PLANNER_PROMPT = (
"You are a helpful research assistant. Given a query, come up with a set of web searches "
"to perform to best answer the query. Output between 5 and 20 terms to query for."
)
class WebSearchItem(BaseModel):
"""
reason(str): Your reasoning for why this search is important to the query.
query(str): The search term to use for the web search.
"""
reason: str
query: str
class WebSearchPlan(BaseModel):
"""
searches(list[WebSearchItem]): A list of web searches to perform to best answer the query.
"""
searches: list[WebSearchItem]
planner_agent = Agent(
name="PlannerAgent",
instructions=PLANNER_PROMPT,
model="gpt-4o",
output_type=WebSearchPlan,
)
# ---------- Search Agent ----------
SEARCH_PROMPT = (
"You are a research assistant. Given a search term, you search the web for that term and "
"produce a concise summary of the results. The summary must 2-3 paragraphs and less than 300 "
"words. Capture the main points. Write succinctly, no need to have complete sentences or good "
"grammar. This will be consumed by someone synthesizing a report, so its vital you capture the "
"essence and ignore any fluff. Do not include any additional commentary other than the summary "
"itself."
)
search_agent = Agent(
name="Search agent",
instructions=SEARCH_PROMPT,
tools=[WebSearchTool()],
model_settings=ModelSettings(tool_choice="required"),
)
# ---------- Writer Agent ----------
WRITER_PROMPT = (
"You are a senior researcher tasked with writing a cohesive report for a research query. "
"You will be provided with the original query, and some initial research done by a research "
"assistant.\n"
"You should first come up with an outline for the report that describes the structure and "
"flow of the report. Then, generate the report and return that as your final output.\n"
"The final output should be in markdown format, and it should be lengthy and detailed. Aim "
"for 5-10 pages of content, at least 1000 words."
)
class ReportData(BaseModel):
"""
short_summary(str): A short 2-3 sentence summary of the findings.
markdown_report(str): The final report
follow_up_questions(list[str]): Suggested topics to research further
"""
short_summary: str
markdown_report: str
follow_up_questions: list[str]
writer_agent = Agent(
name="WriterAgent",
instructions=WRITER_PROMPT,
model="gpt-4o",
output_type=ReportData,
)
# ---------- Manager Agent ----------
MANAGER_PROMPT = (
"You are a senior project manager."
"To complete the task, you need to use handoff to delegate next step of the task."
"Complete the task as followed steps:"
"1. Plan the task"
"2. Search the web for the task"
"3. Write the report"
)
manager_agent = Agent(
name="ManagerAgent",
instructions=MANAGER_PROMPT,
handoffs=[planner_agent, search_agent, writer_agent],
model="gpt-4o",
)
# ---------- Research Bot ----------
class ResearchManager:
def __init__(self):
pass
async def run(self, query: str) -> None:
trace_id = gen_trace_id()
with trace("Research trace", trace_id=trace_id):
inputs: list[TResponseInputItem] = [{"role": "user", "content": f"Query: {query}"}]
while True:
result = Runner.run_streamed(
manager_agent,
inputs,
)
result_stream = result.stream_events()
async for evt in result_stream:
# print(evt.type)
# tool_called イベントを見て検索進行度らしきものを表示
# We'll ignore the raw responses event deltas
if evt.type == "raw_response_event":
if isinstance(evt.data, ResponseTextDeltaEvent):
print(evt.data.delta, end="")
continue
# When the agent updates, print that
elif evt.type == "agent_updated_stream_event":
print(colored(f"Agent updated: {evt.new_agent.name}", "green"))
print("-"*50)
continue
# When items are generated, print them
elif evt.type == "run_item_stream_event":
if evt.item.type == "tool_call_item":
print(colored(f"Tool was called: {evt.item.raw_item}", "blue"))
print("-"*50)
elif evt.item.type == "tool_call_output_item":
print(colored(f"Tool output: {evt.item.output}", "blue"))
print("-"*50)
elif evt.item.type == "message_output_item":
print(colored("output", "blue"))
print("-"*50)
# print(evt.item.output)
else:
pass # Ignore other event types
# print(result.last_agent.name)
# print(colored(result.final_output, "blue"))
# print("-"*50)
inputs = result.to_input_list()
# print(result)
if result.last_agent.name == "WriterAgent":
report = result.final_output_as(ReportData)
break
print("\n\n" + "="*50 + "\n\n")
print("\n\n===== REPORT SUMMARY =====\n")
print(report.short_summary)
print("\n===== FULL REPORT (Markdown) =====\n")
print(report.markdown_report)
print("\n\n=====FOLLOW UP QUESTIONS=====\n\n")
follow_up_questions = "\n".join(report.follow_up_questions)
print(f"Follow up questions: {follow_up_questions}")
with open("report_handoff.md", "w", encoding="utf-8") as f:
f.write("\n\n===== REPORT SUMMARY =====\n")
f.write(report.short_summary)
f.write("\n===== FULL REPORT (Markdown) =====\n")
f.write(report.markdown_report)
f.write("\n===== FOLLOW-UP QUESTIONS =====\n\n")
f.write("Follow up questions: \n" + follow_up_questions + "\n")
async def main() -> None:
query = "2025年4月26日（土）の東京の天気を日本語で教えてください。"
await ResearchManager().run(query)
if __name__ == "__main__":
asyncio.run(main())
結果
planner_agent -> search_agent -> writer_agent の順で呼び出し（handoff）が行われました。しかし、planner_agentが複数の検索クエリを計画したにも関わらず、search_agentは最初の1回しか実行されませんでした。複数ステップにわたる処理（この場合は複数回の検索）をhandoffsで実現させるには、より詳細な指示や工夫が必要になる可能性があります。
一方で、writer_agentのoutput_typeとしてReportDataが指定されていたため、最終的な出力には「short_summary」と「follow_up_questions」が正しく含まれていました。
この結果から、handoffsはエージェントごとの出力形式を厳密に指定したい場合に有効ですが、複数ステップの処理や複雑なデータ連携を伴う場合には、エージェント間の情報伝達を慎重に設計する必要があると言えそうです。
2.3 データ連携のみをtoolsとして定義し、呼び出し元で制御した場合
ここでは、研究プロセス（計画、検索、レポート作成）全体をManagerAgentが統括し、個別のタスク（計画と検索の実行、レポート形式の取得）をfunction_toolとして定義した関数に委任します。
コード全文
from __future__ import annotations
import asyncio
from pydantic import BaseModel
from agents import (
Agent,
Runner,
WebSearchTool,
ModelSettings,
gen_trace_id,
trace,
function_tool,
TResponseInputItem,
)
from agents.model_settings import ModelSettings
from openai.types.responses import ResponseTextDeltaEvent
from termcolor import colored
from dotenv import load_dotenv
load_dotenv()
# ---------- Planner Agent ----------
PLANNER_PROMPT = (
"You are a helpful research assistant. Given a query, come up with a set of web searches "
"to perform to best answer the query. Output between 5 and 20 terms to query for."
)
class WebSearchItem(BaseModel):
"""
reason(str): Your reasoning for why this search is important to the query.
query(str): The search term to use for the web search.
"""
reason: str
query: str
class WebSearchPlan(BaseModel):
"""
searches(list[WebSearchItem]): A list of web searches to perform to best answer the query.
"""
searches: list[WebSearchItem]
planner_agent = Agent(
name="PlannerAgent",
instructions=PLANNER_PROMPT,
model="gpt-4o",
output_type=WebSearchPlan,
)
# ---------- Search Agent ----------
SEARCH_PROMPT = (
"You are a research assistant. Given a search term, you search the web for that term and "
"produce a concise summary of the results. The summary must 2-3 paragraphs and less than 300 "
"words. Capture the main points. Write succinctly, no need to have complete sentences or good "
"grammar. This will be consumed by someone synthesizing a report, so its vital you capture the "
"essence and ignore any fluff. Do not include any additional commentary other than the summary "
"itself."
)
search_agent = Agent(
name="Search agent",
instructions=SEARCH_PROMPT,
tools=[WebSearchTool()],
model_settings=ModelSettings(tool_choice="required"),
)
async def _search(item: WebSearchItem) -> str | None:
input = f"Search term: {item.query}\nReason for searching: {item.reason}"
try:
result = await Runner.run(
search_agent,
input,
)
return str(result.final_output)
except Exception:
return None
@function_tool
async def plan_and_search(query: str) -> str:
"""Web Search
At first, plan the search terms and then search the web for the terms.
Args:
query (str): The query to search the web for.
Returns:
str: The search results.
"""
result = await Runner.run(planner_agent, f"Query: {query}")
search_plan = result.final_output_as(WebSearchPlan)
print(colored(search_plan, "cyan"))
print("-"*50)
tasks = [asyncio.create_task(_search(item)) for item in search_plan.searches]
results = []
for task in asyncio.as_completed(tasks):
result = await task
print(colored(result, "cyan"))
print("-"*50)
if result is not None:
results.append(result)
return results
@function_tool
def get_report_rule() -> str:
"""Get the report rule"""
return (
"When you write a report, you should follow these rules:\n"
"You should write three parts:\n"
"1. Summary of the research\n"
"2. Detailed report\n"
"3. Follow-up questions\n"
"\n"
"The summary should be 2-3 sentences, and the report should be 5-10 pages of content, at least 1000 words.\n"
"The follow-up questions should be 3-5 questions that are related to the research and the query.\n"
)
# ---------- Manager Agent ----------
MANAGER_PROMPT = (
"You are a senior project manager."
"To complete the task, you need to use handoff to delegate next step of the task."
)
class OutputType(BaseModel):
"""
message(str): The message to display to the user.
is_finished(bool): Whether the task is finished.
"""
message: str
is_finished: bool
manager_agent = Agent[OutputType](
name="ManagerAgent",
instructions=MANAGER_PROMPT,
tools=[plan_and_search, get_report_rule],
output_type=OutputType,
model="gpt-4o",
)
# ---------- Research Bot ----------
class ResearchManager:
def __init__(self):
pass
async def run(self, query: str) -> None:
trace_id = gen_trace_id()
with trace("Research trace", trace_id=trace_id):
inputs: list[TResponseInputItem] = [{"role": "user", "content": f"Query: {query}"}]
while True:
result = Runner.run_streamed(
manager_agent,
inputs,
)
result_stream = result.stream_events()
async for evt in result_stream:
# print(evt.type)
# tool_called イベントを見て検索進行度らしきものを表示
# We'll ignore the raw responses event deltas
if evt.type == "raw_response_event":
if isinstance(evt.data, ResponseTextDeltaEvent):
print(evt.data.delta, end="")
continue
# When the agent updates, print that
elif evt.type == "agent_updated_stream_event":
print(colored(f"Agent updated: {evt.new_agent.name}", "green"))
print("-"*50)
continue
# When items are generated, print them
elif evt.type == "run_item_stream_event":
if evt.item.type == "tool_call_item":
print(colored(f"Tool was called: {evt.item.raw_item}", "blue"))
print("-"*50)
elif evt.item.type == "tool_call_output_item":
print(colored(f"Tool output: {evt.item.output}", "blue"))
print("-"*50)
elif evt.item.type == "message_output_item":
print(colored("output", "blue"))
print("-"*50)
# print(evt.item.output)
else:
pass # Ignore other event types
if result.final_output.is_finished:
break
inputs = result.to_input_list()
report = result.final_output.message
print("\n\n" + "="*50 + "\n\n")
print(report)
with open("report_tool.md", "w", encoding="utf-8") as f:
f.write(report)
async def main() -> None:
query = "2025年4月26日（土）の東京の天気について、日本語でレポートをまとめてください。"
await ResearchManager().run(query)
if __name__ == "__main__":
asyncio.run(main())
結果
ManagerAgentは、まずplan_and_searchツールを呼び出し、その中でplanner_agentによる計画立案と、計画に基づいた複数回のsearch_agentによる検索が実行されました。次にget_report_ruleツールを呼び出してレポート形式を取得し、最後にManagerAgent自身が実行結果の情報をもとに最終的なレポートを作成しました。レポート作成自体をManagerAgentが担当するため、「short_summary」や「follow_up_questions」といった情報も欠落なく最終出力に含まれています。このパターンでは、ManagerAgentのoutput_typeを指定していないため結果は文字列で返却されますが、プロンプトでXMLやJSON形式での出力を指示すれば、構造化された情報を得ることも可能です。
複雑な処理フローをPython関数（function_tool）として実装し、それをManagerAgentが必要に応じて呼び出すこの構成は、柔軟性と制御のしやすさのバランスが良いように思われます。
2.4 複数パターンの比較結果
今回の比較結果から、以下の傾向が見られました。
-
すべてtools (パターン2.1):
ManagerAgentが動的にツール（他のエージェント）を呼び出してタスクを進められる手軽さがありますが、ツールとして呼び出したエージェントの出力情報の取捨選択はManagerAgentの判断に委ねられるため、意図しない情報欠落が起こる可能性があります。 - すべてhandoffs (パターン2.2): エージェント間の遷移と各エージェントの出力形式を厳密に定義できますが、複数ステップの処理や複雑なデータ連携には工夫が必要です。
-
必要な処理をfunction_tool化 (パターン2.3): 複雑な処理やデータ連携をPython関数としてカプセル化し、それを
ManagerAgentがツールとして利用する方式です。ManagerAgentが全体の流れを制御しつつ、具体的な処理はツールに任せるため、柔軟性と制御のしやすさのバランスが取れています。
結論として、「情報取得や特定のタスク実行を行う関数（function_tool）を定義し、全体のワークフロー制御と最終的な生成処理はManagerAgentが行う」構成（パターン2.3）が、多くの場合で実装しやすく、かつ柔軟性が高いと考えられます。
厳密なワークフローやエージェントごとの出力形式の指定が重要な場合はhandoffs（パターン2.2）が適していますが、エージェント間の連携ロジック（例: 複数の検索タスクをどう処理するか）をより明確に実装する必要があります。
3. Human in the Loopを入れてみる
情報検索やタスク実行をツールとして定義するのがよさそうなので、ユーザーに質問するツールを定義することでHuman in the Loopを実現できるかを試してみます。Thinkツールも追加して、ユーザーに聞くべき質問を整理できるようにしましょう。
コード全文
from __future__ import annotations
import asyncio
from pydantic import BaseModel
from agents import (
Agent,
Runner,
WebSearchTool,
ModelSettings,
gen_trace_id,
trace,
function_tool,
TResponseInputItem,
)
from agents.model_settings import ModelSettings
from openai.types.responses import ResponseTextDeltaEvent
from termcolor import colored
from dotenv import load_dotenv
load_dotenv()
# ---------- Think Agent ----------
THINK_PROMPT = (
"You are a good research assistant."
"You are responsible for thinking about the query and coming up with a plan to complete the task."
"You should organize the information below:\n"
"1. What is the user saying?\n"
"2. How can we respond to the user's request, and what information do we need?\n"
"3. What information(based on the user's request and prompt) is currently missing?\n"
"4. What information do you need to ask the user?\n"
)
think_agent = Agent(
name="ThinkAgent",
instructions=THINK_PROMPT,
model="gpt-4o",
)
@function_tool
async def think(current_information: str) -> str:
"""Think
Think about the query and come up with a plan to complete the task.
This tool is used when you think about the query and come up with a plan to complete the task.
Thinking is very important to complete the task. So it is better to think carefully each point.
Args:
current_information (str): Information of the goal, the detailed summary of the conversation history and tool/handoffs/mcp information
Returns:
str: The plan to complete the task.
"""
print("thinking...")
print(colored(current_information, "cyan"))
print("-"*50)
result = await Runner.run(think_agent, f"Current information: {current_information}")
print(colored(result.final_output, "cyan"))
print("-"*50)
return result.final_output
# ---------- Planner Agent ----------
PLANNER_PROMPT = (
"You are a helpful research assistant. Given a query, come up with a set of web searches "
"to perform to best answer the query. Output between 5 and 20 terms to query for."
)
class WebSearchItem(BaseModel):
"""
reason(str): Your reasoning for why this search is important to the query.
query(str): The search term to use for the web search.
"""
reason: str
query: str
class WebSearchPlan(BaseModel):
"""
searches(list[WebSearchItem]): A list of web searches to perform to best answer the query.
"""
searches: list[WebSearchItem]
planner_agent = Agent(
name="PlannerAgent",
instructions=PLANNER_PROMPT,
model="gpt-4o",
output_type=WebSearchPlan,
)
# ---------- Search Agent ----------
SEARCH_PROMPT = (
"You are a research assistant. Given a search term, you search the web for that term and "
"produce a concise summary of the results. The summary must 2-3 paragraphs and less than 300 "
"words. Capture the main points. Write succinctly, no need to have complete sentences or good "
"grammar. This will be consumed by someone synthesizing a report, so its vital you capture the "
"essence and ignore any fluff. Do not include any additional commentary other than the summary "
"itself."
)
search_agent = Agent(
name="Search agent",
instructions=SEARCH_PROMPT,
tools=[WebSearchTool()],
model_settings=ModelSettings(tool_choice="required"),
)
async def _search(item: WebSearchItem) -> str | None:
input = f"Search term: {item.query}\nReason for searching: {item.reason}"
try:
result = await Runner.run(
search_agent,
input,
)
return str(result.final_output)
except Exception:
return None
@function_tool
async def web_search(query: str) -> str:
"""Web Search
At first, plan the search terms and then search the web for the terms.
Args:
query (str): Detailed information of what to search for.
Returns:
str: The search results.
"""
result = await Runner.run(planner_agent, f"Query: {query}")
search_plan = result.final_output_as(WebSearchPlan)
print(colored(search_plan, "cyan"))
print("-"*50)
tasks = [asyncio.create_task(_search(item)) for item in search_plan.searches]
results = []
for task in asyncio.as_completed(tasks):
result = await task
print(colored(result, "cyan"))
print("-"*50)
if result is not None:
results.append(result)
return results
@function_tool
def get_report_rule() -> str:
"""Get the report rule
This tool returns the rule and the format of a report.
In the company, we use this rule and format to write a report.
This tool is used when you write a report.
"""
return (
"When you write a report, you should follow these rules:\n"
"You should write three parts:\n"
"1. Summary of the research\n"
"2. Detailed report\n"
"3. Follow-up questions\n"
"\n"
"The summary should be 2-3 sentences, and the report should be 5-10 pages of content, at least 1000 words.\n"
"The follow-up questions should be 3-5 questions that are related to the research and the query.\n"
)
# ---------- Ask Human ----------
@function_tool
async def ask_human(query: str) -> str:
"""Ask Human
Ask the human for more information to complete the task.
Args:
query (str): The message to ask the human for more information.
Returns:
str: The human's response.
"""
print(colored(query, "cyan"))
print("-"*50)
human_input = input("Please enter your response: ")
return human_input
# ---------- Manager Agent ----------
MANAGER_PROMPT = (
"You are a senior project manager."
"You are responsible for completing the task."
"You can use the tools to complete the task."
"First, think about the query and come up with a plan to complete the task."
"Then, tackle the task step by step."
"When you don't have enough information, you can ask the human for more information."
)
manager_agent = Agent(
name="ManagerAgent",
instructions=MANAGER_PROMPT,
tools=[think, web_search, get_report_rule, ask_human],
model="gpt-4o",
)
# ---------- Research Bot ----------
class ResearchManager:
def __init__(self):
self.inputs: list[TResponseInputItem] = []
async def run(self, query: str) -> None:
trace_id = gen_trace_id()
with trace("Research trace", trace_id=trace_id):
self.inputs.append({"role": "user", "content": f"Query: {query}"})
result = Runner.run_streamed(
manager_agent,
self.inputs,
)
result_stream = result.stream_events()
async for evt in result_stream:
# We'll ignore the raw responses event deltas
if evt.type == "raw_response_event":
if isinstance(evt.data, ResponseTextDeltaEvent):
print(evt.data.delta, end="")
continue
# When the agent updates, print that
elif evt.type == "agent_updated_stream_event":
print(colored(f"Agent updated: {evt.new_agent.name}", "green"))
print("-"*50)
continue
# When items are generated, print them
elif evt.type == "run_item_stream_event":
if evt.item.type == "tool_call_item":
print(colored(f"Tool was called: {evt.item.raw_item}", "blue"))
print("-"*50)
elif evt.item.type == "tool_call_output_item":
print(colored(f"Tool output: {evt.item.output}", "blue"))
print("-"*50)
elif evt.item.type == "message_output_item":
print(colored("output", "blue"))
print("-"*50)
# print(evt.item.output)
else:
pass # Ignore other event types
self.inputs = result.to_input_list()
report = result.final_output
print("\n\n" + "="*50 + "\n\n")
print(report)
with open("report_tool_human.md", "w", encoding="utf-8") as f:
f.write(report)
async def main() -> None:
# query = "今日の東京の天気について、日本語でレポートをまとめてください。"
query = "2025/04/27の天気について、日本語でレポートをまとめてください。"
await ResearchManager().run(query)
if __name__ == "__main__":
asyncio.run(main())
結果
期待通り、まずthinkツールが呼び出され、タスクの計画が立てられました。次に、ユーザーの初期入力だけでは情報が不足していると判断され、ask_humanツールが呼び出されました。ユーザーが「どこの天気を知りたいか」という質問に答えると、その情報をもとにweb_searchツールが呼び出され、計画された検索が実行されました。最後に、得られた検索結果とget_report_ruleツールで取得したルール/フォーマットに基づき、ManagerAgentが最終的なレポートを生成しました。
このように、必要な情報をユーザーに確認するステップを自然に組み込むことができました。
4. おわりに
本記事では、OpenAI Agents SDKの基本的な使い方、toolsとhandoffsの機能的な違い、そしてそれらを用いた3つの異なる実装パターンにおける挙動の比較結果を紹介しました。
-
tools: 呼び出し元エージェントが制御を保持し、ツールの実行結果を受け取って処理を継続する場合に適しています。関数の呼び出しや、他のエージェントに一時的に処理を依頼するような場合に便利です。 -
handoffs: 処理の主体を完全に別のエージェントに移譲する場合に適しています。厳密なワークフロー定義や、エージェントごとに明確な役割と出力形式を持たせたい場合に有効です。
今回の検証では、パターン2.3（必要な処理をfunction_tool化し、ManagerAgentが全体を制御）が柔軟性と制御性のバランスが良いと感じました。
OpenAI Agents SDKは、複雑なタスクを複数のエージェントに分割し、連携させるための強力なフレームワークです。まるで小さなAGI（汎用人工知能）のチームを構築しているような感覚で、非常に興味深い技術だと感じました。
ぜひ実際のプロジェクトでこれらのパターンを試しながら、ご自身のユースケースに最適なエージェント連携の形を探求してみてください。
Discussion