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

StreamWriter does not stream chunk in async node #614

New Issue
Closed
opened 2026-02-20 17:40:58 -05:00 by yindo · 3 comments
yindo commented 2026-02-20 17:40:58 -05:00
Owner
Copy Link
Copy Source

Originally created by @sundaraa-deshaw on GitHub (May 9, 2025).

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

from langgraph.graph import MessagesState
from langgraph.types import StreamWriter
from langchain_core.messages import AIMessageChunk

# 1. Define the node, that takes a StreamWriter instance and generates incremental chunks.

async def streaming_node(state, writer: StreamWriter):
    last_message = state['messages'][-1]
    for word in last_message.content.split(' '):
        time.sleep(0.5)
        # Write the chunks to stream
        writer(AIMessageChunk(content=word))

    # Return final message as required.
    return {
        "messages": AIMessage("I am done with streaming the message")
    }

# 2. Build the graph with the node
graph = StateGraph(MessagesState)
graph.add_node("hello", streaming_node)
graph.add_edge(START, "hello")
graph.add_edge("hello", END)
app = graph.compile()

# 3. Invoke the graph and stream the output
question = "hi how are you"

async for stream_mode, chunk in app.astream(
    {"messages": [HumanMessage(question)]},
    # Pass 'custom' to get custom non-LLM chunks, and 'values' to get the final message.
    stream_mode=['custom', 'values']
):
    if stream_mode == 'custom':
        if isinstance(chunk, AIMessageChunk) and chunk.content:
            print(chunk.content, end="|")
    elif stream_mode == 'values':
        for message in chunk['messages'][::-1]:
            if isinstance(message, AIMessage):
                message.pretty_print()
                break

Error Message and Stack Trace (if applicable)

(does not stream the chunks in async mode)

Description

I am using Python 3.11 and trying to use an async node to stream non-LLM chunks using StreamWriter. While this works for sync nodes, this does not stream the chunk for async ones. It only renders the final response.

async node stream writer (does not stream)
Image

sync node stream writer (streams)
Image

System Info

System Information

OS: Linux
OS Version: #1 SMP Thu Feb 6 21:20:51 EST 2025
Python Version: 3.11.8 (main, Sep 17 2024, 14:28:20) [GCC 10.3.1 20210422 (Red Hat 10.3.1-1)]

Package Information

langchain_core: 0.3.48
langchain: 0.3.14
langchain_community: 0.3.14
langsmith: 0.2.10
langchain_anthropic: 0.2.4
langchain_cli: 0.0.31
langchain_experimental: 0.3.4
langchain_openai: 0.2.14
langchain_text_splitters: 0.3.5
langchainplus_sdk: 0.0.21
langgraph_gen: 0.0.6
langgraph_reflection: 0.0.1
langgraph_sdk: 0.1.60
langgraph_supervisor: 0.0.15
langgraph_swarm: 0.0.8

Optional packages not installed

langserve

Other Dependencies

aiohttp: 3.8.4
anthropic: 0.30.0
async-timeout: 4.0.3
dataclasses-json: 0.6.3
defusedxml: 0.7.1
gitpython: 3.1.40
gritql: Installed. No version info available.
httpx: 0.27.0
httpx-sse: 0.4.0
jinja2>=3.1.5: Installed. No version info available.
jsonpatch<2.0,>=1.33: Installed. No version info available.
langchain-core<0.4.0,>=0.3.40: Installed. No version info available.
langchain>=0.1.0: Installed. No version info available.
langgraph: 0.3.21
langgraph-prebuilt<0.2.0,>=0.1.7: Installed. No version info available.
langgraph<0.4.0,>=0.3.5: Installed. No version info available.
langgraph>=0.2.74: Installed. No version info available.
langserve[all]: Installed. No version info available.
langsmith-pyo3: Installed. No version info available.
langsmith<0.4,>=0.1.125: Installed. No version info available.
mypy>=1.8.0: Installed. No version info available.
numpy: 1.26.4+deshaw5
openai: 1.59.7
orjson: 3.10.16+deshaw1
packaging<25,>=23.2: Installed. No version info available.
pydantic: 2.9.2
pydantic-settings: 2.6.0
pydantic<3.0.0,>=2.5.2;: Installed. No version info available.
pydantic<3.0.0,>=2.7.4;: Installed. No version info available.
PyYAML: 6.0.2
PyYAML>=5.3: Installed. No version info available.
pyyaml>=6.0.2: Installed. No version info available.
requests: 2.31.0
requests-toolbelt: 1.0.0
SQLAlchemy: 1.4.50
tenacity: 8.2.3
tenacity!=8.4.0,<10.0.0,>=8.1.0: Installed. No version info available.
tiktoken: 0.7.0
tomlkit: 0.12.3
typer[all]: Installed. No version info available.
typing-extensions>=4.7: Installed. No version info available.
uvicorn: 0.26.0
zstandard: 0.22.0

Originally created by @sundaraa-deshaw on GitHub (May 9, 2025). ### 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 from langgraph.graph import MessagesState from langgraph.types import StreamWriter from langchain_core.messages import AIMessageChunk # 1. Define the node, that takes a StreamWriter instance and generates incremental chunks. async def streaming_node(state, writer: StreamWriter): last_message = state['messages'][-1] for word in last_message.content.split(' '): time.sleep(0.5) # Write the chunks to stream writer(AIMessageChunk(content=word)) # Return final message as required. return { "messages": AIMessage("I am done with streaming the message") } # 2. Build the graph with the node graph = StateGraph(MessagesState) graph.add_node("hello", streaming_node) graph.add_edge(START, "hello") graph.add_edge("hello", END) app = graph.compile() # 3. Invoke the graph and stream the output question = "hi how are you" async for stream_mode, chunk in app.astream( {"messages": [HumanMessage(question)]}, # Pass 'custom' to get custom non-LLM chunks, and 'values' to get the final message. stream_mode=['custom', 'values'] ): if stream_mode == 'custom': if isinstance(chunk, AIMessageChunk) and chunk.content: print(chunk.content, end="|") elif stream_mode == 'values': for message in chunk['messages'][::-1]: if isinstance(message, AIMessage): message.pretty_print() break ``` ### Error Message and Stack Trace (if applicable) ```shell (does not stream the chunks in async mode) ``` ### Description I am using Python 3.11 and trying to use an async node to stream non-LLM chunks using StreamWriter. While this works for sync nodes, this does not stream the chunk for async ones. It only renders the final response. async node stream writer (does not stream) ![Image](https://github.com/user-attachments/assets/9515df07-e6aa-45e7-bc31-f87613488e39) sync node stream writer (streams) ![Image](https://github.com/user-attachments/assets/5ecb1d33-025c-41e9-8853-263a376ec2fa) ### System Info System Information ------------------ > OS: Linux > OS Version: #1 SMP Thu Feb 6 21:20:51 EST 2025 > Python Version: 3.11.8 (main, Sep 17 2024, 14:28:20) [GCC 10.3.1 20210422 (Red Hat 10.3.1-1)] Package Information ------------------- > langchain_core: 0.3.48 > langchain: 0.3.14 > langchain_community: 0.3.14 > langsmith: 0.2.10 > langchain_anthropic: 0.2.4 > langchain_cli: 0.0.31 > langchain_experimental: 0.3.4 > langchain_openai: 0.2.14 > langchain_text_splitters: 0.3.5 > langchainplus_sdk: 0.0.21 > langgraph_gen: 0.0.6 > langgraph_reflection: 0.0.1 > langgraph_sdk: 0.1.60 > langgraph_supervisor: 0.0.15 > langgraph_swarm: 0.0.8 Optional packages not installed ------------------------------- > langserve Other Dependencies ------------------ > aiohttp: 3.8.4 > anthropic: 0.30.0 > async-timeout: 4.0.3 > dataclasses-json: 0.6.3 > defusedxml: 0.7.1 > gitpython: 3.1.40 > gritql: Installed. No version info available. > httpx: 0.27.0 > httpx-sse: 0.4.0 > jinja2>=3.1.5: Installed. No version info available. > jsonpatch<2.0,>=1.33: Installed. No version info available. > langchain-core<0.4.0,>=0.3.40: Installed. No version info available. > langchain>=0.1.0: Installed. No version info available. > langgraph: 0.3.21 > langgraph-prebuilt<0.2.0,>=0.1.7: Installed. No version info available. > langgraph<0.4.0,>=0.3.5: Installed. No version info available. > langgraph>=0.2.74: Installed. No version info available. > langserve[all]: Installed. No version info available. > langsmith-pyo3: Installed. No version info available. > langsmith<0.4,>=0.1.125: Installed. No version info available. > mypy>=1.8.0: Installed. No version info available. > numpy: 1.26.4+deshaw5 > openai: 1.59.7 > orjson: 3.10.16+deshaw1 > packaging<25,>=23.2: Installed. No version info available. > pydantic: 2.9.2 > pydantic-settings: 2.6.0 > pydantic<3.0.0,>=2.5.2;: Installed. No version info available. > pydantic<3.0.0,>=2.7.4;: Installed. No version info available. > PyYAML: 6.0.2 > PyYAML>=5.3: Installed. No version info available. > pyyaml>=6.0.2: Installed. No version info available. > requests: 2.31.0 > requests-toolbelt: 1.0.0 > SQLAlchemy: 1.4.50 > tenacity: 8.2.3 > tenacity!=8.4.0,<10.0.0,>=8.1.0: Installed. No version info available. > tiktoken: 0.7.0 > tomlkit: 0.12.3 > typer[all]: Installed. No version info available. > typing-extensions>=4.7: Installed. No version info available. > uvicorn: 0.26.0 > zstandard: 0.22.0
yindo closed this issue 2026-02-20 17:40:58 -05:00
yindo commented 2026-02-20 17:40:58 -05:00
Author
Owner
Copy Link
Copy Source

@hinthornw commented on GitHub (May 9, 2025):

Thanks for writing this up! If you replace time.sleep with await asyncio.sleep(0.5), it does seem to stream token by token on my side. I believe that python just needs control to be yielded for the token yielding to be scheduled today

Does that match your experience?

@hinthornw commented on GitHub (May 9, 2025): Thanks for writing this up! If you replace `time.sleep` with `await asyncio.sleep(0.5)`, it does seem to stream token by token on my side. I believe that python just needs control to be yielded for the token yielding to be scheduled today Does that match your experience?
yindo commented 2026-02-20 17:40:58 -05:00
Author
Owner
Copy Link
Copy Source

@sundaraa-deshaw commented on GitHub (May 9, 2025):

Thanks for the quick response, the sleep was merely to simulate a pause between non-LLM tokens. Replacing it with await asyncio.sleep works. I am wondering what happens for writes to the stream amidst sync workloads inside async nodes, then.

@sundaraa-deshaw commented on GitHub (May 9, 2025): Thanks for the quick response, the sleep was merely to simulate a pause between non-LLM tokens. Replacing it with `await asyncio.sleep` works. I am wondering what happens for writes to the stream amidst sync workloads inside async nodes, then.
yindo commented 2026-02-20 17:40:58 -05:00
Author
Owner
Copy Link
Copy Source

@nfcampos commented on GitHub (May 23, 2025):

Sync workloads in an async function block the event loop, so nothing else happens in your application while that's happening. This is behavior implemented in Python, not something we can change here. If you want to run a sync workload in an async function you should use asyncio.to_thread to send it to a background thread, ie. my_func(123) becomes await asyncio.to_thread(my_func, 123)

@nfcampos commented on GitHub (May 23, 2025): Sync workloads in an async function block the event loop, so nothing else happens in your application while that's happening. This is behavior implemented in Python, not something we can change here. If you want to run a sync workload in an async function you should use asyncio.to_thread to send it to a background thread, ie. `my_func(123)` becomes `await asyncio.to_thread(my_func, 123)`
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#614
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?