OpenAI Agents SDKのサンプルコードを読む（エージェントの並列実行）

読んだサンプルコード

• parallelization.py
  • OpenAI Agents SDKのCommon agentic patternsのParallelizationのサンプル

Parallelization（エージェントの並列実行）とは

• 複数のエージェントを並列で実行すること
• 以下のような点で役立つ
  • レイテンシーを小さくすることができる
  • その他の理由
    • 例）複数の回答を生成したうえで、そこからベストなものを選択する等

parallelization.py

parallelization.py

import asyncio

from agents import Agent, ItemHelpers, Runner, trace

"""
This example shows the parallelization pattern. We run the agent three times in parallel, and pick
the best result.
"""

• エージェントを3回並列で実行し、ベストな結果を選択するもの

parallelization.py

spanish_agent = Agent(
    name="spanish_agent",
    instructions="You translate the user's message to Spanish",
)

• エージェントの生成
• ユーザーのメッセージをスペイン語に翻訳する

parallelization.py

translation_picker = Agent(
    name="translation_picker",
    instructions="You pick the best Spanish translation from the given options.",
)

• 別のエージェントの生成
• ベストなスペイン語翻訳を選択するエージェント

parallelization.py

async def main():
    msg = input("Hi! Enter a message, and we'll translate it to Spanish.\n\n")

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

parallelization.py

    # Ensure the entire workflow is a single trace
    with trace("Parallel translation"):
        res_1, res_2, res_3 = await asyncio.gather(
            Runner.run(
                spanish_agent,
                msg,
            ),
            Runner.run(
                spanish_agent,
                msg,
            ),
            Runner.run(
                spanish_agent,
                msg,
            ),
        )

• await asyncio.gather()でエージェントを3回並列で実行

parallelization.py

        outputs = [
            ItemHelpers.text_message_outputs(res_1.new_items),
            ItemHelpers.text_message_outputs(res_2.new_items),
            ItemHelpers.text_message_outputs(res_3.new_items),
        ]

• RunResult.new_items()は、このrunで新しく生成されたRunItemのリストを返す。新しいメッセージや、ツールコール、最終アウトプットなどを含む。
• ItemHelpers.text_message_outputs()はlist[RunItem]からテキストメッセージだけを抽出し、結合したstrを返す
• outputsは3つのスペイン語の翻訳文字列のリスト

parallelization.py

        translations = "\n\n".join(outputs)
        print(f"\n\nTranslations:\n\n{translations}")

• 上記のリストを改行区切りで出力

parallelization.py

        best_translation = await Runner.run(
            translation_picker,
            f"Input: {msg}\n\nTranslations:\n{translations}",
        )

• ユーザーが最初にキーボード入力したメッセージと、3つの翻訳結果をインプットとして2つ目のエージェント（ベストな翻訳を選ぶエージェント）を実行

parallelization.py

    print("\n\n-----")

    print(f"Best translation: {best_translation.final_output}")

• 選択された翻訳を表示

parallelization.py

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