Skip to main content

Observability for PydanticAI with Agenta

Learn how to connect Agenta with PydanticAI for complete visibility into your AI agent performance, debugging capabilities, and execution observability.

What is Agenta? Agenta is an open-source LLMOps platform designed to streamline the deployment, management, and scaling of large language models. It offers comprehensive observability, testing, and deployment capabilities for AI applications.

What is PydanticAI? PydanticAI is a Python agent framework designed to make it less painful to build production-grade applications with generative AI. It provides type safety, structured outputs, and dependency injection for building robust AI agents.

info

This tutorial is also available as a Jupyter notebook for interactive learning and experimentation.

Setup Guide​

Follow these steps to integrate PydanticAI with Logfire and Agenta's observability platform for real-time tracing and performance analysis.

Install Required Packages​

First, install the required dependencies. These packages are essential for the integration:

pip install pydantic-ai[examples] logfire agenta

Package Overview:

  • pydantic-ai[examples]: The PydanticAI framework with example dependencies
  • logfire: Pydantic's SDK for structured logging and tracing
  • agenta: The core Agenta SDK for prompt engineering and observability

Setup and Configuration​

Initialize your environment and configure both Agenta and Logfire:

import os
from dataclasses import dataclass
import agenta as ag
import logfire
from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext

# Load configuration from environment
os.environ["AGENTA_API_KEY"] = "your_agenta_api_key"
os.environ["AGENTA_HOST"] = "https://cloud.agenta.ai"  # Optional, defaults to the Agenta cloud API

# Start Agenta SDK
ag.init()

# Configure Logfire
logfire.configure(
    service_name="my_logfire_service", 
    send_to_logfire=False, 
    scrubbing=False
)
What does ag.init() do?

The ag.init() function initializes the Agenta SDK and sets up the necessary configuration for observability. It establishes connection to the Agenta platform, configures tracing and logging settings, and prepares the instrumentation context for your application.

Logfire Configuration:

The logfire.configure() function sets up Logfire's observability features. Setting send_to_logfire=False makes sure that the traces are only sent to Agenta (and not Logfire, Pydantic's observability platform), while scrubbing=False makes sure that all the data is sent as is (scrubbing is a feature that automatically removes sensitive data from the traces).

Build Your Instrumented PydanticAI Application​

Here's a complete example showcasing a banking support agent with Agenta instrumentation:

import os
from dataclasses import dataclass
import agenta as ag
import logfire
from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext


# Load configuration from environment
os.environ["AGENTA_API_KEY"] = "your_agenta_api_key"
os.environ["AGENTA_HOST"] = "https://cloud.agenta.ai"  # Optional, defaults to the Agenta cloud API

# Configuration setup
ag.init()
logfire.configure(
    service_name="banking_support_service", 
    send_to_logfire=False, 
    scrubbing=False
)
logfire.instrument_asyncpg()


# Mock database for demonstration
class DatabaseConn:
    """Fake database for example purposes."""
    
    @classmethod
    async def customer_name(cls, *, id: int) -> str | None:
        if id == 123:
            return "John"
        return None
    
    @classmethod
    async def customer_balance(cls, *, id: int, include_pending: bool) -> float:
        if id == 123 and include_pending:
            return 123.45
        else:
            raise ValueError("Customer not found")


# Dependencies for the support agent
@dataclass
class SupportDependencies:
    customer_id: int
    including_pending: bool
    db: DatabaseConn


# Structured output model
class SupportOutput(BaseModel):
    support_advice: str = Field(description="Advice returned to the customer")
    block_card: bool = Field(description="Whether to block their card or not")
    risk: int = Field(description="Risk level of query", ge=0, le=10)


# Create the support agent with instrumentation
support_agent = Agent(
    "openai:gpt-4o",
    deps_type=SupportDependencies,
    output_type=SupportOutput,
    system_prompt=(
        "You are a support agent in our bank, give the "
        "customer support and judge the risk level of their query. "
        "Reply using the customer's name."
    ),
    instrument=True,  # Enable built-in PydanticAI instrumentation
)


# Dynamic system prompt with customer context
@support_agent.system_prompt
async def add_customer_name(ctx: RunContext[SupportDependencies]) -> str:
    customer_name = await ctx.deps.db.customer_name(id=ctx.deps.customer_id)
    return f"The customer's name is {customer_name!r}"


# Agent tool for balance inquiries
@support_agent.tool
async def customer_balance(ctx: RunContext[SupportDependencies]) -> str:
    """Returns the customer's current account balance."""
    balance = await ctx.deps.db.customer_balance(
        id=ctx.deps.customer_id,
        include_pending=ctx.deps.including_pending,
    )
    return f"${balance:.2f}"


# Agenta-instrumented functions
@ag.instrument()
def bank_balance(customer_id: int, query: str, include_pending: bool = True):
    """Returns the customer's current account balance."""
    deps = SupportDependencies(
        customer_id=customer_id,
        including_pending=include_pending,
        db=DatabaseConn(),
    )
    result = support_agent.run_sync(query, deps=deps)
    return result


@ag.instrument()
def block_card(customer_id: int, query: str, include_pending: bool = True):
    """Blocks the customer's card if they report it lost."""
    deps = SupportDependencies(
        customer_id=customer_id,
        including_pending=include_pending,
        db=DatabaseConn(),
    )
    result = support_agent.run_sync(query, deps=deps)
    return result


# Example usage
if __name__ == "__main__":
    # Agent 1: get user's account balance
    result = bank_balance(123, "What is my balance?", True)
    print("Balance Query Result:", result.output)
    
    # Agent 2: block user's card if they report it lost
    result = block_card(123, "I just lost my card!", True)
    print("Card Block Result:", result.output)

Understanding the @ag.instrument() Decorator​

The @ag.instrument() decorator automatically captures all input and output data from your function, enabling comprehensive observability without manual instrumentation.

Span Classification: Use the spankind parameter to categorize different operations in Agenta WebUI. Available classifications:

  • agent - Autonomous agent operations
  • chain - Sequential processing workflows
  • workflow - End-to-end application flows (default behavior)
  • tool - Utility function executions
  • embedding - Vector embedding operations
  • query - Search and retrieval operations
  • completion - Text generation tasks
  • chat - Conversational interactions
  • rerank - Result reordering operations

View Traces in Agenta​

Once your application runs, access detailed execution traces through Agenta's dashboard. The observability data includes:

  • End-to-end agent workflow execution timeline
  • PydanticAI agent initialization and configuration
  • Tool function calls and dependency injection
  • LLM interactions and structured output validation
  • Database queries and external service calls
  • Performance metrics and execution duration

The trace provides comprehensive visibility into your application's execution, helping you:

  • Debug complex agent interactions and tool usage
  • Monitor dependency injection and context management
  • Analyze LLM performance and structured output validation
  • Track database queries and external API calls

Next Steps​

For more detailed information about Agenta's observability features and advanced configuration options, visit the Agenta Observability SDK Documentation.