OpenAI Agents SDK で シンプルなDeep Researchクローンを実装する

import asyncio
from .manager import ResearchManager

# mainの定義
async def main() -> None:
    query = input("何を研究したいですか?")
    await ResearchManager().run(query)

# mainの実行
if __name__ == "__main__":
    asyncio.run(main())

from __future__ import annotations
import asyncio
import time
from rich.console import Console
from agents import Runner, custom_span, gen_trace_id, trace
from .agents.planner_agent import WebSearchItem, WebSearchPlan, planner_agent
from .agents.search_agent import search_agent
from .agents.writer_agent import ReportData, writer_agent
from .printer import Printer

# リサーチマネージャ
class ResearchManager:
    # 初期化
    def __init__(self):
        self.console = Console()
        self.printer = Printer(self.console)

    # リサーチの実行
    async def run(self, query: str) -> None:
        trace_id = gen_trace_id()
        with trace("Research trace", trace_id=trace_id):
            self.printer.update_item(
                "trace_id",
                f"View trace: https://platform.openai.com/traces/trace?trace_id={trace_id}",
                is_done=True,
                hide_checkmark=True,
            )

            self.printer.update_item(
                "starting",
                "Starting research...",
                is_done=True,
                hide_checkmark=True,
            )

            # Web検索のプランの作成
            search_plan = await self._plan_searches(query)

            # 検索クエリの実行
            search_results = await self._perform_searches(search_plan)

            # レポートの作成
            report = await self._write_report(query, search_results)

            final_report = f"Report summary\n\n{report.short_summary}"
            self.printer.update_item("final_report", final_report, is_done=True)

            self.printer.end()

        print("\n\n=====REPORT=====\n\n")
        print(f"Report: {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}")

    # Web検索のプランの作成
    async def _plan_searches(self, query: str) -> WebSearchPlan:
        self.printer.update_item("planning", "Planning searches...")
        result = await Runner.run(
            planner_agent,
            f"クエリ: {query}",
        )
        self.printer.update_item(
            "planning",
            f"Will perform {len(result.final_output.searches)} searches",
            is_done=True,
        )
        return result.final_output_as(WebSearchPlan)

    # 検索クエリの実行
    async def _perform_searches(self, search_plan: WebSearchPlan) -> list[str]:
        with custom_span("Search the web"):
            self.printer.update_item("searching", "Searching...")
            num_completed = 0
            tasks = [asyncio.create_task(self._search(item)) for item in search_plan.searches]
            results = []
            for task in asyncio.as_completed(tasks):
                result = await task
                if result is not None:
                    results.append(result)
                num_completed += 1
                self.printer.update_item(
                    "searching", f"Searching... {num_completed}/{len(tasks)} completed"
                )
            self.printer.mark_item_done("searching")
            return results

    # 個別の検索クエリの実行
    async def _search(self, item: WebSearchItem) -> str | None:
        input = f"検索ワード: {item.query}\n検索理由: {item.reason}"
        try:
            result = await Runner.run(
                search_agent,
                input,
            )
            return str(result.final_output)
        except Exception:
            return None

    # レポートの作成
    async def _write_report(self, query: str, search_results: list[str]) -> ReportData:
        self.printer.update_item("writing", "Thinking about report...")
        input = f"オリジナルクエリ: {query}\n要約された検索結果: {search_results}"
        result = Runner.run_streamed(
            writer_agent,
            input,
        )
        update_messages = [
            "Thinking about report...",
            "Planning report structure...",
            "Writing outline...",
            "Creating sections...",
            "Cleaning up formatting...",
            "Finalizing report...",
            "Finishing report...",
        ]

        last_update = time.time()
        next_message = 0
        async for _ in result.stream_events():
            if time.time() - last_update > 5 and next_message < len(update_messages):
                self.printer.update_item("writing", update_messages[next_message])
                next_message += 1
                last_update = time.time()

        self.printer.mark_item_done("writing")
        return result.final_output_as(ReportData)

from typing import Any
from rich.console import Console, Group
from rich.live import Live
from rich.spinner import Spinner

# プリンター
class Printer:
    # 初期化
    def __init__(self, console: Console):
        self.live = Live(console=console)
        self.items: dict[str, tuple[str, bool]] = {}
        self.hide_done_ids: set[str] = set()
        self.live.start()

    # 指定した項目の内容の更新
    def update_item(
        self, item_id: str, content: str, is_done: bool = False, hide_checkmark: bool = False
    ) -> None:
        self.items[item_id] = (content, is_done)
        if hide_checkmark:
            self.hide_done_ids.add(item_id)
        self.flush()

    # 指定した項目を完了に変更
    def mark_item_done(self, item_id: str) -> None:
        self.items[item_id] = (self.items[item_id][0], True)
        self.flush()

    # ライブ表示の更新
    def flush(self) -> None:
        renderables: list[Any] = []
        for item_id, (content, is_done) in self.items.items():
            if is_done:
                prefix = "✅ " if item_id not in self.hide_done_ids else ""
                renderables.append(prefix + content)
            else:
                renderables.append(Spinner("dots", text=content))
        self.live.update(Group(*renderables))

    # ライブ表示の停止
    def end(self) -> None:
        self.live.stop()

from pydantic import BaseModel
from agents import Agent

# 指示
INSTRUCTIONS = (
    "あなたは役に立つ研究アシスタントです。"
    "クエリが与えられたら、クエリに最適な回答を得るために実行するWeb検索のセットを考え出してください。"
    "クエリする5から20個の用語を出力してください。"
)

# Web検索アイテム
class WebSearchItem(BaseModel):
    reason: str
    "この検索がクエリにとって重要である理由。"

    query: str
    "Web検索に使用する検索用語。"

# Web検索プラン
class WebSearchPlan(BaseModel):
    searches: list[WebSearchItem]
    """クエリに最適な回答を得るために実行するWeb検索のリスト。"""

# PlannerAgentの準備
planner_agent = Agent(
    name="PlannerAgent",
    instructions=INSTRUCTIONS,
    model="gpt-4o",
    output_type=WebSearchPlan,
)

from agents import Agent, WebSearchTool
from agents.model_settings import ModelSettings

# 指示
INSTRUCTIONS = (
    "あなたは研究助手です。検索語句が与えられたら、その語句で Webを検索し、結果の簡潔な要約を作成します。"
    "要約は2～3段落、300語未満でなければなりません。"
    "要点を押さえてください。簡潔に書きます。完全な文や正しい文法である必要はありません。"
    "これはレポートをまとめる人が読むものなので、要点を押さえ、余分な部分は無視することが重要です。"
    "要約自体以外のコメントは含めないでください。"
)

# SearchAgentの準備
search_agent = Agent(
    name="SearchAgent",
    instructions=INSTRUCTIONS,
    tools=[WebSearchTool()],
    model_settings=ModelSettings(tool_choice="required"),
)

from pydantic import BaseModel
from agents import Agent

# 指示
INSTRUCTIONS = (
    "あなたは、リサーチ クエリに対するまとまりのあるレポートを書くことを任された上級研究者です。"
    "元のクエリと、リサーチアシスタントが行った初期リサーチが提供されます。"
    "まず、レポートの構造と流れを説明するレポートのアウトラインを作成します。"
    "次に、レポートを生成して、最終出力として返します。"
    "最終出力はマークダウン形式で、長くて詳細なものにする必要があります。"
    "5～10 ページ、少なくとも 1,000語のコンテンツを目指します。"
)

# レポート
class ReportData(BaseModel):
    short_summary: str
    """調査結果を2～3文で短く要約"""

    markdown_report: str
    """最終報告書"""

    follow_up_questions: list[str]
    """さらに研究するための提案トピック"""

# WriterAgentの準備
writer_agent = Agent(
    name="WriterAgent",
    instructions=INSTRUCTIONS,
    model="o3-mini",
    output_type=ReportData,
)

export OPENAI_API_KEY=sk-...

set OPENAI_API_KEY=sk-...

pip install openai-agents
pip install rich

python -m research_bot.main