🎸
OpenAI Agents SDK チュートリアル
このノートブックでは、OpenAI Agents SDKを使用してインテリジェントなAIアプリケーションを構築する方法を学びます。ツールの利用、エージェント間のハンドオフ、ガードレールの実装など、実践的な例を通して理解を深めていきましょう。
環境設定
まずは必要なライブラリをインストールします。
# 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を使用して、様々な機能を持つインテリジェントなアプリケーションを構築する方法を学びました。エージェント、ツール、ハンドオフ、コンテキスト管理、ガードレールなどの主要コンセプトを理解し、実装することができます。
次のステップとして、以下を検討してください：
- 公式ドキュメントと例を詳しく調査する
- 実際のAPIを使用して機能を拡張する
- より複雑なエージェントワークフローを設計する
- エージェントのパフォーマンスとセキュリティを最適化する
# 最後のまとめ
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チュートリアルが正常に完了しました！")
参考資料
<script async src="https://platform.twitter.com/widgets.js" charset="utf-8"></script>
Discussion