Errors

from langgraph.graph import StateGraph
from langgraph.errors import GraphRecursionError

builder = StateGraph(State)
# ... add nodes and edges ...
graph = builder.compile()

try:
    # Increase the recursion limit
    result = graph.invoke(
        {"messages": [("user", "Hello, world!")]},
        {"recursion_limit": 1000}  # Allow up to 1000 steps
    )
except GraphRecursionError as e:
    print(f"Graph took too many steps: {e}")
    # Handle the error - perhaps the graph has an infinite loop

class State(TypedDict):
    messages: list
    iteration_count: int

def should_continue(state: State) -> str:
    # Prevent infinite loops by checking iteration count
    if state["iteration_count"] >= 10:
        return "end"
    if needs_more_processing(state):
        return "continue"
    return "end"

builder.add_conditional_edges(
    "process_node",
    should_continue,
    {
        "continue": "process_node",
        "end": END
    }
)

from typing import Annotated
import operator

class State(TypedDict):
    # This will cause InvalidUpdateError if multiple nodes update it
    value: str
    # This is safe for concurrent updates (uses operator.add)
    items: Annotated[list, operator.add]

# Problem: Both nodes run in parallel and update 'value'
def node_a(state: State):
    return {"value": "A"}

def node_b(state: State):
    return {"value": "B"}  # Conflict!

# Solution 1: Use a reducer
class State(TypedDict):
    value: Annotated[str, lambda x, y: y]  # Last write wins

# Solution 2: Don't run nodes in parallel for shared channels
builder.add_edge("node_a", "node_b")  # Sequential

# Solution 3: Use different channels
class State(TypedDict):
    value_a: str
    value_b: str

# Problem: Wrong return type
def bad_node(state: State):
    return "just a string"  # InvalidUpdateError!

# Solution: Return a dict matching the state schema
def good_node(state: State):
    return {"value": "correct format"}

# Problem: Missing required fields
class State(TypedDict):
    required_field: str
    optional_field: str | None

def incomplete_node(state: State):
    return {}  # May cause InvalidUpdateError

def complete_node(state: State):
    return {"required_field": "value"}

from langgraph.types import interrupt, Command

def my_node(state: State):
    # Use interrupt() instead of raising GraphInterrupt
    user_input = interrupt("What should I do next?")
    return {"action": user_input}

# Old (deprecated)
from langgraph.errors import NodeInterrupt

def old_node(state: State):
    raise NodeInterrupt("Need user input")

# New (recommended)
from langgraph.types import interrupt

def new_node(state: State):
    user_input = interrupt("Need user input")
    return {"input": user_input}

# This will raise EmptyInputError
graph.invoke(None)
graph.invoke({})

# Valid inputs
graph.invoke({"messages": []})
graph.invoke({"value": 0})

from langgraph.errors import EmptyInputError

try:
    result = graph.invoke(user_input)
except EmptyInputError:
    print("Please provide valid input")
    result = graph.invoke({"messages": ["default message"]})

class State(TypedDict):
    optional_field: str | None
    required_field: str

def my_node(state: State):
    # This might raise EmptyChannelError if optional_field was never set
    value = state["optional_field"]
    return {"result": value}

# Solution: Provide defaults or check existence
def safe_node(state: State):
    value = state.get("optional_field", "default")
    return {"result": value}

from langgraph.errors import GraphRecursionError, InvalidUpdateError

def run_graph_safely(graph, input_data, config=None):
    config = config or {}
    
    try:
        return graph.invoke(input_data, config)
    except GraphRecursionError:
        # Handle infinite loops
        print("Graph took too many steps, returning partial result")
        state = graph.get_state(config)
        return state.values
    except InvalidUpdateError as e:
        # Handle invalid updates
        print(f"Invalid update: {e}")
        return {"error": "Invalid state update"}
    except EmptyInputError:
        # Handle empty input
        return graph.invoke({"messages": []}, config)

def invoke_with_retry(graph, input_data, max_retries=3):
    recursion_limit = 25
    
    for attempt in range(max_retries):
        try:
            config = {"recursion_limit": recursion_limit}
            return graph.invoke(input_data, config)
        except GraphRecursionError:
            recursion_limit *= 2
            print(f"Retry {attempt + 1}: Increasing limit to {recursion_limit}")
    
    raise Exception("Graph failed after max retries")

from typing import TypedDict
from langgraph.errors import InvalidUpdateError

class State(TypedDict):
    value: int
    status: str

def validate_state(state: dict) -> dict:
    """Validate and clean state before processing."""
    if not isinstance(state.get("value"), int):
        raise InvalidUpdateError("'value' must be an integer")
    
    if state.get("status") not in ["pending", "complete"]:
        state["status"] = "pending"
    
    return state

def my_node(state: State):
    try:
        validated = validate_state(state)
        # Process validated state
        return {"value": validated["value"] + 1}
    except InvalidUpdateError as e:
        print(f"Validation error: {e}")
        return {"value": 0, "status": "error"}

import logging
from langgraph.errors import GraphRecursionError, InvalidUpdateError

logger = logging.getLogger(__name__)

def monitored_invoke(graph, input_data, config=None):
    try:
        logger.info(f"Starting graph execution with input: {input_data}")
        result = graph.invoke(input_data, config)
        logger.info(f"Graph completed successfully")
        return result
    
    except GraphRecursionError as e:
        logger.error(f"Graph recursion limit exceeded: {e}")
        raise
    
    except InvalidUpdateError as e:
        logger.error(f"Invalid state update: {e}")
        raise
    
    except Exception as e:
        logger.exception(f"Unexpected error during graph execution: {e}")
        raise

from langgraph.errors import create_error_message, ErrorCode

message = create_error_message(
    message="Graph exceeded maximum steps",
    error_code=ErrorCode.GRAPH_RECURSION_LIMIT
)

print(message)
# Output:
# Graph exceeded maximum steps
# For troubleshooting, visit: https://docs.langchain.com/oss/python/langgraph/errors/GRAPH_RECURSION_LIMIT

# For simple, short workflows
config = {"recursion_limit": 50}

# For complex workflows with many steps
config = {"recursion_limit": 500}

# For potentially infinite workflows (use with caution)
config = {"recursion_limit": 10000}

import operator
from typing import Annotated

class State(TypedDict):
    # Safe for concurrent updates
    messages: Annotated[list, operator.add]
    count: Annotated[int, operator.add]
    
    # Requires sequential updates or last-write-wins reducer
    status: str

def my_node(state: State) -> dict:
    result = complex_operation(state)
    
    # Validate before returning
    if not isinstance(result, dict):
        raise InvalidUpdateError("Node must return a dict")
    
    return result

from langgraph.types import interrupt, Command

def node_with_interrupt(state: State):
    # Request human input
    user_decision = interrupt("Approve this action?")
    
    if user_decision == "approved":
        return {"status": "approved"}
    else:
        return {"status": "rejected"}

# Resume with Command
command = Command(resume="approved")
graph.stream(command, config)