PydanticAI

import base64
import os
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk import trace as trace_sdk
from opentelemetry.sdk.trace.export import SimpleSpanProcessor

# 環境変数から機密情報を読み込む
WANDB_BASE_URL = "https://trace.wandb.ai"
PROJECT_ID = os.environ.get("WANDB_PROJECT_ID")  # W&B のエンティティ/プロジェクト名（例: "myteam/myproject"）
WANDB_API_KEY = os.environ.get("WANDB_API_KEY")  # W&B APIキー

OTEL_EXPORTER_OTLP_ENDPOINT = f"{WANDB_BASE_URL}/otel/v1/traces"
AUTH = base64.b64encode(f"api:{WANDB_API_KEY}".encode()).decode()

OTEL_EXPORTER_OTLP_HEADERS = {
    "Authorization": f"Basic {AUTH}",
    "project_id": PROJECT_ID,
}

# エンドポイントとヘッダーを指定して OTLP span エクスポーターを作成する
exporter = OTLPSpanExporter(
    endpoint=OTEL_EXPORTER_OTLP_ENDPOINT,
    headers=OTEL_EXPORTER_OTLP_HEADERS,
)

# トレーサープロバイダーを作成してエクスポーターを追加する
tracer_provider = trace_sdk.TracerProvider()
tracer_provider.add_span_processor(SimpleSpanProcessor(exporter))

from pydantic_ai import Agent
from pydantic_ai.models.instrumented import InstrumentationSettings

# OTEL トレースを有効にした PydanticAI エージェントを作成する
agent = Agent(
    "openai:gpt-4o",
    instrument=InstrumentationSettings(tracer_provider=tracer_provider),
)

result = agent.run_sync("What is the capital of France?")
print(result.output)

from pydantic_ai import Agent
from pydantic_ai.models.instrumented import InstrumentationSettings

# system prompt と OTEL トレースを備えた PydanticAI エージェントを作成する
agent = Agent(
    "openai:gpt-4o",
    system_prompt=(
        "You are a helpful assistant that can multiply numbers. "
        "When asked to multiply numbers, use the multiply tool."
    ),
    instrument=InstrumentationSettings(tracer_provider=tracer_provider),
)

# ツールを定義する
@agent.tool_plain
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

# エージェントにツールを使用するよう指示する
result = agent.run_sync("What is 7 multiplied by 8?")
print(result.output)

from pydantic_ai import Agent
from pydantic_ai.models.instrumented import InstrumentationSettings

# すべてのエージェントにデフォルトのインストルメンテーションを設定する
Agent.instrument_all(InstrumentationSettings(tracer_provider=tracer_provider))

# 以降、新しいエージェントはデフォルトでこのインストルメンテーションを使用する
agent1 = Agent("openai:gpt-4o")
agent2 = Agent("openai:gpt-4o", system_prompt="Be helpful.")

result = agent1.run_sync("What is the capital of France?")
print(result.output)