It should be clearer how converging edges should be defined

[shaunhegarty]

I was having an issue with a node getting executed twice unexpectedly.

After combing the examples I spotted that in one specific case an array of nodes is provided to the add_edge method, without explanation. Doing it this way solved my issue.

The examples in the rest of that page don't need it, and I have no idea why. Some explanation would be appreciated!

I managed to boil it all down to this example code. My old approach is commented out. The old approach will print "Finish" twice, the new only once, as intended.

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
from langgraph.graph import StateGraph

class GraphState(TypedDict):       
    messages: Annotated[list, add_messages]

def build_agent():
    workflow = StateGraph(GraphState)         
    workflow.add_node("A", lambda x: print("Start"))
    workflow.add_node("B", lambda x: print("Path I - Step 1"))    
    workflow.add_node("C", lambda x: print("Path I - Step 2"))
    workflow.add_node("D", lambda x: print("Path II"))    
    workflow.add_node("E", lambda x: print("Finish"))
    
    workflow.set_entry_point("A")
    
    workflow.add_edge("A", "B")
    workflow.add_edge("B", "C")    
    workflow.add_edge("A", "D")

    # Defining the Edges individually causes "E" to be executed multiple times
    # workflow.add_edge("C", "E")        
    # workflow.add_edge("D", "E")

    # Need to do this to prevent "E" getting executed multiple times
    workflow.add_edge(["C", "D"], "E")        
    
    workflow.set_finish_point("E")

    return workflow.compile()

from IPython.display import display, Image
agent = build_agent()
display(Image(agent.get_graph().draw_mermaid_png()))
agent.invoke({"messages": []})

[hinthornw]

Thanks for flagging, and thanks for the great code snippet - makes it easy to see what's you're running into. Will try to add more conceptual guide on branching / state updates in the docs beyond the current how-to on branching.

Reason for the behavior is that nodes are executed in consecutive supersteps.

In original case:

Step 1: A
Step 2: B D
Step 3: C E
Step 4: E

In the new case (I'll double check my work later)

Step 1: A
Step 2: B D
Step 3: C # No edge from just D, so the message from D stays
Step 4: E # Now both edges (C and D) have sent outbound messages, so E can execute

[david1542]

Wow, I found myself debugging this myself. Thanks for creating this issue.

Is there any way to add this important piece of knowledge to the docs?

[snopoke]

As far as I can tell there's no way to accomplish this when conditional edges are in play:

builder = StateGraph(State)
builder.add_node("a", ReturnNodeValue("I'm A"))
builder.add_edge(START, "a")
builder.add_node("b", ReturnNodeValue("I'm B"))
builder.add_node("b2a", ReturnNodeValue("I'm B2a"))
builder.add_node("b2b", ReturnNodeValue("I'm B2b"))
builder.add_node("c", ReturnNodeValue("I'm C"))
builder.add_edge("a", "b")
builder.add_edge("a", "c")
builder.add_edge(["b2a", "b2b"], "c")
builder.add_edge("c", END)


def router(state: State) -> Sequence[str]:
    return state["which"]


builder.add_conditional_edges(
    "b",
    router,
    ["b2a", "b2b"],
)

builder.set_entry_point("a")
builder.set_finish_point("c")
graph = builder.compile()
print(graph.invoke({"aggregate": [], "which": "b2a"}))

>> Adding I'm A to []
>> Adding I'm B to ["I'm A"]
>> Adding I'm C to ["I'm A"]
>> Adding I'm B2a to ["I'm A", "I'm B", "I'm C"]
>> {'aggregate': ["I'm A", "I'm B", "I'm C", "I'm B2a"], 'which': 'b2a'}

In this example I would expect C to only execute with inputs from A, B2a and B2b but instead it executes only with input from A.

Making this change:

builder.add_edge(["a", "b2a", "b2b"], "c")

results in C not being executed at all:

Adding I'm A to []
Adding I'm B to ["I'm A"]
Adding I'm B2a to ["I'm A", "I'm B"]
{'aggregate': ["I'm A", "I'm B", "I'm B2a"], 'which': 'b2a'}