error-recovery-in-langgraph

Error Recovery in LangGraph

Error recovery in LangGraph enables workflows to handle failures gracefully through built-in retry policies, checkpointing for fault tolerance, and custom fallback patterns.

Required Imports

from typing import TypedDict, Optional, Any
from langchain_core.messages import BaseMessage, AIMessage, HumanMessage
from langgraph.graph import StateGraph, START, END
from langgraph.types import RetryPolicy
from langgraph.checkpoint.memory import InMemorySaver
import asyncio
# IMPORTANT: asyncio.timeout requires Python 3.11+
# For Python 3.10: pip install async-timeout && from async_timeout import timeout

---

RetryPolicy - Built-in Retry Mechanism (Recommended)

LangGraph provides a built-in RetryPolicy for automatic error recovery with exponential backoff and jitter. This is the primary and recommended approach for handling transient failures.

Basic Usage

from typing import TypedDict, Optional, Any
from langgraph.graph import StateGraph, START, END
from langgraph.types import RetryPolicy

class State(TypedDict):
    query: str
    result: Optional[str]
    error: Optional[str]

def api_call_node(state: State) -> dict[str, Any]:
    """Node that calls an external API.

    NOTE: Replace `fetch_from_api` with your actual API client.
    This is a placeholder function for demonstration.
    """
    def fetch_from_api(query: str) -> str:
        """Placeholder for your API client. Replace with actual implementation."""
        # Example: return requests.get(f"https://api.example.com/search?q={query}").text
        return f"Result for: {query}"

    response = fetch_from_api(state["query"])
    return {"result": response, "error": None}

# Create workflow with retry policy
workflow = StateGraph(State)
workflow.add_node(
    "api_call",
    api_call_node,
    retry_policy=RetryPolicy(max_attempts=5)
)
workflow.add_edge(START, "api_call")
workflow.add_edge("api_call", END)

graph = workflow.compile()

RetryPolicy Parameters