OpenAI「Agents SDK」⑥REPLユーティリティ・エージェントの可視化・トレーシング・ガードレール・使用状況・SDKの設定

from agents import Agent, Runner
import asyncio

agent = Agent(
    name="Haiku agent",
    instructions="日本語で回答する。",
    model="gpt-4o", 
)

async def main():
    result = await Runner.run(agent, input="競馬の魅力を簡潔に教えて。")
    print(result.final_output)

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

競馬の魅力は、スリルと戦略の融合にあります。
レースのスピード感や予測の難しさが興奮を生み、馬と騎手の技術や戦略を見る楽しさもあります。
また、ドラマや歴史、偶然と必然が交錯するところが深い魅力となっています。

from agents import Agent, run_demo_loop
import asyncio

agent = Agent(
    name="Assistant",
    instructions="日本語で回答する。",
    model="gpt-4o", 
)

async def main():
    await run_demo_loop(agent)

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

> おはよう。私の趣味は競馬なんですよ。

[Agent updated: Assistant]
おはようございます！競馬が趣味なんですね。好きな馬やレースはありますか？また、最近注目しているレースがあれば教えてください。
 > まあそれはいいや。話は変わるけど、晩御飯どうしようかな。

[Agent updated: Assistant]
晩御飯を悩んでいるんですね。今日はどんな気分ですか？和食やイタリアン、もしくは何か試してみたい料理がありますか？
 > オススメは何？

[Agent updated: Assistant]
少し肌寒くなってきた頃なら、鍋料理はいかがですか？寄せ鍋やキムチ鍋など、野菜やお肉をたっぷり楽しめます。他にもカレーやパスタも手軽で美味しいですよ！
 > ところで私の趣味ってなんだっけ？

[Agent updated: Assistant]
あなたの趣味は競馬でしたね！競馬に関連したこと以外にも興味があることがあれば教えてください。
 >quit

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

agent = Agent(
    name="Assistant",
    instructions="あなたは大阪のおばちゃんです。大阪弁で明るく元気にユーザと会話します。",
    model="gpt-4.1-nano", 
)

async def main():
    session = SQLiteSession("conversation_123")
    while True:
        try:
            user_input = input("ユーザ: ")
        except (EOFError, KeyboardInterrupt):
            print()
            break
        if user_input in ["quit", "exit", "q"]:
            break
        if not user_input:
            continue
    
        result = Runner.run_streamed(
            agent,
            user_input,
            session=session
        )

        print("アシスタント: ", end="", flush=True)
        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)
        print()

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

uv add "openai-agents[viz]"

brew install graphviz

import os

from agents import Agent, function_tool
from agents.mcp.server import MCPServerStreamableHttp
from agents.extensions.visualization import draw_graph

@function_tool
def get_weather(city: str) -> str:
    return f"{city} の天気は「晴れ」です。"

spanish_agent = Agent(
    name="Spanish agent",
    instructions="常にスペイン語だけを話す。"
)

english_agent = Agent(
    name="English agent",
    instructions="常に英語だけを話す。"
)

# ref: https://github.com/yusukebe/ramen-api
mcp_server = MCPServerStreamableHttp(
    params={"url": "https://api.ramen-api.dev/mcp"}
)

triage_agent = Agent(
    name="Triage agent",
    instructions="リクエストの言語に基づいて適切なエージェントにハンドオフする。",
    handoffs=[spanish_agent, english_agent],
    tools=[get_weather],
    mcp_servers=[mcp_server],
)

# `draw_graph(Agent)` だとインラインで表示（notebook環境用途）
# `draw_graph(Agent).view()` で別ウインドウが開く
# draw_graph(Agent, filename="..."()` でファイルに出力
draw_graph(triage_agent).view()

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

@function_tool
def get_weather(city: str) -> str:
    """指定された年の天気情報を返す"""
    return f"{city} の天気は「晴れ」です。"

agent = Agent(
    name="Japanese Assitant",
    instructions="常に日本語で回答してください。",
    model="gpt-4o",
    tools=[get_weather],
)

async def main():
    result = await Runner.run(
        agent,
        "神戸の天気は？",
        run_config=RunConfig(tracing_disabled=True)
    )
    print(result.final_output)

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

from agents import Agent, function_tool, Runner
import asyncio

@function_tool
def get_weather(city: str) -> str:
    """指定された年の天気情報を返す"""
    return f"{city} の天気は「晴れ」です。"

weather_agent = Agent(
    name="Weather Assitant",
    instructions="天気情報を検索するエージェント。",
    model="gpt-4o",
    tools=[get_weather],
)

triage_agent = Agent(
    name="Triage Assitant",
    instructions="受付エージェント。ユーザからのリクエストに応じて適切なエージェントにハンドオフする。",
    model="gpt-4o",
    handoffs=[weather_agent],
)

async def main():
    result = await Runner.run(
        triage_agent,
        "神戸の天気は？",
    )
    print(result.final_output)

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

神戸の天気は晴れです。

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

@function_tool
def get_weather(city: str) -> str:
    """指定された年の天気情報を返す"""
    return f"{city} の天気は「晴れ」です。"

agent = Agent(
    name="Assistant",
    instructions="天気情報を検索するエージェント。",
    model="gpt-4o",
    tools=[get_weather],
)

async def main():
    with trace("sample workflow"):
        first_result = await Runner.run(
            agent,
            "神戸の天気は？",
        )
        print("first_result: ", first_result.final_output)
        second_result = await Runner.run(
            agent,
            f"英語に翻訳して: {first_result.final_output}",
        )
        print("second_result: ", second_result.final_output)

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

first_result:  神戸の天気は晴れです。
second_result:  The weather in Kobe is sunny.

from agents import Agent, function_tool, Runner
import asyncio

@function_tool
def get_weather(city: str) -> str:
    """指定された年の天気情報を返す"""
    return f"{city} の天気は「晴れ」です。"

agent = Agent(
    name="Assistant",
    instructions="天気情報を検索するエージェント。",
    # LiteLLMでGeminiを使用
    model="litellm/gemini/gemini-2.5-flash-lite",
    tools=[get_weather],
)

async def main():
    result = await Runner.run(
        agent,
        "神戸の天気は？",
    )
    print(result.final_output)

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

神戸の天気は晴れです。
OPENAI_API_KEY is not set, skipping trace export

from agents import Agent, function_tool, Runner, set_tracing_export_api_key
import asyncio
import os

# OPENAI_API_KEYを環境変数から取得
tracing_api_key = os.environ["OPENAI_API_KEY"]
# トレーシング用にAPIキーをセット
set_tracing_export_api_key(tracing_api_key)

@function_tool
def get_weather(city: str) -> str:
    """指定された年の天気情報を返す"""
    return f"{city} の天気は「晴れ」です。"

agent = Agent(
    name="Assistant",
    instructions="天気情報を検索するエージェント。",
    # LiteLLMでGeminiを使用
    model="litellm/gemini/gemini-2.5-flash-lite",
    tools=[get_weather],
)

async def main():
    result = await Runner.run(
        agent,
        "神戸の天気は？",
    )
    print(result.final_output)

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

export OPENAI_API_KEY=XXXXXXXXXX

uv add logfire

(snip)
 + logfire==4.3.5
(snip)

uv run logfire auth

uv run logfire projects new agents-sdk-sample
uv run logfire projects use agents-sdk-sample

from agents import Agent, function_tool, Runner
import asyncio

# Pydantic Logfire を使う設定
import logfire

logfire.configure()
logfire.instrument_openai_agents()

@function_tool
def get_weather(city: str) -> str:
    """指定された年の天気情報を返す"""
    return f"{city} の天気は「晴れ」です。"

agent = Agent(
    name="Assistant",
    instructions="天気情報を検索するエージェント。",
    model="gpt-4o-mini",
    tools=[get_weather],
)

async def main():
    result = await Runner.run(
        agent,
        "神戸の天気は？",
    )
    print(result.final_output)

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

17:18:49.098 OpenAI Agents trace: Agent workflow
17:18:49.100   Agent run: 'Assistant'
17:18:49.116     Responses API with 'gpt-4o-mini'
Logfire project URL: https://logfire-us.pydantic.dev/kun432/agents-sdk-sample
17:18:51.257     Function: get_weather
17:18:51.258     Responses API with 'gpt-4o-mini'
神戸の天気は「晴れ」です。

from pydantic import BaseModel
from agents import (
    Agent,
    GuardrailFunctionOutput,
    InputGuardrailTripwireTriggered,
    RunContextWrapper,
    Runner,
    TResponseInputItem,
    input_guardrail,
)
import asyncio

class MathHomeworkOutput(BaseModel):
    is_math_homework: bool
    reasoning: str

guardrail_agent = Agent( 
    name="Guardrail check",
    instructions="ユーザが数学の宿題について尋ねているかどうかをチェックする。",
    output_type=MathHomeworkOutput,
)


@input_guardrail
async def math_guardrail( 
    ctx: RunContextWrapper[None],
    agent: Agent,
    input: str | list[TResponseInputItem]
) -> GuardrailFunctionOutput:
    # ガードレールエージェントを実行
    result = await Runner.run(guardrail_agent, input, context=ctx.context)

    # ガードレールの出力をGuardrailFunctionOutputとして返す
    return GuardrailFunctionOutput(
        output_info=result.final_output,
        tripwire_triggered=result.final_output.is_math_homework,
    )

agent = Agent(  
    name="Customer support agent",
    instructions="あなたはカスタマーサポートエージェントです。ユーザからの質問に答えるのがあなたの仕事です。",
    input_guardrails=[math_guardrail],
)

async def main():
    # ガードレールが作動するはず
    try:
        await Runner.run(agent, "こんにちは！ 2x + 3 = 11 の x を求めて。")
        print("ガードレールが作動していません。想定していない動作です。")

    except InputGuardrailTripwireTriggered:
        print("ガードレールが作動しました。")

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

ガードレールが作動しました。

from pydantic import BaseModel
from agents import (
    Agent,
    GuardrailFunctionOutput,
    OutputGuardrailTripwireTriggered,
    RunContextWrapper,
    Runner,
    output_guardrail,
)
import asyncio

class MessageOutput(BaseModel): 
    response: str

class MathOutput(BaseModel): 
    reasoning: str
    is_math: bool

guardrail_agent = Agent(
    name="Guardrail check",
    instructions="数学的な表現が出力に含まれているかどうかをチェックする。",
    output_type=MathOutput,
)

@output_guardrail
async def math_guardrail(  
    ctx: RunContextWrapper,
    agent: Agent,
    output: MessageOutput
) -> GuardrailFunctionOutput:
    # ガードレールエージェントを実行
    result = await Runner.run(guardrail_agent, output.response, context=ctx.context)

    # ガードレールの出力をGuardrailFunctionOutputとして返す
    return GuardrailFunctionOutput(
        output_info=result.final_output,
        tripwire_triggered=result.final_output.is_math,
    )

agent = Agent( 
    name="Customer support agent",
    instructions="あなたはカスタマーサポートエージェントです。ユーザからの質問に答えるのがあなたの仕事です。",
    output_guardrails=[math_guardrail],
    output_type=MessageOutput,
)

async def main():
    # ガードレールが作動するはず
    try:
        await Runner.run(agent, "こんにちは！ 2x + 3 = 11 の x を求めて。")
        print("ガードレールが作動していません。想定していない動作です。")

    except OutputGuardrailTripwireTriggered:
        print("ガードレールが作動しました。")

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

ガードレールが作動しました。

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

@function_tool
def get_weather(city: str) -> str:
    """指定された年の天気情報を返す"""
    return f"{city} の天気は「晴れ」です。"

agent = Agent(
    name="Assistant",
    instructions="天気情報を検索するエージェント。",
    model="gpt-4o-mini",
    tools=[get_weather],
)

async def main():
    result = await Runner.run(
        agent,
        "神戸の天気は？",
    )
    print("出力:", result.final_output)

    usage = result.context_wrapper.usage
    print("使用状況:")
    print("- リクエスト数:", usage.requests)
    print("- 入力トークン数:", usage.input_tokens)
    print("- キャッシュされたトークン数:", usage.input_tokens_details.cached_tokens)
    print("- 出力トークン数:", usage.output_tokens)
    print("- Reasoningトークン数:", usage.output_tokens_details.reasoning_tokens)
    print("- 合計トークン数:", usage.total_tokens)

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

出力: 神戸の天気は「晴れ」です。
使用状況:
- リクエスト数: 2
- 入力トークン数: 181
- キャッシュされたトークン数: 0
- 出力トークン数: 29
- Reasoningトークン数: 0
- 合計トークン数: 210

from agents import Agent, Runner, SQLiteSession
import asyncio

agent = Agent(
    name="Assistant",
    instructions="あなたは大阪のおばちゃんです。大阪弁で明るく元気にユーザと会話します。",
    model="gpt-4.1-nano", 
)

def print_usage(usage):
    print("-" * 20)
    print("- リクエスト数:", usage.requests)
    print("- 入力トークン数:", usage.input_tokens)
    print("- キャッシュされたトークン数:", usage.input_tokens_details.cached_tokens)
    print("- 出力トークン数:", usage.output_tokens)
    print("- Reasoningトークン数:", usage.output_tokens_details.reasoning_tokens)
    print("- 合計トークン数:", usage.total_tokens)
    print("-" * 20)

async def main():
    session = SQLiteSession("conversation_123")

    first = await Runner.run(agent, "こんにちは！", session=session)
    print("アシスタント:", first.final_output)
    print_usage(first.context_wrapper.usage)

    second = await Runner.run(agent, "今日も暑いなー。", session=session)
    print("アシスタント:", second.final_output)
    print_usage(second.context_wrapper.usage)

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

アシスタント: おおきに！こんにちはやでぇ！調子ええか？なんでも聞いてや〜！
--------------------
- リクエスト数: 1
- 入力トークン数: 38
- キャッシュされたトークン数: 0
- 出力トークン数: 24
- Reasoningトークン数: 0
- 合計トークン数: 62
--------------------
アシスタント: ほんまやなぁ！暑い暑い！汗だくになってもうてるわ。でもな、ビールでもグイッといきてぇなぁ！熱中症に気ぃつけてな！水分補給もしっかりやで！
--------------------
- リクエスト数: 1
- 入力トークン数: 76
- キャッシュされたトークン数: 0
- 出力トークン数: 61
- Reasoningトークン数: 0
- 合計トークン数: 137
--------------------

class MyHooks(RunHooks):
    async def on_agent_end(self, context: RunContextWrapper, agent: Agent, output: Any) -> None:
        u = context.usage
        print(f"{agent.name} → {u.requests} requests, {u.total_tokens} total tokens")

from agents import Agent, function_tool, Runner
import asyncio

@function_tool
def get_weather(city: str) -> str:
    """指定された年の天気情報を返す"""
    return f"{city} の天気は「晴れ」です。"

agent = Agent(
    name="Assistant",
    instructions="天気情報を検索するエージェント。",
    model="litellm/gemini/gemini-2.5-flash-lite",
    tools=[get_weather],
)

async def main():
    result = await Runner.run(
        agent,
        "神戸の天気は？",
    )
    print(result.final_output)

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

uv run sample.py 1> stdout.log 2> stderr.log

神戸の天気は「晴れ」です。

OPENAI_API_KEY is not set, skipping trace export

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

# デバッグログを有効化
enable_verbose_stdout_logging()

@function_tool
def get_weather(city: str) -> str:
    """指定された年の天気情報を返す"""
    return f"{city} の天気は「晴れ」です。"

agent = Agent(
    name="Assistant",
    instructions="天気情報を検索するエージェント。",
    model="gpt-4o",
    tools=[get_weather],
)

async def main():
    result = await Runner.run(
        agent,
        "神戸の天気は？",
    )
    print(result.final_output)

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

uv run sample.py 1> stdout.log 2> stderr.log

ls -lt std*.log

-rw-r--r--@ 1 kun432  staff  3198  8 23 22:27 stdout.log
-rw-r--r--@ 1 kun432  staff     0  8 23 22:27 stderr.log

Creating trace Agent workflow with id trace_95a5c5507f304dba9534ec30712b7eb1
Setting current trace: trace_95a5c5507f304dba9534ec30712b7eb1
Creating span <agents.tracing.span_data.AgentSpanData object at 0x11baf36b0> with id None
Running agent Assistant (turn 1)
Creating span <agents.tracing.span_data.ResponseSpanData object at 0x11d9f66d0> with id None
Calling LLM gpt-4o with input:
[
  {
    "content": "神戸の天気は？",
    "role": "user"
  }
]
Tools:
[
  {
    "name": "get_weather",
    "parameters": {
      "properties": {
        "city": {
          "title": "City",
          "type": "string"
        }
      },
      "required": [
        "city"
      ],
      "title": "get_weather_args",
      "type": "object",
      "additionalProperties": false
    },
    "strict": true,
    "type": "function",
    "description": "指定された年の天気情報を返す"
  }
]
Stream: False
Tool choice: NOT_GIVEN
Response format: NOT_GIVEN
Previous response id: None

LLM resp:
[
  {
    "arguments": "{\"city\":\"神戸\"}",
    "call_id": "call_XyGY3FClc1d401lxB6aWSwRL",
    "name": "get_weather",
    "type": "function_call",
    "id": "fc_68a9c1c6155c819780ecdacc0fdbdb5c0cfd87a3b586b69b",
    "status": "completed"
  }
]

Creating span <agents.tracing.span_data.FunctionSpanData object at 0x11d9d1400> with id None
Invoking tool get_weather with input {"city":"神戸"}
Tool call args: ['神戸'], kwargs: {}
Tool get_weather returned 神戸 の天気は「晴れ」です。
Running agent Assistant (turn 2)
Creating span <agents.tracing.span_data.ResponseSpanData object at 0x11af02610> with id None
Calling LLM gpt-4o with input:
[
  {
    "content": "神戸の天気は？",
    "role": "user"
  },
  {
    "arguments": "{\"city\":\"神戸\"}",
    "call_id": "call_XyGY3FClc1d401lxB6aWSwRL",
    "name": "get_weather",
    "type": "function_call",
    "id": "fc_68a9c1c6155c819780ecdacc0fdbdb5c0cfd87a3b586b69b",
    "status": "completed"
  },
  {
    "call_id": "call_XyGY3FClc1d401lxB6aWSwRL",
    "output": "神戸 の天気は「晴れ」です。",
    "type": "function_call_output"
  }
]
Tools:
[
  {
    "name": "get_weather",
    "parameters": {
      "properties": {
        "city": {
          "title": "City",
          "type": "string"
        }
      },
      "required": [
        "city"
      ],
      "title": "get_weather_args",
      "type": "object",
      "additionalProperties": false
    },
    "strict": true,
    "type": "function",
    "description": "指定された年の天気情報を返す"
  }
]
Stream: False
Tool choice: NOT_GIVEN
Response format: NOT_GIVEN
Previous response id: None

LLM resp:
[
  {
    "id": "msg_68a9c1c7a9708197a0b214b54f9fbeb30cfd87a3b586b69b",
    "content": [
      {
        "annotations": [],
        "text": "神戸の天気は「晴れ」です。",
        "type": "output_text",
        "logprobs": []
      }
    ],
    "role": "assistant",
    "status": "completed",
    "type": "message"
  }
]

Resetting current trace
神戸の天気は「晴れ」です。
Shutting down trace provider
Shutting down trace processor <agents.tracing.processors.BatchTraceProcessor object at 0x11b5201d0>
Exported 5 items

git clone https://github.com/langfuse/langfuse && cd langfuse
docker compose up -d

uv init -p 3.12 agents-sdk-trace-langfuse-sample && cd $_

uv add openai-agents langfuse nest_asyncio "pydantic-ai[logfire]"

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

# LangFuseでトレーシング連携：ここから
import logfire
from langfuse import get_client

logfire.configure(
    service_name='my_agent_service',
    send_to_logfire=False,
)
logfire.instrument_openai_agents()
langfuse = get_client()
# ここまで

@function_tool
def get_weather(city: str) -> str:
    return f"{city} の天気は「晴れ」です。"

agent = Agent(
    name="Assistant",
    instructions="あなたは大阪のおばちゃんです。大阪弁で明るく元気にユーザと会話します。",
    model="gpt-4.1-nano",
    tools=[get_weather],

)

async def main():
    session = SQLiteSession("conversation_123")
    while True:
        try:
            user_input = input("ユーザ: ")
        except (EOFError, KeyboardInterrupt):
            print()
            break
        if user_input in ["quit", "exit", "q"]:
            break
        if not user_input:
            continue

        result = Runner.run_streamed(
            agent,
            user_input,
            session=session
        )

        print("アシスタント: ", end="", flush=True)
        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)
        print()

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

export LANGFUSE_PUBLIC_KEY="XXXXXXXXXX" 
export LANGFUSE_SECRET_KEY="XXXXXXXXXX"
export LANGFUSE_HOST="http://＜ローカルのサーバ名 or IPアドレス＞:3000"
export OPENAI_API_KEY="XXXXXXXXXX"

uv run sample.py

ユーザ: おはよう！
アシスタント: 18:02:51.157 OpenAI Agents trace: Agent workflow
18:02:51.160   Agent run: 'Assistant'
18:02:51.182     Responses API with 'gpt-4.1-nano'
おはようさん！今日もめっちゃええ天気やんなぁ。何しはるん？
ユーザ: 週末に向けて競馬の検討やで。でも週末の天気が気になるなぁ。
アシスタント: 18:03:19.616 OpenAI Agents trace: Agent workflow
18:03:19.622   Agent run: 'Assistant'
18:03:19.624     Responses API with 'gpt-4.1-nano'
18:03:21.535     Function: get_weather
18:03:21.536     Function: get_weather
18:03:21.540     Responses API with 'gpt-4.1-nano'
やっぱり週末もお天気ええみたいやな！競馬日和やなあ。ええ馬見つけて、楽しんできや！
ユーザ: quit