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

Error with Astream_events and AsyncSqliteSaver after migrating to v0.3 #235

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

Originally created by @mingxuan-he on GitHub (Sep 18, 2024).

Checked other resources

  • I added a very descriptive title to this issue.
  • I searched the LangGraph/LangChain documentation with the integrated search.
  • I used the GitHub search to find a similar question and didn't find it.
  • I am sure that this is a bug in LangGraph/LangChain rather than my code.
  • I am sure this is better as an issue rather than a GitHub discussion, since this is a LangGraph bug and not a design question.

Example Code

import sqlite3
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
from langgraph.graph import StateGraph

from typing import Literal
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.runnables import ConfigurableField
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from langgraph.prebuilt import ToolNode


@tool
def get_weather(city: Literal["nyc", "sf"]):
    """Use this to get weather information."""
    if city == "nyc":
        return "It might be cloudy in nyc"
    elif city == "sf":
        return "It's always sunny in sf"
    else:
        raise AssertionError("Unknown city")


tools = [get_weather]
model = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0)
final_model = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0)

model = model.bind_tools(tools)
# NOTE: this is where we're adding a tag that we'll can use later to filter the model stream events to only the model called in the final node.
# This is not necessary if you call a single LLM but might be important in case you call multiple models within the node and want to filter events
# from only one of them.
final_model = final_model.with_config(tags=["final_node"])
tool_node = ToolNode(tools=tools)
from typing import TypedDict, Annotated

from langgraph.graph import END, StateGraph, START
from langgraph.graph.message import MessagesState
from langchain_core.messages import BaseMessage, SystemMessage, HumanMessage


def should_continue(state: MessagesState) -> Literal["tools", "final"]:
    messages = state["messages"]
    last_message = messages[-1]
    # If the LLM makes a tool call, then we route to the "tools" node
    if last_message.tool_calls:
        return "tools"
    # Otherwise, we stop (reply to the user)
    return "final"


def call_model(state: MessagesState):
    messages = state["messages"]
    response = model.invoke(messages)
    # We return a list, because this will get added to the existing list
    return {"messages": [response]}


def call_final_model(state: MessagesState):
    messages = state["messages"]
    last_ai_message = messages[-1]
    response = final_model.invoke(
        [
            SystemMessage("Rewrite this in the voice of Al Roker"),
            HumanMessage(last_ai_message.content),
        ]
    )
    # overwrite the last AI message from the agent
    response.id = last_ai_message.id
    return {"messages": [response]}

workflow = StateGraph(MessagesState)

workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)
# add a separate final node
workflow.add_node("final", call_final_model)

workflow.add_edge(START, "agent")
workflow.add_conditional_edges(
    "agent",
    should_continue,
)

workflow.add_edge("tools", "agent")
workflow.add_edge("final", END)

# Memory
memory = AsyncSqliteSaver.from_conn_string("test.db")
app = workflow.compile(checkpointer=memory)

# Astream_events for token streaming
inputs = {"messages": [("human", "what's the weather in nyc?")]}
config = {"configurable": {"thread_id": "1"}}
async for event in app.astream_events(inputs, config, version="v2"):
    kind = event["event"]
    tags = event.get("tags", [])
    # filter on the langgraph node name
    if kind == "on_chat_model_stream" and event["metadata"].get("langgraph_node") == "final":
        data = event["data"]
        if data["chunk"].content:
            # Empty content in the context of OpenAI or Anthropic usually means
            # that the model is asking for a tool to be invoked.
            # So we only print non-empty content
            print(data["chunk"].content, end="|", flush=True)



### Error Message and Stack Trace (if applicable)

```shell
trace (sync version): https://smith.langchain.com/public/7742def4-35e5-4578-af7f-c1b418c53eee/r

AttributeError                            Traceback (most recent call last)
Cell In[36], line 3
      1 inputs = {"messages": [("human", "what's the weather in nyc?")]}
      2 config = {"configurable": {"thread_id": "1"}}
----> 3 async for event in app.astream_events(inputs, config, version="v2"):
      4     kind = event["event"]
      5     tags = event.get("tags", [])

File /opt/miniconda3/envs/yuichan/lib/python3.12/site-packages/langchain_core/runnables/base.py:1377, in Runnable.astream_events(self, input, config, version, include_names, include_types, include_tags, exclude_names, exclude_types, exclude_tags, **kwargs)
   1372     raise NotImplementedError(
   1373         'Only versions "v1" and "v2" of the schema is currently supported.'
   1374     )
   1376 async with aclosing(event_stream):
-> 1377     async for event in event_stream:
   1378         yield event

File /opt/miniconda3/envs/yuichan/lib/python3.12/site-packages/langchain_core/tracers/event_stream.py:1006, in _astream_events_implementation_v2(runnable, input, config, include_names, include_types, include_tags, exclude_names, exclude_types, exclude_tags, **kwargs)
   1004 # Await it anyway, to run any cleanup code, and propagate any exceptions
   1005 try:
-> 1006     await task
   1007 except asyncio.CancelledError:
   1008     pass

File /opt/miniconda3/envs/yuichan/lib/python3.12/site-packages/langchain_core/tracers/event_stream.py:966, in _astream_events_implementation_v2.<locals>.consume_astream()
...
--> 741     self.checkpointer_get_next_version = checkpointer.get_next_version
    742     self.checkpointer_put_writes = checkpointer.aput_writes
    743 else:

AttributeError: '_AsyncGeneratorContextManager' object has no attribute 'get_next_version'



### Description

I've been using AsyncSqliteSaver with astream_events in my langgraph app for a while. The recent upgrade to langchain v0.3 and langgraph v0.2 broke it. The message is not very descriptive but looks like a bug from the checkpointer class.
The code snippet above can reproduce the error.

### System Info

runtime
langchain_core_version: "0.3.1"

langchain_version: "0.3.0"

library: "langsmith"

platform: "Linux-6.5.0-1021-aws-x86_64-with-glibc2.36"

py_implementation: "CPython"

RENDER_GIT_COMMIT: "16d83d0c987788b782d3c8bb27c8ce34eaba911f"

runtime: "python"

runtime_version: "3.12.2"

sdk: "langsmith-py"

sdk_version: "0.1.122"
Originally created by @mingxuan-he on GitHub (Sep 18, 2024). ### Checked other resources - [X] I added a very descriptive title to this issue. - [X] I searched the [LangGraph](https://langchain-ai.github.io/langgraph/)/LangChain documentation with the integrated search. - [X] I used the GitHub search to find a similar question and didn't find it. - [X] I am sure that this is a bug in LangGraph/LangChain rather than my code. - [X] I am sure this is better as an issue [rather than a GitHub discussion](https://github.com/langchain-ai/langgraph/discussions/new/choose), since this is a LangGraph bug and not a design question. ### Example Code ```python import sqlite3 from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver from langgraph.graph import StateGraph from typing import Literal from langchain_community.tools.tavily_search import TavilySearchResults from langchain_core.runnables import ConfigurableField from langchain_core.tools import tool from langchain_openai import ChatOpenAI from langgraph.prebuilt import create_react_agent from langgraph.prebuilt import ToolNode @tool def get_weather(city: Literal["nyc", "sf"]): """Use this to get weather information.""" if city == "nyc": return "It might be cloudy in nyc" elif city == "sf": return "It's always sunny in sf" else: raise AssertionError("Unknown city") tools = [get_weather] model = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0) final_model = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0) model = model.bind_tools(tools) # NOTE: this is where we're adding a tag that we'll can use later to filter the model stream events to only the model called in the final node. # This is not necessary if you call a single LLM but might be important in case you call multiple models within the node and want to filter events # from only one of them. final_model = final_model.with_config(tags=["final_node"]) tool_node = ToolNode(tools=tools) from typing import TypedDict, Annotated from langgraph.graph import END, StateGraph, START from langgraph.graph.message import MessagesState from langchain_core.messages import BaseMessage, SystemMessage, HumanMessage def should_continue(state: MessagesState) -> Literal["tools", "final"]: messages = state["messages"] last_message = messages[-1] # If the LLM makes a tool call, then we route to the "tools" node if last_message.tool_calls: return "tools" # Otherwise, we stop (reply to the user) return "final" def call_model(state: MessagesState): messages = state["messages"] response = model.invoke(messages) # We return a list, because this will get added to the existing list return {"messages": [response]} def call_final_model(state: MessagesState): messages = state["messages"] last_ai_message = messages[-1] response = final_model.invoke( [ SystemMessage("Rewrite this in the voice of Al Roker"), HumanMessage(last_ai_message.content), ] ) # overwrite the last AI message from the agent response.id = last_ai_message.id return {"messages": [response]} workflow = StateGraph(MessagesState) workflow.add_node("agent", call_model) workflow.add_node("tools", tool_node) # add a separate final node workflow.add_node("final", call_final_model) workflow.add_edge(START, "agent") workflow.add_conditional_edges( "agent", should_continue, ) workflow.add_edge("tools", "agent") workflow.add_edge("final", END) # Memory memory = AsyncSqliteSaver.from_conn_string("test.db") app = workflow.compile(checkpointer=memory) # Astream_events for token streaming inputs = {"messages": [("human", "what's the weather in nyc?")]} config = {"configurable": {"thread_id": "1"}} async for event in app.astream_events(inputs, config, version="v2"): kind = event["event"] tags = event.get("tags", []) # filter on the langgraph node name if kind == "on_chat_model_stream" and event["metadata"].get("langgraph_node") == "final": data = event["data"] if data["chunk"].content: # Empty content in the context of OpenAI or Anthropic usually means # that the model is asking for a tool to be invoked. # So we only print non-empty content print(data["chunk"].content, end="|", flush=True) ``` ``` ### Error Message and Stack Trace (if applicable) ```shell trace (sync version): https://smith.langchain.com/public/7742def4-35e5-4578-af7f-c1b418c53eee/r AttributeError Traceback (most recent call last) Cell In[36], line 3 1 inputs = {"messages": [("human", "what's the weather in nyc?")]} 2 config = {"configurable": {"thread_id": "1"}} ----> 3 async for event in app.astream_events(inputs, config, version="v2"): 4 kind = event["event"] 5 tags = event.get("tags", []) File /opt/miniconda3/envs/yuichan/lib/python3.12/site-packages/langchain_core/runnables/base.py:1377, in Runnable.astream_events(self, input, config, version, include_names, include_types, include_tags, exclude_names, exclude_types, exclude_tags, **kwargs) 1372 raise NotImplementedError( 1373 'Only versions "v1" and "v2" of the schema is currently supported.' 1374 ) 1376 async with aclosing(event_stream): -> 1377 async for event in event_stream: 1378 yield event File /opt/miniconda3/envs/yuichan/lib/python3.12/site-packages/langchain_core/tracers/event_stream.py:1006, in _astream_events_implementation_v2(runnable, input, config, include_names, include_types, include_tags, exclude_names, exclude_types, exclude_tags, **kwargs) 1004 # Await it anyway, to run any cleanup code, and propagate any exceptions 1005 try: -> 1006 await task 1007 except asyncio.CancelledError: 1008 pass File /opt/miniconda3/envs/yuichan/lib/python3.12/site-packages/langchain_core/tracers/event_stream.py:966, in _astream_events_implementation_v2.<locals>.consume_astream() ... --> 741 self.checkpointer_get_next_version = checkpointer.get_next_version 742 self.checkpointer_put_writes = checkpointer.aput_writes 743 else: AttributeError: '_AsyncGeneratorContextManager' object has no attribute 'get_next_version' ``` ``` ### Description I've been using AsyncSqliteSaver with astream_events in my langgraph app for a while. The recent upgrade to langchain v0.3 and langgraph v0.2 broke it. The message is not very descriptive but looks like a bug from the checkpointer class. The code snippet above can reproduce the error. ### System Info runtime langchain_core_version: "0.3.1" langchain_version: "0.3.0" library: "langsmith" platform: "Linux-6.5.0-1021-aws-x86_64-with-glibc2.36" py_implementation: "CPython" RENDER_GIT_COMMIT: "16d83d0c987788b782d3c8bb27c8ce34eaba911f" runtime: "python" runtime_version: "3.12.2" sdk: "langsmith-py" sdk_version: "0.1.122"
yindo closed this issue 2026-02-20 17:33:17 -05:00
yindo commented 2026-02-20 17:33:19 -05:00
Author
Owner
Copy Link
Copy Source

@isahers1 commented on GitHub (Sep 18, 2024):

Thanks for the question! The issue is with how you are initializing the checkpointer. You can solve this in 2 ways:

Option 1:

async with AsyncSqliteSaver.from_conn_string("test.db") as memory:
    # The rest of your code

Option 2

conn = await aiosqlite.connect("test.db")
memory = AsyncSqliteSaver(conn)
# The rest of your code

Let me know if you still have issues!

@isahers1 commented on GitHub (Sep 18, 2024): Thanks for the question! The issue is with how you are initializing the checkpointer. You can solve this in 2 ways: **Option 1**: ``` async with AsyncSqliteSaver.from_conn_string("test.db") as memory: # The rest of your code ``` **Option 2** ``` conn = await aiosqlite.connect("test.db") memory = AsyncSqliteSaver(conn) # The rest of your code ``` Let me know if you still have issues!
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#235
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?