> ## Documentation Index
> Fetch the complete documentation index at: https://docs.obiguard.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# Pydantic AI
> Use Obiguard with PydanticAI to take your AI Agents to production
## Introduction
PydanticAI is a Python agent framework designed to make it less painful to build production-grade applications with Generative AI.
It brings the same ergonomic design and developer experience to GenAI that FastAPI brought to web development.
Obiguard enhances PydanticAI with production-readiness features, turning your experimental agents into robust systems by providing:
* **Complete observability** of every agent step, tool use, and interaction
* **Cost tracking and optimization** to manage your AI spend
* **Access to 200+ LLMs** through a single integration
* **Guardrails** to keep agent behavior safe and compliant
Learn more about PydanticAI's core concepts and features
### Installation & Setup
```bash theme={null}
pip install -U pydantic-ai obiguard
```
Create a Obiguard API key with optional budget/rate limits from the [Obiguard dashboard](https://app.obiguard.ai/).
You can attach configurations for reliability, caching, and more to this key.
For a simple setup, first configure the Obiguard client that will be used with PydanticAI:
```python theme={null}
from obiguard import AsyncObiguard
# Set up Obiguard client with appropriate metadata for tracking
obiguard_client = AsyncObiguard(
obiguard_api_key="vk-obg***", # Your Obiguard virtual key
)
```
**What are Virtual Keys?** Virtual keys in Obiguard securely store your LLM provider API keys (OpenAI, Anthropic,
etc.) in an encrypted vault. They allow for easier key rotation and budget management. [Learn more about virtual
keys here](/virtual-keys).
After setting up your Obiguard client, you can integrate it with PydanticAI by connecting it to a model provider:
```python theme={null}
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel
from pydantic_ai.providers.openai import OpenAIProvider
# Connect Obiguard client to a PydanticAI model via provider
agent = Agent(
model=OpenAIModel(
model_name="gpt-4o",
provider=OpenAIProvider(openai_client=obiguard_client),
),
system_prompt="You are a helpful assistant."
)
```
## Basic Agent Implementation
Let's create a simple structured output agent with PydanticAI and Obiguard. This agent will respond to a query about Formula 1 and return structured data:
```python theme={null}
from obiguard import AsyncObiguard
from pydantic import BaseModel, Field
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel
from pydantic_ai.providers.openai import OpenAIProvider
# Set up Obiguard client with tracing and metadata
obiguard_client = AsyncObiguard(
obiguard_api_key="vk-obg***", # Your Obiguard virtual key
)
# Define structured output using Pydantic
class F1GrandPrix(BaseModel):
gp_name: str = Field(description="Grand Prix name, e.g. `Emilia Romagna Grand Prix`")
year: int = Field(description="The year of the Grand Prix")
constructor_winner: str = Field(description="The winning constructor of the Grand Prix")
podium: list[str] = Field(description="Names of the podium drivers (1st, 2nd, 3rd)")
# Create the agent with structured output type
f1_gp_agent = Agent[None, F1GrandPrix](
model=OpenAIModel(
model_name="gpt-4o",
provider=OpenAIProvider(openai_client=obiguard_client),
),
output_type=F1GrandPrix,
system_prompt="Assist the user by providing data about the specified Formula 1 Grand Prix"
)
# Run the agent
async def main():
result = await f1_gp_agent.run("Las Vegas 2023")
print(result.output)
if __name__ == "__main__":
import asyncio
asyncio.run(main())
```
The output will be a structured `F1GrandPrix` object with all fields properly typed and validated:
```json theme={null}
gp_name='Las Vegas Grand Prix'
year=2023
constructor_winner='Red Bull Racing'
podium=['Max Verstappen', 'Charles Leclerc', 'Sergio Pérez']
```
You can also use the synchronous API if preferred:
```python theme={null}
result = f1_gp_agent.run_sync("Las Vegas 2023")
print(result.output)
```
## Advanced Features
### Working with Images
PydanticAI supports multimodal inputs including images. Here's how to use Obiguard with a vision model:
```python theme={null}
from obiguard import AsyncObiguard
from pydantic_ai import Agent, ImageUrl
from pydantic_ai.models.openai import OpenAIModel
from pydantic_ai.providers.openai import OpenAIProvider
# Set up Obiguard client
obiguard_client = AsyncObiguard(
obiguard_api_key="sk-obg***", # Your Obiguard API key
)
# Create a vision-capable agent
vision_agent = Agent(
model=OpenAIModel(
model_name="gpt-4o", # Vision-capable model
provider=OpenAIProvider(openai_client=obiguard_client),
),
system_prompt="Analyze images and provide detailed descriptions."
)
# Process an image
result = vision_agent.run_sync([
'What company is this logo from?',
ImageUrl(url='https://iili.io/3Hs4FMg.png'),
])
print(result.output)
```
Visit your Obiguard dashboard to see detailed logs of this image analysis request, including token usage and costs.
### Tools and Tool Calls
PydanticAI provides a powerful tools system that integrates seamlessly with Obiguard. Here's how to create an agent with tools:
```python theme={null}
import random
from obiguard import AsyncObiguard
from pydantic_ai import Agent, RunContext
from pydantic_ai.models.openai import OpenAIModel
from pydantic_ai.providers.openai import OpenAIProvider
# Set up Obiguard client
obiguard_client = AsyncObiguard(
obiguard_api_key="sk-obg***", # Your Obiguard API key
)
# Create an agent with dependency injection (player name)
dice_agent = Agent(
model=OpenAIModel(
model_name="gpt-4o",
provider=OpenAIProvider(openai_client=obiguard_client),
),
deps_type=str, # Dependency type (player name as string)
system_prompt=(
"You're a dice game host. Roll the die and see if it matches "
"the user's guess. If so, tell them they're a winner. "
"Use the player's name in your response."
),
)
# Define a plain tool (no context needed)
@dice_agent.tool_plain
def roll_die() -> str:
"""Roll a six-sided die and return the result."""
return str(random.randint(1, 6))
# Define a tool that uses the dependency
@dice_agent.tool
def get_player_name(ctx: RunContext[str]) -> str:
"""Get the player's name."""
return ctx.deps
# Run the agent
dice_result = dice_agent.run_sync('My guess is 4', deps='Anne')
print(dice_result.output)
```
Obiguard logs each tool call separately, allowing you to analyze the full execution path of your agent, including both
LLM calls and tool invocations.
### Multi-agent Applications
PydanticAI excels at creating multi-agent systems where agents can call each other. Here's how to integrate Obiguard with a multi-agent setup:
This multi-agent system uses three specialized agents:
`search_agent` - Orchestrates the flow and validates flight selections
`extraction_agent` - Extracts structured flight data from raw text
`seat_preference_agent` - Interprets user's seat preferences
With Obiguard integration, you get:
* Unified tracing across all three agents
* Token and cost tracking for the entire workflow
* Ability to set usage limits across the entire system
* Observability of both AI and human interaction points
Here's a diagram of how these agents interact:
```mermaid theme={null}
graph TD
A[START] --> B[search agent]
B --> C[extraction agent]
C --> B
B --> D[human confirm]
D --> E[find seat function]
B --> F[FAILED]
E --> G[human seat choice]
G --> H[find seat agent]
H --> E
E --> I[buy flights]
I --> J[SUCCESS]
```
```python [expandable] theme={null}
import datetime
from dataclasses import dataclass
from typing import Literal
from pydantic import BaseModel, Field
from rich.prompt import Prompt
from pydantic_ai import Agent, ModelRetry, RunContext
from pydantic_ai.messages import ModelMessage
from pydantic_ai.usage import Usage, UsageLimits
from obiguard import AsyncObiguard
# Set up Obiguard clients with shared trace ID for connected tracing
obiguard_client = AsyncObiguard(
obiguard_api_key="sk-obg***", # Your Obiguard API key
)
# Define structured output types
class FlightDetails(BaseModel):
"""Details of the most suitable flight."""
flight_number: str
price: int
origin: str = Field(description='Three-letter airport code')
destination: str = Field(description='Three-letter airport code')
date: datetime.date
class NoFlightFound(BaseModel):
"""When no valid flight is found."""
class SeatPreference(BaseModel):
row: int = Field(ge=1, le=30)
seat: Literal['A', 'B', 'C', 'D', 'E', 'F']
class Failed(BaseModel):
"""Unable to extract a seat selection."""
# Dependencies for flight search
@dataclass
class Deps:
web_page_text: str
req_origin: str
req_destination: str
req_date: datetime.date
# This agent is responsible for controlling the flow of the conversation
from pydantic_ai.models.openai import OpenAIModel
from pydantic_ai.providers.openai import OpenAIProvider
search_agent = Agent[Deps, FlightDetails | NoFlightFound](
model=OpenAIModel(
model_name="gpt-4o",
provider=OpenAIProvider(openai_client=obiguard_client),
),
output_type=FlightDetails | NoFlightFound, # type: ignore
retries=4,
system_prompt=(
'Your job is to find the cheapest flight for the user on the given date. '
),
instrument=True, # Enable instrumentation for better tracing
)
# This agent is responsible for extracting flight details from web page text
extraction_agent = Agent(
model=OpenAIModel(
model_name="gpt-4o",
provider=OpenAIProvider(openai_client=obiguard_client),
),
output_type=list[FlightDetails],
system_prompt='Extract all the flight details from the given text.',
)
# This agent is responsible for extracting the user's seat selection
seat_preference_agent = Agent[None, SeatPreference | Failed](
model=OpenAIModel(
model_name="gpt-4o",
provider=OpenAIProvider(openai_client=obiguard_client),
),
output_type=SeatPreference | Failed, # type: ignore
system_prompt=(
"Extract the user's seat preference. "
'Seats A and F are window seats. '
'Row 1 is the front row and has extra leg room. '
'Rows 14, and 20 also have extra leg room. '
),
)
@search_agent.tool
async def extract_flights(ctx: RunContext[Deps]) -> list[FlightDetails]:
"""Get details of all flights."""
# Pass the usage to track nested agent calls
result = await extraction_agent.run(ctx.deps.web_page_text, usage=ctx.usage)
return result.output
@search_agent.output_validator
async def validate_output(
ctx: RunContext[Deps], output: FlightDetails | NoFlightFound
) -> FlightDetails | NoFlightFound:
"""Procedural validation that the flight meets the constraints."""
if isinstance(output, NoFlightFound):
return output
errors: list[str] = []
if output.origin != ctx.deps.req_origin:
errors.append(
f'Flight should have origin {ctx.deps.req_origin}, not {output.origin}'
)
if output.destination != ctx.deps.req_destination:
errors.append(
f'Flight should have destination {ctx.deps.req_destination}, not {output.destination}'
)
if output.date != ctx.deps.req_date:
errors.append(f'Flight should be on {ctx.deps.req_date}, not {output.date}')
if errors:
raise ModelRetry('\n'.join(errors))
else:
return output
# Sample flight data (in a real application, this would be from a web scraper)
flights_web_page = """
1. Flight SFO-AK123
- Price: $350
- Origin: San Francisco International Airport (SFO)
- Destination: Ted Stevens Anchorage International Airport (ANC)
- Date: January 10, 2025
2. Flight SFO-AK456
- Price: $370
- Origin: San Francisco International Airport (SFO)
- Destination: Fairbanks International Airport (FAI)
- Date: January 10, 2025
... more flights ...
"""
# Main application flow
async def main():
# Restrict how many requests this app can make to the LLM
usage_limits = UsageLimits(request_limit=15)
deps = Deps(
web_page_text=flights_web_page,
req_origin='SFO',
req_destination='ANC',
req_date=datetime.date(2025, 1, 10),
)
message_history: list[ModelMessage] | None = None
usage: Usage = Usage()
# Run the agent until a satisfactory flight is found
while True:
result = await search_agent.run(
f'Find me a flight from {deps.req_origin} to {deps.req_destination} on {deps.req_date}',
deps=deps,
usage=usage,
message_history=message_history,
usage_limits=usage_limits,
)
if isinstance(result.output, NoFlightFound):
print('No flight found')
break
else:
flight = result.output
print(f'Flight found: {flight}')
answer = Prompt.ask(
'Do you want to buy this flight, or keep searching? (buy/*search)',
choices=['buy', 'search', ''],
show_choices=False,
)
if answer == 'buy':
seat = await find_seat(usage, usage_limits)
await buy_tickets(flight, seat)
break
else:
message_history = result.all_messages(
output_tool_return_content='Please suggest another flight'
)
async def find_seat(usage: Usage, usage_limits: UsageLimits) -> SeatPreference:
message_history: list[ModelMessage] | None = None
while True:
answer = Prompt.ask('What seat would you like?')
result = await seat_preference_agent.run(
answer,
message_history=message_history,
usage=usage,
usage_limits=usage_limits,
)
if isinstance(result.output, SeatPreference):
return result.output
else:
print('Could not understand seat preference. Please try again.')
message_history = result.all_messages()
async def buy_tickets(flight_details: FlightDetails, seat: SeatPreference):
print(f'Purchasing flight {flight_details=!r} {seat=!r}...')
```
Obiguard preserves all the type safety of PydanticAI while adding production monitoring and reliability.
## Production Features
### 1. Enhanced Observability
Obiguard provides comprehensive observability for your PydanticAI agents, helping you understand exactly what's happening during each execution.
Traces provide a hierarchical view of your agent's execution, showing the sequence of LLM calls, tool invocations,
and state transitions.
Obiguard logs every interaction with LLMs, including:
* Complete request and response payloads
* Latency and token usage metrics
* Cost calculations
* Tool calls and function executions
All logs can be filtered by metadata, trace IDs, models, and more, making it easy to debug specific agent runs.
Obiguard provides built-in dashboards that help you:
* Track cost and token usage across all agent runs
* Analyze performance metrics like latency and success rates
* Identify bottlenecks in your agent workflows
* Compare different agent configurations and LLMs
You can filter and segment all metrics by custom metadata to analyze specific agent types, user groups, or use
cases.
### 2. Guardrails for Safe Agents
Guardrails ensure your PydanticAI agents operate safely and respond appropriately in all situations.
**Why Use Guardrails?**
PydanticAI agents can experience various failure modes:
* Generating harmful or inappropriate content
* Leaking sensitive information like PII
* Hallucinating incorrect information
* Generating outputs in incorrect formats
While PydanticAI provides type safety for outputs, Obiguard's guardrails add additional protections for both inputs and outputs.
Obiguard's guardrails can:
* Detect and redact PII in both inputs and outputs
* Filter harmful or inappropriate content
* Validate response formats against schemas
* Check for hallucinations against ground truth
* Apply custom business logic and rules
Explore Obiguard's guardrail features to enhance agent safety.
### 3. Model Interoperability
PydanticAI supports multiple LLM providers, and Obiguard extends this capability by providing access to over 200 LLMs through a unified interface.
You can easily switch between different models without changing your core agent logic:
Obiguard provides access to LLMs from providers including:
* OpenAI (GPT-4o, GPT-4 Turbo, etc.)
* Anthropic (Claude 3.5 Sonnet, Claude 3 Opus, etc.)
* Mistral AI (Mistral Large, Mistral Medium, etc.)
* Google Vertex AI (Gemini 1.5 Pro, etc.)
* Cohere (Command, Command-R, etc.)
* AWS Bedrock (Claude, Titan, etc.)
* Local/Private Models
See the full list of LLM providers supported by Obiguard.
## Set Up Enterprise Governance for PydanticAI
**Why Enterprise Governance?**
If you are using PydanticAI inside your organization, you need to consider several governance aspects:
* **Cost Management**: Controlling and tracking AI spending across teams
* **Access Control**: Managing which teams can use specific models
* **Usage Analytics**: Understanding how AI is being used across the organization
* **Security & Compliance**: Maintaining enterprise security standards
* **Reliability**: Ensuring consistent service across all users
Obiguard adds a comprehensive governance layer to address these enterprise needs.
### Enterprise Features Now Available
**Your PydanticAI integration now has:**
* Departmental budget controls
* Model access governance
* Usage tracking & attribution
* Security guardrails
* Reliability features
## Frequently Asked Questions
Obiguard adds production-readiness to PydanticAI through comprehensive observability (traces, logs, metrics),
reliability features (fallbacks, retries, caching), and access to 200+ LLMs through a unified interface. This makes
it easier to debug, optimize, and scale your agent applications, all while preserving PydanticAI's strong type
safety.
Yes! Obiguard integrates seamlessly with existing PydanticAI applications. You just need to replace your client
initialization code with the Obiguard-enabled version. The rest of your agent code remains unchanged and continues
to benefit from PydanticAI's strong typing.
Obiguard supports all PydanticAI features, including structured outputs, tool use, multi-agent systems, and more. It
adds observability and reliability without limiting any of the framework's functionality.
Yes, Obiguard allows you to use a consistent `trace_id` across multiple agents and requests to track the entire
workflow. This is especially useful for multi-agent systems where you want to understand the full execution path.
Yes! Obiguard uses your own API keys for the various LLM providers. It securely stores them as virtual keys,
allowing you to easily manage and rotate keys without changing your code.
## Resources