OpenAI Agent SDKではじめる高度なAIエージェント開発

こんにちは！今回は、OpenAIが提供する「Agent SDK」を使って、高度なAIエージェントを簡単に構築・カスタマイズする方法をご紹介します。

はじめに

大規模言語モデル（LLM）の進化により、単なる質問応答だけでなく、複雑なタスクを実行できるAIエージェントの開発が可能になりました。OpenAI Agent SDKは、そのようなエージェントを簡単に構築するためのフレームワークです。

このSDKを使うことで、以下のような機能を持つエージェントを簡単に作成できます：

• 自然言語による指示に基づいた応答生成
• カスタム関数ツールの実行
• 複数エージェント間の連携（ハンドオフ）
• コンテキスト管理
• 構造化データの出力
• 動的な指示変更
• ライフサイクルイベントの監視
• ガードレールによる安全性確保
• リアルタイムストリーミング応答

デモ動画

実際の動作を確認したい方は、以下のデモ動画をご覧ください：

OpenAI Agent SDKデモ動画

環境準備

まずは必要なライブラリをインストールしましょう：

pip install agents

次に、OpenAI APIキーを設定します。.envファイルを作成し、以下のように記述します：

OPENAI_API_KEY=your_api_key_here

Pythonコードからこのキーを読み込むには：

from dotenv import load_dotenv
import os
from agents import set_default_openai_key

load_dotenv()
openai_api_key = os.getenv("OPENAI_API_KEY")
set_default_openai_key(openai_api_key)

基本的な使い方

最もシンプルなエージェントの作成と実行方法は以下の通りです：

from agents import Agent, Runner

agent = Agent(
    name="Assistant", 
    instructions="You are a helpful assistant", 
    model="gpt-4o"
)

result = Runner.run_sync(
    agent,
    "Write the haiku about recursion in programming. Japanese language.",
)
print(result.final_output)

このコードでは：

1. Agentクラスを使ってエージェントを作成
2. エージェントに名前、指示、使用するモデルを設定
3. Runner.run_syncメソッドでエージェントを実行
4. 結果をfinal_outputから取得して表示

高度な機能

1. Function Tools

カスタム関数をエージェントのツールとして定義し、エージェントがそれらを呼び出せるようにします：

from agents import Agent, function_tool, Runner

@function_tool
def get_weather(city: str) -> str:
    # 実際のAPIを呼び出す代わりに、簡易的な応答を返す
    return f"{city} の天気は晴れです"

agent = Agent(
    name="Haiku Agent",
    instructions="常に俳句形式で回答してください。",
    model="o3-mini",
    tools=[get_weather],
)

result = Runner.run_sync(agent, "東京の天気を教えてください")
print(result.final_output)

function_toolデコレータを使うことで、通常のPython関数をエージェントが使用できるツールとして定義できます。エージェントは自動的にこの関数の存在を認識し、適切なタイミングで呼び出します。

2. Agent Handoffs（エージェント間の連携）

複数のエージェントを連携させ、特定の条件に基づいて別のエージェントに処理を委譲できます：

booking_agent = Agent(
    name="Booking Agent",
    instructions="航空券やホテルの予約に関する質問に答えてください。",
    model="o3-mini",
)

refund_agent = Agent(
    name="Refund Agent",
    instructions="返金に関する質問に答えてください。",
    model="o3-mini",
)

triage_agent = Agent(
    name="Triage Agent",
    instructions="質問内容に応じて、適切なエージェントへ手渡ししてください。",
    handoffs=[booking_agent, refund_agent],
    model="o3-mini",
)

この例では、triage_agentが質問内容を分析し、予約に関する質問ならbooking_agentへ、返金に関する質問ならrefund_agentへと処理を委譲します。

3. コンテキスト管理

エージェントが会話の履歴や状態を保持するためのコンテキスト機能を活用できます：

def get_instructions(context_wrapper, agent):
    conversation_history = context_wrapper.context.get("conversation_history", [])
    instructions = "ユーザーとの会話履歴を参照して、一貫性のある応答をしてください。"
    
    if conversation_history:
        instructions += "\n\n会話履歴:\n"
        for entry in conversation_history:
            role = "ユーザー" if entry["role"] == "user" else "アシスタント"
            instructions += f"{role}: {entry['content']}\n"
    
    return instructions

agent = Agent(
    name="Context Agent",
    instructions=get_instructions,
    model="o3-mini",
)

context = {"conversation_history": []}

この例では、会話履歴をコンテキストに保存し、それを基にエージェントの指示を動的に生成しています。

4. 構造化データの出力

Pydanticモデルを使用して、エージェントからの応答を構造化データとして受け取ることができます：

from pydantic import BaseModel, Field
from typing import List, Dict

class ProductReview(BaseModel):
    product_name: str = Field(description="レビュー対象の商品名")
    rating: int = Field(description="評価（1-5の整数）", ge=1, le=5)
    pros: List[str] = Field(description="商品の良い点のリスト")
    cons: List[str] = Field(description="商品の改善点のリスト")
    summary: str = Field(description="レビューの要約")
    recommendation: bool = Field(description="他の人にお勧めするかどうか")

review_agent = Agent(
    name="Review Agent",
    instructions="商品レビューリクエストに対して、詳細な構造化レビューを提供してください。",
    model="o3-mini",
)

この例では、商品レビューを構造化データとして受け取るためのPydanticモデルを定義しています。

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

エージェントからの応答をリアルタイムでトークンごとに受け取ることができます：

import asyncio
import time

async def run_streaming():
    start_time = time.time()
    
    # ストリーミングモードでエージェントを実行
    result = Runner.run_streamed(agent, query)
    
    # ストリーミングイベントを処理
    async for event in result.stream_events():
        if (
            hasattr(event, "data")
            and hasattr(event.data, "delta")
            and event.data.delta
        ):
            # deltaが文字列の場合
            if isinstance(event.data.delta, str) and event.data.delta:
                print(event.data.delta, end="", flush=True)
            # deltaがオブジェクトでcontentプロパティを持つ場合
            elif hasattr(event.data.delta, "content") and event.data.delta.content:
                print(event.data.delta.content, end="", flush=True)
    
    end_time = time.time()
    print(f"\n\n実行時間: {end_time - start_time:.2f}秒")

# 非同期関数を実行
asyncio.run(run_streaming())

この例では、エージェントの応答をリアルタイムで受け取り、トークンごとに表示しています。

実用的な応用例

Agent SDKを使った実用的な応用例をいくつか紹介します：

1. カスタマーサポートボット: 質問の種類に応じて適切な部門のエージェントに転送するシステム
2. データ分析アシスタント: データベースからの情報取得と分析を行うツールを持つエージェント
3. コンテンツ生成ツール: 特定のフォーマットや制約に従ってコンテンツを生成するエージェント
4. タスク管理アシスタント: タスクの追加、更新、完了を管理するエージェント
5. 多言語翻訳サービス: 様々な言語間で翻訳を行うエージェント

まとめ

OpenAI Agent SDKを使うことで、高度なAIエージェントを簡単に構築できることがわかりました。基本的な応答生成から、カスタム関数の実行、複数エージェント間の連携、構造化データの出力まで、様々な機能を活用することができます。

特に注目すべき点は、エージェントの振る舞いをカスタマイズする柔軟性と、複雑なタスクを実行するための豊富なツールセットです。これらを組み合わせることで、特定のユースケースに最適化されたAIソリューションを構築できます。

ぜひ、OpenAI Agent SDKを使って、あなた自身のAIエージェントを開発してみてください！

参考リソース

• OpenAI Agent SDK 公式ドキュメント
• OpenAI API リファレンス

---

この記事で紹介したコードサンプルは、GitHubで公開されています。スターやフォローをいただけると嬉しいです！