OpenAI Agents SDK 入門 (2) - Agent ・ Runner ・ Result

from agents import Agent, ModelSettings, function_tool

# Tool
@function_tool
def get_weather(city: str) -> str:
    return f"The weather in {city} is sunny"

# Agent
agent = Agent(
    name="Haiku agent",
    instructions="常に俳句形式で応答する",
    model="o3-mini",
    tools=[get_weather],
)

# Context
@dataclass
class UserContext:
  uid: str
  is_pro_user: bool

  async def fetch_purchases() -> list[Purchase]:
     return ...

# Agent
agent = Agent[UserContext](
    ...,
)

from pydantic import BaseModel
from agents import Agent

# Output Type
class CalendarEvent(BaseModel):
    name: str
    date: str
    participants: list[str]

# Agent
agent = Agent(
    name="Calendar extractor",
    instructions="テキストからカレンダーイベントを抽出する",
    output_type=CalendarEvent,
)

from agents import Agent

# Handoff
booking_agent = Agent(...)
refund_agent = Agent(...)

# Agent
triage_agent = Agent(
    name="Triage agent",
    instructions=(
        "ユーザーの質問に対応する。"
        "予約に関する質問の場合は、booking agentにハンドオフする。"
        "返金に関する質問の場合は、refund agentにハンドオフする。"
    ),
    handoffs=[booking_agent, refund_agent],
)

# Dynamic Instruction
def dynamic_instructions(
    context: RunContextWrapper[UserContext], agent: Agent[UserContext]
) -> str:
    return f"The user's name is {context.context.name}. Help them with their questions."

# Agent
agent = Agent[UserContext](
    name="Triage agent",
    instructions=dynamic_instructions,
)

# Agent
pirate_agent = Agent(
    name="Pirate",
    instructions="Write like a pirate",
    model="o3-mini",
)

# クローンしたAgent
robot_agent = pirate_agent.clone(
    name="Robot",
    instructions="Write like a robot",
)

from agents import Agent, Runner

async def main():
    # Agentの作成
    agent = Agent(name="Assistant", instructions="あなたは役に立つアシスタントです。")

    # Agentの実行
    result = await Runner.run(agent, "プログラミングにおける再帰について俳句を書いてください。")
    print(result.final_output)

async def main():
    agent = Agent(name="Assistant", instructions="簡潔に返信してください。")

    with trace(workflow_name="Conversation", group_id=thread_id):
        # 最初のターン
        result = await Runner.run(agent, "ゴールデンゲートブリッジはどの都市にありますか?")
        print(result.final_output)
        # San Francisco

        # 2番目のターン
        new_input = result.to_input_list() + [{"role": "user", "content": "それはどの州にありますか?"}]
        result = await Runner.run(agent, new_input)
        print(result.final_output)
        # California

import asyncio
from openai.types.responses import ResponseTextDeltaEvent
from agents import Agent, Runner

async def main():
    # Agent
    agent = Agent(
        name="Joker",
        instructions="あなたは役に立つアシスタントです。",
    )

    # Agentの実行
    result = Runner.run_streamed(agent, input="ジョークを5つ教えてください。")
    async for event in result.stream_events():
        if event.type == "raw_response_event" and isinstance(event.data, ResponseTextDeltaEvent):
            print(event.data.delta, end="", flush=True)

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

import asyncio
import random
from agents import Agent, ItemHelpers, Runner, function_tool

# Tool
@function_tool
def how_many_jokes() -> int:
    return random.randint(1, 10)

async def main():
    # Agent
    agent = Agent(
        name="Joker",
        instructions="まず how_many_jokes ツールを呼び出し、その数だけジョークを言います。",
        tools=[how_many_jokes],
    )

    # Agentの実行
    result = Runner.run_streamed(
        agent,
        input="こんにちは",
    )
    print("=== Run starting ===")

    async for event in result.stream_events():
        # 生の応答イベントのdeltasは無視
        if event.type == "raw_response_event":
            continue
        # Agentが更新されたら印刷
        elif event.type == "agent_updated_stream_event":
            print(f"Agent updated: {event.new_agent.name}")
            continue
        # アイテムが生成されたら出力
        elif event.type == "run_item_stream_event":
            if event.item.type == "tool_call_item":
                print("-- Tool was called")
            elif event.item.type == "tool_call_output_item":
                print(f"-- Tool output: {event.item.output}")
            elif event.item.type == "message_output_item":
                print(f"-- Message output:\n {ItemHelpers.text_message_output(event.item)}")
            else:
                pass  # 他のイベントタイプを無視

    print("=== Run complete ===")


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