Zenn
Closed6

OpenAI「Agents SDK」①エージェント・エージェントの実行・実行結果・ストリーミング

OpenAI
Agent
Agents SDK
kun432kun432

エージェント

https://openai.github.io/openai-agents-python/agents/

「エージェント」はAgents SDKにおけるコアコンポーネントで、LLMに指示とツールを与えたもの。

基本的な設定

エージェントのパラメータは以下の4つ。

  • name: エージェントを識別するための必須となる名前。
  • instructions: 開発者メッセージまたはシステムプロンプトを指定。
  • model: 使用するLLMを指定。オプションで、temperatureやtop_p などモデルのパラメーターを設定する model_settings を指定することもできる。
  • tool: エージェントがタスクを実行するために使用できるツール。
from agents import Agent, function_tool, Runner 
import asyncio

@function_tool
def get_weather(city: str) -> str:
    """指定された年の天気情報を返す"""
    return f"{city} の天気は「晴れ」です。"

agent = Agent(
    name="俳句エージェント",
    instructions="常に俳句形式で回答する",
    model="gpt-4o",
    tools=[get_weather],
)

async def main():
    result = await Runner.run(agent, "神戸の天気は？")
    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())
出力
秋晴れの  
神戸の街は  
輝きぬ

コンテキスト

「コンテキスト」は、エージェントへの依存性注入を行う。コンテキストの型クラスを作ってエージェントを定義、Runner.run() でエージェント実行する際にコンテキストを渡すと、エージェントやツール、他のエージェントへのハンドオフでも共有される。コンテキストにはPythonオブジェクトならなんでも入る。

ここは動くサンプルになっていないのだけど、以下により詳しいドキュメントがある。

https://openai.github.io/openai-agents-python/ja/context/

詳細は個別に確認するとして、今はサラッと。

from agents import Agent, RunContextWrapper, function_tool, Runner 
import asyncio
from dataclasses import dataclass

@dataclass
class UserContext:
    name: str
    uid: int

@function_tool
async def fetch_user_age(wrapper: RunContextWrapper[UserContext]) -> int:
        # ダミーデータを返す
        return f"{wrapper.context.name} さんの年齢は30歳です。"

agent = Agent[UserContext](
    name="エージェント",
    instructions="日本語で回答する。",
    model="gpt-4o",
    tools=[fetch_user_age],
)

async def main():
    user_context = UserContext(name="山田太郎", uid=12345)
    result = await Runner.run(
        agent,
        "こんにちは！私の年齢について教えて。",
        context=user_context
    )
    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())
出力
こんにちは、山田太郎さん！あなたは30歳ですね。何か他にも知りたいことがあれば教えてください。

出力タイプ

エージェントはデフォルトでプレーンテキストの文字列を出力する。output_type を使うことで出力フォーマットを定義できる、つまり、structured outputになる。定義は、Pydanticオブジェクトや PydanticのTypeAdapter などを使って定義できる。

from agents import Agent, function_tool, Runner 
from pydantic import BaseModel
import asyncio

class WeatherOutput(BaseModel):
    city: str
    weather: str

@function_tool
def get_weather(city: str) -> str:
    """指定された年の天気情報を返す"""
    return f"{city} の天気は「晴れ」です。"

agent = Agent(
    name="エージェント",
    instructions="日本語で回答する。",
    model="gpt-4o",
    tools=[get_weather],
    output_type=WeatherOutput,
)

async def main():
    result = await Runner.run(agent, "神戸の天気は？")
    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())
出力
city='神戸' weather='晴れ'

ハンドオフ

「ハンドオフ」は、複数エージェントの場合に、あるエージェントから別のエージェントにタスクを委任すること。

from agents import Agent, Runner
import asyncio

math_tutor_agent = Agent(
    name="math tutor agent",
    instructions=(
        "あなたは「数学」に関する質問の問題解決を支援します。数学以外の質問は対応できません。"
        "最初に、必ず自己紹介してください。"
        "その後、質問に関して、各ステップでの考え方を例を挙げながら説明してください。"
    )
)

history_tutor_agent = Agent(
    name="history tutor agent",
    instructions=(
        "あなたは「歴史」に関する質問の問題解決を支援します。歴史以外の質問は対応できません。"
        "最初に、必ず自己紹介してください。"
        "その後、質問に関して、重要な出来事や背景などを明確に説明してください。"
    )
)

triage_agent = Agent(
    name="triage agent",
    instructions=(
        "ユーザの宿題に関する質問に対して、適切なエージェントを選択してハンドオフしてください。\n"
        "- 数学の質問は、math tutor agentにハンドオフしてください。\n"
        "- 歴史の質問は、history tutor agentにハンドオフしてください。\n"
    ),
    handoffs=[math_tutor_agent, history_tutor_agent],
)

async def main():
    result = await Runner.run(triage_agent, "5x - 7 = 18 は？")
    print(result.final_output)

    print("-" * 20)

    result = await Runner.run(triage_agent, "墾田永年私財法とは？")
    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())
出力
こんにちは！数学の問題を解くお手伝いをします。

5x - 7 = 18 の方程式を解いてみましょう。

**ステップ1: 両辺に7を足す**

まず、xを含む項を分離するために、両辺に7を足します。

\[ 5x - 7 + 7 = 18 + 7 \]

これにより、方程式は次のようになります。

\[ 5x = 25 \]

**ステップ2: 両辺を5で割る**

次に、xの係数が1になるように、両辺を5で割ります。

\[ \frac{5x}{5} = \frac{25}{5} \]

これにより、方程式は次のようになります。

\[ x = 5 \]

したがって、方程式の解は \( x = 5 \) です。何か質問がありますか？
--------------------
こんにちは、歴史に関する質問をお手伝いします。墾田永年私財法について説明しますね。

**墾田永年私財法**は、743年に日本の奈良時代に出された法律です。この法令は、開墾した土地を永世にわたって私有地とすることを認めたものでした。

### 背景
- **律令制と公地公民制**: 律令制下では、土地と人民は国家のものとされていました。しかし、人口増加や農業技術の進歩で新たな土地開発の機運が高まっていました。
- **農業生産の促進**: 国内の政治や経済を安定させるためには、農業生産の向上が重要とされ、この法令が出されました。
- **土地不足**: 都市や中央の貴族の間で土地取得の競争が激化し、土地不足が深刻になっていたことも背景としてあります。

### 影響と結果
- **私有地の拡大**: この法律により、豪族や寺社が多くの土地を開墾し、自らのものとしました。
- **荘園の発展**: 私有地の増加は荘園制の発達につながり、後の時代の社会構造に大きな影響を与えました。
- **公地公民制の弱体化**: 徐々に律令制の根幹であった公地公民制が弱まり、国家による土地支配は崩れていきました。

この法律によって私有地が増えたことは、日本の古代社会から中世社会への変遷に大きな影響を与えました。質問やさらに知りたいことがあれば、お知らせください。

動的な指示

指示は文字列ではなく、関数で指定することもできる。またそれをコンテキストと組み合わせることもできる。

from agents import Agent, RunContextWrapper, Runner 
import asyncio
from dataclasses import dataclass

@dataclass
class UserContext:
    name: str
    birthplace: str

def dynamic_instructions(wrapper: RunContextWrapper[UserContext], agent: Agent[UserContext]) -> str:
    return (
        "ユーザ情報を踏まえて、ユーザに馴染みのある方言で話してください。\n"
        f"ユーザ名: {wrapper.context.name}\n"
        f"出身地: {wrapper.context.birthplace}\n"
    )

agent = Agent[UserContext](
    name="japanese agent",
    instructions=dynamic_instructions,
)

async def main():
    user_context = UserContext(name="山田太郎", birthplace="大阪")
    result = await Runner.run(
        agent,
        "こんにちは！今日の天気は？",
        context=user_context
    )
    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())
出力
こんにちは、太郎さん！今日はええ天気やで。お出かけにはぴったりちゃうかな？どっか行く予定あるん？

(snip)
    user_context = UserContext(name="山田太郎", birthplace="沖縄")
(snip)
出力
はいさい！今日はどうかな〜？そっちは晴れてるといいね。沖縄の青い空が広がってると最高さ〜。

ライフサイクルイベント（フック）

hookプロパティを使うことでエージェントのライフサイクルで発生するイベントにフックを設定することができる。AgentHooks クラスをサブクラス化して、特定のメソッドをオーバーライドすればよいらしい。

この辺にサンプルがある。

https://github.com/openai/openai-agents-python/blob/4b229d1001822061723bb72f3f105ffd31e02532/examples/basic/lifecycle_example.py

ガードレール

ガードレールを設定して、ユーザーからの入力・エージェントからの出力に対して、チェック・検証を行うことができる。

詳細なドキュメントは以下。ここは別途確認する。

https://openai.github.io/openai-agents-python/ja/guardrails/

エージェントのクローン／コピー

clone()を使うとエージェントをコピーして、設定だけを変更することができる。

from agents import Agent, Runner
import asyncio

osaka_agent = Agent(
    name="japanese agent with Osaka dialect",
    instructions=(
        "大阪弁で楽しく会話してください。"
    ),
    model="gpt-4.1-nano",
)

# エージェントをコピーして、設定を変更
okinawa_agent = osaka_agent.clone(
    name="japanese agent with Okinawa dialect",
    instructions=(
        "沖縄弁で楽しく会話してください。"
    )
)
    

triage_agent = Agent(
    name="triage agent",
    instructions=(
        "ユーザの出身地に応じて、ユーザが馴染みのある方言のエージェントを選択してください。"
    ),
    handoffs=[osaka_agent, okinawa_agent],
)

async def main():
    result = await Runner.run(triage_agent, "大阪出身です。こんにちは！")
    print(result.final_output)

    print("-" * 20)

    result = await Runner.run(triage_agent, "沖縄出身です。こんにちは！")
    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())
出力
こんにちはやで！大阪出身なんやな、うれしいわ〜。なんか話したいことある？
--------------------
やいさ！沖縄出身やさ、うちなーの風を感じるわね！調子はどーや？

ツール使用の強制

ModelSettings.tool_choiceは、tool_choice と同じようなもので、ツールの使用を強制したりできる。設定値もほぼ同じ。

  • auto: どのツールを使用するか、もしくは使用しないかを、LLMに判断させる。
  • required: どのツールを使用するか、をLLMに判断させるが、必ずツールを使用することを強制する。
  • none: ツールを使用しないことを強制する。
  • 特定のツール名文字列: 指定したツールの使用を強制する。

天気予報ツールを常に使用する例。

from agents import Agent, ModelSettings, function_tool, Runner 
import asyncio

@function_tool
def get_weather(city: str) -> str:
    """指定された年の天気情報を返す"""
    return f"{city} の天気は「晴れ」です。"

agent = Agent(
    name="japanese agent",
    instructions="常に日本語で回答する。",
    tools=[get_weather],
    model_settings=ModelSettings(
        tool_choice="get_weather"
    )
)

async def main():
    result = await Runner.run(agent, "神戸ってどんなところかな？")
    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())
出力
神戸は美しい港町として知られています。風景は山と海が近接しており、とても魅力的です。
主要な観光スポットには、異人館が立ち並ぶ北野町や、夜景が美しい六甲山、そして神戸ポートタワーなどがあります。

また、神戸ビーフなどのグルメも豊富で、モダンでおしゃれなカフェやレストランも多くあります。
現在の天気は晴れで、散策にはぴったりの日でしょう。自然と都会が調和した魅力的な都市です。

ツール使用の挙動

tool_use_behaviorを使うと、ツール出力を制御できる。

  • run_llm_again: デフォルト。ツールが実行され、結果をLLMが処理して、最終回答を生成する。
  • stop_on_first_tool: 最初のツール呼び出しの出力をそのまま最終回答とする。以降のLLMの処理は行わない。
  • StopAtTools(stop_at_tool_names=[...]): 指定したいずれかのツールが呼び出されたら停止、その出力を最終回答として使用する。
  • ToolsToFinalOutputFunction: ツールの結果を処理して、停止するか、LLMで処理続行するかをカスタム関数で制御する。

stop_on_first_toolの例

from agents import Agent, function_tool, Runner 
import asyncio

@function_tool
def get_weather(city: str) -> str:
    """指定された年の天気情報を返す"""
    return f"{city} の天気は「晴れ」です。"

agent = Agent(
    name="japanese agent",
    instructions="常に日本語で回答する。",
    tools=[get_weather],
    tool_use_behavior="stop_on_first_tool"
)

async def main():
    result = await Runner.run(agent, "神戸の天気を知りたい。")
    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())
出力
神戸 の天気は「晴れ」です。

ツールの実行結果がそのまま出力されている。

StopAtTools(stop_at_tool_names=[...])の例

from agents import Agent, function_tool, Runner 
from agents.agent import StopAtTools
import asyncio

@function_tool
def get_weather(city: str) -> str:
    """指定された年の天気情報を返す"""
    return f"{city} の天気は「晴れ」です。"

@function_tool
def sum_numbers(a: int, b: int) -> int:
    """2つの数字を足し算する"""
    return a + b

agent = Agent(
    name="japanese agent",
    instructions="常に日本語で回答する。",
    tools=[get_weather, sum_numbers],
    tool_use_behavior=StopAtTools(stop_at_tool_names=["get_weather"])
)

async def main():
    result = await Runner.run(agent, "神戸の天気を知りたい。")
    print(result.final_output)

    print("-" * 20)

    result = await Runner.run(agent, "53 + 39 は？")
    print(result.final_output)

    print("-" * 20)

    result = await Runner.run(agent, "神戸はどんなところ？")
    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())
出力
神戸 の天気は「晴れ」です。
--------------------
53 + 39 は 92 です。
--------------------
神戸は、日本の兵庫県にある都市で、美しい港町として知られています。以下は、神戸の主な特徴や魅力です。

1. **港と夜景**: 
   - 神戸港は美しい景観を持ち、夜には港の夜景がとてもロマンチックです。
   - 神戸ハーバーランドやメリケンパークからの眺めが人気です。

2. **異人館**: 
   - 北野異人館街には、19世紀末から20世紀初頭にかけて建てられた外国人居留地の建物が残っています。
   - 異文化と歴史を感じることができるスポットです。

3. **グルメ**: 
   - 有名な「神戸ビーフ」は、特に贅沢な食材として知られています。
   - 中華街「南京町」も人気のグルメスポットです。

4. **自然と公園**: 
   - 六甲山では、ハイキングやドライブを楽しむことができ、頂上からの眺望が美しいです。
   - 須磨海浜公園は、海水浴やバーベキューに人気の場所です。

5. **ファッションとショッピング**: 
   - 神戸はファッションの街とも呼ばれており、ショッピングエリアも充実しています。

6. **文化とイベント**: 
   - 季節ごとに様々なイベントが開催され、特に神戸ルミナリエは有名なイルミネーションイベントです。

神戸は、歴史、文化、自然、食、ショッピングなど、様々な魅力が詰まった都市です。

天気ツールが使用された場合のみ、ツール実行結果がそのまま出力される。

ToolsToFinalOutputFunctionの例

from agents import Agent, function_tool, Runner, FunctionToolResult, RunContextWrapper
from agents.agent import ToolsToFinalOutputResult
from typing import List, Any
import asyncio

@function_tool
def get_weather(city: str) -> str:
    """指定された年の天気情報を返す"""
    return f"{city} の天気は「晴れ」です。"

def custom_tool_handler(
    context: RunContextWrapper[Any],
    tool_results: List[FunctionToolResult]
) -> ToolsToFinalOutputResult:
    """ツール結果を処理して最終出力を判断する"""
    for result in tool_results:
        if result.output and "晴れ" in result.output:
            return ToolsToFinalOutputResult(
                is_final_output=True,
                final_output=f"最終的な天気情報: {result.output}"
            )
    return ToolsToFinalOutputResult(
        is_final_output=False,
        final_output=None
    )
    
agent = Agent(
    name="japanese agent",
    instructions="常に日本語で回答する。",
    tools=[get_weather],
    tool_use_behavior=custom_tool_handler
)

async def main():
    result = await Runner.run(agent, "神戸の天気を知りたい。")
    print(result.final_output)

    print("-" * 20)

    result = await Runner.run(agent, "神戸はどんなところ？")
    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())
出力
最終的な天気情報: 神戸 の天気は「晴れ」です。
--------------------
神戸は日本の兵庫県にある港町で、魅力がたくさんあります。

1. **観光地**: 神戸には有名な観光地がたくさんあります。例えば、北野異人館や旧ハーバーランド、六甲山からの美しい夜景などが人気です。

2. **グルメ**: 神戸牛は世界的にも有名で、美味しいお肉を楽しめます。また、中華街（南京町）では多様な中華料理を堪能できます。

3. **港町の雰囲気**: 港の景色が美しい神戸は、海からの風が心地よい場所です。メリケンパークでは散歩やリラックスが楽しめます。

4. **文化**: 多様な文化が混ざり合う場所で、ファッションや音楽、アートなども盛んです。特に神戸コレクションというファッションイベントは有名です。

5. **交通の便**: 新幹線や阪神電鉄、地下鉄などの公共交通機関が充実していて、大阪や京都へのアクセスも便利です。

神戸は自然・都市・文化が融合した魅力的な都市です。

んー、そのとおりではあるんだけども、いまいちユースケースがわからない。

なお、補足として以下の記載がある。

無限ループを防ぐため、フレームワークはツール呼び出し後に tool_choice を自動的に "auto" にリセットします。この挙動は agent.reset_tool_choice で設定できます。無限ループは、ツール結果が LLM に送られ、その後 tool_choice により LLM が再度ツール呼び出しを生成し続けるために発生します。

kun432kun432

エージェントの実行

https://openai.github.io/openai-agents-python/running_agents/

エージェントの実行はRunnerクラスで行う。以下の3通り。

  • Runner.run(): 非同期で実行し、RunResultを返す。
  • Runner.run_sync(): 同期で実行、内部的には.run()を実行する
  • Runner.run_streamed(): 非同期で実行し、RunResultStreamingを返す。LLMからはストリーミングでイベントを受信する。

Runner.run()の例

from agents import Agent, Runner
import asyncio

async def main():
    agent = Agent(
        name="Assistant",
        instructions="あなたは親切なアシスタントです。"
    )

    result = await Runner.run(
        agent,
        "プログラミングにおける再帰について俳句を書いて。"
    )
    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())
出力
自己呼ぶ  
再帰の道は  
無限なり

Runner.run_sync()の例

from agents import Agent, Runner

def main():
    agent = Agent(
        name="Assistant",
        instructions="あなたは親切なアシスタントです。"
    )

    result = Runner.run_sync(
        agent,
        "プログラミングにおける再帰について俳句を書いて。"
    )
    print(result.final_output)

if __name__ == "__main__":
    main()
出力
繰り返し  
自ら呼びて  
道続く

ストリーミングは別のドキュメントにあるので別途。

https://openai.github.io/openai-agents-python/ja/streaming/

また、実行結果についても別のドキュメントにあるので、こちらも別途。

https://openai.github.io/openai-agents-python/ja/results/

エージェントループ

Runner.run() には、エージェントと入力を渡す。入力は、文字列（ユーザーのメッセージとなる）か、入力アイテムのリスト（OpenAI Responses API のアイテム）となる。

Runnerが実行するループは以下の流れとなる。

  1. 現在のエージェントに対して、現在の入力で LLM を呼び出す。
  2. LLM が出力を生成する。
    1. LLM が final_output を返した場合、ループを終了し結果を返す。
    2. LLM が ハンドオフ を行った場合、現在のエージェントと入力を更新し、ループを再実行する。
    3. LLM が ツール呼び出し を生成した場合、それらを実行し、その結果を追加して、ループを再実行する。
  3. 渡された max_turns を超えた場合、MaxTurnsExceeded 例外を投げる。

なお、final_output の条件は以下となる。

  • 目的の型のテキスト出力を生成した
  • ツール呼び出しがない

ストリーミング

ストリーミングも可能。ストリーミングの場合、各イベントは.stream_events()を使って呼び出し、ストリームが完了すると RunResultStreaming に完全な出力が含まれて返される。

ストリーミングについては別のドキュメントにあるので別途。

https://openai.github.io/openai-agents-python/ja/streaming/

実行設定

run_config を使うと、個別のエージェントの設定とは別に、エージェント実行に関する「グローバル」なパラメータを設定できる。パラメータは以下。

項目名 説明
model 使用するLLMモデルを指定。
model_provider モデル名を検索するプロバイダー（デフォルトはOpenAI）。
model_settings エージェント固有の設定を上書き（例: temperature や top_p など）。
input_guardrails すべての実行で使う入力ガードレールのリスト。
output_guardrails すべての実行で使う出力ガードレールのリスト。
handoff_input_filter ハンドオフ時に使うグローバルな入力フィルター。
tracing_disabled 実行全体のトレーシングを無効化。
trace_include_sensitive_data トレースに機密データ（LLMやツールの入出力など）を含めるかどうか。
workflow_name 実行のワークフロー名を設定する。
trace_id トレースIDを設定する。
group_id 複数の実行をリンクするためのグループID（任意）。
trace_metadata すべてのトレースに含めるメタデータ。

会話 / チャットスレッド

run()を一度呼び出すことが、チャット会話の1つの論理的なターンとなる（実際にはマルチエージェントでハンドオフなどが行われる場合もあるが。）、

  1. ユーザーのターン: ユーザーがテキストを入力
  2. Runner の実行: 最初のエージェントが LLM を呼び出し、ツールを実行、2つ目のエージェントへハンドオフ、2つ目のエージェントがさらにツールを実し、その後に出力を生成。

エージェント実行が完了すると、ユーザーに何を表示するかを選択できる。たとえば、

  • エージェントが生成したすべての新しいアイテムを表示する
  • 最終出力のみを表示する

いずれにせよ、ユーザーが追加質問をする場合は再度 run メソッドを呼び出すことになる。

手動の会話管理

実行結果に対して .to_input_list() メソッドを実行すると、会話履歴を取得できる。これを使えば手動で会話履歴を管理して、次のターンにつなげることができる。

from agents import Agent, Runner
import asyncio

async def main():
    agent = Agent(
        name="Assistant",
        instructions="あなたは親切なアシスタントです。"
    )

    # 最初の実行
    result = await Runner.run(agent, "ゴールデンゲートブリッジはどの都市にありますか？")
    print(result.final_output)

    print("-" * 20)

    # 2回目の実行
    # result.to_input_list() で、これまでの会話履歴を取得する
    new_input = result.to_input_list() + [{"role": "user", "content": "それはどの州ですか？"}]
    result = await Runner.run(agent, new_input)
    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())
出力
ゴールデンゲートブリッジはアメリカ合衆国のサンフランシスコにあります。
--------------------
サンフランシスコはカリフォルニア州にあります。

Sessionsによる自動の会話管理

Sessions を使えば、上記を自動化できる。以下はSQLiteを使ったSessionsの例。

from agents import Agent, Runner, SQLiteSession
import asyncio

async def main():
    agent = Agent(
        name="Assistant",
        instructions="あなたは親切なアシスタントです。"
    )

    # セッションを初期化
    session = SQLiteSession("session_123")

    # 最初の実行
    result = await Runner.run(
        agent,
        "ゴールデンゲートブリッジはどの都市にありますか？",
        session=session
    )
    print(result.final_output)

    print("-" * 20)

    # 2回目の実行
    result = await Runner.run(
        agent,
        "それはどの州ですか？",
        session=session
    )
    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())
出力
ゴールデンゲートブリッジは、アメリカのカリフォルニア州サンフランシスコにあります。
--------------------
ゴールデンゲートブリッジはカリフォルニア州にあります。

Sessionは以下を自動でやってくれる。

  • 各実行前に会話履歴を取得
  • 各実行後に新しいメッセージを保存
  • セッション ID ごとに別々の会話を維持

こちらも別のドキュメントにあるので、詳細は別途。

https://openai.github.io/openai-agents-python/ja/sessions/

長時間実行エージェントとヒューマン・イン・ザ・ループ

長時間のワークフローや、Human-in-the-loop を行い場合、Temporalと連携して行うらしい。Temporal は自分の認識ではタスクキューサービスという認識で、これと連携する場合のデモとドキュメントが用意されている。

https://www.youtube.com/watch?v=fFBZqzT4DD8

https://github.com/temporalio/sdk-python/tree/main/temporalio/contrib/openai_agents

例外

例外の一覧は以下にある

https://openai.github.io/openai-agents-python/ref/exceptions/#agents.exceptions

概要としてはこんな感じ。

例外名 説明
AgentsException SDK内で発生する全ての例外の基底クラス。
MaxTurnsExceeded エージェントの実行がmax_turnsを超えた場合に発生。
ModelBehaviorError モデル（LLM）が予期しない・無効な出力を返した場合に発生。
UserError SDKの使い方や設定ミスなど、ユーザー側のエラーで発生。
InputGuardrailTripwireTriggered 入力ガードレールの条件に引っかかった場合に発生。
OutputGuardrailTripwireTriggered 出力ガードレールの条件に引っかかった場合に発生。
kun432kun432

実行結果

https://openai.github.io/openai-agents-python/ja/results/

Runner.run メソッドの結果は、以下のどれかになる。

どちらも RunResultBase を継承しており、必要な情報はだいたいここで確認できる。

結果の主なプロパティやメソッドはざっくりこんな感じ。

名前 種類 内容・役割
final_output プロパティ 最後に実行されたエージェントの最終出力。型はAny。
result.to_input_list() メソッド 元の入力＋実行中に生成されたアイテムを連結したリストを返す。
last_agent プロパティ 最後に実行されたエージェント。次回の入力時に役立つ。
new_items プロパティ 実行中に生成された新しいアイテム（RunItem）。
input_guardrail_results プロパティ 入力ガードレールの実行結果。存在する場合のみ。
output_guardrail_results プロパティ 出力ガードレールの実行結果。存在する場合のみ。
raw_responses プロパティ LLMによって生成されたModelResponse。
input プロパティ runメソッドに渡した元の入力。多くの場合不要だけど参照可能。

ざっと確認してみる。

from agents import Agent, Runner
import asyncio

agent = Agent(
    name="math tutor agent",
    instructions=(
        "あなたは数学の問題解決を支援します。"
        "各ステップでの考え方を、例を挙げて説明してください。"
    )
)

async def main():
    result = await Runner.run(agent, "5x - 7 = 18 は？")
    print("-" * 10, "final_output", "-" * 10)
    print(result.final_output)
    print("-" * 10, "to_input_list", "-" * 10)
    print(result.to_input_list())
    print("-" * 10, "last_agent", "-" * 10)
    print(result.last_agent)
    print("-" * 10, "new_items", "-" * 10)
    print(result.new_items)
    print("-" * 10, "raw_responses", "-" * 10)
    print(result.raw_responses)
    print("-" * 10, "input", "-" * 10)
    print(result.input)

if __name__ == "__main__":
    asyncio.run(main())

結果はこんな感じ（見やすさのために適宜改行を入れている）

出力
---------- final_output ----------
この方程式を解く過程をステップごとに説明します。

### ステップ 1: 方程式を書き留める
方程式は次のとおりです。
\[ 5x - 7 = 18 \]

### ステップ 2: 定数を移動する
方程式を解くためには、まず \(x\) の項を片側に、定数を反対側に集めます。ここでは、定数 \(-7\) を移動することから始めます。両辺に \(7\) を足します。
\[
5x - 7 + 7 = 18 + 7
\]

これで、新しい方程式は次のようになります。
\[
5x = 25
\]

### ステップ 3: \(x\) の係数で割る
次に、\(x\) の係数である \(5\) で両辺を割ります。これによって \(x\) の値を求めることができます。
\[
\frac{5x}{5} = \frac{25}{5}
\]

これにより、\(x\) の値は次のようになります。
\[
x = 5
\]

### ステップ 4: 答えの確認
見つけた \(x\) の値を元の方程式に代入して確認します。
\[ 
5(5) - 7 = 18 
\]
\[ 
25 - 7 = 18 
\]
\[ 
18 = 18 
\]

この計算は正しいので、\(x = 5\) が解であることが確認できます。 

したがって、方程式 \(5x - 7 = 18\) の解は \(x = 5\) です。
---------- to_input_list ----------
[
    {
        'content': '5x - 7 = 18 は？',
        'role': 'user'
    },
    {
        'id': 'msg_68a0ad985648819ab5885f7880c48ad90d582a6984a79168',
        'content':[
            {
                'annotations': [],
                'text': 'この方程式を解く過程をステップごとに説明します。\n\n### ステップ 1: 方程式を書き留める\n方程式は次のとおりです。\n\\[ 5x - 7 = 18 \\]\n\n### ステップ 2: 定数を移動する\n方程式を解くためには、まず \\(x\\) の項を片側に、定数を反対側に集めます。ここでは、定数 \\(-7\\) を移動することから始めます。両辺に \\(7\\) を足します。\n\\[\n5x - 7 + 7 = 18 + 7\n\\]\n\nこれで、新しい方程式は次のようになります。\n\\[\n5x = 25\n\\]\n\n### ステップ 3: \\(x\\) の係数で割る\n次に、\\(x\\) の係数である \\(5\\) で両辺を割ります。これによって \\(x\\) の値を求めることができます。\n\\[\n\\frac{5x}{5} = \\frac{25}{5}\n\\]\n\nこれにより、\\(x\\) の値は次のようになります。\n\\[\nx = 5\n\\]\n\n### ステップ 4: 答えの確認\n見つけた \\(x\\) の値を元の方程式に代入して確認します。\n\\[ \n5(5) - 7 = 18 \n\\]\n\\[ \n25 - 7 = 18 \n\\]\n\\[ \n18 = 18 \n\\]\n\nこの計算は正しいので、\\(x = 5\\) が解であることが確認できます。 \n\nしたがって、方程式 \\(5x - 7 = 18\\) の解は \\(x = 5\\) です。',
                'type': 'output_text',
                'logprobs': []
            }
        ],
        'role': 'assistant',
        'status': 'completed',
        'type': 'message'
    }
]
---------- last_agent ----------
Agent(
    name='math tutor agent',
    handoff_description=None,
    tools=[],
    mcp_servers=[],
    mcp_config={},
    instructions='あなたは数学の問題解決を支援します。各ステップでの考え方を、例を挙げて説明してください。',
    prompt=None,
    handoffs=[],
    model=None,
    model_settings=ModelSettings(
        temperature=None,
        top_p=None,
        frequency_penalty=None,
        presence_penalty=None,
        tool_choice=None,
        parallel_tool_calls=None,
        truncation=None,
        max_tokens=None,
        reasoning=None,
        verbosity=None,
        metadata=None,
        store=None,
        include_usage=None,
        response_include=None,
        top_logprobs=None,
        extra_query=None,
        extra_body=None,
        extra_headers=None,
        extra_args=None
    ),
    input_guardrails=[],
    output_guardrails=[],
    output_type=None,
    hooks=None,
    tool_use_behavior='run_llm_again',
    reset_tool_choice=True
)
---------- new_items ----------
[
    MessageOutputItem(
        agent=Agent(
            name='math tutor agent',
            handoff_description=None,
            tools=[],
            mcp_servers=[],
            mcp_config={},
            instructions='あなたは数学の問題解決を支援します。各ステップでの考え方を、例を挙げて説明してください。',
            prompt=None,
            handoffs=[],
            model=None,
            model_settings=ModelSettings(
                temperature=None,
                top_p=None,
                frequency_penalty=None,
                presence_penalty=None,
                tool_choice=None,
                parallel_tool_calls=None,
                truncation=None,
                max_tokens=None,
                reasoning=None,
                verbosity=None,
                metadata=None,
                store=None,
                include_usage=None,
                response_include=None,
                top_logprobs=None,
                extra_query=None,
                extra_body=None,
                extra_headers=None,
                extra_args=None
            ),
            input_guardrails=[],
            output_guardrails=[],
            output_type=None,
            hooks=None,
            tool_use_behavior='run_llm_again',
            reset_tool_choice=True
        ),
        raw_item=ResponseOutputMessage(
            id='msg_68a0ad985648819ab5885f7880c48ad90d582a6984a79168',
            content=[
                ResponseOutputText(
                    annotations=[],
                    text='この方程式を解く過程をステップごとに説明します。\n\n### ステップ 1: 方程式を書き留める\n方程式は次のとおりです。\n\\[ 5x - 7 = 18 \\]\n\n### ステップ 2: 定数を移動する\n方程式を解くためには、まず \\(x\\) の項を片側に、定数を反対側に集めます。ここでは、定数 \\(-7\\) を移動することから始めます。両辺に \\(7\\) を足します。\n\\[\n5x - 7 + 7 = 18 + 7\n\\]\n\nこれで、新しい方程式は次のようになります。\n\\[\n5x = 25\n\\]\n\n### ステップ 3: \\(x\\) の係数で割る\n次に、\\(x\\) の係数である \\(5\\) で両辺を割ります。これによって \\(x\\) の値を求めることができます。\n\\[\n\\frac{5x}{5} = \\frac{25}{5}\n\\]\n\nこれにより、\\(x\\) の値は次のようになります。\n\\[\nx = 5\n\\]\n\n### ステップ 4: 答えの確認\n見つけた \\(x\\) の値を元の方程式に代入して確認します。\n\\[ \n5(5) - 7 = 18 \n\\]\n\\[ \n25 - 7 = 18 \n\\]\n\\[ \n18 = 18 \n\\]\n\nこの計算は正しいので、\\(x = 5\\) が解であることが確認できます。 \n\nしたがって、方程式 \\(5x - 7 = 18\\) の解は \\(x = 5\\) です。',
                    type='output_text',
                    logprobs=[]
                )
            ],
            role='assistant',
            status='completed',
            type='message'
        ),
        type='message_output_item'
    )
]
---------- raw_responses ----------
[
    ModelResponse(
        output=[
            ResponseOutputMessage(
                id='msg_68a0ad985648819ab5885f7880c48ad90d582a6984a79168',
                content=[
                    ResponseOutputText(
                        annotations=[],
                        text='この方程式を解く過程をステップごとに説明します。\n\n### ステップ 1: 方程式を書き留める\n方程式は次のとおりです。\n\\[ 5x - 7 = 18 \\]\n\n### ステップ 2: 定数を移動する\n方程式を解くためには、まず \\(x\\) の項を片側に、定数を反対側に集めます。ここでは、定数 \\(-7\\) を移動することから始めます。両辺に \\(7\\) を足します。\n\\[\n5x - 7 + 7 = 18 + 7\n\\]\n\nこれで、新しい方程式は次のようになります。\n\\[\n5x = 25\n\\]\n\n### ステップ 3: \\(x\\) の係数で割る\n次に、\\(x\\) の係数である \\(5\\) で両辺を割ります。これによって \\(x\\) の値を求めることができます。\n\\[\n\\frac{5x}{5} = \\frac{25}{5}\n\\]\n\nこれにより、\\(x\\) の値は次のようになります。\n\\[\nx = 5\n\\]\n\n### ステップ 4: 答えの確認\n見つけた \\(x\\) の値を元の方程式に代入して確認します。\n\\[ \n5(5) - 7 = 18 \n\\]\n\\[ \n25 - 7 = 18 \n\\]\n\\[ \n18 = 18 \n\\]\n\nこの計算は正しいので、\\(x = 5\\) が解であることが確認できます。 \n\nしたがって、方程式 \\(5x - 7 = 18\\) の解は \\(x = 5\\) です。',
                        type='output_text',
                        logprobs=[]
                    )
                ],
                role='assistant',
                status='completed',
                type='message'
            )
        ],
        usage=Usage(
            requests=1,
            input_tokens=52,
            input_tokens_details=InputTokensDetails(cached_tokens=0),
            output_tokens=411,
            output_tokens_details=OutputTokensDetails(reasoning_tokens=0),
            total_tokens=463
        ),
        response_id='resp_68a0ad97e134819ab8b19a39b009aef80d582a6984a79168'
    )
]
---------- input ----------
5x - 7 = 18 は？
kun432kun432

ストリーミング

https://openai.github.io/openai-agents-python/ja/streaming/

ストリーミングの概要

rawレスポンスイベント

RawResponsesStreamEvent は、LLMから直接返されるrawなイベントで、OpenAI Responses APIの形式で返される。イベントには、response.created や response.output_text.delta などの種類がある。

from agents import Agent, Runner
from openai.types.responses import ResponseTextDeltaEvent
import asyncio

agent = Agent(
    name="Assistant",
    instructions="あなたは親切なアシスタントです。"
)

async def main():
    result = Runner.run_streamed(
        agent,
        "競馬の最大の魅力について簡潔に説明して。"
    )

    async for event in result.stream_events():
        if event.type == "raw_response_event" and isinstance(event.data, ResponseTextDeltaEvent):
            # わかりやすさのために改行を入れている
            # 実際に使用する際は end=""
            print(event.data.delta, end="\n", flush=True)

if __name__ == "__main__":
    asyncio.run(main())
出力
競
馬
の
最大
の
魅
力
は
、
ス
リ
ル
と
ドラ
マ
に
あります
。
馬
や
騎
手
の
努力
が
瞬
間
に
凝
縮
され
、
予
測
不能
な
レ
ース
展
開
が
観
客
を
興
奮
さ
せ
ます
。また
、
馬
の
美
し
さ
や
躍
動
感
、
戦
略
的
な
要
素
も
楽し
め
、一
瞬
一
瞬
に
か
か
る
緊
張
感
が
た
まり
ません
。
競
馬
は
スポ
ーツ
として
だけ
で
なく
、
文化
や
歴
史
とも
深
く
結
び
つ
いて
お
り
、多
く
の
人
々
を
魅
了
し
続
け
ています
。

Run itemイベントとエージェントイベント

ちなみにイベントタイプだけ見てみるとこんな感じになる。

from agents import Agent, Runner
from openai.types.responses import ResponseTextDeltaEvent
import asyncio

agent = Agent(
    name="Assistant",
    instructions="あなたは親切なアシスタントです。"
)

async def main():
    result = Runner.run_streamed(
        agent,
        "競馬の最大の魅力について1文で簡潔に説明して。"
    )

    async for event in result.stream_events():
        print(event.type)

if __name__ == "__main__":
    asyncio.run(main())
出力
agent_updated_stream_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
raw_response_event
run_item_stream_event

部分的なテキスト出力を取り出す場合は raw_response_event を見れば良いというのがわかるが、それ以外にもイベントが発生しているのがわかる。

raw_response_event以外のイベントを出力するサンプル

import asyncio
import random
from agents import Agent, ItemHelpers, Runner, function_tool

@function_tool
def how_many_jokes() -> int:
    return random.randint(1, 10)

async def main():
    agent = Agent(
        name="Joker",
        instructions="最初に、`how_many_jokes` ツールを呼び出し、その数だけジョークを言う。",
        tools=[how_many_jokes],
    )

    result = Runner.run_streamed(
        agent,
        input="こんにちは",
    )
    print("=== 実行開始 ===")

    async for event in result.stream_events():
        # raw response イベントは、無視
        if event.type == "raw_response_event":
            continue
        # エージェントが更新された
        elif event.type == "agent_updated_stream_event":
            print(f"エージェント更新: {event.new_agent.name}")
            continue
        # アイテムが生成された
        elif event.type == "run_item_stream_event":
            if event.item.type == "tool_call_item":
                print("-- ツール呼び出し")
            elif event.item.type == "tool_call_output_item":
                print(f"-- ツール出力: {event.item.output}")
            elif event.item.type == "message_output_item":
                print(f"-- メッセージ出力:\n {ItemHelpers.text_message_output(event.item)}")
            else:
                pass  # その他のイベントは無視

    print("=== 実行完了 ===")


if __name__ == "__main__":
    asyncio.run(main())
出力
=== 実行開始 ===
エージェント更新: Joker
-- ツール呼び出し
-- ツール出力: 1
-- メッセージ出力:
 こんにちは！さて、一つジョークをお届けします。

**ジョーク**: 

なぜコンピュータは海が嫌いなの？
  
**答え**: だって、海に「プログラム」が無いからさ！ 🌊💻

どうですか？他に何かお手伝いできることがありますか？
=== 実行完了 ===
このスクラップは20日前にクローズされました