OpenAI Agents SDK 0.3 : エージェントの実行

by

OpenAI Agents SDK のエージェントの実行方法についての説明です。Runner クラスを通してエージェントを実行できて、3 つのオプションがあります。

OpenAI Agents SDK 0.3 : エージェントの実行

作成 : クラスキャット・セールスインフォメーション
作成日時 : 10/16/2025
バージョン : v0.3.3

* 本記事は以下のページを参考にしています :

* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。

 

 

OpenAI Agents SDK 0.3 : エージェントの実行

Runner クラスを通してエージェントを実行できます。3 つのオプションがあります :

  1. Runner.run(), これは非同期に実行し、RunResult を返します。

  2. Runner.run_sync(), これは同期メソッドで内部的には単に .run() を実行するだけです。

  3. Runner.run_streamed(), これは非同期で実行し RunResultStreaming を返します。LLM をストリーミング・モードで呼び出し、それらのイベントを受け取りながらストリーミングします。 
from agents import Agent, Runner

async def main():
    agent = Agent(name="Assistant", instructions="You are a helpful assistant")

    result = await Runner.run(agent, "Write a haiku about recursion in programming.")
    print(result.final_output)
    # Code within the code,
    # Functions calling themselves,
    # Infinite loop's dance

Read more in the results guide.

 

エージェント・ループ

Runner の run メソッドを使用する場合、開始エージェントと入力を渡します。入力は文字列 (ユーザメッセージと考えられます) か (OpenAI Responses API の items である) 入力項目のリストのいずれかです。

そして runner はループを実行します :

  1. 現在の入力を使用して、現在のエージェント用の LLM を呼び出します。

  2. LLM は出力を生成します。

    1. LLM が final_output を返す場合、ループは終了して結果を返します。
    2. LLM がハンドオフを行う場合、現在のエージェントと入力を更新し、ループを再実行します。
    3. LLM がツール呼び出しを生成する場合、それらのツール呼び出しを実行し、結果を追加し、そしてループを再実行します。

  3. 渡された max_turns をを超える場合、MaxTurnsExceeded 例外を上げます。

Note : LLM 出力が「最終出力」として見なされるかどうかのルールは、それが希望の型を持つテキスト出力を生成し、ツール呼び出しがないことです。

 

ストリーミング

ストリーミングは LLM を実行するときストリーミング・イベントを追加で受け取ることができます。ストリーミングが完了すると、RunResultStreaming が (すべての生成された新しい出力も含む) 実行について完全な情報を含みます。ストリーミングイベントについては .stream_events() を呼び出すことができます。Read more in the streaming guide.

 

Run config

run_config パラメータはエージェント実行のためのグローバル設定を構成可能にします :

  • model : 各エージェントがどのモデルを持つかに関係なく、使用するグローバルな LLM モデルの設定を可能にします。

  • model_provider : モデル名を検索するためのモデルプロバイダー、デフォルトは OpenAI です。

  • model_settings : エージェント固有の設定をオーバーライドします。例えば、global temperature や top_p を設定できます。

  • input_guardrails, output_guardrails : すべての実行に含める入力または出力ガードレールのリスト。

  • handoff_input_filter : ハンドオフがまだ入力フィルターを持たない場合、グローバル入力フィルターはすべてのハンドオフに適用されます。入力フィルターは新しいエージェントに送られる入力の編集を可能にします。See the documentation in Handoff.input_filter for more details.

  • tracing_disabled : 実行全体に対する トレース を無効にできます。

  • trace_include_sensitive_data : トレースに、LLM やツール呼び出しの入出力のような、潜在的に機密なデータを含めるか否かを設定します。

  • workflow_name, trace_id, group_id : 実行のためのトレース・ワークフロー名、トレース ID とグループ ID を設定します。少なくとも workflow_name の設定を勧めます。グループ ID はオプションのフィールドで、複数の実行に渡るトレースのリンクを可能にします。

  • trace_metadata : すべてのトレース上に含めるメタデータ。

 

会話/チャット・スレッド

実行メソッドのいずれかを呼び出すと一つ以上のエージェントの実行 (従って一つ以上の LLM 呼び出し) という結果になりますが、これはチャット会話における単一の論理的ターンを表します。例えば :

  1. ユーザ・ターン : ユーザのテキスト入力

  2. Runner の実行 : 最初のエージェントが LLM を呼び出し、ツールを実行し、2 番目のエージェントにハンドオフし、2 番目のエージェントが更にツールを実行し、それから出力を生成します。

エージェント実行の最後に、ユーザに表示するものを選択できます。例えば、エージェントにより生成されたすべての新しい項目をユーザに表示したり、最終結果だけを表示することもできるでしょう。いずれの場合でも、ユーザはフォローアップの質問をするかもしれません、その場合 run メソッドを再び呼び出すことができます。

 

会話の手動管理

次のターンの入力を取得するために、RunResultBase.to_input_list() メソッドを使用して会話履歴を手動で管理できます :

async def main():
    agent = Agent(name="Assistant", instructions="Reply very concisely.")

    thread_id = "thread_123"  # Example thread ID
    with trace(workflow_name="Conversation", group_id=thread_id):
        # First turn
        result = await Runner.run(agent, "What city is the Golden Gate Bridge in?")
        print(result.final_output)
        # San Francisco

        # Second turn
        new_input = result.to_input_list() + [{"role": "user", "content": "What state is it in?"}]
        result = await Runner.run(agent, new_input)
        print(result.final_output)
        # California

 

セッションによる自動会話管理

より単純なアプローチとして、Session を使用して .to_input_list() を手動で呼び出すことなく、会話履歴を自動的に処理できます :

from agents import Agent, Runner, SQLiteSession

async def main():
    agent = Agent(name="Assistant", instructions="Reply very concisely.")

    # Create session instance
    session = SQLiteSession("conversation_123")

    thread_id = "thread_123"  # Example thread ID
    with trace(workflow_name="Conversation", group_id=thread_id):
        # First turn
        result = await Runner.run(agent, "What city is the Golden Gate Bridge in?", session=session)
        print(result.final_output)
        # San Francisco

        # Second turn - agent automatically remembers previous context
        result = await Runner.run(agent, "What state is it in?", session=session)
        print(result.final_output)
        # California

セッションは自動的に :

  • 各実行前に会話履歴を取得します

  • 各実行後に新しいメッセージを保存します

  • 異なるセッション ID のために会話を分離して維持します

See the Sessions documentation for more details.

 

サーバ管理の会話

to_input_list() やセッションを使用して会話状態をローカルで処理する代わりに、OpenAI 会話状態機能にサーバ側で会話状態を管理させることもできます。これは、過去のメッセージすべてを手動で再送信することなく、会話履歴を保持することを可能にします。See the OpenAI Conversation state guide for more details.

OpenAI はターンに渡り状態を追跡する 2 つの方法を提供しています :

 
1. conversation_id の使用

まず OpenAI Conversations API を使用して会話を作成し、そしてすべての続く呼び出しでその ID を再利用します :

from agents import Agent, Runner
from openai import AsyncOpenAI

client = AsyncOpenAI()

async def main():
    # Create a server-managed conversation
    conversation = await client.conversations.create()
    conv_id = conversation.id    

    agent = Agent(name="Assistant", instructions="Reply very concisely.")

    # First turn
    result1 = await Runner.run(agent, "What city is the Golden Gate Bridge in?", conversation_id=conv_id)
    print(result1.final_output)
    # San Francisco

    # Second turn reuses the same conversation_id
    result2 = await Runner.run(
        agent,
        "What state is it in?",
        conversation_id=conv_id,
    )
    print(result2.final_output)
    # California

 
2. previous_response_id の使用

もう一つのオプションは レスポンス連鎖 (chaining) です、そこでは各ターンは前のターンからのレスポンス ID に明示的にリンクします。

from agents import Agent, Runner

async def main():
    agent = Agent(name="Assistant", instructions="Reply very concisely.")

    # First turn
    result1 = await Runner.run(agent, "What city is the Golden Gate Bridge in?")
    print(result1.final_output)
    # San Francisco

    # Second turn, chained to the previous response
    result2 = await Runner.run(
        agent,
        "What state is it in?",
        previous_response_id=result1.last_response_id,
    )
    print(result2.final_output)
    # California

 

長時間実行エージェント & human-in-the-loop

Agents SDK Temporal 統合を使用して、human-in-the-loop タスクを含む、耐久性の高い、長時間実行のワークフローを実行できます。この動画 で、Temporal と Agents SDK が実際に動作して長時間実行タスクを完了するデモを見て、ここのドキュメント をご覧ください。

 

例外

SDK は特定のケースで例外を上げます。完全なリストは agents.exceptions にあります。概要は :

  • AgentsException は SDK で上がるすべての例外の基底クラスです。

  • MaxTurnsExceeded は、実行が run メソッドに渡された max_turns を超えた場合に発生します。

  • ModelBehaviorError は、モデルが無効な出力 (e.g. 不正な形式の JSON や存在しないツールの使用) を生成する場合に発生します。

  • UserError は、SDK を使用してコードを書く人が SDK を使用してエラーを起こした場合に発生します。

  • InputGuardrailTripwireTriggered, OutputGuardrailTripwireTriggered はガードレールが作動する (trip) 場合に発生します。

 

以上