Langgraph: Задача Parallel MapReduce, реализованная Langgraph, имеет неверный заказ на выполнение ⇐ Python

[Гость]

фон < /h1>
Отношения узла на графике следующие:
Введите описание изображения здесь < /p>

• Параллельная работа узлов B и C < /li>
  нет блокировки. />Question:
  
  Why should b2 be executed after C?
• Why can't b execute b2 immediately after execution?

ENV

python: 3.13.5 < /li>
langchain: 0.3.27 < /li>
langchain-core: 0.3.75 < /li>
0.2.3 < /li>
langgraph-checkpoint: 2.1.1 < /li>
< /ul>
code < /h1>
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())
< /code>
Примечание < /h2>
Я попробовал операции в комментариях < /p>

Отменить операцию отложения узла d < /li>
Узел B2 и узлы C напрямую связаны с конец < /li>
*li>. /> Все печати одинаковы < /strong> < /p>
output < /h1>
====== 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']
< /code>
Пожалуйста, помогите мне решить. Как получить правильную последовательность выполнения? < /P>
Я попробовал операции в комментариях < /p>

Отменить операцию отсрочки узла d < /li>
Узел B2 и узлы C напрямую связаны с End < /li>
phori>. Отпечатки одинаковы ， все неправильно


Подробнее здесь: https://stackoverflow.com/questions/797 ... -incorrect