I Can't Handle Errors by Both Updating the State and Short-Circuiting the Graph in LangGraph

OSS Product Help LangGraph
1

TL;DR,
Is there a way to catch node errors, while adding these errors to the state and short-circuit the graph.
I should add these errors to the state because I want to provide it to my end-users to explain why the app isn’t working. I also want to short circuit the graph because my nodes are dependent to each other.

A perfect solution would be like:
Whenever an exception is caught, skip everything in the graph and go to the handle_error node, then END.
I then could be able to both update the state and short-circuit the graph. Which’d be awesome. And I only can imagine this solution with something like Command(jump_to=“handle_error”).
If it aligns with maintainers’ view, I’d be glad to work on that.
If there is a much more easy way you know that I can solve this problem, I’d love to hear from you.

Look at this basic example:

from langgraph.graph import START, END, StateGraph, MessagesState

class State(MessagesState):

	text: str
 

def node_a(state: State) -> State:
	modified_text = state["text"] + "a"
	
	return {
	
		"text": modified_text,
		
		"messages": [
		
			{
			
				"type": "tool",
				
				"content": "node_a successfully modified the text",
				
				"tool_call_id": "1",
				
				"ok": True,
			
			}
		
		],
	
	}

  
  
def node_b(state: State) -> State:

	modified_text = state["text"] + "b"
	
	return {
		
		"text": modified_text,
		
		"messages": [
		
			{
			
				"type": "tool",
				
				"content": "node_b successfully modified the text",
				
				"tool_call_id": "2",
				
				"ok": True,
			
			}
		
		],
	
	}

  
  

def node_c(state: State) -> State:

	modified_text = state["text"] + "c"
	
	return {
	
		"text": modified_text,
		
		"messages": [
		
			{
			
				"type": "tool",
				
				"content": "node c successfully modified the text",
				
				"tool_call_id": "3",
				
				"ok": True,
			
			}
		
		],
	
	}

  

graph = StateGraph(State)

graph.add_node("node_a", node_a)

graph.add_node("node_b", node_b)

graph.add_node("node_c", node_c)

graph.add_edge(START, "node_a")

graph.add_edge("node_a", "node_b")

graph.add_edge("node_b", "node_c")

  

print(graph.compile().invoke({"text": ""}))

# Output:
"""

{
	"messages": [
	
		ToolMessage(
		
		content="node_a successfully modified the text",
		
		id="130bc0d7-ded7-4983-906a-7ba2fdf83755",
		
		tool_call_id="1",
		
		),
		
		ToolMessage(
		
		content="node_b successfully modified the text",
		
		id="697287c7-ba4f-4afd-a4f7-a3c9a5be5d17",
		
		tool_call_id="2",
		
		),
		
		ToolMessage(
		
		content="node c successfully modified the text",
		
		id="f7cda129-ae86-4c42-957a-fbe7221e6f68",
		
		tool_call_id="3",
		
		),
		
		],
	
	"text": "abc",
}

"""

Assume that every node here represents some logic that generates/retrieves necessary information to give the end-user what they want. For example, you can think that node_a generates excel files, node_b performs fuzzy match and node_c retrieves google drive files from drive api. What I’m trying to say is, these nodes can be used as tools without being given to a react agent.

Now consider we need to catch the exceptions thrown from these nodes, and append a ToolMessage to the messages with ok arg is set to False, for further logging or to explain the end-user why our app isn’t working. Since the nodes are sequential and -assume that- dependent on each other, we want our graph to short circuit if an exception occurs -because in our graph, “ac” means nothing without “b”, you got the idea-

We can’t throw exceptions from our graph and catch them in our code which calls invoke on the graph, because we still want to append to the messages channel. Writing to a checkpointer from outside of a graph is tricky, unnecessary and kills the elegance of LangGraph.

Adding an error handler decorator is one intuitive way to go, as discussed in the #6170
and #6571.

I learned from #6571 that Command() adds a dynamic edge, static edges still execute. So trying to make Command() to override static edges is not a valid solution.

I tried a lot of things, but I can’t seem to find a way -without hacky solutions- to implement the logic I explained. I think if Command had a parameter like jump_to, it’d be pretty easy to do. I’d create an error handler decorator, if I catch an exception, I’d goto handle_error node, by updating the state.

2

Hi, as far as I understand, you should use Command instead of a static edge. Returning a Command from a node lets you decide the next step dynamically at runtime while also updating the state (for example, propagating error information).

The main thing to be careful about is that if you define a static edge that determines where a node continues, that edge can override or conflict with Command.goto. For nodes that return a Command, it’s usually better not to declare a fixed outgoing edge and let the command control the flow instead.
Reference: https://github.com/langchain-ai/langgraph/issues/5829

You should also type the node’s return value as a Command and explicitly list the possible destinations.

def example_node(state: SomeState) -> Command[Literal["happy_path_node", "__end__"]]:
    try:
        data = fetch_data()
        return Command(
            goto="happy_path_node",
            update={"data": data},
        )
    except Exception as e:
        return Command(
            goto="__end__",
            update={"error": str(e)},
        )

graph = StateGraph(SomeState)

graph.add_node("first_node", first_node)
graph.add_node("example", example_node)
graph.add_node("happy_path_node", happy_path_node)

graph.set_entry_point("first_node")
graph.add_edge("first_node", "example")
graph.set_finish_point("happy_path_node")

In this setup, the transition out of example_node is fully controlled by the returned Command, without a static edge forcing a specific path during graph compilation.

You can also see this pattern used for error handling in the public LangChain Deep Research Agent repository:
https://github.com/langchain-ai/open_deep_research/blob/main/src/open_deep_research/deep_researcher.py

2 Likes
3

Thank you for taking the time to answer. I understand your approach, which is the only documented way to use Command: as a conditional edge with update superpower.

But I think this method is only useful if you need to control specific nodes’ errors. It’d be pretty messy, if not impossible, to add error handling for each node in a graph that consists of tens of nodes with this approach.

In the public LangChain Deep Research Agent repository, only one node supervisor_tools used this method.

I already have a working logic that uses edges and conditional edges. I don’t want to rewrite it from scratch in a much more complex way so that I can add error handling.
What I want to do is to say “preserve my logic, and wrap all of my nodes with this error handler decorator”. I am currently able to do it, by writing a decorator like:

def error_handling_decorator(f):
    @wraps(f)
    async def wrapper(state, *args, **kwargs):
        try:
            return await f(state, *args, **kwargs)
        except Exception as e:
           # handle error, then update state
            return {"a_state_channel": "update"}

    return wrapper

What is missing is I need to short circuit the graph in case of error.
And we know that the following won’t work because Command adds dynamic edges, static edges will execute anyway.

def error_handling_decorator(f):
    @wraps(f)
    async def wrapper(state, *args, **kwargs):
        try:
            return await f(state, *args, **kwargs)
        except Exception as e:
           # handle error, then update state
            return Command(goto="__end__", update={"a_state_channel": "update"})

    return wrapper

That’s exactly where I’m stuck.

I have two rough ideas:
First, we can define a handle_error node that directly connects to the END.
and do either

def handle_error(state: State):
    # handle error
    return {"a_state_channel": "update"}

def error_handling_decorator(f):
    @wraps(f)
    async def wrapper(state, *args, **kwargs):
        try:
            return await f(state, *args, **kwargs)
        except Exception as e:
           # send to handle error node, without running any node in between.
            return Command(goto="handle_error", jump_to_goto=True)

    return wrapper

This requires to implement jump_to_goto arg to Command, I am not sure if it is technically possible.
or

def handle_error(state: State):
    # handle error conditional edge
    if "error" in state:
        return Command(goto=END, update={"a_state_channel": "update"})

    return None

# this adds a global conditional edge that is placed at the end of every node.
graph.compile(global_conditional_edge=handle_error)
# a structure like start -> a -> b -> c -> end would be:
# start -> a -> handle_error - > b -> handle_error - > c -> handle_error -> end

This requires to implement global_conditional_edge arg to graph.compile, I am not sure if it is technically possible, too.

Again, if you have any solutions to my problem I’d love to hear.