langchain-ai/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-07-15 12:35:43 -04:00
Code Issues 314 Packages Projects Releases 17 Wiki Activity

Subgraph not working parallelly #363

New Issue
Closed
opened 2026-02-20 17:38:32 -05:00 by yindo · 1 comment
yindo commented 2026-02-20 17:38:32 -05:00
Owner
Copy Link
Copy Source

Originally created by @owenzhw on GitHub (Dec 19, 2024).

Checked other resources

  • This is a bug, not a usage question. For questions, please use GitHub Discussions.
  • I added a clear and detailed title that summarizes the issue.
  • I read what a minimal reproducible example is (https://stackoverflow.com/help/minimal-reproducible-example).
  • I included a self-contained, minimal example that demonstrates the issue INCLUDING all the relevant imports. The code run AS IS to reproduce the issue.

Example Code

# this work as expected in parallel
import operator
import time
from functools import partial
from typing import Annotated

from langgraph.graph.state import END, START, CompiledStateGraph, StateGraph
from typing_extensions import TypedDict

class ChildState(TypedDict):
    question: str
    results: Annotated[list[str], operator.add]


class ChildOutputState(TypedDict):
    results: Annotated[list[str], operator.add]


class ParentState(TypedDict):
    question: str
    results: Annotated[list[str], operator.add]


def child_node(state: ChildState, i: int) -> dict:
    print(i,"sleeping,,,")
    time.sleep(i * 5)
    print(i,"awake,,,")
    return {"results": [f"{state['question']} FROM {i}"]}


def create_child(i: int) -> CompiledStateGraph:
    child = StateGraph(ChildState, output=ChildOutputState)
    child.add_node("sub_child1", partial(child_node, i=i))
    child.add_node("sub_child2", partial(child_node, i=i))
    child.add_edge(START, "sub_child1")
    child.add_edge("sub_child1", "sub_child2")
    child.add_edge("sub_child2", END)
    return child.compile()


async def parent_question_generation_node(state: dict) -> dict:
    return {"question": "what's up?"}


async def parent_sink_node(state: dict) -> dict:
    return {"results": state["results"]}


parent = StateGraph(ParentState)
parent.add_node("question_generation", parent_question_generation_node)
parent.add_node("child1", create_child(1))
parent.add_node("child2", create_child(3))
parent.add_node("sink", parent_sink_node)

parent.add_edge(START, "question_generation")
parent.add_edge("question_generation", "child1")
parent.add_edge("question_generation", "child2")
parent.add_edge("child1", "sink")
parent.add_edge("child2", "sink")
parent.add_edge("sink", END)

parent_graph = parent.compile()
state = ParentState(question="hi", results=[])


async def safe_invoke(parent_graph, state):
    result = await parent_graph.ainvoke(state)
    return result


import asyncio

async def main():
    result = await safe_invoke(parent_graph, state)

asyncio.run(main())



# this is not working as expected, it is more like running synchronously
import operator
import time
from functools import partial
from typing import Annotated

from langgraph.graph.state import END, START, CompiledStateGraph, StateGraph
from typing_extensions import TypedDict


class ChildState(TypedDict):
    question: str
    results: Annotated[list[str], operator.add]


class ChildOutputState(TypedDict):
    results: Annotated[list[str], operator.add]


class ParentState(TypedDict):
    question: str
    results: Annotated[list[str], operator.add]


async def child_node(state: ChildState, i: int) -> dict:
    print(i,"sleeping,,,")
    time.sleep(i * 5)
    print(i,"awake,,,")
    return {"results": [f"{state['question']} FROM {i}"]}


def create_child(i: int) -> CompiledStateGraph:
    child = StateGraph(ChildState, output=ChildOutputState)
    child.add_node("sub_child1", partial(child_node, i=i))
    child.add_node("sub_child2", partial(child_node, i=i))
    child.add_edge(START, "sub_child1")
    child.add_edge("sub_child1", "sub_child2")
    child.add_edge("sub_child2", END)
    return child.compile()


async def parent_question_generation_node(state: dict) -> dict:
    return {"question": "what's up?"}


async def parent_sink_node(state: dict) -> dict:
    return {"results": state["results"]}


parent = StateGraph(ParentState)
parent.add_node("question_generation", parent_question_generation_node)
parent.add_node("child1", create_child(1))
parent.add_node("child2", create_child(3))
parent.add_node("sink", parent_sink_node)

parent.add_edge(START, "question_generation")
parent.add_edge("question_generation", "child1")
parent.add_edge("question_generation", "child2")
parent.add_edge("child1", "sink")
parent.add_edge("child2", "sink")
parent.add_edge("sink", END)

parent_graph = parent.compile()
state = ParentState(question="hi", results=[])


async def safe_invoke(parent_graph, state):
    result = await parent_graph.ainvoke(state)
    return result


import asyncio

async def main():
    result = await safe_invoke(parent_graph, state)
    # print(result)

asyncio.run(main())

Error Message and Stack Trace (if applicable)

No response

Description

I'm trying to use LangGraph to implement a parallel running subgraph.

However, when subgraph include "async def" the graph is more like running synchronously, not in parallel.

In the first script, the output is like:

13 sleeping,,,
 sleeping,,,
1 awake,,,
1 sleeping,,,
1 awake,,,
3 awake,,,
3 sleeping,,,
3 awake,,,

The second script looks like this:

1 sleeping,,,
1 awake,,,
3 sleeping,,,
3 awake,,,
1 sleeping,,,
1 awake,,,
3 sleeping,,,
3 awake,,,

This does not like working in parallel.

System Info

System Information

OS: Darwin
OS Version: Darwin Kernel Version 22.6.0: Fri Sep 15 13:41:28 PDT 2023; root:xnu-8796.141.3.700.8~1/RELEASE_ARM64_T6000
Python Version: 3.11.11 (main, Dec 11 2024, 10:25:04) [Clang 14.0.6 ]

Package Information

langchain_core: 0.3.27
langchain: 0.3.13
langchain_community: 0.3.12
langsmith: 0.1.147
langchain_openai: 0.2.13
langchain_text_splitters: 0.3.3
langgraph_sdk: 0.1.48

Optional packages not installed

langserve

Other Dependencies

aiohttp: 3.11.11
async-timeout: Installed. No version info available.
dataclasses-json: 0.6.7
httpx: 0.27.0
httpx-sse: 0.4.0
jsonpatch: 1.33
langsmith-pyo3: Installed. No version info available.
numpy: 1.26.4
openai: 1.58.1
orjson: 3.10.7
packaging: 24.2
pydantic: 2.8.2
pydantic-settings: 2.7.0
PyYAML: 6.0.2
requests: 2.32.3
requests-toolbelt: 1.0.0
SQLAlchemy: 2.0.36
tenacity: 9.0.0
tiktoken: 0.8.0
typing-extensions: 4.11.0

Originally created by @owenzhw on GitHub (Dec 19, 2024). ### Checked other resources - [X] This is a bug, not a usage question. For questions, please use GitHub Discussions. - [X] I added a clear and detailed title that summarizes the issue. - [X] I read what a minimal reproducible example is (https://stackoverflow.com/help/minimal-reproducible-example). - [X] I included a self-contained, minimal example that demonstrates the issue INCLUDING all the relevant imports. The code run AS IS to reproduce the issue. ### Example Code ```python # this work as expected in parallel import operator import time from functools import partial from typing import Annotated from langgraph.graph.state import END, START, CompiledStateGraph, StateGraph from typing_extensions import TypedDict class ChildState(TypedDict): question: str results: Annotated[list[str], operator.add] class ChildOutputState(TypedDict): results: Annotated[list[str], operator.add] class ParentState(TypedDict): question: str results: Annotated[list[str], operator.add] def child_node(state: ChildState, i: int) -> dict: print(i,"sleeping,,,") time.sleep(i * 5) print(i,"awake,,,") return {"results": [f"{state['question']} FROM {i}"]} def create_child(i: int) -> CompiledStateGraph: child = StateGraph(ChildState, output=ChildOutputState) child.add_node("sub_child1", partial(child_node, i=i)) child.add_node("sub_child2", partial(child_node, i=i)) child.add_edge(START, "sub_child1") child.add_edge("sub_child1", "sub_child2") child.add_edge("sub_child2", END) return child.compile() async def parent_question_generation_node(state: dict) -> dict: return {"question": "what's up?"} async def parent_sink_node(state: dict) -> dict: return {"results": state["results"]} parent = StateGraph(ParentState) parent.add_node("question_generation", parent_question_generation_node) parent.add_node("child1", create_child(1)) parent.add_node("child2", create_child(3)) parent.add_node("sink", parent_sink_node) parent.add_edge(START, "question_generation") parent.add_edge("question_generation", "child1") parent.add_edge("question_generation", "child2") parent.add_edge("child1", "sink") parent.add_edge("child2", "sink") parent.add_edge("sink", END) parent_graph = parent.compile() state = ParentState(question="hi", results=[]) async def safe_invoke(parent_graph, state): result = await parent_graph.ainvoke(state) return result import asyncio async def main(): result = await safe_invoke(parent_graph, state) asyncio.run(main()) # this is not working as expected, it is more like running synchronously import operator import time from functools import partial from typing import Annotated from langgraph.graph.state import END, START, CompiledStateGraph, StateGraph from typing_extensions import TypedDict class ChildState(TypedDict): question: str results: Annotated[list[str], operator.add] class ChildOutputState(TypedDict): results: Annotated[list[str], operator.add] class ParentState(TypedDict): question: str results: Annotated[list[str], operator.add] async def child_node(state: ChildState, i: int) -> dict: print(i,"sleeping,,,") time.sleep(i * 5) print(i,"awake,,,") return {"results": [f"{state['question']} FROM {i}"]} def create_child(i: int) -> CompiledStateGraph: child = StateGraph(ChildState, output=ChildOutputState) child.add_node("sub_child1", partial(child_node, i=i)) child.add_node("sub_child2", partial(child_node, i=i)) child.add_edge(START, "sub_child1") child.add_edge("sub_child1", "sub_child2") child.add_edge("sub_child2", END) return child.compile() async def parent_question_generation_node(state: dict) -> dict: return {"question": "what's up?"} async def parent_sink_node(state: dict) -> dict: return {"results": state["results"]} parent = StateGraph(ParentState) parent.add_node("question_generation", parent_question_generation_node) parent.add_node("child1", create_child(1)) parent.add_node("child2", create_child(3)) parent.add_node("sink", parent_sink_node) parent.add_edge(START, "question_generation") parent.add_edge("question_generation", "child1") parent.add_edge("question_generation", "child2") parent.add_edge("child1", "sink") parent.add_edge("child2", "sink") parent.add_edge("sink", END) parent_graph = parent.compile() state = ParentState(question="hi", results=[]) async def safe_invoke(parent_graph, state): result = await parent_graph.ainvoke(state) return result import asyncio async def main(): result = await safe_invoke(parent_graph, state) # print(result) asyncio.run(main()) ``` ### Error Message and Stack Trace (if applicable) _No response_ ### Description I'm trying to use LangGraph to implement a parallel running subgraph. However, when subgraph include "async def" the graph is more like running synchronously, not in parallel. In the first script, the output is like: ``` 13 sleeping,,, sleeping,,, 1 awake,,, 1 sleeping,,, 1 awake,,, 3 awake,,, 3 sleeping,,, 3 awake,,, ``` The second script looks like this: ``` 1 sleeping,,, 1 awake,,, 3 sleeping,,, 3 awake,,, 1 sleeping,,, 1 awake,,, 3 sleeping,,, 3 awake,,, ``` This does not like working in parallel. ### System Info System Information ------------------ > OS: Darwin > OS Version: Darwin Kernel Version 22.6.0: Fri Sep 15 13:41:28 PDT 2023; root:xnu-8796.141.3.700.8~1/RELEASE_ARM64_T6000 > Python Version: 3.11.11 (main, Dec 11 2024, 10:25:04) [Clang 14.0.6 ] Package Information ------------------- > langchain_core: 0.3.27 > langchain: 0.3.13 > langchain_community: 0.3.12 > langsmith: 0.1.147 > langchain_openai: 0.2.13 > langchain_text_splitters: 0.3.3 > langgraph_sdk: 0.1.48 Optional packages not installed ------------------------------- > langserve Other Dependencies ------------------ > aiohttp: 3.11.11 > async-timeout: Installed. No version info available. > dataclasses-json: 0.6.7 > httpx: 0.27.0 > httpx-sse: 0.4.0 > jsonpatch: 1.33 > langsmith-pyo3: Installed. No version info available. > numpy: 1.26.4 > openai: 1.58.1 > orjson: 3.10.7 > packaging: 24.2 > pydantic: 2.8.2 > pydantic-settings: 2.7.0 > PyYAML: 6.0.2 > requests: 2.32.3 > requests-toolbelt: 1.0.0 > SQLAlchemy: 2.0.36 > tenacity: 9.0.0 > tiktoken: 0.8.0 > typing-extensions: 4.11.0
yindo closed this issue 2026-02-20 17:38:33 -05:00
yindo commented 2026-02-20 17:38:34 -05:00
Author
Owner
Copy Link
Copy Source

@owenzhw commented on GitHub (Dec 19, 2024):

Oh I forgot to change sleep to asyncio.sleep. Sorry for the inconvenience....

@owenzhw commented on GitHub (Dec 19, 2024): Oh I forgot to change sleep to asyncio.sleep. Sorry for the inconvenience....
Sign in to join this conversation.
No Branch/Tag Specified
Branches Tags
Labels
Clear labels
agents
awaiting author revision
bug
change
chat models
checkpointer
ci
code quality
code quality
dependencies
documentation
drawing
duplicate
enhancement
good first issue
help wanted
invalid
maintainer
meta
missing information
not maintained by langchain
open-swe-auto
open-swe-max
open-swe-max-auto
pending
performance
pull-request
Mirrored from GitHub Pull Request
 question
requires design
sdk
server
stale
studio
tests
type checking
usage question
v0.5
v1.1
No Label
Milestone
No items
No Milestone
Projects
Clear projects
No project
Assignees
Clear assignees
yindo
No Assignees
1 Participants
Notifications
Due Date
No due date set.
Dependencies

No dependencies set.

Reference: langchain-ai/langgraph#363
Delete Branch "%!s()"

Deleting a branch is permanent. Although the deleted branch may continue to exist for a short time before it actually gets removed, it CANNOT be undone in most cases. Continue?