OpenAI Agents SDKのサンプルコードを読む（Handoffs and routing）

読んだサンプルコード

• routing.py
  • OpenAI Agents SDKのCommon agentic patternsのHandoffs and routingのサンプル

Handoffsとroutingとは

特定のタスクを担当する専門のサブエージェントが存在する場合に、Handoffsを使って適切なエージェントにタスクをルーティングすることができる。

In many situations, you have specialized sub-agents that handle specific tasks. You can use handoffs to route the task to the right agent.
For example, you might have a frontline agent that receives a request, and then hands off to a specialized agent based on the language of the request. See the routing.py file for an example of this.

（Handoffs and routingより）

routing.py

routing.py

import asyncio
import uuid

from openai.types.responses import ResponseContentPartDoneEvent, ResponseTextDeltaEvent

from agents import Agent, RawResponsesStreamEvent, Runner, TResponseInputItem, trace

"""
This example shows the handoffs/routing pattern. The triage agent receives the first message, and
then hands off to the appropriate agent based on the language of the request. Responses are
streamed to the user.
"""

• importとdocstringの記載

routing.py

french_agent = Agent(
    name="french_agent",
    instructions="You only speak French",
)

spanish_agent = Agent(
    name="spanish_agent",
    instructions="You only speak Spanish",
)

english_agent = Agent(
    name="english_agent",
    instructions="You only speak English",
)

• フランス語、スペイン語、英語のみを話すエージェントのインスタンスをそれぞれ生成。
• nameはエージェント名であり、instructionsはシステムプロンプトにあたる。

routing.py

triage_agent = Agent(
    name="triage_agent",
    instructions="Handoff to the appropriate agent based on the language of the request.",
    handoffs=[french_agent, spanish_agent, english_agent],
)

• このサンプルコードの肝の部分であり、トリアージュを行うエージェントを生成している
• handoffsには引き継ぎ先のエージェントとして、上記で生成されたフランス語、スペイン語、英語のエージェントを登録している
• Handoffsを使用することで、エージェントがタスクを別のエージェントに委譲することができる

routing.py

async def main():
    # We'll create an ID for this conversation, so we can link each trace
    conversation_id = str(uuid.uuid4().hex[:16])

• Trace用のconversation_idを作成
• これを後述のtrace()指定し、複数回のエージェント実行（複数ターンの会話）を一つの会話単位で紐付けることができる
  • Trace DashboardではGroup IDとして記録される
  • conversation_idを固定することで、同じ会話に属するトレースをUI上で追いやすくなる

routing.py

    msg = input("Hi! We speak French, Spanish and English. How can I help? ")

• ユーザーからのキーボード入力を受け付けている

routing.py

    agent = triage_agent
    inputs: list[TResponseInputItem] = [{"content": msg, "role": "user"}]

• TResponseInputItemは、Agents SDKが使用する入力メッセージを表す型エイリアス
• OpenAIのResponses APIにおける入力アイテム型（ResponseInputItemParam）に対応しており、エージェントへの入力として受け付けられるデータ形式を示す
  • 具体的には、ユーザやシステムからエージェントに与えるメッセージ1件分の構造で、{"role": ロール, "content": メッセージ内容}という形を取る。
  • Agents SDKでは、Runner.runやRunner.run_streamedに渡すinput引数として、文字列 もしくはTResponseInputItemのリスト のいずれかを受け付ける
    • 文字列で渡した場合は自動的に「ユーザからの単一メッセージ」と解釈され、内部でユーザロールのアイテムに変換される
    • 一方、過去の対話履歴を含めて渡したい場合は、複数のTResponseInputItem（辞書形式のメッセージ）をリストで渡すことで会話のコンテキストを与えることができる
  • このサンプルコードでは初回入力時にinputs: list[TResponseInputItem] = [{"content": msg, "role": "user"}]という形でリストを準備し、これをRunner.run_streamedに渡している。このリストにはユーザからのメッセージ1件だけが含まれているが、後述のように対話が進むごとにこのリストにメッセージを追加していく運用をしている。

routing.py

    while True:
        # Each conversation turn is a single trace. Normally, each input from the user would be an
        # API request to your app, and you can wrap the request in a trace()
        with trace("Routing example", group_id=conversation_id):

• while True:のブロック内では以下の処理が行われており、会話のターンを繰り返せるようになっている
  1. ユーザーがキーボードに入力した内容（inputs）をrun_streamed()に渡す
  2. サブエージェントが回答を生成しその内容を出力
  3. 再度ユーザーのキーボード入力を受け付けてその内容をinputsにappendする

routing.py

            result = Runner.run_streamed(
                agent,
                input=inputs,
            )

• ユーザーがキーボードに入力した内容（inputs）をrun_streamed()に渡す

routing.py

            async for event in result.stream_events():
                if not isinstance(event, RawResponsesStreamEvent):
                    continue
                data = event.data
                if isinstance(data, ResponseTextDeltaEvent):
                    print(data.delta, end="", flush=True)
                elif isinstance(data, ResponseContentPartDoneEvent):
                    print("\n")

• eventがRawResponsesStreamEvent（モデルの生ストリーム応答を包むイベント）である場合
  • event.dataがResponseTextDeltaEvent（テキスト増分イベント）であれば、data.deltaを出力し表示を更新
    • printの引数end=""は文末に何も出力しないことを明示している。この指定がなければデフォルトが改行コードとなり、data.delta同士の間に不要な改行が出力されてしまうため。
    • flush=Trueはバッファを即時フラッシュし、リアルタイムに文字を出力するための指定。（デフォルトはFalse）
  • ResponseContentPartDoneEvent（コンテンツ完了イベント）であればメッセージ全体が出力し終わった合図なので改行を出力する

routing.py

        inputs = result.to_input_list()
        print("\n")

• to_input_listはもともとのinputと新しく生成されたすべてのitemsを統合したinputリストを作成するメソッド

  Creates a new input list, merging the original input with all the new items generated.

  リファレンス

• これによりinputsには最新の会話履歴が格納される

routing.py

        user_msg = input("Enter a message: ")
        inputs.append({"content": user_msg, "role": "user"})
        agent = result.current_agent

• 再度ユーザーのキーボード入力を受け付けてその内容をinputsにappendする
• agent = result.current_agentで、agentをHandoffされたagentで更新している。これを書かない場合は常にtriage_agentから始めることになり、毎回言語判定をやり直すことになる。

routing.py

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