OpenAI Agents SDKを用いたマルチエージェントのポートフォリオコラボレーション

Posted at 2025-06-09

こちらのノートブックをDatabricksで動かしてみます。サーバレスコンピュートを使います。

コード自体はこちらに。

OpenAI Agents SDKを使用したマルチエージェントオーケストレーション: 金融ポートフォリオ分析の例

はじめに

このガイドは、OpenAIモデルとLLMエージェントに既に精通しており、現実世界の複雑なタスクのためにエージェントチームをオーケストレーションする方法を知りたい読者向けです。

学べること

このノートブックでは、OpenAI Agents SDKを使用して複雑なマルチエージェントコラボレーションシステムを設計および実装する方法を学びます。具体的には、次のことがわかります:

複数の専門エージェント（マクロ、ファンダメンタル、定量）がポートフォリオマネージャーエージェントの下で協力して、難しい投資研究問題を解決するワークフローを構築する方法。
中央エージェントが他のエージェントを特定のサブタスクのツールとしてオーケストレーションおよび呼び出す「ツールとしてのエージェント」アプローチを使用する方法。
SDKがサポートするすべての主要なツールタイプ（カスタムPython関数、Code InterpreterやWebSearchのような管理ツール、外部MCPサーバー）を単一の統合ワークフローで活用する方法。
エージェントパターンにおけるモジュール性、並列性、観測性のベストプラクティスを適用する方法。

なぜ重要か

「ツールとしてのエージェント」パターンは、透明性が高く、監査可能で、スケーラブルなマルチエージェントコラボレーションを構築する強力な方法です。この例では、OpenAI Agents SDKを使用して、深い専門性、並列実行、および堅牢なオーケストレーションを組み合わせる方法を示します。

このガイドの終わりまでに、研究、分析、または専門家の協力が有益な複雑なタスクのための独自のマルチエージェントワークフローを構築するための明確な青写真を得ることができます。

マルチエージェントコラボレーションとは？

マルチエージェントコラボレーションとは、複数の自律エージェント（LLM「ノード」）が協力して、単一のエージェントでは対処が難しい全体的な目標を達成することを意味します。一つの大きなプロンプトではなく、各エージェントが特定のサブタスクや専門分野を担当し、オーケストレーション層がこれらのエージェント「ノード」を一貫したワークフローに接続します。このアプローチは複雑なシステムに有用です。例えば、金融分析はマクロ経済分析、企業のファンダメンタル分析、定量的シグナル分析に分けられ、それぞれが異なるエージェントスペシャリストによって処理されます。エージェントは情報を共有し、その結果が組み合わさって最終的な成果を生み出します。

コラボレーションパターン: ハンドオフ vs. エージェント・アズ・ツール

OpenAI Agents SDKは、エージェントが協力するための複数のパターンをサポートしています。

ハンドオフコラボレーション: 一つのエージェントが問題の途中で他のエージェントに制御を「ハンドオフ」することができます。ハンドオフアーキテクチャでは、各エージェントは他のエージェントについて知っており、より適切なエージェントに委ねるタイミングを決定できます。これはオープンエンドや会話型のワークフローに柔軟ですが、タスクの全体像を維持するのが難しくなることがあります。SDKドキュメントで詳細を読む
エージェント・アズ・ツール: このアプローチでは、一つのエージェント（通常は中央のプランナーやマネージャー）が他のエージェントをツールのように呼び出します。サブエージェントは会話を引き継ぐことはなく、メインエージェントが特定のサブタスクのために呼び出し、その結果を取り入れます。このモデルは単一の制御スレッドを維持し（メインエージェントがすべてをオーケストレーション）、調整を簡素化する傾向があります。このリポジトリはエージェント・アズ・ツールモデルを使用しています: ポートフォリオマネージャーエージェントが主導権を握り、他の専門エージェントを必要に応じてツールとして使用します。この選択により、全体の推論が透明になり、サブタスクの並行実行が可能になり、複雑な分析に最適です。

これらのコラボレーションパターンの詳細については、OpenAI Agents SDKドキュメントを参照してください。

アーキテクチャ概要

私たちのシステムはハブアンドスポーク設計に従っています。ポートフォリオマネージャーエージェントがハブ（中央のコーディネーター）であり、専門エージェントがスポークです。ユーザーのクエリ（例：「予定されている金利引き下げが私のGOOGL保有にどのように影響するか？」）は最初にポートフォリオマネージャーに送られます。ポートフォリオマネージャーエージェントは問題を分解し、適切な専門エージェントに委任するように促されます。各専門エージェントを呼び出し可能なツールとして扱い、それぞれの分析部分を実行します。全てのエージェントがポートフォリオマネージャーに報告し、最終的な回答をユーザーに統合して提供します。

エージェントシステムにおいては、こういうシーケンス図を用いた設計が重要な気がします。

サポートされているツールタイプ

エージェントSDKの大きな利点は、エージェントが使用できるツールの定義に柔軟性があることです。ツールは、シンプルなPython関数から外部サービスまでさまざまです。本プロジェクトでは、以下を使用しています。

MCP（Model Context Protocol）サーバー: エージェントを外部ツールやデータソースに標準化された方法で接続するために使用します。このプロジェクトでは、Yahoo Financeデータ用のローカルMCPサーバー（mcp/yahoo_finance_server.py）を使用しています。詳細はこちら：OpenAI MCPドキュメント | MCP仕様
OpenAI管理ツール: OpenAIが提供する組み込みのホステッドツールで、カスタム実装は不要です。強力な機能をすぐに利用でき、Code Interpreter（定量・統計分析用）やWebSearch（最新ニュースやデータ取得用）などがあります。これらのツールは簡単に統合でき、OpenAIによって管理されており、コード実行やリアルタイム情報取得など高度な操作が可能です。
カスタムツール: 独自に定義し、ツールとして登録したPython関数です。エージェントSDKは、関数にデコレーターを付けるだけで、名前、ドキュメンテーション、入力スキーマを自動抽出します。これは、ドメイン固有のロジックやデータアクセス、ワークフロー拡張に最適です。本プロジェクトでは、FRED経済データ（FRED API）へのアクセスやファイルシステム操作にカスタムツールを使用しています。

カスタムツールは、組み込みや管理ツールを超えたエージェントの機能拡張に完全に柔軟です。関数ツールに関するSDKドキュメントを見る

ツールを追加したいですか？ SDKは、Web検索、ファイル検索、コード実行など、多彩なツールタイプをサポートしています。SDKドキュメントのサポートツール一覧もご覧ください。

セットアップ

エラーに遭遇したので、requirements.txtを以下のように変更しています。

requirements.txt

openai
openai-agents
fredapi
yfinance

# 必要な依存関係のインストール
%pip install -r requirements.txt
%pip install -U mlflow
%restart_python

ワークフローを実行する前に、環境変数を設定してください:

OPENAI_API_KEY (OpenAIアクセス用)
FRED_API_KEY (FRED経済データ用、FRED APIキーの取得方法を参照)

import os
os.environ['FRED_API_KEY'] = "FRED APIキー"
os.environ['OPENAI_API_KEY'] = dbutils.secrets.get(scope="demo-token-takaaki.yayoi", key="openai_api_key")

missing = []
if not os.environ.get('OPENAI_API_KEY'):
    missing.append('OPENAI_API_KEY')
if not os.environ.get('FRED_API_KEY'):
    missing.append('FRED_API_KEY')

if missing:
    print(f"Missing environment variable(s): {', '.join(missing)}. Please set them before running the workflow.")
else:
    print("All required API keys are set.")

あと、コードではgpt-4.1を使っていますが、私はTPMの制限に引っかかったので、gpt-4.1-miniに切り替えています。editor.pyやfundamental.pyなどのファイルを変更しています。

#default_model = "gpt-4.1"
default_model = "gpt-4.1-mini"

ワークフローの実行

質問を自由に編集してください。ただし、日付フィールドは精度を高めるために保持してください！

免責事項: この例は教育目的のためのものです。投資判断を行う前に、資格のある金融専門家に相談してください。

ワークフローは、ユーザーリクエストをヘッドポートフォリオマネージャー（PM）エージェントに送信することで開始されます。PMエージェントは、必要に応じて専門エージェントやツールに委任しながら、全体のプロセスを調整します。OpenAI Tracesを使用して、各エージェントおよびツールの呼び出しに関する詳細な可視性をリアルタイムで監視できます。

以下のコード内のquestionを好きなように編集してください。ただし、精度を高めるために日付フィールドは保持してください！

注意: タスクの複雑さに応じて、このリクエストには最大で10分かかる場合があります。

import datetime
import json
import os
from pathlib import Path
from contextlib import AsyncExitStack
from agents import Runner, add_trace_processor, trace
from agents.tracing.processors import BatchTraceProcessor
from utils import FileSpanExporter, output_file
from investment_agents.config import build_investment_agents
import asyncio

add_trace_processor(BatchTraceProcessor(FileSpanExporter()))

async def run_workflow():
    if "OPENAI_API_KEY" not in os.environ:
        raise EnvironmentError("OPENAI_API_KEYが設定されていません — 実行する前に環境変数として設定してください。")

    today_str = datetime.date.today().strftime("%Y年%m月%d日")
    question = (
        f"今日は{today_str}です。"
        "予定されている金利引き下げがGOOGLの保有株にどのような影響を与えるか？"
        "現在の価格に影響を与えるすべての要因（マクロ、テクニカル、ファンダメンタルなど）を考慮して、年末までの現実的な価格目標は何ですか？"
    )
    bundle = build_investment_agents()

    async with AsyncExitStack() as stack:
        for agent in [getattr(bundle, "fundamental", None), getattr(bundle, "quant", None)]:
            if agent is None:
                continue
            for server in getattr(agent, "mcp_servers", []):
                await server.connect()
                await stack.enter_async_context(server)

        print("トレースを有効にしてマルチエージェントワークフローを実行中...\n")
        with trace(
            "投資リサーチワークフロー",
            metadata={"question": question[:512]}
        ) as workflow_trace:
            print(
                f"\n🔗 OpenAIコンソールでトレースを表示: "
                f"https://platform.openai.com/traces/trace?trace_id={workflow_trace.trace_id}\n"
            )

            response = None
            try:
                response = await asyncio.wait_for(
                    Runner.run(bundle.head_pm, question, max_turns=40),
                    timeout=1200
                )
            except asyncio.TimeoutError:
                print("\n❌ ワークフローが20分後にタイムアウトしました。")

            report_path = None
            try:
                if hasattr(response, 'final_output'):
                    output = response.final_output
                    if isinstance(output, str):
                        data = json.loads(output)
                        if isinstance(data, dict) and 'file' in data:
                            report_path = output_file(data['file'])
            except Exception as e:
                print(f"投資レポートパスを解析できませんでした: {e}")

            print(f"ワークフロー完了 エージェントからの応答: {response.final_output if hasattr(response, 'final_output') else response}, 作成された投資レポート: {report_path if report_path else '[不明]'}")

# In a Jupyter notebook cell, run:
await run_workflow()

トレースを有効にしてマルチエージェントワークフローを実行中...


🔗 OpenAIコンソールでトレースを表示: https://platform.openai.com/traces/trace?trace_id=trace_xxxxx

ワークフロー完了 エージェントからの応答: {"file": "investment_report.md"}, 作成された投資レポート: /Workspace/Users/takaaki.yayoi@databricks.com/20250609_multi_agent_orchestration/openai-cookbook/examples/agents_sdk/multi-agent-portfolio-collaboration/outputs/investment_report.md

途中の様子をトレースで確認できます。並列に異なる分析エージェントが呼び出されていることがわかります。

ヘッドポートフォリオマネージャーエージェントの分解

ヘッドポートフォリオマネージャー（PM）エージェントは、全体のワークフローのオーケストレーターです。各専門分野に焦点を当てた4つの専門エージェントのセットを調整します。この設計は意図的です。単一のエージェントにすべての責任を負わせると、浅くて一般的な出力になり、システムの維持や改善が難しくなります。

この設計の理由

問題を明確な役割を持つ専門エージェントに分割することで、以下の利点があります：

より深く、高品質なリサーチ: 各エージェントは自分の分野に集中し、適切なツールとプロンプトを使用できます。PMエージェントはこれらの視点をまとめ、より微妙で堅牢な回答を提供します。
モジュール性と明確さ: 一つのエージェントを更新、テスト、改善しても他のエージェントに影響を与えません。これにより、システムの維持と拡張が容易になります。
並列処理による迅速な結果: 独立したエージェントが同時に作業できるため、複雑で多部分の分析を完了する時間が大幅に短縮されます。
一貫性と監査可能性: 構造化されたプロンプト駆動のワークフローにより、すべての実行がベストプラクティスに従い、デバッグが容易で、信頼できる出力が得られます。

このアプローチは、リサーチアシスタント、意思決定支援ツール、専門家の協力とオーケストレーションが利益をもたらすシステムなど、深さ、専門性、信頼性を求めるあらゆるアプリケーションに最適です。

実際の実装方法:

各専門エージェント（ファンダメンタル、マクロ、定量）は、SDKのfunction_toolデコレータを使用して呼び出し可能なツールとしてラップされ、カスタム名と説明が付けられます。これにより、PMエージェントのツールセットが明示的かつLLMフレンドリーになります。
ヘッドPMエージェントは、run_all_specialists_parallelツールを使用して3つの専門家を同時に呼び出し、parallel_tool_calls=Trueを活用して最大の速度と効率を実現します。
エージェントのプロンプトはマークダウンファイル（pm_base.md）から読み込まれ、企業の哲学だけでなく、詳細なツール使用ルールとステップバイステップのワークフローもエンコードされます。これにより、すべての実行が一貫性があり、監査可能で、ベストプラクティスに従うことが保証されます。
専門家の出力を収集してレビューした後、PMエージェントは専用のメモエディターツールを使用して投資レポートを組み立て、フォーマットし、最終化します。この関心の分離により、ワークフローがモジュール化され、拡張が容易になります。
システムは拡張性を考慮して設計されています。新しい専門エージェントを追加したり、ツールを交換したり、プロンプトを更新したりしても、全体のオーケストレーションロジックを壊すことはありません。すべてのツール呼び出し、エージェントの決定、および出力は、完全な透明性とデバッグのためにOpenAI Tracesにキャプチャされます。

これらの実装選択は、上記の利点を直接サポートし、深く、モジュール化され、信頼性の高いマルチエージェントリサーチワークフローを実現し、維持、監査、および改善が容易です。

ヘッドポートフォリオマネージャーエージェント: コード

from agents import Agent, ModelSettings, function_tool
from utils import load_prompt, DISCLAIMER

def build_head_pm_agent(fundamental, macro, quant, memo_edit_tool):
    def make_agent_tool(agent, name, description):
        @function_tool(name_override=name, description_override=description)
        async def agent_tool(input):
            return await specialist_analysis_func(agent, input)
        return agent_tool
    fundamental_tool = make_agent_tool(fundamental, "fundamental_analysis", "Generate the Fundamental Analysis section.")
    macro_tool = make_agent_tool(macro, "macro_analysis", "Generate the Macro Environment section.")
    quant_tool = make_agent_tool(quant, "quantitative_analysis", "Generate the Quantitative Analysis section.")

    @function_tool(name_override="run_all_specialists_parallel", description_override="Run all three specialist analyses (fundamental, macro, quant) in parallel and return their results as a dict.")
    async def run_all_specialists_tool(fundamental_input, macro_input, quant_input):
        return await run_all_specialists_parallel(
            fundamental, macro, quant,
            fundamental_input, macro_input, quant_input
        )

    return Agent(
        name="Head Portfolio Manager Agent",
        instructions=(load_prompt("pm_base.md") + DISCLAIMER),
        model="gpt-4.1",
        tools=[fundamental_tool, macro_tool, quant_tool, memo_edit_tool, run_all_specialists_tool],
        model_settings=ModelSettings(parallel_tool_calls=True, tool_choice="auto", temperature=0)
    )

ヘッドPMシステムプロンプト: ベストプラクティスの強制

PMエージェントのシステムプロンプト（prompts/pm_base.md参照）は、ワークフローの中心です。これには以下がエンコードされています：

会社の哲学（独創性、リスク意識、コンセンサスへの挑戦）
明確なツール使用ルール（並列ツールの使用タイミング、入力の構造化方法）
堅牢なマルチステップワークフロー（タスクタイプの決定、ガイダンスの提供、出力のレビュー、メモの組み立て、データ欠落の処理）

このプロンプトにより、すべての実行が以下を保証します：

一貫性: 毎回同じ高い基準とプロセスが守られます。
監査可能: 各ステップ、ツール呼び出し、決定がトレースに表示されます。
高品質: 出力は独創的で、リスクを意識し、厳密にレビューされます。

# ヘッドポートフォリオマネージャーエージェントが使用する実際のシステムプロンプトを表示
from pathlib import Path
from IPython.display import Markdown, display

pm_prompt_path = Path("prompts/pm_base.md")
if pm_prompt_path.exists():
    with pm_prompt_path.open("r", encoding="utf-8") as f:
        content = f.read()
    display(Markdown(content))
else:
    print("System prompt not found at prompts/pm_base.md")

この辺りも翻訳すれば日本語でいけそうです。

例の出力

以下は、ワークフローを通じて生成された投資レポートの例です。出力はディレクトリ内の outputs フォルダーに書き込まれます。

エージェント構築のベストプラクティス

最も効果的なエージェントシステムは、モジュール化されたエージェント設計、明確なツール定義、並列実行、構造化されたプロンプトを組み合わせています。このアプローチはOpenAI Agents SDKの中心であり、ワークフローを堅牢でスケーラブル、かつデバッグや拡張が容易にします。

OpenAI Agents SDKがこれらのベストプラクティスを可能にする主な特徴：

エージェントループ: ツール呼び出し、LLMの推論、ワークフロー制御を自動で処理。
Pythonファーストのオーケストレーション: 親しみやすいPythonパターンでエージェントの連結、合成、オーケストレーションが可能。
ハンドオフ: エージェント間でタスクを委譲し、専門性とモジュール性を実現。
ガードレール: 入出力の検証とエラー時の早期停止で信頼性を確保。
関数ツール: 任意のPython関数をツールとして登録し、自動スキーマと検証を提供。
トレーシング: ワークフローの全ステップを可視化、デバッグ、監視して完全な透明性を実現。

効果的なエージェントシステムを構築するには、よく設計されたツール、慎重なオーケストレーション、適切なモデル選択の組み合わせが重要です。この例では、強力な分析力とツール利用能力を持つGPT-4.1ファミリーのモデルを使用しています（GPT-4.1プロンプティングガイド参照）。より深いアーキテクチャのベストプラクティスについては、同梱のA Practical Guide to Building Agents (PDF)をご覧ください。これらの要素を組み合わせることで、堅牢でスケーラブル、かつデバッグや拡張が容易なシステムが得られます。

ぜひご自身の投資質問でサンプルを試し、フィードバックをお寄せください。楽しい構築を！