エージェントの実行
エージェントは Runner
クラスで実行できます。方法は 3 つあります。
Runner.run()
: 非同期で実行し、RunResult
を返します。Runner.run_sync()
: 同期メソッドで、内部的には.run()
を実行します。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
詳細は results ガイドをご覧ください。
エージェントループ
Runner
の run メソッドを使うとき、開始するエージェントと入力を渡します。入力は文字列(ユーザーからのメッセージとみなされます)または入力アイテムのリスト(OpenAI Responses API のアイテム)です。
Runner は次のループを実行します。
- 現在のエージェントと現在の入力で LLM を呼び出します。
- LLM が出力を生成します。
- LLM が
final_output
を返した場合、ループを終了して結果を返します。 - LLM がハンドオフを行った場合、現在のエージェントと入力を更新し、ループを再実行します。
- LLM がツールコールを生成した場合、それらを実行して結果を追記し、ループを再実行します。
- LLM が
- 渡された
max_turns
を超えた場合、MaxTurnsExceeded
例外を送出します。
Note
LLM の出力が「最終出力」と見なされる条件は、希望する型のテキスト出力を生成し、ツールコールがないことです。
ストリーミング
ストリーミングを使うと、LLM の実行中にストリーミングイベントも受け取れます。ストリームが完了すると、RunResultStreaming
に、生成されたすべての新しい出力を含む実行の完全な情報が含まれます。ストリーミングイベントは .stream_events()
を呼び出して取得できます。詳細は ストリーミングガイドをご覧ください。
実行設定
run_config
パラメーターでは、エージェント実行のグローバル設定を構成できます。
model
: 各 Agent のmodel
に関係なく、使用するグローバルな LLM モデルを設定できます。model_provider
: モデル名を解決するためのモデルプロバイダーで、デフォルトは OpenAI です。model_settings
: エージェント固有の設定を上書きします。たとえば、グローバルなtemperature
やtop_p
を設定できます。input_guardrails
,output_guardrails
: すべての実行に含める入力/出力ガードレールのリストです。handoff_input_filter
: すべてのハンドオフに適用するグローバル入力フィルター(ハンドオフですでに指定されていない場合)。入力フィルターでは、新しいエージェントに送信される入力を編集できます。詳細はHandoff.input_filter
のドキュメントを参照してください。tracing_disabled
: 実行全体のトレーシングを無効化できます。trace_include_sensitive_data
: LLM やツールコールの入出力など、機微なデータをトレースに含めるかどうかを構成します。workflow_name
,trace_id
,group_id
: 実行のトレーシング用ワークフロー名、トレース ID、トレースグループ ID を設定します。少なくともworkflow_name
の設定を推奨します。グループ ID は任意で、複数の実行にまたがるトレースを関連付けできます。trace_metadata
: すべてのトレースに含めるメタデータです。
会話/チャットスレッド
いずれの run メソッドを呼び出しても、1 つ以上のエージェントが実行される(つまり 1 回以上の LLM 呼び出しが行われる)可能性がありますが、チャット会話における 1 つの論理的なターンを表します。例:
- ユーザーのターン: ユーザーがテキストを入力
- 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
Sessions による自動会話管理
より簡単な方法として、Sessions を使うと、.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
Sessions は自動で次を行います。
- 各実行前に会話履歴を取得
- 各実行後に新しいメッセージを保存
- 異なるセッション ID ごとに別個の会話を維持
詳細は Sessions のドキュメントをご覧ください。
サーバー管理の会話
to_input_list()
や Sessions
でローカルに管理する代わりに、OpenAI の conversation state 機能により、サーバー側で会話状態を管理させることもできます。これにより、過去のすべてのメッセージを手動で再送信せずに会話履歴を保存できます。詳細は OpenAI Conversation state ガイドをご覧ください。
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
の使用
もう 1 つの方法は、各ターンが前のターンのレスポンス 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
長時間実行エージェントとヒューマン・イン・ザ・ループ
Agents SDK の Temporal 連携を使うと、ヒューマン・イン・ザ・ループのタスクを含む、耐久性のある長時間実行ワークフローを実行できます。Temporal と Agents SDK が連携して長時間タスクを完了するデモはこの動画で、ドキュメントはこちらをご覧ください。
例外
SDK は特定の状況で例外を送出します。完全な一覧は agents.exceptions
にあります。概要は次のとおりです。
AgentsException
: SDK 内で送出されるすべての例外の基底クラスです。他の特定の例外はすべてこれを継承します。MaxTurnsExceeded
: エージェントの実行がRunner.run
、Runner.run_sync
、Runner.run_streamed
メソッドに渡したmax_turns
制限を超えた場合に送出されます。指定されたインタラクションターン数内にタスクを完了できなかったことを示します。ModelBehaviorError
: 基盤となるモデル(LLM)が予期しない、または無効な出力を生成した場合に発生します。例:- 不正な JSON: 特定の
output_type
が定義されている場合に特に、ツールコールや直接の出力として不正な JSON 構造を返す場合。 - 予期しないツール関連の失敗: モデルが期待どおりの方法でツールを使用できない場合
- 不正な JSON: 特定の
UserError
: SDK を使用する際に(SDK を用いたコードを書く)あなたがエラーを起こした場合に送出されます。これは通常、誤ったコード実装、無効な構成、SDK の API の誤用が原因です。InputGuardrailTripwireTriggered
,OutputGuardrailTripwireTriggered
: それぞれ入力ガードレールまたは出力ガードレールの条件が満たされた場合に送出されます。入力ガードレールは処理前に受信メッセージを確認し、出力ガードレールはエージェントの最終応答を配信前に確認します。