OpenAI Agents SDK 入門 (3) - Tool ・ Handoff ・ Tracing

from agents import Agent, FileSearchTool, Runner, WebSearchTool

# Agent
agent = Agent(
    name="Assistant",
    tools=[ # Tool
        WebSearchTool(),
        FileSearchTool(
            max_num_results=3,
            vector_store_ids=["VECTOR_STORE_ID"],
        ),
    ],
)

async def main():
    result = await Runner.run(agent, "私の好みと今日のサンフランシスコの天気を考慮すると、どのコーヒーショップに行くべきでしょうか?")
    print(result.final_output)

import json
from typing_extensions import TypedDict, Any
from agents import Agent, FunctionTool, RunContextWrapper, function_tool


class Location(TypedDict):
    lat: float
    long: float

# Tool
@function_tool  
async def fetch_weather(location: Location) -> str:
    
    """指定された場所の天気を取得します。

    Args:
        location: 天気を取得する場所。
    """
    # 実際には、天気APIから天気情報を取得します。
    return "sunny"

# Tool
@function_tool(name_override="fetch_data")  
def read_file(ctx: RunContextWrapper[Any], path: str, directory: str | None = None) -> str:
    """ファイルの内容を読み取ります。

    Args:
        path: 読み取るファイルへのパス。
        directory: ファイルを読み込むディレクトリ。
    """
    # 実際には、ファイルシステムからファイルを読み取ります
    return "<file contents>"

# Agent
agent = Agent(
    name="Assistant",
    tools=[fetch_weather, read_file], # Tool 
)

for tool in agent.tools:
    if isinstance(tool, FunctionTool):
        print(tool.name)
        print(tool.description)
        print(json.dumps(tool.params_json_schema, indent=2))
        print()

from typing import Any
from pydantic import BaseModel
from agents import RunContextWrapper, FunctionTool


def do_some_work(data: str) -> str:
    return "done"


class FunctionArgs(BaseModel):
    username: str
    age: int


async def run_function(ctx: RunContextWrapper[Any], args: str) -> str:
    parsed = FunctionArgs.model_validate_json(args)
    return do_some_work(data=f"{parsed.username} is {parsed.age} years old")


tool = FunctionTool(
    name="process_user",
    description="抽出されたユーザーデータを処理します。",
    params_json_schema=FunctionArgs.model_json_schema(),
    on_invoke_tool=run_function,
)

from agents import Agent, Runner
import asyncio

# Spanish Agent
spanish_agent = Agent(
    name="Spanish agent",
    instructions="ユーザーのメッセージをスペイン語に翻訳します。",
)

# French Agent
french_agent = Agent(
    name="French agent",
    instructions="ユーザーのメッセージをフランス語に翻訳します。",
)

# Orchestrator Agent
orchestrator_agent = Agent(
    name="orchestrator_agent",
    instructions=(
        "あなたは翻訳Agentです。与えられたToolを使用して翻訳を行います。"
        "複数の翻訳を要求された場合は、関連するToolを呼び出します。"
    ),
    tools=[
        spanish_agent.as_tool(
            tool_name="translate_to_spanish",
            tool_description="ユーザーのメッセージをスペイン語に翻訳する。",
        ),
        french_agent.as_tool(
            tool_name="translate_to_french",
            tool_description="ユーザーのメッセージをフランス語に翻訳する。",
        ),
    ],
)

# メイン
async def main():
    result = await Runner.run(orchestrator_agent, input="スペイン語で「こんにちは、お元気ですか？」と言いましょう。")
    print(result.final_output)

from agents import Agent, handoff

billing_agent = Agent(name="Billing agent")
refund_agent = Agent(name="Refund agent")


triage_agent = Agent(name="Triage agent", handoffs=[billing_agent, handoff(refund_agent)])

from agents import Agent, handoff, RunContextWrapper

# Handoffのコールバック関数
def on_handoff(ctx: RunContextWrapper[None]):
    print("Handoff called")

# Agent
agent = Agent(name="My agent")

# Handoff
handoff_obj = handoff(
    agent=agent,
    on_handoff=on_handoff,
    tool_name_override="custom_handoff_tool",
    tool_description_override="Custom description",
)

from pydantic import BaseModel
from agents import Agent, handoff, RunContextWrapper

class EscalationData(BaseModel):
    reason: str

# Handoffのコールバック
async def on_handoff(ctx: RunContextWrapper[None], input_data: EscalationData):
    print(f"Escalation agent called with reason: {input_data.reason}")

# Agent
agent = Agent(name="Escalation agent")

# Handoff
handoff_obj = handoff(
    agent=agent,
    on_handoff=on_handoff,
    input_type=EscalationData,
)

from agents import Agent, handoff
from agents.extensions import handoff_filters

# Agent
agent = Agent(name="FAQ agent")

# Handoff
handoff_obj = handoff(
    agent=agent,
    input_filter=handoff_filters.remove_all_tools, 
)

from agents import Agent
from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX

# Agent
billing_agent = Agent(
    name="Billing agent",
    instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
    <Fill in the rest of your prompt here>.""",
)

from agents import Agent, Runner, trace

async def main():
    agent = Agent(name="Joke generator", instructions="Tell funny jokes.")

    with trace("Joke workflow"): 
        first_result = await Runner.run(agent, "Tell me a joke")
        second_result = await Runner.run(agent, f"Rate this joke: {first_result.final_output}")
        print(f"Joke: {first_result.final_output}")
        print(f"Rating: {second_result.final_output}")

import asyncio
from dataclasses import dataclass
from agents import Agent, RunContextWrapper, Runner, function_tool

@dataclass
class UserInfo:  
    name: str
    uid: int

# Tool
@function_tool
async def fetch_user_age(wrapper: RunContextWrapper[UserInfo]) -> str:  
    return f"ユーザー {wrapper.context.name} は47歳です"

# メイン
async def main():
    user_info = UserInfo(name="ジョン", uid=123)  

    # Agent
    agent = Agent[UserInfo](  
        name="Assistant",
        tools=[fetch_user_age],
    )

    # Agentの実行
    result = await Runner.run(
        starting_agent=agent,
        input="ユーザーの年齢は何歳ですか?",
        context=user_info,
    )

    print(result.final_output)  

# メインの実行
if __name__ == "__main__":
    asyncio.run(main())