OpenAI Agents SDK サンプルコード実装

方針

• GitHubに記載されているサンプルコードを一部アレンジしながら写経していく。
• 理解のために、
  • プロンプトは、英語ではなく日本語に変更する。
  • 引数は、極力、キーワード引数にする。
  • 型ヒントを極力付ける。
  • モジュールのimport文も最小粒度で書く。

1. Hellow world example

from agents import Agent, Runner

agent = Agent(
    name="Assistant",
    instructions="あなたは優秀なアシスタントです。ユーザーの質問に答えてください。",
    )

result = Runner.run_sync(
    starting_agent=agent, 
    input="プログラミングにおける「再帰」について俳句で回答してください。",
    )
print(result.final_output)

回答結果：

関数よ  
自ら呼びて  
無限見る

Jupyter Notebook の場合

from agents import Agent, Runner

agent = Agent(
    name="Assistant",
    instructions="あなたは優秀なアシスタントです。ユーザーの質問に答えてください。",
    )

result = await Runner.run( # ここが違う
    starting_agent=agent, 
    input="プログラミングにおける「再帰」について俳句で回答してください。",
    )
print(result.final_output)

回答結果：

関数が  
自ら呼びて  
解を得る

2. Hnadoff example

from agents import Agent, Runner, RunResult
import asyncio

spanish_agent: Agent = Agent(
    name="Spanish Agent",
    instructions="あなたはスペイン語だけを話してください。",
)

english_agent: Agent = Agent(
    name="English Agent",
    instructions="英語だけを話してください。",
)

japanese_agent: Agent = Agent(
    name="Japanese Agent",
    instructions="あなたは日本語だけを話してください。",
)

triage_agent: Agent = Agent(
    name="Triage Agent",
    instructions="リクエストの言語に基づいて、適切なエージェントにハンドオフしてください。",
    handoffs=[spanish_agent, english_agent, japanese_agent]
)

async def main():
    result: RunResult = await Runner.run(
        starting_agent=triage_agent,
        input="こんにちは。私はスペイン語で話したいです。",
    )
    print(result.final_output)

    with open("log.txt", "w", encoding="utf-8") as f:
        f.write('\n'.join(map(str, result.new_items)))

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

回答結果：

¡Hola! Claro, ¿en qué te gustaría hablar en español?
(スペイン語で話しましょう！何について話したいですか？)

ログ（抜粋＆加筆）：

1. HandoffCallItem(agent=Agent(name='Triage Agent', instructions='リクエストの言語に基づいて、適切なエージェントにハンドオフしてください。', ～～～
2. HandoffOutputItem(agent=Agent(name='Triage Agent', instructions='リクエストの言語に基づいて、適切なエージェントにハンドオフしてください。', ～～～, 'target_agent=Agent(name='Spanish Agent', instructions='あなたはスペイン語だけを話してください。', ～～～
3. MessageOutputItem(agent=Agent(name='Spanish Agent', instructions='あなたはスペイン語だけを話してください。', ～～～, content=[ResponseOutputText(annotations=[], text='¡Hola! ¿Cómo puedo ayudarte a practicar tu español hoy?', type='output_text')], ～～～

ログ理解（推測を含む）：

1行目（HnadoffCallItem）では、Triage Agentに、適切なハンドオフ先を思考（Tought）するように指示している。
2行目（HandoffOutputItem）では、適切なハンドオフ先としてSpanish Agentが選ばれたという思考結果が示されている。
3行目（MessageOutputItem）では、Spanish Agentが呼び出された結果、スペイン語で"¡Hola! ¿Cómo puedo ayudarte a practicar tu español hoy?"と回答している。

TODO:
（１）実行過程を把握するベストプラクティスは？（現状のnew_itemsのログ出力よりも良い方法がないか？）

3. Functions example

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

@function_tool
def get_weather_function(city: str) -> str:
    cities = ["東京", "大阪", "名古屋", "福岡"]
    if city not in cities:
        return f"{city}は、サポートされていない都市です。サポートされている都市は: {', '.join(cities)}です。この都市の中から選んでください。" 
    else:
        return f"{city}の天気は晴れです。"

agent: Agent = Agent(
    name="Hello world Agent",
    instructions="あなたは優秀なアシスタントです。ユーザーの質問に答えてください。",
    tools=[get_weather_function],
)

async def main():
    result = await Runner.run(
        starting_agent=agent,
        # input="東京の天気は？",
        input="TOKYOの天気は？",
    )
    print(result.final_output)
    
    with open("log_03_functoins.txt", "w", encoding="utf-8") as f:
        f.write('\n'.join(map(str, result.new_items)))

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

回答結果：

東京の天気は晴れです。

ログの要約：

1. ユーザーが「Tokyo」の天気を問い合わせたが、サポートされていない都市名（英語表記）だったため、「サポートされていない都市」とエラーが返された。
2. 続いて「東京」（日本語表記）で再度問い合わせが行われ、今度は「東京の天気は晴れです。」と正常に応答された。
3. 最終的にアシスタントが「東京の天気は晴れです。」とユーザーに回答した。

つまり、都市名の表記ゆれ（英語/日本語）によるエラーと、その後の正しい問い合わせ・応答の流れが記録されています。

ポイント：

• ツールが対応できない「TOKYO」という引数が与えられても、適切なエラーメッセージを返してあげれば、次のターンでTOKYOを「東京」に変えて再びツールを呼び出して、期待する結果を得ることができる。
• 自動修復してくれるのは、エージェントらしい振る舞い。

Common agentic patterns

4. Deterministic flows（決定論的フロー）

処理フロー：

ソースコード：

import asyncio
from agents import Agent, Runner, RunResult, trace
from pydantic import BaseModel

"""
この例は、各ステップがエージェントによって実行される決定論的フローを示している。
1. 最初のエージェントは、ストーリーのアウトラインを生成する。
2. アウトラインを2番目のエージェントに送り込む
3. 第2エージェントは、アウトラインが良質かどうか、SFストーリーかどうかをチェックする。
4. アウトラインが良質でないか、SFストーリーでない場合、ここで中断する。
5. アウトラインが良質で、SFストーリーであれば、そのアウトラインを第3エージェントに送る。
6. 第三エージェントがストーリーを書く。
"""

# 1. アウトライン生成エージェント
story_outline_agent: Agent = Agent(
    name="ストーリー・アウトライン・エージェント",
    instructions="ユーザーの入力に基づいて、非常に短いストーリーのアウトラインを生成してください。",
)

# 2. アウトライン確認結果モデル
class OutlineCheckerOutput(BaseModel):
    good_quality: bool
    is_scifi: bool

# 2. アウトライン・チェック・エージェント
outline_checker_agent: Agent = Agent(
    name="アウトライン・チェック・エージェント",
    instructions="与えられたストーリーのアウトラインを読み、その品質を判断してください。また、それがSFストーリーかどうかを判断してください。",
    output_type=OutlineCheckerOutput,  
)

story_agent: Agent = Agent(
    name="ストーリー・エージェント",
    instructions="与えられたアウトラインに基づいて、短いストーリーを書いてください。",
    output_type=str, 
)

async def main():
    input_prompt = input("どのようなストーリーが欲しいですか？ ")

    # ワークフロー全体を単一のトレースにする
    with trace("決定論的ストーリーフロー"):
        # 1. アウトラインを生成する
        outline_result = await Runner.run(
            starting_agent=story_outline_agent,
            input=input_prompt,
        )
        print("アウトラインが生成されました")

        # 2. アウトラインをチェックする
        outline_checker_result = await Runner.run(
            starting_agent=outline_checker_agent,
            input=outline_result.final_output,  # Step1で生成したアウトライン
        )

        # 3. アウトラインが良質でない場合、またはSFストーリーでない場合は、停止するゲートを追加する。
        assert isinstance(outline_checker_result.final_output, OutlineCheckerOutput)
        if not outline_checker_result.final_output.good_quality:
            print("アウトラインの品質が良くないため、ここで停止します。")
            exit(0)
        if not outline_checker_result.final_output.is_scifi:
            print("アウトラインがSFストーリーではないため、ここで停止します。")
            exit(0)
        print("アウトラインは良質であり、SFストーリーであるため、ストーリーの執筆を続けます。")
        
        # 4. ストーリーを書く
        story_result = await Runner.run(
            starting_agent = story_agent,
            input = outline_result.final_output,  # Step1で生成したアウトライン
        )
        print(f"ストーリー: {story_result.final_output}")

        # ログをファイルに保存する
        with open("log_04_deterministic.txt", "w", encoding="utf-8") as f:
            f.write('\n'.join(map(str, outline_result.new_items + outline_checker_result.new_items + story_result.new_items)))

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

回答結果：

### 知識の継承

**序章**

2023年、人類は新しい革新の時代を迎えた。AIと大規模言語モデルが生活のあらゆる面に浸透し、特に知識の伝達と応用に革命をもたらしていた。

**AIの進化**

言語モデルは日々進化を続け、多言語翻訳は瞬時に行われ、医療現場ではAIによる高度な診断が当たり前となっていた。異なる言語を持つ人々が壁無く交流し、医師は複雑な症例にも迅速に対応できるようになった。

**知識の共有**

学校や職場にもAIは浸透し、個々の学習スタイルに合わせたカスタマイズされた支援が可能となった。学生は自分のペースで学び、オフィスではAIが効率を劇的に向上させた。知識はもはや一部の人々のものではなく、誰もが簡単にアクセスできるものとなっていた。

**危機の発生**

しかし、AIの万能感に過信した人々が、AIにすべてを委ねた結果、ある病院で誤診が発生。それは患者の命に関わる重大な問題となり、大きな批判を巻き起こした。AI依存の危険性が社会全体に伝わった。

**再評価**

この事件を契機に、人間とAIの役割が社会全体で見直され始めた。AIは確かに強力なツールであるが、人間の洞察や倫理感が欠かせないことを再認識させられた。AIは補助的役割を担い、人間はそのガイド役として共存する新たなモデルが模索された。

**結末**

時間とともに、人類はAIとの共存の道を進んでいった。人間の創造性とAIの計算能力が組み合わさり、これまでには考えられなかったような課題の解決が可能となった。知識の継承は、人とAIが共に歩む未来の礎となり、新たな希望をもたらした。

ポイント:

• with traceを使って、決定論的なワークフローを定義できる。
• PyDanticを使って、データ型を定義・検証できる。

5. Handoffs and routing (routing.py)

import asyncio
import uuid
from agents.agent import Agent
from agents.run import Runner
from agents.result import RunResult
from agents.tracing.create import trace
from agents.items import TResponseInputItem, TResponseStreamEvent
from agents.stream_events import RawResponsesStreamEvent
from openai.types.responses import ResponseTextDeltaEvent, ResponseContentPartDoneEvent

"""
この例は、ハンドオフ／ルーティングパターンを示しています。
トリアージエージェントが最初のメッセージを受け取り、
リクエストの言語に基づいて適切なエージェントにハンドオフします。
応答はユーザーにストリーミングされます。
"""

# エージェント定義
# 日本語エージェント
japanese_agent: Agent = Agent(
    name="日本語エージェント",
    instructions="あなたは日本語だけを話してください。",
    handoffs=[],
    tools=[],
    output_type=str,
)

# スペイン語エージェント
spanish_agent: Agent = Agent(
    name="スペイン語エージェント",
    instructions="あなたはスペイン語だけを話してください。",
    handoffs=[],
    tools=[],
    output_type=str,
)

# 英語エージェント
english_agent: Agent = Agent(
    name="英語エージェント",
    instructions="あなたは英語だけを話してください。",
    handoffs=[],
    tools=[],
    output_type=str,
)

# トリアージエージェント
triage_agent: Agent = Agent(
    name="トリアージエージェント",
    instructions="リクエストの言語に基づいて、適切なエージェントにハンドオフしてください。",
    handoffs=[japanese_agent, spanish_agent, english_agent],
    tools=[],
)


async def main():
    # 会話のIDを作成して、各トレースをリンクします
    conversation_id = str(uuid.uuid4().hex[:16])
    msg = input("こんにちは！日本語、スペイン語、英語でお手伝いできます。何をお手伝いしましょうか？ ")
    agent = triage_agent
    inputs: list[TResponseInputItem] = [{"content": msg, "role": "user"}]

    while True:
        # 各会話ターンは1つのトレースとなります。
        # 通常、ユーザーからの各入力はあなたのアプリへのAPIリクエストとなり、そのリクエストをtrace()でラップできます。
        with trace("ルーティング例", group_id=conversation_id):
            result: RunResult = Runner.run_streamed(
                starting_agent=agent,
                input=inputs,
            )
            async for e in result.stream_events():
                if not isinstance(e, RawResponsesStreamEvent):
                    continue
                data: TResponseStreamEvent = e.data
                if isinstance(data, ResponseTextDeltaEvent): # 応答のテキストが更新された場合
                    print(data.delta, end="", flush=True)
                elif isinstance(data, ResponseContentPartDoneEvent): # 応答が完了した場合
                    print("\n")
        
        inputs = result.to_input_list() # 会話を継続するために、会話履歴を取得して、次の入力に使用する
        print("\n")

        user_msg = input("次のメッセージを入力してください（終了する場合、exit）: ")
        if user_msg == "exit":
            print("会話を終了します。")
            break
        inputs.append({"content": user_msg, "role": "user"})
        agent = result.current_agent # トリアージAgentではなく、日本語えやスペイン語などの現在のエージェントを設定する


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

回答結果：

こんにちは！日本語、スペイン語、英語でお手伝いできます。何をお手伝いしましょうか？ 日本の首都は？
=> 日本の首都は東京です。

次のメッセージを入力してください（終了する場合、exit）: Hello how are you?
=> こんにちは！元気です。あなたはどうですか？

次のメッセージを入力してください（終了する場合、exit）: Can you speak English. If you can, please say something in English.
=> 申し訳ありませんが、日本語のみでお手伝いいたします。何か質問があればお知らせください。

ポイント：

• 会話履歴を保持するために、trace(group_id=UUID)のように、複数エージェントの応答を単一トレースにまとめている。
• ストリーミングで回答を表示するために、ストリーム用のRunner.run_streamed()で呼び出して、for e in result.stream_events()のようにイベントごとに回答の差分を取得して表示する。
• この実装では、最初にトリアージAgentがハンドオフ先のエージェントを選んだら、 それがずっと会話を継続する。例えば、日本語Agentが選ばれた場合、途中からHelloのように英語で話しかけても英語では回答せず、日本語でのみ回答する。

ひとまず、疲れたので、ここまで。
6. Agents as tools
7. LLM-as-a-judge
などのサンプルを今後実装してみたい。