Pydantic AI Agent Development in Practice: 5 Core Patterns for Type-Safe Production AI Applications

AI与大数据

In 2026, AI Agents have moved from experimental projects to production-grade deployments. However, AI development in the Python ecosystem has long faced a core contradiction: LLM outputs are uncontrollable strings, while production systems need type-safe data structures. Pydantic AI was born to solve this — it deeply combines Pydantic's type validation capabilities with the flexibility of AI Agents, allowing developers to define Agent behavior contracts using Python type annotations alone.

This article distills 5 core patterns from real-world development practice, covering the complete development chain from Agent definition to streaming responses. Each pattern comes with runnable code to help you quickly implement these patterns in your projects.

Core Concepts at a Glance

Concept Description Core Value
Agent Pydantic AI core abstraction, encapsulating LLM interaction logic Unified Agent definition interface
RunContext Runtime context carrying dependency injection objects Type-safe dependency management
Tool Function tools — external capabilities the Agent can invoke Standardized Function Calling
ResultType Structured output type constraining LLM return format Output type safety guarantee
StreamedRunResult Streaming response result supporting token-by-token output Real-time interaction experience

Problem Analysis: Why Traditional AI Development Falls Short

  1. Missing Types: LLMs return raw strings requiring manual parsing and validation — runtime errors are frequent
  2. Dependency Chaos: Database connections, API keys, and other dependencies are passed through layers of functions, causing tight coupling
  3. Tedious Tool Registration: Writing Function Calling JSON Schema by hand is error-prone and disconnected from code logic
  4. Uncontrollable Output: LLMs may return unexpected formats with no compile-time guarantees
  5. Complex Streaming: SSE/WebSocket streaming response parsing and error handling code is verbose

Pattern 1: Agent Basics and Type-Safe Models

Pydantic AI's Agent definition abstracts LLM interactions into type-safe function calls. Through Python type annotations, you define the Agent's input/output contract at code-writing time.

"""Pattern 1: Agent Basics and Type-Safe Models"""
from __future__ import annotations

from dataclasses import dataclass
from typing import Optional

from pydantic import BaseModel, Field
from pydantic_ai import Agent


# 1. Define structured output model
class CodeReviewResult(BaseModel):
    """Code review result — LLM output will be constrained to this structure"""

    quality_score: float = Field(
        ...,
        ge=0.0,
        le=10.0,
        description="Code quality score, 0-10",
    )
    issues: list[str] = Field(
        default_factory=list,
        description="List of discovered issues",
    )
    suggestions: list[str] = Field(
        default_factory=list,
        description="List of improvement suggestions",
    )
    summary: str = Field(
        ...,
        min_length=10,
        max_length=500,
        description="Review summary",
    )
    language: str = Field(
        ...,
        description="Programming language",
    )


# 2. Define Agent — specify output type via generic parameter
code_review_agent = Agent(
    "openai:gpt-4o",
    result_type=CodeReviewResult,
    system_prompt="""You are a senior code review expert.
Please professionally review the code submitted by the user, providing a quality score, issues list, improvement suggestions, and review summary.
Scoring criteria: code readability, performance, security, best practices.""",
)


# 3. Use the Agent
async def review_code(code: str) -> CodeReviewResult:
    """Review code and return structured result"""
    result = await code_review_agent.run(code)
    return result.data


# 4. Synchronous execution
def review_code_sync(code: str) -> CodeReviewResult:
    """Review code synchronously"""
    result = code_review_agent.run_sync(code)
    return result.data


# 5. Agent definition with retries
robust_review_agent = Agent(
    "openai:gpt-4o",
    result_type=CodeReviewResult,
    system_prompt="You are a senior code review expert. Please output strictly according to JSON Schema.",
    retries=3,  # Auto-retry on output validation failure
    model_settings={
        "temperature": 0.1,  # Low temperature for stable output
        "max_tokens": 2000,
    },
)


# 6. Run and check usage information
async def review_with_usage(code: str) -> tuple[CodeReviewResult, dict]:
    """Review code and return usage statistics"""
    result = await robust_review_agent.run(code)
    usage = result.usage()
    return result.data, {
        "request_tokens": usage.request_tokens,
        "response_tokens": usage.response_tokens,
        "total_tokens": usage.total_tokens,
    }


# Run example
if __name__ == "__main__":
    import asyncio

    sample_code = """
def calculate_total(items):
    total = 0
    for i in range(len(items)):
        total = total + items[i]['price'] * items[i]['qty']
    return total
    """

    result = asyncio.run(review_code(sample_code))
    print(f"Quality Score: {result.quality_score}")
    print(f"Issues: {result.issues}")
    print(f"Suggestions: {result.suggestions}")
    print(f"Summary: {result.summary}")
    print(f"Language: {result.language}")

Key Takeaways:

  • The result_type parameter binds LLM output directly to a Pydantic model with automatic validation
  • The retries parameter auto-retries on output validation failure, improving robustness
  • model_settings controls LLM parameters; low temperature ensures stable structured output

Pattern 2: Dependency Injection and System Prompts

Production-grade Agents need access to databases, caches, configurations, and other external dependencies. Pydantic AI's dependency injection system lets these dependencies be automatically injected through type annotations, completely eliminating the chaos of global variables and parameter passing.

"""Pattern 2: Dependency Injection and System Prompts"""
from __future__ import annotations

from dataclasses import dataclass, field
from typing import Optional

from pydantic import BaseModel
from pydantic_ai import Agent, RunContext


# 1. Define dependency types
@dataclass
class DatabaseDep:
    """Database dependency"""

    connection_string: str

    async def query(self, sql: str) -> list[dict]:
        """Simulate database query"""
        print(f"[DB] Executing query: {sql}")
        return [{"id": 1, "name": "John", "score": 95}]


@dataclass
class CacheDep:
    """Cache dependency"""

    backend: str = "redis"
    _cache: dict = field(default_factory=dict)

    async def get(self, key: str) -> Optional[str]:
        return self._cache.get(key)

    async def set(self, key: str, value: str, ttl: int = 3600) -> None:
        self._cache[key] = value
        print(f"[Cache] Set {key} (TTL={ttl}s)")


@dataclass
class UserContext:
    """User context dependency"""

    user_id: str
    username: str
    role: str = "user"
    preferences: dict = field(default_factory=dict)


# 2. Compose dependencies
@dataclass
class AgentDeps:
    """All Agent dependencies"""

    db: DatabaseDep
    cache: CacheDep
    user: UserContext


# 3. Define output model
class AnalysisReport(BaseModel):
    title: str
    findings: list[str]
    confidence: float
    recommendation: str


# 4. Create Agent with dependencies
analysis_agent = Agent(
    "openai:gpt-4o",
    result_type=AnalysisReport,
    deps_type=AgentDeps,
)


# 5. Dynamic system prompt — generated based on dependencies
@analysis_agent.system_prompt
async def build_system_prompt(ctx: RunContext[AgentDeps]) -> str:
    """Dynamically generate system prompt based on user role and preferences"""
    role_instructions = {
        "admin": "You have admin privileges and can access all data and perform all operations.",
        "analyst": "You have analyst privileges and can view data but cannot modify it.",
        "user": "You have regular user privileges and can only view your own data.",
    }

    base_prompt = f"""You are a data analysis assistant.
Current user: {ctx.deps.user.username} (Role: {ctx.deps.user.role})
{role_instructions.get(ctx.deps.user.role, role_instructions['user'])}"""

    if ctx.deps.user.preferences.get("detail_level") == "verbose":
        base_prompt += "\nPlease provide detailed analysis process and reasoning steps."
    else:
        base_prompt += "\nPlease provide concise and clear analysis conclusions."

    return base_prompt


# 6. Use dependencies in tool functions
@analysis_agent.tool
async def query_user_data(
    ctx: RunContext[AgentDeps],
    query: str,
) -> str:
    """Query user-related data"""
    cache_key = f"user:{ctx.deps.user.user_id}:{query}"
    cached = await ctx.deps.cache.get(cache_key)
    if cached:
        return f"[Cache hit] {cached}"

    sql = f"SELECT * FROM user_data WHERE user_id = '{ctx.deps.user.user_id}' AND query LIKE '%{query}%'"
    results = await ctx.deps.db.query(sql)

    await ctx.deps.cache.set(cache_key, str(results))

    return str(results)


@analysis_agent.tool
async def get_user_preferences(
    ctx: RunContext[AgentDeps],
) -> str:
    """Get user preference settings"""
    return str(ctx.deps.user.preferences)


# 7. Run the Agent
async def run_analysis(user_question: str) -> AnalysisReport:
    """Run analysis Agent"""
    deps = AgentDeps(
        db=DatabaseDep(connection_string="postgresql://localhost/mydb"),
        cache=CacheDep(backend="redis"),
        user=UserContext(
            user_id="u_001",
            username="John",
            role="analyst",
            preferences={"detail_level": "verbose", "language": "en"},
        ),
    )

    result = await analysis_agent.run(user_question, deps=deps)
    return result.data


# Run example
if __name__ == "__main__":
    import asyncio

    report = asyncio.run(run_analysis("Analyze the sales data trend for the past month"))
    print(f"Title: {report.title}")
    print(f"Findings: {report.findings}")
    print(f"Confidence: {report.confidence}")
    print(f"Recommendation: {report.recommendation}")

Key Takeaways:

  • deps_type declares the Agent's dependency type, automatically injected into RunContext at runtime
  • The @agent.system_prompt decorator supports dynamic generation of system prompts, customized based on dependency content
  • Tool functions access dependencies through ctx: RunContext[AgentDeps] — type-safe with IDE autocompletion

Pattern 3: Tool Registration and Function Calling

Function Calling is the bridge connecting AI Agents to the external world. Pydantic AI simplifies tool registration to decorator + type annotations, automatically generating JSON Schema so LLMs can precisely call your Python functions.

"""Pattern 3: Tool Registration and Function Calling"""
from __future__ import annotations

import json
from datetime import datetime, timezone
from typing import Optional

from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext, Tool


# 1. Define output model
class TravelPlan(BaseModel):
    """Travel planning result"""

    destination: str
    start_date: str
    end_date: str
    estimated_cost: float
    itinerary: list[str]
    tips: list[str]


# 2. Create Agent
travel_agent = Agent(
    "openai:gpt-4o",
    result_type=TravelPlan,
    system_prompt="You are a travel planning assistant. Plan trips based on user needs using the provided tools to query real-time information.",
)


# 3. Basic tool registration — decorator approach
@travel_agent.tool
async def search_flights(
    ctx: RunContext[None],
    origin: str = Field(description="Departure city"),
    destination: str = Field(description="Destination city"),
    date: str = Field(description="Departure date, format YYYY-MM-DD"),
) -> str:
    """Search flight information"""
    flights = [
        {"flight": "AA1234", "price": 450, "departure": "08:00"},
        {"flight": "UA5678", "price": 380, "departure": "14:30"},
        {"flight": "DL9012", "price": 520, "departure": "19:00"},
    ]
    return json.dumps(flights)


@travel_agent.tool
async def get_weather(
    ctx: RunContext[None],
    city: str = Field(description="City name"),
    date: Optional[str] = Field(default=None, description="Date, defaults to today if not specified"),
) -> str:
    """Query weather information"""
    weather_data = {
        "city": city,
        "date": date or datetime.now(timezone.utc).strftime("%Y-%m-%d"),
        "temperature": "77°F",
        "condition": "Partly cloudy",
        "humidity": "65%",
    }
    return json.dumps(weather_data)


@travel_agent.tool
async def search_hotels(
    ctx: RunContext[None],
    city: str = Field(description="City name"),
    check_in: str = Field(description="Check-in date"),
    check_out: str = Field(description="Check-out date"),
    budget: Optional[float] = Field(default=None, description="Budget limit per night (USD)"),
) -> str:
    """Search hotel information"""
    hotels = [
        {"name": "Downtown Grand Hotel", "price": 180, "rating": 4.5, "location": "Downtown"},
        {"name": "Lakeside Resort", "price": 280, "rating": 4.8, "location": "Lakeside"},
        {"name": "Budget Inn", "price": 89, "rating": 4.0, "location": "Near station"},
    ]
    if budget:
        hotels = [h for h in hotels if h["price"] <= budget]
    return json.dumps(hotels)


# 4. Advanced tool registration — Tool class approach (more control)
async def calculate_exchange_rate(
    amount: float,
    from_currency: str,
    to_currency: str,
) -> str:
    """Currency conversion"""
    rates = {"USD": 1.0, "EUR": 0.92, "GBP": 0.79, "JPY": 155.0}
    if from_currency not in rates or to_currency not in rates:
        return f"Unsupported currency: {from_currency} -> {to_currency}"
    usd_amount = amount / rates[from_currency]
    result = usd_amount * rates[to_currency]
    return f"{amount} {from_currency} = {result:.2f} {to_currency}"


exchange_tool = Tool(
    name="calculate_exchange_rate",
    description="Currency conversion tool, supports USD, EUR, GBP, JPY",
    function=calculate_exchange_rate,
)

travel_agent.register_tool(exchange_tool)


# 5. Tools with dependencies
@dataclass
class TravelDeps:
    user_id: str
    membership_level: str


travel_agent_with_deps = Agent(
    "openai:gpt-4o",
    result_type=TravelPlan,
    deps_type=TravelDeps,
    system_prompt="You are a travel planning assistant.",
)


@travel_agent_with_deps.tool
async def get_member_discount(
    ctx: RunContext[TravelDeps],
) -> str:
    """Get member discount information"""
    discounts = {
        "gold": "20% off + free upgrade opportunity",
        "silver": "10% off",
        "bronze": "5% off",
        "regular": "No discount",
    }
    level = ctx.deps.membership_level
    return f"Membership level: {level}, Discount: {discounts.get(level, 'No discount')}"


# 6. Tool execution tracking
class ToolCallLogger:
    """Tool call logger"""

    def __init__(self):
        self.calls: list[dict] = []

    def log(self, tool_name: str, args: dict, result: str) -> None:
        self.calls.append(
            {
                "tool": tool_name,
                "args": args,
                "result_preview": result[:100],
                "timestamp": datetime.now(timezone.utc).isoformat(),
            }
        )


# 7. Run the Agent
async def plan_travel(user_request: str) -> TravelPlan:
    """Plan travel"""
    result = await travel_agent.run(user_request)

    for msg in result.all_messages():
        if hasattr(msg, "tool_calls") and msg.tool_calls:
            for tc in msg.tool_calls:
                print(f"Tool call: {tc.function.name}({tc.function.arguments})")

    return result.data


# Run example
if __name__ == "__main__":
    import asyncio
    from dataclasses import dataclass

    plan = asyncio.run(plan_travel("I want to go from New York to Tokyo for 5 days next month, budget $3000"))
    print(f"Destination: {plan.destination}")
    print(f"Dates: {plan.start_date} ~ {plan.end_date}")
    print(f"Budget: ${plan.estimated_cost}")
    print(f"Itinerary: {plan.itinerary}")
    print(f"Tips: {plan.tips}")

Key Takeaways:

  • The @agent.tool decorator automatically generates JSON Schema from function signatures and type annotations
  • Field(description=...) provides parameter descriptions for the LLM, improving Function Calling accuracy
  • The Tool class approach provides finer-grained control, such as custom tool names and descriptions
  • Tool functions can be async or sync — Pydantic AI handles both automatically

Pattern 4: Structured Output and Result Validation

Structured output is the core value of Pydantic AI. When the LLM returns JSON that doesn't match the model definition, Pydantic automatically validates and raises errors, and with the retry mechanism, ensures the final output meets expectations.

"""Pattern 4: Structured Output and Result Validation"""
from __future__ import annotations

import json
from enum import Enum
from typing import Literal, Optional, Union

from pydantic import BaseModel, Field, field_validator, model_validator
from pydantic_ai import Agent


# 1. Basic structured output
class SentimentType(str, Enum):
    POSITIVE = "positive"
    NEGATIVE = "negative"
    NEUTRAL = "neutral"


class SentimentResult(BaseModel):
    """Sentiment analysis result"""

    text: str = Field(description="The analyzed text")
    sentiment: SentimentType = Field(description="Sentiment type")
    confidence: float = Field(ge=0.0, le=1.0, description="Confidence score")
    keywords: list[str] = Field(description="Keyword list")

    @field_validator("keywords")
    @classmethod
    def keywords_not_empty(cls, v: list[str]) -> list[str]:
        if not v:
            raise ValueError("Keyword list cannot be empty")
        return v


# 2. Nested structured output
class Address(BaseModel):
    state: str
    city: str
    district: str
    street: Optional[str] = None


class BusinessInfo(BaseModel):
    name: str
    category: str
    address: Address
    rating: float = Field(ge=0.0, le=5.0)
    price_range: str


class BusinessExtraction(BaseModel):
    """Business information extraction result"""

    businesses: list[BusinessInfo] = Field(description="List of extracted businesses")
    total_count: int = Field(ge=0, description="Total business count")
    source_text_summary: str = Field(description="Source text summary")

    @model_validator(mode="after")
    def validate_count(self) -> "BusinessExtraction":
        if self.total_count != len(self.businesses):
            raise ValueError(
                f"total_count({self.total_count}) doesn't match actual business count({len(self.businesses)})"
            )
        return self


# 3. Union type output — let the LLM choose the most appropriate output type
class TextSummary(BaseModel):
    type: Literal["summary"] = "summary"
    content: str
    word_count: int


class TextOutline(BaseModel):
    type: Literal["outline"] = "outline"
    sections: list[str]
    total_sections: int


class TextAnalysis(BaseModel):
    type: Literal["analysis"] = "analysis"
    key_points: list[str]
    tone: str
    audience: str


TextOutput = Union[TextSummary, TextOutline, TextAnalysis]


# 4. Create multiple specialized Agents
sentiment_agent = Agent(
    "openai:gpt-4o",
    result_type=SentimentResult,
    system_prompt="You are a sentiment analysis expert. Analyze the sentiment of text, providing classification, confidence, and keywords.",
    retries=3,
)

extraction_agent = Agent(
    "openai:gpt-4o",
    result_type=BusinessExtraction,
    system_prompt="You are an information extraction expert. Extract business information from text, including name, category, address, and rating.",
    retries=3,
)

flexible_agent = Agent(
    "openai:gpt-4o",
    result_type=TextOutput,
    system_prompt="You are a text analysis assistant. Choose the most appropriate analysis method based on the text: summary, outline, or deep analysis.",
    retries=3,
)


# 5. Result validation and post-processing
class ValidatedResult(BaseModel):
    """Result wrapper with validation logic"""

    raw_result: BaseModel
    is_valid: bool = True
    validation_errors: list[str] = []

    @classmethod
    def from_agent_result(cls, result: BaseModel) -> "ValidatedResult":
        """Create validated result from Agent result"""
        errors = []
        is_valid = True

        if hasattr(result, "confidence") and result.confidence < 0.5:
            errors.append("Low confidence — result may be unreliable")
            is_valid = False

        if hasattr(result, "keywords") and len(result.keywords) < 2:
            errors.append("Too few keywords — analysis may not be thorough enough")
            is_valid = False

        return cls(
            raw_result=result,
            is_valid=is_valid,
            validation_errors=errors,
        )


# 6. Batch processing and result aggregation
async def batch_sentiment_analysis(texts: list[str]) -> list[ValidatedResult]:
    """Batch sentiment analysis"""
    results = []
    for text in texts:
        try:
            result = await sentiment_agent.run(text)
            validated = ValidatedResult.from_agent_result(result.data)
            results.append(validated)
        except Exception as e:
            error_result = ValidatedResult(
                raw_result=SentimentResult(
                    text=text,
                    sentiment=SentimentType.NEUTRAL,
                    confidence=0.0,
                    keywords=["error"],
                ),
                is_valid=False,
                validation_errors=[str(e)],
            )
            results.append(error_result)

    return results


# 7. Structured output and JSON Schema export
def export_result_schema() -> dict:
    """Export the result model's JSON Schema"""
    return SentimentResult.model_json_schema()


# Run example
if __name__ == "__main__":
    import asyncio

    text = "This product is absolutely amazing! Powerful performance, beautiful design, incredible value!"
    result = asyncio.run(sentiment_agent.run(text))
    print(f"Sentiment: {result.data.sentiment.value}")
    print(f"Confidence: {result.data.confidence}")
    print(f"Keywords: {result.data.keywords}")

    business_text = """
    Joe's Pizza located at 123 Main St, Downtown, NY. Italian restaurant, rating 4.2, $15-25 per person.
    Sakura Sushi at 456 Oak Ave, Midtown, NY. Japanese restaurant, rating 4.8, $25-40 per person.
    """
    extraction = asyncio.run(extraction_agent.run(business_text))
    print(f"Business count: {extraction.data.total_count}")
    for b in extraction.data.businesses:
        print(f"  {b.name} - {b.category} - {b.rating} stars")

    schema = export_result_schema()
    print(f"JSON Schema: {json.dumps(schema, indent=2)[:200]}...")

Key Takeaways:

  • Pydantic's field_validator and model_validator play a critical role in LLM output validation
  • Union types let the LLM choose the most appropriate output format based on input
  • The retries parameter works with validators to auto-retry LLM calls on validation failure
  • Use ValidatedResult wrapper for unified error handling in batch processing

Pattern 5: Streaming Responses and Multi-Turn Conversations

Streaming responses let users see the AI's generation process in real-time, significantly improving the interaction experience. Pydantic AI's streaming API is fully compatible with structured output and supports multi-turn conversation context management.

"""Pattern 5: Streaming Responses and Multi-Turn Conversations"""
from __future__ import annotations

from typing import Optional

from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext


# 1. Define output model
class ChatResponse(BaseModel):
    """Chat response"""

    content: str = Field(description="Reply content")
    mood: str = Field(default="neutral", description="Mood state")
    follow_up_questions: list[str] = Field(
        default_factory=list, description="Suggested follow-up questions"
    )


class CodeExplanation(BaseModel):
    """Code explanation result"""

    explanation: str
    complexity: str
    key_concepts: list[str]
    related_topics: list[str]


# 2. Create streaming Agent
stream_agent = Agent(
    "openai:gpt-4o",
    result_type=ChatResponse,
    system_prompt="You are a friendly AI assistant skilled at explaining complex concepts in simple language.",
)


# 3. Streaming text output (token by token)
async def stream_text_response(user_message: str) -> None:
    """Stream text response output"""
    print("User:", user_message)
    print("Assistant: ", end="", flush=True)

    async with stream_agent.run_stream(user_message) as stream:
        # Method 1: Token-by-token output
        async for token in stream.stream_text():
            print(token, end="", flush=True)

    print("\n")


# 4. Streaming structured output (field by field)
async def stream_structured_response(user_message: str) -> ChatResponse:
    """Stream structured response output"""
    async with stream_agent.run_stream(user_message) as stream:
        # Method 2: Stream building of structured object
        async for partial in stream.stream_structured():
            if partial.content:
                print(f"\rContent: {partial.content[:50]}...", end="", flush=True)

        # Get the final complete result
        result = await stream.get_result()
        return result.data


# 5. Multi-turn conversation management
class ConversationManager:
    """Multi-turn conversation manager"""

    def __init__(self):
        self.agent = Agent(
            "openai:gpt-4o",
            result_type=ChatResponse,
            system_prompt="You are a friendly AI assistant. Remember previous conversation content and maintain context continuity.",
        )
        self.message_history: list = []

    async def chat(self, user_input: str) -> ChatResponse:
        """Have a conversation turn"""
        result = await self.agent.run(
            user_input,
            message_history=self.message_history,
        )

        # Update message history
        self.message_history = result.all_messages()

        return result.data

    async def chat_stream(self, user_input: str) -> None:
        """Streaming multi-turn conversation"""
        print(f"User: {user_input}")
        print("Assistant: ", end="", flush=True)

        async with self.agent.run_stream(
            user_input,
            message_history=self.message_history,
        ) as stream:
            async for token in stream.stream_text():
                print(token, end="", flush=True)

        # Update message history
        result = await stream.get_result()
        self.message_history = result.all_messages()
        print("\n")

    def reset(self) -> None:
        """Reset conversation"""
        self.message_history = []

    def get_history_summary(self) -> dict:
        """Get conversation history summary"""
        return {
            "total_messages": len(self.message_history),
            "user_messages": sum(
                1 for m in self.message_history if hasattr(m, "content") and m.role == "user"
            ),
        }


# 6. Streaming Agent with tools
from pydantic_ai import Tool


class StreamingTravelAgent:
    """Streaming travel assistant"""

    def __init__(self):
        self.agent = Agent(
            "openai:gpt-4o",
            result_type=ChatResponse,
            system_prompt="You are a travel assistant. Use tools to query real-time information and provide professional advice.",
        )
        self.message_history: list = []
        self._register_tools()

    def _register_tools(self) -> None:
        @self.agent.tool
        async def search_flights(
            ctx: RunContext[None],
            origin: str,
            destination: str,
        ) -> str:
            """Search flights"""
            return f"Found 3 flights from {origin} to {destination}: AA1234($450), UA5678($380), DL9012($520)"

        @self.agent.tool
        async def get_weather(ctx: RunContext[None], city: str) -> str:
            """Query weather"""
            return f"{city}: Partly cloudy today, 77°F, great for travel"

    async def chat_stream(self, user_input: str) -> ChatResponse:
        """Streaming conversation"""
        async with self.agent.run_stream(
            user_input,
            message_history=self.message_history,
        ) as stream:
            async for token in stream.stream_text():
                print(token, end="", flush=True)

            result = await stream.get_result()
            self.message_history = result.all_messages()
            print()
            return result.data


# 7. Streaming output and progress tracking
class StreamProgress:
    """Streaming output progress tracker"""

    def __init__(self):
        self.token_count = 0
        self.tool_calls = 0
        self.start_time: Optional[float] = None

    def report(self) -> dict:
        """Generate progress report"""
        import time

        elapsed = time.time() - self.start_time if self.start_time else 0
        return {
            "token_count": self.token_count,
            "tool_calls": self.tool_calls,
            "elapsed_seconds": round(elapsed, 2),
            "tokens_per_second": round(self.token_count / elapsed, 1) if elapsed > 0 else 0,
        }


# Run example
if __name__ == "__main__":
    import asyncio

    # Streaming text output
    print("=== Streaming Text Output ===")
    asyncio.run(stream_text_response("What is quantum computing?"))

    # Multi-turn conversation
    print("\n=== Multi-Turn Conversation ===")
    manager = ConversationManager()

    async def demo_conversation():
        await manager.chat_stream("I want to learn Python, where should I start?")
        await manager.chat_stream("What about data structures? Any recommended learning paths?")
        await manager.chat_stream("Can you give me a quicksort example?")
        print(f"Conversation stats: {manager.get_history_summary()}")

    asyncio.run(demo_conversation())

    # Streaming travel assistant
    print("\n=== Streaming Travel Assistant ===")
    travel = StreamingTravelAgent()
    asyncio.run(travel.chat_stream("I want to go from New York to Tokyo, how's the weather?"))

Key Takeaways:

  • run_stream() returns an async context manager supporting both stream_text() and stream_structured() modes
  • The message_history parameter passes historical messages to implement multi-turn conversation context
  • Tool calls execute automatically in streaming mode; the LLM continues generating after receiving tool results
  • ConversationManager encapsulates conversation management logic, suitable for production environments

Pitfall Guide

Pitfall 1: Not Specifying result_type Results in String Output

# ❌ Wrong: No result_type, output is raw string
agent = Agent("openai:gpt-4o")
result = await agent.run("Analyze this code")
print(type(result.data))  # <class 'str'>, no type-safe operations possible

# ✅ Correct: Specify result_type, output is auto-validated as structured object
class AnalysisResult(BaseModel):
    score: float
    issues: list[str]

agent = Agent("openai:gpt-4o", result_type=AnalysisResult)
result = await agent.run("Analyze this code")
print(type(result.data))  # <class 'AnalysisResult'>, type-safe

Pitfall 2: Dependency Type Mismatch Causes Runtime Errors

# ❌ Wrong: deps_type doesn't match the actual dependency type passed
@dataclass
class DepsA:
    api_key: str

agent = Agent("openai:gpt-4o", deps_type=DepsA)

@dataclass
class DepsB:
    token: str  # Different field

result = await agent.run("hello", deps=DepsB(token="xxx"))  # Runtime type error

# ✅ Correct: Ensure deps_type matches the dependency type passed
result = await agent.run("hello", deps=DepsA(api_key="sk-xxx"))

Pitfall 3: Tool Functions Missing Type Annotations

# ❌ Wrong: Tool function without type annotations, LLM can't understand parameter meaning
@agent.tool
async def search(query):  # Missing type annotations and descriptions
    return "results"

# ✅ Correct: Complete type annotations and Field descriptions
@agent.tool
async def search(
    ctx: RunContext[None],
    query: str = Field(description="Search keywords"),
    limit: int = Field(default=10, description="Maximum number of results to return"),
) -> str:
    return "results"

Pitfall 4: Forgetting to Handle Exceptions in Streaming Responses

# ❌ Wrong: No exception handling in streaming response, program crashes on network disconnect
async with agent.run_stream(prompt) as stream:
    async for token in stream.stream_text():
        print(token, end="")

# ✅ Correct: Add exception handling and timeout control
import asyncio

try:
    async with asyncio.timeout(30):
        async with agent.run_stream(prompt) as stream:
            async for token in stream.stream_text():
                print(token, end="")
except asyncio.TimeoutError:
    print("\n[Timeout] Response took too long, please retry")
except Exception as e:
    print(f"\n[Error] {type(e).__name__}: {e}")

Pitfall 5: message_history Not Properly Passed Causes Context Loss

# ❌ Wrong: Not passing history each time, Agent can't remember previous conversation
result1 = await agent.run("My name is John")
result2 = await agent.run("What's my name?")  # Agent doesn't know your name

# ✅ Correct: Pass message_history to maintain context
result1 = await agent.run("My name is John")
history = result1.all_messages()
result2 = await agent.run("What's my name?", message_history=history)  # Agent knows your name

Error Troubleshooting Table

Error Message Cause Solution
ValidationError: field required LLM output missing required field Check model definition, ensure fields have defaults or increase retries
UnexpectedModelBehavior LLM output cannot be parsed as result_type Add format instructions in system_prompt, lower temperature
ModelHTTPError: 429 API call rate limit exceeded Add request intervals, use exponential backoff retry
ToolRunError Tool function execution exception Check tool function logic, add try/except
RunContextError: deps not provided Dependencies not provided at runtime Ensure deps parameter is passed when calling run()
StreamCompleteError Streaming response interrupted Check network connection, add reconnection logic
SchemaGenerationError result_type contains unsupported types Avoid complex nested Unions, simplify model definition
TimeoutError LLM response timeout Increase timeout setting, optimize prompt to reduce token consumption
AuthenticationError API key invalid or expired Check API key configuration in environment variables
JSONDecodeError in tool args LLM generated tool arguments are not valid JSON Specify parameter format in tool description, add examples

Advanced Optimization

  1. Model Fallback Strategy: Configure a fallback model for the Agent, automatically switching when the primary model is unavailable
from pydantic_ai.models import Model

agent = Agent(
    "openai:gpt-4o",
    result_type=AnalysisResult,
    model_settings={"fallback": "openai:gpt-4o-mini"},
)
  1. Tool Call Caching: Cache results of idempotent tool calls to reduce duplicate API requests
import hashlib
import json

tool_cache: dict[str, str] = {}

async def cached_tool_call(tool_name: str, args: dict) -> str:
    cache_key = hashlib.md5(
        json.dumps({"tool": tool_name, "args": args}, sort_keys=True).encode()
    ).hexdigest()
    if cache_key in tool_cache:
        return tool_cache[cache_key]
    result = await execute_tool(tool_name, args)
    tool_cache[cache_key] = result
    return result
  1. Structured Output Schema Optimization: Simplify Schema descriptions to reduce the probability of LLM generating incorrect formats
class OptimizedResult(BaseModel):
    score: float = Field(ge=0, le=10, description="0-10 score")
    issues: list[str] = Field(max_length=5, description="Max 5 issues")
    category: str = Field(default="general", description="Category")
  1. Concurrent Agent Calls: Use asyncio.gather to execute multiple Agent tasks in parallel
import asyncio

async def parallel_analysis(texts: list[str]) -> list[SentimentResult]:
    tasks = [sentiment_agent.run(text) for text in texts]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return [
        r.data if isinstance(r, AgentRunResult) else SentimentResult(text=t, sentiment=SentimentType.NEUTRAL, confidence=0, keywords=[])
        for r, t in zip(results, texts)
    ]
  1. Message History Trimming: Summarize and trim overly long conversation histories to control token consumption
def trim_message_history(
    messages: list,
    max_messages: int = 20,
) -> list:
    if len(messages) <= max_messages:
        return messages
    system_msgs = [m for m in messages if hasattr(m, "role") and m.role == "system"]
    recent_msgs = messages[-max_messages:]
    return system_msgs + recent_msgs

Comparison

Feature Raw OpenAI SDK LangChain Pydantic AI
Type Safety ❌ None ⚠️ Partial ✅ Full Pydantic validation
Dependency Injection ❌ Manual management ⚠️ Global variables ✅ Auto-injection via type annotations
Structured Output ⚠️ Manual parsing ⚠️ Requires OutputParser ✅ Auto-validation + retries
Tool Registration ⚠️ Hand-written JSON Schema ⚠️ Decorator + manual Schema ✅ Auto-generated from type annotations
Streaming Response ⚠️ Manual handling ✅ Supported ✅ Native support + structured streaming
Learning Curve Low High Medium
Code Volume High Medium Low
Production Ready ⚠️ Requires extensive wrapping ⚠️ Over-abstracted ✅ Ready out of the box

Summary

Pydantic AI brings the power of Python's type system into AI Agent development. Through the 5 core patterns of result_type for output constraints, deps_type for dependency management, @agent.tool for tool registration, and run_stream() for streaming responses, developers can build production-grade AI applications in a type-safe manner. Remember: type annotations are not documentation — they are contracts. Pydantic AI ensures these contracts are strictly enforced in LLM interactions.

Online Tools Recommendation

  • JSON Formatter - Format and validate JSON data from Agent output
  • cURL to Code - Convert API debugging cURL commands to Python code
  • Hash Calculator - Calculate hash values for API keys or data, useful for cache key generation

Try these browser-local tools — no sign-up required →

#Pydantic AI#AI Agent#类型安全#Python#2026#AI与大数据