Parallel node execution process slow?

[tianshangwuyun]

import asyncio
from typing import Annotated, Any

from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END
from langgraph.types import Command,interrupt
from langgraph.checkpoint.memory import MemorySaver
checkpoint = MemorySaver()


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

def one(state:State):
    print("1")
    return {"aggregate":["one"]}

def two(state:State):
    print("2")
    return {"aggregate":["two"]}

async def three(state:State):
    print("3")
    await asyncio.sleep(3)
    print("3 end")
    return {"aggregate":["three"]}

def four(state:State):
    print("4")
    # res =interrupt("4")
    # print(f"res:{res}")
    return {"aggregate":["four"]}

async def five(state:State):
    await asyncio.sleep(4)
    print("5")
    await asyncio.sleep(4)
    print("5 end")
    return {"aggregate":["five"]}

def six(state:State):
    print("6")
    return {"aggregate":["six"]}

def seven(state:State):
    print("7")
    return {"aggregate":["seven"]}

def eight(state:State):
    print("8")
    return {"aggregate":["eight"]}

def nine(state:State):
    print("9")
    return {"aggregate":["nine"]}

def ten(state:State):
    print("10")
    return {"aggregate":["ten"]}


builder = StateGraph(State)

builder.add_node(one)
builder.add_node(two)
builder.add_node(three)

builder.add_node(five)
builder.add_node(seven)
builder.add_node(nine)

builder.add_node(four)
builder.add_node(six)

builder.add_node(eight)

builder.add_node(ten)






builder.add_edge(START,"one")
builder.add_edge("one","three")
builder.add_edge("one","two")


builder.add_edge("three","five")
builder.add_edge("five","seven")
builder.add_edge("seven","nine")
builder.add_edge("nine","ten")

builder.add_edge("two","four")

builder.add_edge("four","six")

builder.add_edge("six","eight")

builder.add_edge("eight","ten")

builder.add_edge("ten",END)
graph = builder.compile(checkpointer=checkpoint)
config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}
async def main():
    res = await graph.ainvoke({"aggregate": [],"num":1},config=config)
    print(f"res:{res}")
    await asyncio.sleep(5)    
    

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

Nodes 2 and 3 are on the same superstep They execute concurrently, with other nodes executing alternately. This results in the execution of 2, 4, 6, and 8, which could have been done quickly, but may have been slow due to prolonged code running in 5, 7, and 9

[pedrojrv]

Same observation here