Tekko

Language

Get in Touch →

Usually respond within 24 hours

Back to BlogAI & ML

Self-Healing Agentic Workflows with LangGraph and PydanticAI

7 min read
LangGraphPydanticAIAgentic WorkflowsPythonLLMOps
Self-Healing Agentic Workflows with LangGraph and PydanticAI

Building production-grade AI applications is less about the prompt and more about the architecture. In a laboratory setting, a one-shot prompt might look impressive. In production, however, non-deterministic outputs, schema violations, and failed tool calls are the norm. If your agent crashes the moment an LLM provides a malformed JSON string or an invalid API parameter, you don't have a product; you have a prototype.

To move beyond fragile scripts, we need Self-Healing Agentic Workflows. This architectural pattern treats errors not as exceptions to be caught, but as state transitions to be managed. By combining LangGraph for orchestration and PydanticAI for structured validation, we can build systems that detect their own mistakes, reflect on the failure, and autonomously correct their path.

The Anatomy of a Self-Healing Loop

A self-healing workflow is essentially a feedback loop. Instead of a linear flow (Input -> Agent -> Tool -> Output), we implement a cyclic graph. The core components include:

  1. State Management: A persistent object that tracks the conversation history, tool outputs, and error logs.
  2. Strict Schema Enforcement: Using Pydantic models to define exactly what the LLM should output.
  3. Validation Nodes: Logic that checks the LLM's output against the schema or business rules.
  4. Reflection & Retry: A conditional edge that routes failures back to the agent with a descriptive error message, allowing it to "heal" the output.

Why LangGraph and PydanticAI?

While there are many frameworks in the ecosystem, the combination of LangGraph and PydanticAI offers a unique balance of control and type safety.

LangGraph: The Orchestrator

LangGraph is designed for building stateful, multi-agent applications using graphs. Unlike standard linear chains, LangGraph allows for cycles. This is the fundamental requirement for self-healing. If a tool call fails, the graph can route the flow back to the agent node. It treats the agent's execution as a series of state updates, making it easy to implement "Human-in-the-loop" or "Validation-in-the-loop" patterns.

PydanticAI: The Validator

PydanticAI (built by the team behind Pydantic) brings the power of Python type hints and validation to LLM interactions. It ensures that the data passed into and out of your agents is typed and validated at runtime. When an LLM attempts to call a tool with an incorrect type, PydanticAI catches it immediately, providing a structured error that can be fed back into the self-healing loop.

Implementing the Workflow

Let's walk through a practical implementation of a self-healing tool-calling loop. Imagine an agent responsible for updating a database record. The database requires a specific UUID format and a strict set of enums.

1. Defining the State and Schema

First, we define our data structures. We use Pydantic to ensure the LLM knows exactly what we expect.

from pydantic import BaseModel, Field, UUID4 from typing import Literal, Annotated from typing_extensions import TypedDict class UserUpdate(BaseModel): user_id: UUID4 status: Literal["active", "suspended", "archived"] reason: str = Field(description="Brief explanation for the status change") class AgentState(TypedDict): messages: list attempts: int last_error: str | None

2. Setting up the PydanticAI Agent

PydanticAI allows us to wrap the LLM logic with built-in validation. If the LLM generates a UserUpdate that fails Pydantic's validation, we get a clear error message.

from pydantic_ai import Agent # Define the agent with the expected result type update_agent = Agent( 'openai:gpt-4o', result_type=UserUpdate, system_prompt="You are a database admin agent. Update user statuses accurately." )

3. Constructing the LangGraph

Now, we define the nodes of our graph. We'll have an agent_node that attempts the task and a validator_node that checks the work.

from langgraph.graph import StateGraph, END def call_agent(state: AgentState): # We pass the history and the last error back to the agent prompt = "Please process the update." if state['last_error']: prompt += f"\nYour previous attempt failed with: {state['last_error']}. Please fix it." # In a real scenario, we'd call the agent here # For brevity, assume result is the output of update_agent.run_sync() # result = update_agent.run_sync(prompt, message_history=state['messages']) return {"messages": [prompt], "attempts": state['attempts'] + 1} def validate_output(state: AgentState): # Business logic validation or schema check # If invalid, return the error; if valid, go to END if state['attempts'] < 3: # Example: simulate one failure return "retry" return "complete" workflow = StateGraph(AgentState) workflow.add_node("agent", call_agent) workflow.set_entry_point("agent") workflow.add_conditional_edges( "agent", validate_output, { "retry": "agent", "complete": END } ) app = workflow.compile()

The Power of the "Self-Correction" Prompt

The secret sauce of self-healing is the error feedback. When a validation fails—whether it's a Pydantic ValidationError or a failed API call—you shouldn't just tell the agent "Try again." You need to provide the specific traceback or validation error.

For example, if the LLM provides "status": "deleted" (which isn't in our Literal), PydanticAI will raise an error: 1 validation error for UserUpdate: status: input should be 'active', 'suspended' or 'archived'.

Feeding this specific string back into the next iteration of the graph allows the LLM to recognize its mistake. Modern models like GPT-4o or Claude 3.5 Sonnet are exceptionally good at fixing their own code or JSON when presented with a compiler error.

Advanced Pattern: Multi-Agent Validation

In more complex systems, the entity performing the task shouldn't be the one validating it. This is the "Checker-Producer" pattern.

  • The Producer: Uses PydanticAI to generate a structured plan or tool call.
  • The Checker: A separate node in LangGraph (potentially using a different, cheaper model) that critiques the output against edge cases.
  • The Feedback Loop: If the Checker finds an issue, it sends the report back to the Producer.

This separation of concerns reduces "sunk cost bias" in the LLM, where it tends to repeat the same mistake because it generated the logic itself.

Practical Considerations for Engineering Teams

Implementing these loops comes with trade-offs that senior engineers must manage:

1. Token Usage and Latency

Self-healing loops consume more tokens and increase latency because you're essentially running multiple LLM inferences for a single task. To mitigate this:

  • Set a max_attempts limit in your graph state to prevent infinite loops.
  • Use smaller, faster models for the initial validation and only escalate to larger models if correction fails.

2. Idempotency

Since a tool call might be retried multiple times within a loop, your tools must be idempotent. If an agent calls a charge_credit_card tool, fails at the validation step, and retries, you must ensure the card isn't charged twice. Use transaction IDs or idempotency keys in your tool definitions.

3. Observability

Standard logging isn't enough for cyclic graphs. You need to track the "path" the agent took. Tools like LangSmith or Arize Phoenix are essential here to visualize the graph execution and see exactly where the healing triggered. If you notice a specific tool is triggering the healing loop 40% of the time, it’s a signal that your system prompt or tool description needs refinement.

Conclusion: Building for Reliability

Self-healing agentic workflows shift the burden of reliability from the prompt engineer to the software architect. By using LangGraph to manage the cyclic state and PydanticAI to enforce data integrity, we create systems that are resilient to the inherent fuzziness of Large Language Models.

Actionable Takeaways:

  • Stop catching and crashing: Design your LangGraph nodes to return error states instead of raising exceptions.
  • Enforce schemas early: Use PydanticAI to wrap all tool calls and LLM outputs in strict Pydantic models.
  • Provide rich feedback: When an error occurs, pass the specific validation failure back to the LLM to facilitate accurate correction.
  • Monitor loop frequency: Treat the activation of a self-healing loop as a metric for prompt health and model performance.