Pydantic AI (tracing)

1
import os
2
3
from opentelemetry import trace
4
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
5
from opentelemetry.sdk.resources import Resource
6
from opentelemetry.sdk.trace import TracerProvider
7
from opentelemetry.sdk.trace.export import BatchSpanProcessor
8
from pydantic_ai import Agent
9
10
exporter = OTLPSpanExporter(
11
    endpoint="https://api.respan.ai/api/v2/traces",
12
    headers={"Authorization": f"Bearer {os.environ['RESPAN_API_KEY']}"},
13
)
14
provider = TracerProvider(
15
    resource=Resource.create({"service.name": "pydantic-ai-app"}),
16
)
17
provider.add_span_processor(BatchSpanProcessor(exporter))
18
trace.set_tracer_provider(provider)
19
20
Agent.instrument_all()
21
22
agent = Agent(
23
    model="openai:gpt-4o",
24
    system_prompt="You are a helpful assistant.",
25
)
26
result = agent.run_sync("What is the capital of France?")
27
print(result.output)
28
provider.force_flush()

1
from pydantic_ai import Agent
2
from respan import Respan
3
from respan_instrumentation_pydantic_ai import PydanticAIInstrumentor
4
5
agent = Agent(model="openai:gpt-4o")
6
7
respan = Respan(
8
    instrumentations=[PydanticAIInstrumentor(agent=agent)],
9
)

1
from pydantic_ai import Agent
2
from respan import Respan, workflow, task
3
from respan_instrumentation_pydantic_ai import PydanticAIInstrumentor
4
5
respan = Respan(
6
    instrumentations=[PydanticAIInstrumentor()],
7
)
8
9
agent = Agent(
10
    model="openai:gpt-4o",
11
    system_prompt="You are a helpful travel assistant.",
12
)
13
14
@task(name="fetch_destination_info")
15
def fetch_destination_info(destination: str) -> str:
16
    result = agent.run_sync(f"Give me a one-sentence summary of {destination}.")
17
    return result.output
18
19
@workflow(name="travel_planning_workflow")
20
def travel_planning_workflow(destination: str) -> str:
21
    return fetch_destination_info(destination)
22
23
print(travel_planning_workflow("Paris"))

1
from pydantic_ai import Agent
2
3
agent = Agent(
4
    model="openai:gpt-4o",
5
    system_prompt=(
6
        "You are a calculator assistant. You must use the provided tools for any arithmetic. "
7
        "Never compute numbers yourself; always call the add tool when asked to add numbers."
8
    ),
9
)
10
11
@agent.tool_plain
12
def add(a: int, b: int) -> int:
13
    return a + b
14
15
result = agent.run_sync(
16
    "Use your add tool to compute 15 + 27, then reply with the result."
17
)
18
print(result.output)

1
from pydantic import BaseModel
2
from pydantic_ai import Agent
3
4
class TravelAnswer(BaseModel):
5
    city: str
6
    summary: str
7
8
agent = Agent(
9
    model="openai:gpt-4o",
10
    system_prompt="You are a helpful travel assistant.",
11
    output_type=TravelAnswer,
12
)
13
14
result = agent.run_sync("Recommend a weekend trip to Paris.")
15
print(result.output)