Building Self-Healing AI Agents with LangGraph and PydanticAI

Building a basic LLM-powered application is easy. Building a production-grade agentic system that doesn't fall apart the moment it encounters unexpected input or a transient API error is significantly harder. Most developers start with simple linear chains, only to realize that real-world workflows require loops, state management, and robust error recovery.

In the current AI ecosystem, two tools have emerged as a powerful combination for building these resilient systems: LangGraph and PydanticAI. LangGraph provides the cyclical orchestration needed for complex state machines, while PydanticAI offers the type-safe, model-agnostic framework for defining the agents themselves. Together, they allow us to implement self-healing agentic workflows—systems that can detect their own mistakes, validate tool outputs, and re-attempt tasks without human intervention.

The Architecture of Resilience

Traditional software relies on deterministic logic: if $A$ happens, do $B$. LLMs are non-deterministic. They might hallucinate a tool argument, return improperly formatted JSON, or fail to follow a complex constraint. A self-healing workflow acknowledges this uncertainty and builds a feedback loop directly into the architecture.

In a self-healing system, the workflow follows a specific pattern:

1. Plan/Act: The agent receives a task and selects a tool.
2. Execute: The tool runs and returns a result.
3. Validate: A dedicated validation layer (often using Pydantic models) checks the result against a schema or business logic.
4. Heal: If validation fails, the error message is fed back to the agent as new context, and the agent is asked to correct its mistake.

Why LangGraph and PydanticAI?

Before we dive into the implementation, it's important to understand why this specific stack is so effective.

LangGraph: The Orchestrator

LangGraph is built on top of LangChain but introduces a critical concept: cycles. Unlike standard Directed Acyclic Graphs (DAGs), LangGraph allows you to create loops. This is essential for self-healing because it enables the "retry" logic. If an agent's output is invalid, you can route the flow back to the same agent node with the error context.

PydanticAI: The Type-Safe Boundary

PydanticAI is a relatively new framework that brings the power of Pydantic to the LLM world. It treats LLM interactions as structured data operations. By defining your agent's inputs, outputs, and tool dependencies using Pydantic models, you gain immediate validation and IDE support. It ensures that the data flowing into your LangGraph nodes is exactly what you expect.

Implementing the Self-Healing Loop

Let’s walk through a practical example: a data extraction agent that must pull specific financial metrics from a raw text report and ensure they meet strict validation criteria.

1. Defining the State

In LangGraph, the State is a shared object that moves between nodes. We’ll track the messages, the current extracted data, and any validation errors.

from typing import Annotated, TypedDict, List, Optional
from langgraph.graph.message import add_messages
from pydantic import BaseModel, Field

class ExtractionResult(BaseModel):
    revenue: float
    growth_rate: float
    currency: str = Field(pattern="^[A-Z]{3}$")

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    data: Optional[ExtractionResult]
    errors: List[str]
    retry_count: int

2. Setting Up the PydanticAI Agent

PydanticAI allows us to define an agent with a clear output schema. If the LLM returns data that doesn't match ExtractionResult, PydanticAI will raise a validation error, which we can then catch and handle in our loop.

from pydantic_ai import Agent

# Define the agent
extraction_agent = Agent(
    'openai:gpt-4o',
    result_type=ExtractionResult,
    system_prompt="Extract financial metrics from the provided text accurately."
)

3. Creating the LangGraph Nodes

We need two primary nodes: one to run the agent and one to validate the output. While PydanticAI handles basic schema validation, we might have complex business logic (e.g., "growth_rate cannot exceed 500% unless specifically explained") that requires a separate validation step.

async def call_agent_node(state: AgentState):
    # Pass the conversation history and errors to the agent
    prompt = "Please correct the previous errors and try again." if state['errors'] else "Extract data."
    
    result = await extraction_agent.run(
        prompt, 
        message_history=state['messages']
    )
    
    return {
        "messages": [result.new_messages()],
        "data": result.data,
        "retry_count": state['retry_count'] + 1
    }

def validation_node(state: AgentState):
    errors = []
    data = state['data']
    
    if data.growth_rate > 5.0: # 500%
        errors.append("Growth rate seems anomalously high. Please verify.")
    
    return {"errors": errors}

4. Defining the Logic Flow (The Self-Healing Edge)

This is where the "healing" happens. We define a conditional edge that decides whether to finish the workflow or loop back to the agent based on the presence of errors.

from langgraph.graph import StateGraph, END

workflow = StateGraph(AgentState)

workflow.add_node("extractor", call_agent_node)
workflow.add_node("validator", validation_node)

workflow.set_entry_point("extractor")
workflow.add_edge("extractor", "validator")

# The Router
def should_continue(state: AgentState):
    if not state['errors'] or state['retry_count'] > 3:
        return END
    return "extractor"

workflow.add_conditional_edges("validator", should_continue)

app = workflow.compile()

Advanced Validation: Tool-Validation Loops

In more complex scenarios, agents don't just return data; they call tools (e.g., searching a database, executing code). A common failure point is an agent passing the wrong arguments to a tool.

With PydanticAI, you can define tools using Pydantic models for their arguments. If the agent calls a tool with invalid types, the tool itself can return a validation error message to the agent. This creates a Tool-Validation Loop.

Example: A Database Query Tool

from pydantic_ai.models.openai import OpenAIModel

@extraction_agent.tool
def query_database(sql_query: str) -> str:
    """Executes a SQL query against the financial DB."""
    # Imagine a logic here that checks for forbidden keywords
    if "DROP" in sql_query.upper():
        return "Error: DROP commands are forbidden. Please use SELECT."
    # Execute query...
    return "Results..."

When the agent attempts to run a forbidden command, the tool doesn't just crash the program. It returns a string that explains why it failed. LangGraph sees this as a successful node execution, but the agent receives the error message as context in the next turn, allowing it to rewrite the query.

Handling the "Loop of Death"

One risk with self-healing workflows is the infinite loop. If an agent is stuck on a logic error it cannot solve, it will keep retrying until your API bill hits the ceiling. To prevent this, always implement:

1. Max Retries: As shown in the should_continue function above, hard-cap the number of cycles.
2. State Reset: If an agent fails three times, consider routing to a different, more powerful model (e.g., moving from GPT-4o-mini to GPT-4o) or a human-in-the-loop node.
3. Explicit Error Context: Don't just tell the agent "it failed." Pass the specific Pydantic validation error message. This gives the model the "debugging" information it needs to fix the specific field.

Real-World Benefits

Implementing these loops transformed a recent project I worked on—a document processing pipeline for insurance claims. Initially, the system had a 15% failure rate due to complex table structures that the LLM couldn't parse correctly on the first try. By adding a PydanticAI validation layer and a LangGraph feedback loop, we reduced the failure rate to under 2%. The agent learned to "look again" at specific coordinates of the PDF when the initial extraction failed validation.

Conclusion

Self-healing workflows represent the shift from "prompt engineering" to "system engineering." By using LangGraph to manage the lifecycle of your agents and PydanticAI to enforce strict data boundaries, you can build systems that are far more reliable than the sum of their parts.

Actionable Next Steps:

• Audit your current chains: Identify points where LLM output is used directly in a database query or business logic.
• Introduce PydanticAI: Wrap those outputs in Pydantic models to catch schema mismatches early.
• Implement the Loop: Use LangGraph to catch those validation errors and route them back to the agent with a "Please fix this" instruction.

The goal isn't to build a perfect agent, but to build a system that is smart enough to know when it has failed and capable enough to fix itself.