Help! The langgraph node is executed repeatedly.

[Anonymous-H]

I built a graph to test langgraph nodes, but during testing, I noticed that Node D was executed repeatedly.
Here's my graph construction code where Node D gets executed multiple times, and I don't understand why.

import time
import asyncio
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator

from tools.show_tools import print_png


# Define the state for our graph
class AgentState(TypedDict):
    messages: Annotated[List[str], operator.add]
    run_f_branch: bool # Added to control F branch

# Define the nodes
def node_a(state: AgentState):
    print("--- Starting Node A ---")
    time.sleep(1) # Simulate some work
    print("--- Finished Node A ---")
    # Decide to run F branch (e.g., always True for this demo, can be changed in initial_state for testing)
    # Or, for more dynamic control, this could be based on input or other state
    return {"messages": ["A completed"], "run_f_branch": True}

async def node_b(state: AgentState):
    print("--- Starting Node B (Parallel) ---")
    await asyncio.sleep(5) # Simulate 5 seconds of work
    print("--- Finished Node B (Parallel) ---")
    return {"messages": ["B completed"]}

async def node_c(state: AgentState):
    print("--- Starting Node C (Parallel) ---")
    await asyncio.sleep(10) # Simulate 10 seconds of work
    print("--- Finished Node C (Parallel) ---")
    return {"messages": ["C completed"]}

async def node_f(state: AgentState): # New node F
    print("--- Starting Node F (Conditional & Parallel with B, C) ---")
    await asyncio.sleep(3) # Simulate 3 seconds of work
    print("--- Finished Node F ---")
    return {"messages": ["F completed"]}

async def node_g(state: AgentState): # New node F
    print("--- Starting Node G ---")
    await asyncio.sleep(7) # Simulate 3 seconds of work
    print("--- Finished Node G ---")
    return {"messages": ["G completed"]}

def node_d_prep_no_f(state: AgentState): # New dummy node for when F is skipped
    print("--- Node D_PREP_NO_F (F was skipped by conditional edge from A) ---")
    return {} # No message change, just a pass-through state

def node_d(state: AgentState):
    print("--- Starting Node D (After B and C) ---")
    # This node runs after B and C. 
    # LangGraph's add_edge from a list of nodes to a single node ensures D waits for all predecessors.
    time.sleep(1) # Simulate some work
    print("--- Finished Node D ---")
    return {"messages": ["D completed"]}

def node_e(state: AgentState):
    print("--- Starting Node E (Serial) ---")
    time.sleep(1) # Simulate some work
    print("--- Finished Node E ---")
    return {"messages": ["E completed"]}

# Conditional routing function after Node A for F branch
def route_after_a_for_f(state: AgentState):
    # Access 'run_f_branch' from the state. Ensure it's present.
    # Node A is responsible for setting this value in the state.
    if state.get("run_f_branch", False):
        print("--- Routing decision after A: Path to F will be taken. ---")
        return "F"
    else:
        print("--- Routing decision after A: Path to F will be skipped. Routing to D_PREP_NO_F. ---")
        return "D_PREP_NO_F"

# Define the graph
workflow = StateGraph(AgentState)

# Add nodes
workflow.add_node("A", node_a)
workflow.add_node("B", node_b)
workflow.add_node("G", node_g)
workflow.add_node("C", node_c)
workflow.add_node("D", node_d)
workflow.add_node("E", node_e)
workflow.add_node("F", node_f) # Add node F
workflow.add_node("D_PREP_NO_F", node_d_prep_no_f) # Add dummy node for skipping F

# Set up the edges
workflow.set_entry_point("A")
workflow.add_edge("A", "B") # A always triggers B
workflow.add_edge("B", "G") # A always triggers B
workflow.add_edge("A", "C") # A always triggers C

# Conditional edge from A for the F branch
workflow.add_conditional_edges(
    "A",
    route_after_a_for_f,
    {
        "F": "F",  # If route_after_a_for_f returns "F", go to node F
        "D_PREP_NO_F": "D_PREP_NO_F"  # If it returns "D_PREP_NO_F", go to this node
    }
)

# F (if run) goes to D
workflow.add_edge("F", "D")
# D_PREP_NO_F (if F is skipped) also goes to D, ensuring D always has this path resolved
workflow.add_edge("D_PREP_NO_F", "D")

# B and C run in parallel, then both go to D
# LangGraph handles this by waiting for all direct predecessors of D to complete.
# D will wait for B, C, and (F or D_PREP_NO_F).
workflow.add_edge(["G", "C"], "D")

workflow.add_edge("D", "E")
workflow.add_edge("E", END)

# Compile the graph
app = workflow.compile()

print_png(app)

async def main():
    print("Starting graph execution...")
    # To test the F-skipped path, set run_f_branch to False in node_a or here.
    # For this demo, node_a sets it to True. You can override it here for testing:
    # initial_state = {"messages": [], "run_f_branch": False} 
    initial_state = {"messages": [], "run_f_branch": True} # Default to run_f_branch being True as per node_a's current logic
    async for event in app.astream(initial_state):
        for key, value in event.items():
            print(f"Node: {key}, Output: {value}")
        print("---")
    print("Graph execution finished.")

if __name__ == "__main__":
    # To run async code in a script
    import platform
    if platform.system() == 'Windows':
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    asyncio.run(main())

Here's the printed log:

Starting graph execution...

--- Starting Node A ---

--- Finished Node A ---

--- Routing decision after A: Path to F will be taken. ---

Node: A, Output: {'messages': ['A completed'], 'run_f_branch': True}

---

--- Starting Node B (Parallel) ---

--- Starting Node C (Parallel) ---

--- Starting Node F (Conditional & Parallel with B, C) ---

--- Finished Node F ---

Node: F, Output: {'messages': ['F completed']}

---

--- Finished Node B (Parallel) ---

Node: B, Output: {'messages': ['B completed']}

---

--- Finished Node C (Parallel) ---

Node: C, Output: {'messages': ['C completed']}

---

--- Starting Node G ---

--- Starting Node D (After B and C) ---

--- Finished Node D ---

Node: D, Output: {'messages': ['D completed']}

---

--- Finished Node G ---

Node: G, Output: {'messages': ['G completed']}

---

--- Starting Node D (After B and C) ---

--- Starting Node E (Serial) ---

--- Finished Node D ---

--- Finished Node E ---

Node: D, Output: {'messages': ['D completed']}

---

Node: E, Output: {'messages': ['E completed']}

---

--- Starting Node E (Serial) ---

--- Finished Node E ---

Node: E, Output: {'messages': ['E completed']}

---

Graph execution finished.