[Python] The parallel mapReduce task implemented by langGraph has an incorrect execution order

Background


The node relationships in the graph are as follows: enter image description here

• Parallel operation of nodes b and c
• No blocking in nodes b and b2
• Node C has sleep

Question:

• Why should b2 be executed after C?
• Why can't b execute b2 immediately after execution?

ENV

• python: 3.13.5
• langchain: 0.3.27
• langchain-core: 0.3.75
• langgraph:0.6.6
• langgraph-prebuilt: 0.6.4
• langgraph-sdk: 0.2.3
• langgraph-checkpoint: 2.1.1

Code


import operator
from typing import Annotated, Any
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
import threading
import time
import asyncio

class State(TypedDict):
# The operator.add reducer fn makes this append-only
aggregate: Annotated[list, operator.add]

async def a(state: State):
print(f'====== thread_id: {threading.get_ident()} time:{time.strftime("%H:%M:%S", time.localtime())} Adding "A" to {state["aggregate"]}')
return {"aggregate": ["A"]}

async def b(state: State):
#time.sleep(3)
print(f'====== thread_id: {threading.get_ident()} time:{time.strftime("%H:%M:%S", time.localtime())} Adding "B" to {state["aggregate"]}')
return {"aggregate": ["B"]}

async def b_2(state: State):
print(f'====== thread_id: {threading.get_ident()} time:{time.strftime("%H:%M:%S", time.localtime())} Adding "B_2" to {state["aggregate"]}')
return {"aggregate": ["B_2"]}

async def c(state: State):
await asyncio.sleep(3) # sleep operation
print(f'====== thread_id: {threading.get_ident()} time:{time.strftime("%H:%M:%S", time.localtime())} Adding "C" to {state["aggregate"]}')
return {"aggregate": ["C"]}

async def d(state: State):
print(f'====== thread_id: {threading.get_ident()} time:{time.strftime("%H:%M:%S", time.localtime())} Adding "D" to {state["aggregate"]}')
return {"aggregate": ["D"]}


async def condition_a2bc(state: State):
return [Send("b", state), Send("c", state)]

builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)
builder.add_node(b_2)
builder.add_node(c)
builder.add_node(d, defer=True) # defer=True
#builder.add_node(d)
builder.set_entry_point("a")
builder.add_conditional_edges("a", condition_a2bc, ["b", "c"])
builder.add_edge("b", "b_2")
builder.add_edge("b_2", "d")
builder.add_edge("c", "d")
#builder.add_edge(["b_2", "c"], "d")
builder.add_edge("d", END)
#builder.add_edge("b_2", END)
#builder.add_edge("c", END)
graph = builder.compile()

async def main():
async for msg, metadata in graph.astream({"aggregate": []}, stream_mode="messages"):
print(msg)
print(metadata)

if __name__ == "__main__":
asyncio.run(main())

NOTE


I tried the operations in the comments

• Cancel the defer operation of node d
• Node b2 and Node c are directly connected to the end
• ...

All the prints are the same

Output


====== thread_id: 8715454528 time:21:47:56 Adding "A" to []
====== thread_id: 8715454528 time:21:47:56 Adding "B" to ['A']
====== thread_id: 8715454528 time:21:47:59 Adding "C" to ['A']
====== thread_id: 8715454528 time:21:47:59 Adding "B_2" to ['A', 'B', 'C']
====== thread_id: 8715454528 time:21:47:59 Adding "D" to ['A', 'B', 'C', 'B_2']


Please help me solve it. How to obtain the correct execution sequence?

I tried the operations in the comments

• Cancel the defer operation of node d
• Node b2 and Node c are directly connected to the end
• ...

But the prints are the same，all wrong

Continue reading...