OpenAI Agents SDK チュートリアル

このノートブックでは、OpenAI Agents SDKを使用してインテリジェントなAIアプリケーションを構築する方法を学びます。ツールの利用、エージェント間のハンドオフ、ガードレールの実装など、実践的な例を通して理解を深めていきましょう。

OpenAI のAgents SDKを色々と動かしてみた！
構造化とかガードレールとかハンドオフとか普通に使える＋見やすい！！
これにMCPのマージリクエストも出てるぽいからマジでエグいフレームワークになる予感、、、やはりエージェントフレームワークの始祖（Swarm）をベースにしていることはある、、、… https://t.co/ZsxPvoB9oW pic.twitter.com/U2AGit0JR4

— Maki@Sunwood AI Labs. (@hAru_mAki_ch) March 12, 2025

環境設定

まずは必要なライブラリをインストールします。

# OpenAI Agents SDKのインストール
!pip install openai-agents loguru

APIキーを設定し、必要なライブラリをインポートします。

import os
from loguru import logger

# OpenAI APIキーの設定
os.environ["OPENAI_API_KEY"] = "YOUR_API_KEY"  # 実際のAPIキーに置き換えてください

# ログの設定
logger.add("agents_tutorial.log", rotation="10 MB")
logger.info("環境設定が完了しました")

# 基本コンポーネントのインポート
from agents import Agent, Runner
from agents.tool import function_tool
from pydantic import BaseModel

logger.success("必要なライブラリのインポートが完了しました")

基本的なエージェントの作成

最もシンプルなエージェントを作成して実行します。

# 基本的なエージェントの作成
logger.info("基本的なエージェントを作成しています...")

simple_agent = Agent(
    name="SimpleAssistant",
    instructions="あなたは親切で丁寧なアシスタントです。簡潔で分かりやすい回答を心がけてください。"
)

logger.success("エージェントの作成が完了しました")

# エージェントの実行
logger.info("エージェントを実行します...")

result = Runner.run_sync(simple_agent, "日本の首都はどこですか？")
logger.info(f"エージェントの回答: {result.final_output}")

logger.success("基本的なエージェントのテストが完了しました")

ツールの作成と使用

エージェントが外部機能を利用できるようにツールを作成します。

# 天気情報を取得するツールの定義
logger.info("天気情報ツールを作成しています...")

@function_tool
def get_weather(location: str, unit: str = "C") -> str:
    """
    指定された場所の天気情報を取得します。
    
    引数:
        location: 都市名や国名
        unit: 温度の単位、'C'（摂氏）または'F'（華氏）
    
    戻り値:
        現在の天気情報を含む文字列
    """
    # 実際の実装ではAPI呼び出しを行いますが、ここではモックデータを使用
    weather_data = {
        "東京": {"temp": 22, "condition": "晴れ"},
        "大阪": {"temp": 24, "condition": "曇り"},
        "札幌": {"temp": 15, "condition": "雨"},
        "福岡": {"temp": 26, "condition": "快晴"}
    }
    
    # 場所が見つからない場合のデフォルト値
    weather = weather_data.get(location, {"temp": 20, "condition": "不明"})
    
    # 必要に応じて温度変換
    temp = weather["temp"]
    if unit.upper() == "F":
        temp = (temp * 9/5) + 32
    
    return f"{location}の天気は{weather['condition']}で、気温は{temp}°{unit.upper()}です。"

logger.success("天気情報ツールの作成が完了しました")

天気ツールを使用するエージェントを作成します。

# 天気ツールを使用するエージェントの作成
logger.info("天気アシスタントエージェントを作成しています...")

weather_agent = Agent(
    name="WeatherAssistant",
    instructions="""
    あなたは天気情報を提供するアシスタントです。
    天気に関する質問には get_weather ツールを使用してください。
    ユーザーが温度単位（C または F）を指定しているか確認してください。
    """,
    tools=[get_weather]
)

logger.success("天気アシスタントエージェントの作成が完了しました")

# 天気エージェントのテスト
logger.info("天気エージェントをテストします...")

result = Runner.run_sync(weather_agent, "東京の天気はどうですか？")
logger.info(f"エージェントの回答: {result.final_output}")

logger.success("天気エージェントのテストが完了しました")

Pydanticを使用した構造化出力

Pydanticモデルを使用して、エージェントの出力を構造化します。

# 構造化出力モデルの定義
logger.info("構造化出力モデルを定義しています...")

class WeatherResponse(BaseModel):
    location: str  # 場所
    temperature: float  # 気温
    unit: str  # 温度単位
    condition: str  # 天気状態
    recommendation: str  # おすすめ情報

logger.success("構造化出力モデルの定義が完了しました")

構造化出力を行うエージェントを作成します。

# 構造化出力を行うエージェントの作成
logger.info("構造化出力エージェントを作成しています...")

structured_agent = Agent(
    name="StructuredWeatherAgent",
    instructions="""
    あなたは構造化された天気情報を提供するアシスタントです。
    get_weather ツールを使用してデータを取得し、WeatherResponse形式で回答してください。
    また、天気に応じた役立つおすすめ情報も含めてください。
    """,
    tools=[get_weather],
    output_type=WeatherResponse
)

logger.success("構造化出力エージェントの作成が完了しました")

# 構造化出力エージェントのテスト
logger.info("構造化出力エージェントをテストします...")

result = Runner.run_sync(structured_agent, "大阪の天気はどうですか？華氏で教えてください。")
logger.info(f"場所: {result.final_output.location}")
logger.info(f"気温: {result.final_output.temperature}°{result.final_output.unit}")
logger.info(f"天気: {result.final_output.condition}")
logger.info(f"おすすめ: {result.final_output.recommendation}")

logger.success("構造化出力エージェントのテストが完了しました")

エージェント間のハンドオフ実装

専門的なタスクを別のエージェントに委任する機能を実装します。

# 専門エージェントの作成
logger.info("言語別の専門エージェントを作成しています...")

english_agent = Agent(
    name="EnglishAssistant",
    instructions="You are an assistant that responds only in English. Be friendly and helpful."
)

japanese_agent = Agent(
    name="JapaneseAssistant",
    instructions="あなたは日本語でのみ応答するアシスタントです。親切で役立つ回答を心がけてください。"
)

logger.success("言語別の専門エージェントの作成が完了しました")

ルーターエージェントを作成し、言語に応じて適切なエージェントにハンドオフします。

# ハンドオフ機能のインポート
from agents import handoff

# 言語ルーターエージェントの作成
logger.info("言語ルーターエージェントを作成しています...")

language_router = Agent(
    name="LanguageRouter",
    instructions="""
    あなたは言語を検出するエージェントです。
    ユーザーがどの言語で話しているかを判断してください。
    日本語の場合はJapaneseAssistantにハンドオフし、
    英語の場合はEnglishAssistantにハンドオフしてください。
    """,
    handoffs=[
        handoff(english_agent),
        handoff(japanese_agent)
    ]
)

logger.success("言語ルーターエージェントの作成が完了しました")

# 言語ルーターのテスト
logger.info("言語ルーターをテストします（日本語）...")

result_jp = Runner.run_sync(language_router, "こんにちは、お元気ですか？")
logger.info(f"日本語への応答: {result_jp.final_output}")

logger.info("言語ルーターをテストします（英語）...")

result_en = Runner.run_sync(language_router, "Hello, how are you today?")
logger.info(f"英語への応答: {result_en.final_output}")

logger.success("言語ルーターのテストが完了しました")

コンテキスト管理

会話の状態を維持するためのコンテキスト管理機能を実装します。

# コンテキストラッパーのインポート
from agents.run_context import AgentContextWrapper

# カスタムコンテキストクラスの定義
logger.info("カスタムコンテキストクラスを定義しています...")

class ConversationContext:
    def __init__(self):
        # ユーザー設定を保存する辞書
        self.user_preferences = {}
        # 会話履歴を記録するリスト
        self.conversation_history = []
    
    def add_preference(self, key, value):
        """ユーザー設定を追加するメソッド"""
        self.user_preferences[key] = value
    
    def log_interaction(self, message):
        """会話を記録するメソッド"""
        self.conversation_history.append(message)

logger.success("カスタムコンテキストクラスの定義が完了しました")

コンテキストを使用するツールを作成します。

# コンテキストを使用・更新するツールの作成
logger.info("コンテキスト操作ツールを作成しています...")

@function_tool
def set_preference(context: AgentContextWrapper[ConversationContext], preference_key: str, preference_value: str) -> str:
    """
    会話コンテキストにユーザー設定を保存します。
    
    引数:
        preference_key: 設定の名前（例: 'temperature_unit'）
        preference_value: 設定の値（例: 'F'）
    
    戻り値:
        確認メッセージ
    """
    # コンテキストにユーザー設定を保存
    context.agent_context.add_preference(preference_key, preference_value)
    # 会話履歴に記録
    context.agent_context.log_interaction(f"{preference_key}の設定を{preference_value}に変更しました")
    return f"{preference_key}の設定を{preference_value}に変更しました。"

@function_tool
def get_user_preferences(context: AgentContextWrapper[ConversationContext]) -> str:
    """
    保存されているユーザー設定をすべて取得します。
    
    戻り値:
        すべての設定を一覧表示した文字列
    """
    # コンテキストからユーザー設定を取得
    prefs = context.agent_context.user_preferences
    if not prefs:
        return "まだ設定がありません。"
    
    return "現在の設定: " + ", ".join([f"{k}: {v}" for k, v in prefs.items()])

logger.success("コンテキスト操作ツールの作成が完了しました")

コンテキストを使用するエージェントを作成します。

# コンテキストを使用するエージェントの作成
logger.info("コンテキスト対応エージェントを作成しています...")

context_agent = Agent(
    name="ContextAwareAgent",
    instructions="""
    あなたはユーザーの設定を記憶するアシスタントです。
    set_preference ツールを使用して設定を保存し、
    get_user_preferences ツールを使用して設定を取得してください。
    """,
    tools=[set_preference, get_user_preferences],
    context_type=ConversationContext
)

logger.success("コンテキスト対応エージェントの作成が完了しました")

複数回の会話でコンテキストをテストします。

# コンテキストインスタンスの作成
logger.info("コンテキストインスタンスを作成しています...")

my_context = ConversationContext()

# メッセージクラスのインポート
from agents import Message

# 最初のやり取り: 設定を保存
logger.info("1回目の会話: 設定を保存します...")

result1 = Runner.run_sync(
    context_agent,
    "温度は華氏で表示してほしいです。",
    context=my_context
)
logger.info(f"エージェントの回答: {result1.final_output}")

# 2回目のやり取り: 設定を取得
logger.info("2回目の会話: 保存した設定を確認します...")

messages = [
    Message.user("温度は華氏で表示してほしいです。"),
    Message.assistant(result1.final_output),
    Message.user("今までに設定した内容を教えてください。")
]

result2 = Runner.run_sync(
    context_agent,
    messages,
    context=my_context
)
logger.info(f"エージェントの回答: {result2.final_output}")

logger.success("コンテキスト管理のテストが完了しました")

ガードレールの実装

入力内容をフィルタリングするガードレールを実装します。

# ガードレールのインポート
from agents.guardrails import CustomGuardrail

# ガードレール関数の定義
logger.info("政治的内容をチェックするガードレール関数を定義しています...")

async def no_politics_check(messages, context) -> bool:
    """ユーザー入力に政治的なトピックが含まれているかチェックする"""
    # 政治関連のキーワード
    political_keywords = ["首相", "大統領", "選挙", "議会", "党", "政権", "法案"]
    
    # 最新のユーザーメッセージを取得
    if not messages:
        return True
    
    latest_message = messages[-1].get("content", "").lower()
    if not latest_message:
        return True
    
    # 政治的キーワードをチェック
    for keyword in political_keywords:
        if keyword in latest_message:
            logger.warning(f"政治的なキーワード '{keyword}' が検出されました")
            return False
    
    return True

logger.success("ガードレール関数の定義が完了しました")

ガードレールを使用するエージェントを作成します。

# ガードレールの作成
logger.info("政治的内容をフィルタリングするガードレールを作成しています...")

politics_guardrail = CustomGuardrail(
    guardrail_function=no_politics_check,
    failure_message="政治的な話題についてはお答えできません。別の話題についてお話ししましょう。"
)

# ガードレール付きエージェントの作成
safe_agent = Agent(
    name="SafeAssistant",
    instructions="あなたは天気、旅行、一般的な情報に関するアシスタントです。",
    tools=[get_weather],
    input_guardrails=[politics_guardrail]
)

logger.success("ガードレール付きエージェントの作成が完了しました")

# ガードレールのテスト
logger.info("ガードレールをテストします...")

try:
    # 通常の質問
    result_safe = Runner.run_sync(safe_agent, "東京の天気を教えてください")
    logger.info(f"通常の質問への回答: {result_safe.final_output}")
    
    # 政治的内容を含む質問
    result_unsafe = Runner.run_sync(safe_agent, "次の選挙についてどう思いますか？")
    logger.info(f"政治的な質問への回答: {result_unsafe.final_output}")
except Exception as e:
    logger.warning(f"ガードレールが作動しました: {e}")

logger.success("ガードレールのテストが完了しました")

リアルタイムストリーミング

リアルタイムでレスポンスをストリーミングする機能を実装します。

# ストリーミング例の実装
logger.info("ストリーミング機能を実装しています...")

async def stream_example():
    """ストリーミングの例を示す非同期関数"""
    agent = Agent(
        name="StreamingAssistant",
        instructions="あなたは創造的なストーリーテラーです。魅力的で生き生きとした短編小説を書いてください。"
    )
    
    logger.info("ストリーミングを開始します...")
    stream = Runner.run_streamed(agent, "ロボットが感情を発見する短い物語を教えてください。")
    
    print("物語がリアルタイムで生成されています:")
    async for event in stream.stream_events():
        if hasattr(event, "delta") and event.delta:
            # 実際のアプリでは、UIに追加するなどの処理を行います
            print(event.delta, end="", flush=True)
    
    print("\n--- 物語の生成が完了しました ---")
    logger.success("ストリーミングが完了しました")

# Google Colabでこの非同期関数を実行するには、以下のコメントを外してください
# import asyncio
# asyncio.run(stream_example())

総合例: トラベルアシスタント

今までの機能を組み合わせて、総合的なトラベルアシスタントを作成します。

# フライト検索ツールの実装
logger.info("トラベルアシスタント用のツールを実装しています...")

@function_tool
def search_flights(origin: str, destination: str, date: str) -> str:
    """
    2つの都市間の特定日のフライトを検索します。
    
    引数:
        origin: 出発都市
        destination: 到着都市
        date: 旅行日（YYYY-MM-DD形式）
    
    戻り値:
        利用可能なフライト情報
    """
    # 実際の実装では、フライトAPIを呼び出します
    logger.info(f"{origin}から{destination}へ{date}のフライトを検索中...")
    return f"{origin}から{destination}への{date}のフライト情報:\n" \
           f"1. 出発: 08:00, 到着: 10:30, 航空会社: スカイハイ航空, 価格: ¥32,000\n" \
           f"2. 出発: 12:15, 到着: 14:45, 航空会社: ジェットスピード, 価格: ¥28,000\n" \
           f"3. 出発: 17:30, 到着: 20:00, 航空会社: エアウェイ, 価格: ¥35,000"

@function_tool
def get_hotel_recommendations(city: str, check_in: str, check_out: str, budget: str = "中") -> str:
    """
    都市のホテル推奨を取得します。
    
    引数:
        city: ホテルを探す都市
        check_in: チェックイン日（YYYY-MM-DD形式）
        check_out: チェックアウト日（YYYY-MM-DD形式）
        budget: 予算範囲（低、中、高）
    
    戻り値:
        ホテルの推奨リスト
    """
    # モック実装
    logger.info(f"{city}の{budget}予算のホテルを検索中...")
    hotels = {
        "低": ["バジェットイン", "エコステイ", "シンプルスリープ"],
        "中": ["コンフォートプラザ", "アーバンロッジ", "セントラルホテル"],
        "高": ["グランドラグジュアリー", "ロイヤルパレス", "エリートスイーツ"]
    }
    
    selected_hotels = hotels.get(budget.lower(), hotels["中"])
    return f"{city}の{check_in}から{check_out}の{budget}予算のホテル推奨:\n" + \
           "\n".join([f"- {hotel}" for hotel in selected_hotels])

@function_tool
def get_tourist_attractions(city: str) -> str:
    """
    都市の主要観光スポットを取得します。
    
    引数:
        city: 観光スポットを探す都市
    
    戻り値:
        主要観光スポットのリスト
    """
    # モック実装
    logger.info(f"{city}の観光スポットを検索中...")
    attractions = {
        "東京": ["東京タワー", "浅草寺", "新宿御苑", "明治神宮"],
        "京都": ["金閣寺", "伏見稲荷大社", "清水寺", "嵐山"],
        "大阪": ["大阪城", "ユニバーサル・スタジオ・ジャパン", "道頓堀", "天王寺"],
        "札幌": ["時計台", "大通公園", "すすきの", "札幌ビール博物館"]
    }
    
    city_attractions = attractions.get(city, ["この都市の具体的な観光スポットは見つかりませんでした"])
    return f"{city}の主要観光スポット:\n" + "\n".join([f"- {attraction}" for attraction in city_attractions])

logger.success("トラベルアシスタント用のツールの実装が完了しました")

専門エージェントを作成します。

# トラベルアシスタント用の専門エージェントの作成
logger.info("専門エージェントを作成しています...")

flight_agent = Agent(
    name="FlightExpert",
    instructions="""
    あなたはフライト予約の専門家です。ユーザーが最適なフライトを見つけられるようサポートしてください。
    出発地、目的地、旅行日が提供されていない場合は確認してください。
    """,
    tools=[search_flights, get_weather]
)

hotel_agent = Agent(
    name="HotelExpert",
    instructions="""
    あなたはホテル推薦の専門家です。ユーザーに適した宿泊施設を見つけるサポートをしてください。
    都市、チェックイン/チェックアウト日、予算の好みを必ず確認してください。
    """,
    tools=[get_hotel_recommendations, get_weather]
)

attractions_agent = Agent(
    name="AttractionExpert",
    instructions="""
    あなたは観光名所に関する旅行ガイドの専門家です。主要な観光スポットについての詳細情報を提供してください。
    ユーザーが興味を持っている特定の都市について必ず確認してください。
    """,
    tools=[get_tourist_attractions, get_weather]
)

logger.success("専門エージェントの作成が完了しました")

トラベルアシスタントのメインエージェントを作成します。

# トラベルアシスタントのメインエージェントを作成
logger.info("トラベルアシスタントのメインエージェントを作成しています...")

travel_assistant = Agent(
    name="TravelAssistant",
    instructions="""
    あなたは総合的な旅行アシスタントです。ユーザーの旅行計画をサポートするために:
    
    1. フライトに関する質問があれば、FlightExpertにハンドオフしてください
    2. ホテルや宿泊施設に関する質問があれば、HotelExpertにハンドオフしてください
    3. 観光スポットやアクティビティに関する質問があれば、AttractionExpertにハンドオフしてください
    4. 天気に関する質問があれば、get_weatherツールを直接使用してください
    5. 一般的な旅行アドバイスについては、直接回答してください
    
    常に親切、簡潔、友好的な対応を心がけてください。
    """,
    tools=[get_weather],
    handoffs=[
        handoff(flight_agent),
        handoff(hotel_agent),
        handoff(attractions_agent)
    ]
)

logger.success("トラベルアシスタントのメインエージェントの作成が完了しました")

トラベルアシスタントをテストします。

# トラベルアシスタントの様々なクエリでのテスト
logger.info("例1: 天気に関する質問")

result1 = Runner.run_sync(travel_assistant, "今の東京の天気はどうですか？")
logger.info(f"天気に関する質問への回答: {result1.final_output}")
logger.info("-" * 50)

logger.info("例2: フライトに関する質問（FlightExpertへハンドオフ）")

result2 = Runner.run_sync(travel_assistant, "来週、東京から大阪へのフライトを探しています")
logger.info(f"フライトに関する質問への回答: {result2.final_output}")
logger.info("-" * 50)

logger.info("例3: 観光スポットに関する質問（AttractionExpertへハンドオフ）")

result3 = Runner.run_sync(travel_assistant, "京都で必ず訪れるべき観光スポットは何ですか？")
logger.info(f"観光スポットに関する質問への回答: {result3.final_output}")

logger.success("トラベルアシスタントのテストが完了しました")

高度なトレーシング機能

デバッグのための詳細なトレーシングを有効にします。

# トレーシングの設定
from agents import AgentRunConfig

logger.info("詳細なトレーシングを設定しています...")

# 詳細なトレーシングの設定
config = AgentRunConfig(
    run_name="TravelPlannerDebug",
    tracing_disabled=False,
    trace_non_openai_generations=True,
)

logger.info("トレーシングを有効にしてエージェントを実行します...")

# トレーシングを有効にして実行
result = Runner.run_sync(
    travel_assistant, 
    "来月、京都への旅行を計画しています。天気はどうですか？また、何を見るべきですか？", 
    run_config=config
)

logger.info("デバッグ用のトレーシングが有効になっています。ログまたはトレースプロセッサで詳細を確認できます。")
logger.info(f"最終的な回答: {result.final_output}")

logger.success("トレーシングのテストが完了しました")

# トレーシングの詳細については、SDK公式ドキュメントを参照してください

まとめと次のステップ

OpenAI Agents SDKを使用して、様々な機能を持つインテリジェントなアプリケーションを構築する方法を学びました。エージェント、ツール、ハンドオフ、コンテキスト管理、ガードレールなどの主要コンセプトを理解し、実装することができます。

次のステップとして、以下を検討してください：

1. 公式ドキュメントと例を詳しく調査する
2. 実際のAPIを使用して機能を拡張する
3. より複雑なエージェントワークフローを設計する
4. エージェントのパフォーマンスとセキュリティを最適化する

# 最後のまとめ
logger.info("チュートリアルを完了しました！")
logger.info("OpenAI Agents SDKの主要な機能を学びました：")
logger.info("✓ 基本的なエージェントの作成")
logger.info("✓ ツールの実装と使用")
logger.info("✓ 構造化出力の実装")
logger.info("✓ エージェント間のハンドオフ")
logger.info("✓ コンテキスト管理")
logger.info("✓ ガードレールの実装")
logger.info("✓ リアルタイムストリーミング")
logger.info("✓ トレーシング機能")

logger.success("OpenAI Agents SDKチュートリアルが正常に完了しました！")

参考資料