Observability and Tracing with Pydantic AI: Complete Guide

Observability and Tracing with Pydantic AI: Complete Guide

Building production-grade applications with LLMs requires more than just prompt engineering. When agents start chaining multiple tool calls, model invocations, and validation steps, debugging becomes a black box. Pydantic AI addresses this challenge head-on by integrating first-class observability and tracing capabilities, primarily through its tight integration with Logfire and OpenTelemetry-compatible backends.

This guide walks you through everything you need to know to instrument, trace, and debug your Pydantic AI applications effectively.

What Is Observability in Pydantic AI?

Observability in Pydantic AI refers to the ability to inspect what happens inside your agents during execution. This includes capturing:

• Every LLM call with its prompt, response, tokens used, and latency
• Tool invocations, their arguments, and return values
• Validation steps and retry attempts
• Model selection and fallback behavior
• Structured outputs and any validation failures

Pydantic AI automatically instruments these events as spans — units of work in a distributed trace. Each agent run produces a hierarchical trace that you can visualize and query.

Why Observability Matters for AI Applications

LLM applications behave differently from traditional software. The same input can produce different outputs, models hallucinate, tool calls fail silently, and costs can spiral out of control. Without observability, you are flying blind. Here is why tracing is essential:

• Debugging unpredictable behavior: When an agent returns a wrong answer, traces show exactly which step went wrong.
• Cost monitoring: Token usage across calls adds up fast. Traces expose per-call token counts.
• Performance optimization: Identify slow tool calls or redundant model invocations.
• Compliance and auditing: Maintain a record of what your AI did and why.
• Evaluation and regression testing: Compare traces across model versions or prompt changes.

Setting Up Logfire Integration

Pydantic AI is built by the same team behind Logfire, an observability platform designed for Python and AI workloads. The integration is seamless and requires minimal setup.

Installation

Install Pydantic AI with the logfire extra, or install Logfire separately:

pip install pydantic-ai logfire

# Or install with the logfire extra
pip install "pydantic-ai[logfire]"

Authenticating with Logfire

After installing, authenticate your local environment with your Logfire account:

logfire auth

This opens a browser to complete authentication. Once done, you can configure your application to send traces to Logfire.

Basic Instrumentation

Here is the minimal setup to start tracing your Pydantic AI agents:

import logfire
from pydantic_ai import Agent

# Configure Logfire — this enables automatic instrumentation
logfire.configure()

# Auto-instrument common libraries (optional but recommended)
logfire.instrument_pydantic_ai()
logfire.instrument_httpx()

# Define and run your agent
agent = Agent(
    'openai:gpt-4o',
    system_prompt='You are a helpful assistant that answers concisely.',
)

result = agent.run_sync('What is the capital of France?')
print(result.data)

The logfire.instrument_pydantic_ai() call automatically instruments all agent runs, tool calls, and model interactions. Every execution now produces a structured trace visible in your Logfire dashboard.

Understanding the Trace Structure

When you run an agent, Pydantic AI generates a hierarchical trace. Understanding this structure helps you navigate and debug effectively.

Trace Hierarchy

A typical agent run produces the following span hierarchy:

• Agent run span: The top-level span representing the entire agent execution
• Model request span: Each LLM API call, including retries
• Tool call span: Each tool invocation with arguments and results
• Validation span: Output validation and retry attempts

Inspecting a Trace Programmatically

You can access trace information directly from the result object without a full observability backend:

import logfire
from pydantic_ai import Agent

logfire.configure()
logfire.instrument_pydantic_ai()

agent = Agent('openai:gpt-4o', system_prompt='You are a math tutor.')

result = agent.run_sync('What is 15 * 23?')

# Access the run history
print("All messages exchanged:")
for message in result.all_messages():
    print(f"  Role: {message.role}")
    if hasattr(message, 'content'):
        print(f"  Content: {message.content}")

# Access usage statistics
print(f"\nUsage: {result.usage()}")

The result.usage() method returns token counts for the entire run, while result.all_messages() gives you the complete conversation history including tool calls and responses.

Tracing Tool Calls in Detail

Tool calls are often the most complex part of an agent's execution. Pydantic AI traces each tool invocation with full argument and return value visibility.

Defining and Tracing Tools

import logfire
from pydantic_ai import Agent, RunContext
from dataclasses import dataclass

logfire.configure()
logfire.instrument_pydantic_ai()

@dataclass
class Dependencies:
    db_connection_string: str

agent = Agent(
    'openai:gpt-4o',
    deps_type=Dependencies,
    system_prompt='You can query a database for user information.',
)

@agent.tool
async def get_user_info(ctx: RunContext[Dependencies], user_id: int) -> dict:
    """Retrieve user information by ID."""
    # This tool call will appear in the trace with:
    # - The user_id argument
    # - The return value
    # - Execution duration
    logfire.info('Querying user', user_id=user_id)
    
    # Simulated database lookup
    users = {
        1: {"name": "Alice", "email": "alice@example.com"},
        2: {"name": "Bob", "email": "bob@example.com"},
    }
    return users.get(user_id, {"error": "User not found"})

@agent.tool
async def update_user_email(
    ctx: RunContext[Dependencies], 
    user_id: int, 
    new_email: str
) -> dict:
    """Update a user's email address."""
    logfire.info('Updating email', user_id=user_id, new_email=new_email)
    return {"status": "updated", "user_id": user_id, "email": new_email}

deps = Dependencies(db_connection_string="postgresql://localhost/mydb")
result = agent.run_sync(
    'Find user 1 and update their email to alice.new@example.com',
    deps=deps,
)
print(result.data)

In your Logfire dashboard, you will see each tool call as a child span under the agent run, with the full arguments and return values captured for inspection.

Custom Spans and Logging

Beyond automatic instrumentation, you can add custom spans to mark important business logic boundaries within your tools or agent workflows.

Adding Custom Spans

import logfire
from pydantic_ai import Agent, RunContext

logfire.configure()
logfire.instrument_pydantic_ai()

agent = Agent('openai:gpt-4o', system_prompt='You are a research assistant.')

@agent.tool
async def search_documents(ctx: RunContext[None], query: str) -> str:
    """Search through documents."""
    # Create a custom span for the search phase
    with logfire.span('document_search', query=query):
        # Simulate search logic
        logfire.info('Performing vector search')
        results = f"Found 3 documents matching '{query}'"
        
        # Log structured data
        logfire.info('Search completed', result_count=3, query=query)
        return results

@agent.tool
async def summarize(ctx: RunContext[None], text: str) -> str:
    """Summarize the given text."""
    with logfire.span('summarization', text_length=len(text)):
        logfire.info('Generating summary')
        return f"Summary of: {text[:50]}..."

result = agent.run_sync('Search for "machine learning" and summarize the results.')
print(result.data)

Custom spans nest within the agent run trace, giving you granular visibility into specific operations. This is especially useful for identifying bottlenecks in multi-step workflows.

Using OpenTelemetry Instead of Logfire

If you prefer a different observability backend — such as Jaeger, Zipkin, Datadog, or Honeycomb — Pydantic AI supports OpenTelemetry natively. Since Logfire uses OpenTelemetry under the hood, you can redirect traces to any OTLP-compatible backend.

Configuring OpenTelemetry Export

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
import logfire

# Set up OpenTelemetry with a custom backend
resource = Resource.create({
    "service.name": "my-pydantic-ai-app",
    "deployment.environment": "production",
})

provider = TracerProvider(resource=resource)
otlp_exporter = OTLPSpanExporter(
    endpoint="http://localhost:4317",  # Jaeger, Zipkin, etc.
    insecure=True,
)
provider.add_span_processor(BatchSpanProcessor(otlp_exporter))
trace.set_tracer_provider(provider)

# Configure Logfire to use the existing OpenTelemetry provider
logfire.configure(send_to_logfire=False)

# Still instrument Pydantic AI — traces go to your OTLP backend
logfire.instrument_pydantic_ai()

from pydantic_ai import Agent

agent = Agent('openai:gpt-4o', system_prompt='You are a helpful assistant.')
result = agent.run_sync('Explain quantum computing in one sentence.')
print(result.data)

This setup sends all Pydantic AI traces to your chosen OpenTelemetry backend while disabling Logfire's own collection. You get the same rich instrumentation without vendor lock-in.

Streaming and Async Tracing

Tracing works identically for streaming and asynchronous agent runs. Each streamed token chunk is captured, and the full trace is completed when the stream finishes.

Tracing a Streaming Run

import asyncio
import logfire
from pydantic_ai import Agent

logfire.configure()
logfire.instrument_pydantic_ai()

agent = Agent('openai:gpt-4o', system_prompt='You are a creative storyteller.')

async def main():
    # Streaming with full trace capture
    async with agent.run_stream('Write a short story about a robot learning to paint.') as result:
        async for chunk in result.stream_text(delta=True):
            print(chunk, end='', flush=True)
        print()  # Newline after streaming completes
    
    # The complete trace is available after the stream ends
    print(f"\nToken usage: {result.usage()}")

asyncio.run(main())

Tracing Async Tool Calls

import asyncio
import logfire
import httpx
from pydantic_ai import Agent, RunContext

logfire.configure()
logfire.instrument_pydantic_ai()
logfire.instrument_httpx()  # Trace HTTP calls made by httpx

agent = Agent('openai:gpt-4o', system_prompt='You can fetch weather data.')

@agent.tool
async def get_weather(ctx: RunContext[None], city: str) -> dict:
    """Fetch current weather for a city."""
    async with httpx.AsyncClient() as client:
        # This HTTP call will appear as a nested span
        response = await client.get(
            f"https://wttr.in/{city}",
            params={"format": "j1"},
        )
        data = response.json()
        current = data["current_condition"][0]
        return {
            "city": city,
            "temp_c": current["temp_C"],
            "description": current["weatherDesc"][0]["value"],
        }

async def main():
    result = await agent.run('What is the weather in London?')
    print(result.data)

asyncio.run(main())

With logfire.instrument_httpx(), every HTTP request made inside a tool call appears as a child span, showing the URL, status code, and duration. This is invaluable for debugging external API integrations.

Working with Multi-Agent Systems

When agents delegate to other agents, tracing becomes critical for understanding the full execution flow. Pydantic AI captures the delegation chain in the trace hierarchy.

Tracing Agent Delegation

import logfire
from pydantic_ai import Agent, RunContext

logfire.configure()
logfire.instrument_pydantic_ai()

# Specialist agent for math
math_agent = Agent(
    'openai:gpt-4o',
    system_prompt='You are a math expert. Solve problems step by step.',
)

# Specialist agent for writing
writing_agent = Agent(
    'openai:gpt-4o',
    system_prompt='You are a professional writer. Create polished content.',
)

# Orchestrator agent that delegates to specialists
orchestrator = Agent(
    'openai:gpt-4o',
    system_prompt='You coordinate between specialists. Delegate tasks appropriately.',
)

@orchestrator.tool
async def solve_math(ctx: RunContext[None], problem: str) -> str:
    """Delegate a math problem to the math specialist."""
    logfire.info('Delegating to math agent', problem=problem)
    result = await math_agent.run(problem, usage=ctx.usage)
    return result.data

@orchestrator.tool
async def write_content(ctx: RunContext[None], topic: str) -> str:
    """Delegate writing to the writing specialist."""
    logfire.info('Delegating to writing agent', topic=topic)
    result = await writing_agent.run(topic, usage=ctx.usage)
    return result.data

result = orchestrator.run_sync(
    'Calculate the area of a circle with radius 5, then write a paragraph about it.'
)
print(result.data)

In the trace, you will see the orchestrator's run as the top-level span, with each delegation appearing as a nested agent run span. This makes it easy to see which specialist handled which part and how long each took.

Monitoring Costs and Token Usage

One of the most practical uses of observability is tracking token consumption. Pydantic AI captures usage data at every model call.

Extracting Usage Data

import logfire
from pydantic_ai import Agent

logfire.configure()
logfire.instrument_pydantic_ai()

agent = Agent('openai:gpt-4o', system_prompt='You are a helpful assistant.')

result = agent.run_sync('Explain the theory of relativity.')

usage = result.usage()
print(f"Request tokens:  {usage.request_tokens}")
print(f"Response tokens: {usage.response_tokens}")
print(f"Total tokens:    {usage.total_tokens}")
print(f"Details:         {usage.details}")

# For multi-turn conversations, track cumulative usage
result2 = agent.run_sync(
    'Now explain it in simpler terms.',
    message_history=result.all_messages(),
)
combined_usage = result2.usage()
print(f"\nCumulative total tokens: {combined_usage.total_tokens}")

In Logfire, you can create dashboards and alerts based on token usage metrics, helping you catch cost anomalies before they become problems.

Best Practices for Observability

1. Instrument Early and Always

Add observability from day one of development. Retrofitting tracing into a complex agent system is painful and error-prone. The overhead is negligible, and the insights are invaluable.

2. Use Meaningful Span Names

When creating custom spans, use descriptive names that reflect business logic, not implementation details:

# Good — describes what the operation does
with logfire.span('invoice_validation', invoice_id=inv_id):
    ...

# Bad — too generic
with logfire.span('process'):
    ...

3. Log Structured Data, Not Strings

Structured logging enables filtering and aggregation in your observability backend:

# Good — structured and queryable
logfire.info('tool_executed', tool_name='search', duration_ms=142, success=True)

# Bad — unstructured, hard to query
logfire.info(f'Tool search executed in 142ms, success=True')

4. Set Up Alerts for Anomalies

Configure alerts for:

• Token usage spikes that indicate runaway agents
• High error rates in tool calls
• Latency exceeding acceptable thresholds
• Validation failure rates trending upward

5. Use Distributed Tracing for Microservices

If your agent calls external services, propagate trace context so you can see the full request path:

import logfire
from opentelemetry.propagate import inject
import httpx

logfire.configure()
logfire.instrument_pydantic_ai()

async def call_external_service(url: str, data: dict) -> dict:
    headers = {}
    # Inject trace context into headers for distributed tracing
    inject(headers)
    
    async with httpx.AsyncClient() as client:
        response = await client.post(url, json=data, headers=headers)
        return response.json()

6. Sample in Production, Trace Everything in Development

In production, consider sampling to reduce costs while maintaining visibility:

import logfire

# In development — capture everything
logfire.configure(environment='development')

# In production — use sampling
logfire.configure(
    environment='production',
    # Logfire handles sampling automatically based on your plan,
    # but you can also configure custom sampling rates
)

7. Tag Traces with Business Context

Add tags and metadata to traces so you can filter by user, feature, or experiment:

import logfire
from pydantic_ai import Agent

logfire.configure()
logfire.instrument_pydantic_ai()

agent = Agent('openai:gpt-4o', system_prompt='You are a customer support agent.')

# Add business context to the trace
logfire.log('info', 'support_request_started', 
    user_id='user_12345',
    tier='premium',
    feature='billing_support',
)

result = agent.run_sync('How do I upgrade my plan?')
print(result.data)

Debugging Common Issues with Traces

Identifying Infinite Loops

Agents can sometimes get stuck in tool-call loops. Traces make this immediately visible — you will see repeated tool call spans with similar arguments. Set a maximum number of result retries to prevent this:

from pydantic_ai import Agent

agent = Agent(
    'openai:gpt-4o',
    system_prompt='You are a helpful assistant.',
    retries=3,  # Maximum validation retries
)

# In traces, if you see more than 3 retry spans, something is wrong

Diagnosing Validation Failures

When structured output validation fails, traces show the raw model response alongside the validation error, making it easy to understand why the model's output did not match your schema:

from pydantic import BaseModel, field_validator
from pydantic_ai import Agent

class UserResponse(BaseModel):
    name: str
    age: int
    
    @field_validator('age')
    @classmethod
    def validate_age(cls, v):
        if v < 0 or v > 150:
            raise ValueError('Age must be between 0 and 150')
        return v

agent = Agent(
    'openai:gpt-4o',
    output_type=UserResponse,
    system_prompt='Extract user information from text.',
)

# If validation fails, the trace shows:
# 1. The raw model response
# 2. The validation error
# 3. Any retry attempts with corrected output
result = agent.run_sync('John is 25 years old.')
print(result.data)

Conclusion

Observability and tracing are not optional features for production AI applications — they are fundamental requirements. Pydantic AI's deep integration with Logfire and OpenTelemetry makes it straightforward to gain full visibility into your agents' behavior, from individual LLM calls to complex multi-agent delegation chains. By instrumenting early, logging structured data, and leveraging the trace hierarchy for debugging, you can build reliable, cost-efficient, and debuggable AI systems. Start with the basic logfire.configure() and logfire.instrument_pydantic_ai() calls, then gradually add custom spans, alerts, and business context tags as your application grows. The investment in observability pays off the first time you need to debug an unexpected agent response in production.