persistence_postgres error #268

Closed
opened 2026-02-20 17:34:44 -05:00 by yindo · 15 comments
Owner

Originally created by @naturesh on GitHub (Oct 7, 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

https://langchain-ai.github.io/langgraph/how-tos/persistence_postgres/ 

My code is all the same as above and the difference is that I used tool_calling_agent(langgraph) not create_react_agent,

Error Message and Stack Trace (if applicable)

---> 26 for event in graph.stream(
     27     *request,stream_mode="values",
     28 ):
     29     event["messages"][-1].pretty_print()
     31 pool.close()

File /Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/site-packages/langgraph/pregel/__init__.py:1248, in Pregel.stream(self, input, config, stream_mode, output_keys, interrupt_before, interrupt_after, debug, subgraphs)
   1244 if "custom" in stream_modes:
   1245     config[CONF][CONFIG_KEY_STREAM_WRITER] = lambda c: stream.put(
   1246         ((), "custom", c)
   1247     )
-> 1248 with SyncPregelLoop(
   1249     input,
   1250     stream=StreamProtocol(stream.put, stream_modes),
   1251     config=config,
   1252     store=store,
   1253     checkpointer=checkpointer,
   1254     nodes=self.nodes,
   1255     specs=self.channels,
   1256     output_keys=output_keys,
   1257     stream_keys=self.stream_channels_asis,
   1258     debug=debug,
   1259 ) as loop:
   1260     # create runner
   1261     runner = PregelRunner(
   1262         submit=loop.submit,
   1263         put_writes=loop.put_writes,
   1264     )
   1265     # enable subgraph streaming

File /Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/site-packages/langgraph/pregel/loop.py:727, in SyncPregelLoop.__enter__(self)
    725         raise CheckpointNotLatest
    726 elif self.checkpointer:
--> 727     saved = self.checkpointer.get_tuple(self.checkpoint_config)
    728 else:
    729     saved = None

File /Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/site-packages/langgraph/checkpoint/postgres/__init__.py:228, in PostgresSaver.get_tuple(self, config)
    225     where = "WHERE thread_id = %s AND checkpoint_ns = %s ORDER BY checkpoint_id DESC LIMIT 1"
    227 with self._cursor() as cur:
--> 228     cur.execute(
    229         self.SELECT_SQL + where,
    230         args,
    231         binary=True,
    232     )
    234     for value in cur:
    235         return CheckpointTuple(
    236             {
    237                 "configurable": {
   (...)
    260             self._load_writes(value["pending_writes"]),
    261         )

File /Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/site-packages/psycopg/cursor.py:97, in Cursor.execute(self, query, params, prepare, binary)
     93         self._conn.wait(
     94             self._execute_gen(query, params, prepare=prepare, binary=binary)
     95         )
     96 except e._NO_TRACEBACK as ex:
---> 97     raise ex.with_traceback(None)
     98 return self

DuplicatePreparedStatement: prepared statement "_pg3_0" already exists

Description

https://langchain-ai.github.io/langgraph/how-tos/persistence_postgres/

kwargs={
"autocommit": True,
"prepare_threshold": 0,
}

In , when threshold is a large number, the typical llm agent outputs well, but errors occur after using the tool.

If threshold 0, llm agent output will not come out and an error will occur.

System Info

langgraph==0.2.34
langgraph-checkpoint==2.0.0
langgraph-checkpoint-postgres==2.0.0

I am using Postgres in supabase project

Originally created by @naturesh on GitHub (Oct 7, 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 https://langchain-ai.github.io/langgraph/how-tos/persistence_postgres/ My code is all the same as above and the difference is that I used tool_calling_agent(langgraph) not create_react_agent, ``` ### Error Message and Stack Trace (if applicable) ```shell ---> 26 for event in graph.stream( 27 *request,stream_mode="values", 28 ): 29 event["messages"][-1].pretty_print() 31 pool.close() File /Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/site-packages/langgraph/pregel/__init__.py:1248, in Pregel.stream(self, input, config, stream_mode, output_keys, interrupt_before, interrupt_after, debug, subgraphs) 1244 if "custom" in stream_modes: 1245 config[CONF][CONFIG_KEY_STREAM_WRITER] = lambda c: stream.put( 1246 ((), "custom", c) 1247 ) -> 1248 with SyncPregelLoop( 1249 input, 1250 stream=StreamProtocol(stream.put, stream_modes), 1251 config=config, 1252 store=store, 1253 checkpointer=checkpointer, 1254 nodes=self.nodes, 1255 specs=self.channels, 1256 output_keys=output_keys, 1257 stream_keys=self.stream_channels_asis, 1258 debug=debug, 1259 ) as loop: 1260 # create runner 1261 runner = PregelRunner( 1262 submit=loop.submit, 1263 put_writes=loop.put_writes, 1264 ) 1265 # enable subgraph streaming File /Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/site-packages/langgraph/pregel/loop.py:727, in SyncPregelLoop.__enter__(self) 725 raise CheckpointNotLatest 726 elif self.checkpointer: --> 727 saved = self.checkpointer.get_tuple(self.checkpoint_config) 728 else: 729 saved = None File /Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/site-packages/langgraph/checkpoint/postgres/__init__.py:228, in PostgresSaver.get_tuple(self, config) 225 where = "WHERE thread_id = %s AND checkpoint_ns = %s ORDER BY checkpoint_id DESC LIMIT 1" 227 with self._cursor() as cur: --> 228 cur.execute( 229 self.SELECT_SQL + where, 230 args, 231 binary=True, 232 ) 234 for value in cur: 235 return CheckpointTuple( 236 { 237 "configurable": { (...) 260 self._load_writes(value["pending_writes"]), 261 ) File /Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/site-packages/psycopg/cursor.py:97, in Cursor.execute(self, query, params, prepare, binary) 93 self._conn.wait( 94 self._execute_gen(query, params, prepare=prepare, binary=binary) 95 ) 96 except e._NO_TRACEBACK as ex: ---> 97 raise ex.with_traceback(None) 98 return self DuplicatePreparedStatement: prepared statement "_pg3_0" already exists ``` ### Description https://langchain-ai.github.io/langgraph/how-tos/persistence_postgres/ kwargs={ "autocommit": True, "prepare_threshold": 0, } In , when threshold is a large number, the typical llm agent outputs well, but errors occur after using the tool. If threshold 0, llm agent output will not come out and an error will occur. ### System Info langgraph==0.2.34 langgraph-checkpoint==2.0.0 langgraph-checkpoint-postgres==2.0.0 I am using Postgres in supabase project
yindo closed this issue 2026-02-20 17:34:44 -05:00
Author
Owner

@vbarda commented on GitHub (Oct 7, 2024):

@naturesh could you please provide a full reproducible code snippet? what is tool_calling_agent?

@vbarda commented on GitHub (Oct 7, 2024): @naturesh could you please provide a full reproducible code snippet? what is `tool_calling_agent`?
Author
Owner

@naturesh commented on GitHub (Oct 8, 2024):

from typing import Annotated, List

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import Runnable, RunnableConfig
from langgraph.graph.message import AnyMessage, add_messages
from typing_extensions import TypedDict

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

class Assistant:
    def __init__(self, runnable: Runnable):
        """
        Initialize the Assistant with a runnable object.

        Args:
            runnable (Runnable): The runnable instance to invoke.
        """
        self.runnable = runnable

    def __call__(self, state: State, config: RunnableConfig):
        """
        Call method to invoke the LLM and handle its responses.
        Re-prompt the assistant if the response is not a tool call or meaningful text.

        Args:
            state (State): The current state containing messages.
            config (RunnableConfig): The configuration for the runnable.

        Returns:
            dict: The final state containing the updated messages.
        """
        while True:
            result = self.runnable.invoke(state)  # Invoke the LLM
            if not result.tool_calls and (
                not result.content
                or isinstance(result.content, list)
                and not result.content[0].get("text")
            ):
                messages = state["messages"] + [("user", "Respond with a real output.")]
                state = {**state, "messages": messages}
            else:
                break
        return {"messages": result}


# Create the primary assistant prompt template
primary_assistant_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You are a helpful assistant tasked with answering user questions. "
            "You have access to two tools: retrieve_documents and web_search. "
            "For any user questions about LLM agents, use the retrieve_documents tool to get information for a vectorstore. "
            "For any other questions, such as questions about current events, use the web_search tool to get information from the web. ",
        ),
        ("placeholder", "{messages}"),
    ]
)

# Prompt our LLM and bind tools
assistant_runnable = primary_assistant_prompt | llm.bind_tools(tools)

from langchain_core.messages import ToolMessage
from langchain_core.runnables import RunnableLambda
from langgraph.prebuilt import ToolNode


def create_tool_node_with_fallback(tools: list) -> dict:
    return ToolNode(tools).with_fallbacks(
        [RunnableLambda(handle_tool_error)], exception_key="error"
    )


def handle_tool_error(state: State) -> dict:
    error = state.get("error")
    tool_calls = state["messages"][-1].tool_calls
    return {
        "messages": [
            ToolMessage(
                content=f"Error: {repr(error)}\n please fix your mistakes.",
                tool_call_id=tc["id"],
            )
            for tc in tool_calls
        ]
    }


from IPython.display import Image, display
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.prebuilt import tools_condition

# Graph
builder = StateGraph(State)

# Define nodes: these do the work
builder.add_node("assistant", Assistant(assistant_runnable))
builder.add_node("tools", create_tool_node_with_fallback(tools))

# Define edges: these determine how the control flow moves
builder.add_edge(START, "assistant")
builder.add_conditional_edges(
    "assistant",
    # If the latest message (result) from assistant is a tool call -> tools_condition routes to tools
    # If the latest message (result) from assistant is a not a tool call -> tools_condition routes to END
    tools_condition,
)
builder.add_edge("tools", "assistant")

# The checkpointer lets the graph persist its state
memory = MemorySaver()
react_graph = builder.compile(checkpointer=memory)

# Show
display(Image(react_graph.get_graph(xray=True).draw_mermaid_png()))

this is my code ( langgraph tutorial example )

@naturesh commented on GitHub (Oct 8, 2024): ```python from typing import Annotated, List from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import Runnable, RunnableConfig from langgraph.graph.message import AnyMessage, add_messages from typing_extensions import TypedDict class State(TypedDict): messages: Annotated[list[AnyMessage], add_messages] class Assistant: def __init__(self, runnable: Runnable): """ Initialize the Assistant with a runnable object. Args: runnable (Runnable): The runnable instance to invoke. """ self.runnable = runnable def __call__(self, state: State, config: RunnableConfig): """ Call method to invoke the LLM and handle its responses. Re-prompt the assistant if the response is not a tool call or meaningful text. Args: state (State): The current state containing messages. config (RunnableConfig): The configuration for the runnable. Returns: dict: The final state containing the updated messages. """ while True: result = self.runnable.invoke(state) # Invoke the LLM if not result.tool_calls and ( not result.content or isinstance(result.content, list) and not result.content[0].get("text") ): messages = state["messages"] + [("user", "Respond with a real output.")] state = {**state, "messages": messages} else: break return {"messages": result} # Create the primary assistant prompt template primary_assistant_prompt = ChatPromptTemplate.from_messages( [ ( "system", "You are a helpful assistant tasked with answering user questions. " "You have access to two tools: retrieve_documents and web_search. " "For any user questions about LLM agents, use the retrieve_documents tool to get information for a vectorstore. " "For any other questions, such as questions about current events, use the web_search tool to get information from the web. ", ), ("placeholder", "{messages}"), ] ) # Prompt our LLM and bind tools assistant_runnable = primary_assistant_prompt | llm.bind_tools(tools) from langchain_core.messages import ToolMessage from langchain_core.runnables import RunnableLambda from langgraph.prebuilt import ToolNode def create_tool_node_with_fallback(tools: list) -> dict: return ToolNode(tools).with_fallbacks( [RunnableLambda(handle_tool_error)], exception_key="error" ) def handle_tool_error(state: State) -> dict: error = state.get("error") tool_calls = state["messages"][-1].tool_calls return { "messages": [ ToolMessage( content=f"Error: {repr(error)}\n please fix your mistakes.", tool_call_id=tc["id"], ) for tc in tool_calls ] } from IPython.display import Image, display from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import END, START, StateGraph from langgraph.prebuilt import tools_condition # Graph builder = StateGraph(State) # Define nodes: these do the work builder.add_node("assistant", Assistant(assistant_runnable)) builder.add_node("tools", create_tool_node_with_fallback(tools)) # Define edges: these determine how the control flow moves builder.add_edge(START, "assistant") builder.add_conditional_edges( "assistant", # If the latest message (result) from assistant is a tool call -> tools_condition routes to tools # If the latest message (result) from assistant is a not a tool call -> tools_condition routes to END tools_condition, ) builder.add_edge("tools", "assistant") # The checkpointer lets the graph persist its state memory = MemorySaver() react_graph = builder.compile(checkpointer=memory) # Show display(Image(react_graph.get_graph(xray=True).draw_mermaid_png())) ``` this is my code ( langgraph tutorial example )
Author
Owner

@vbarda commented on GitHub (Oct 8, 2024):

@naturesh i cannot reproduce the issue when invoking the graph from your example with postgres checkpointer, e.g.

config = {"configurable": {"thread_id": "2"}}
DB_URI = "postgres://postgres:postgres@localhost:5442/postgres?sslmode=disable"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    react_graph = builder.compile(checkpointer=checkpointer)
    res = react_graph.invoke({"messages": [("user", "hi")]}, config)

have you called PostgresSaver.setup()? also, which postgres version are you running locally? we recommend postgres >= 16

also, i would recommend rerunning a notebook in a fresh virtual environment and see if the error is still there. let me know if you're still running into issues after that

@vbarda commented on GitHub (Oct 8, 2024): @naturesh i cannot reproduce the issue when invoking the graph from your example with postgres checkpointer, e.g. ```python config = {"configurable": {"thread_id": "2"}} DB_URI = "postgres://postgres:postgres@localhost:5442/postgres?sslmode=disable" with PostgresSaver.from_conn_string(DB_URI) as checkpointer: react_graph = builder.compile(checkpointer=checkpointer) res = react_graph.invoke({"messages": [("user", "hi")]}, config) ``` have you called `PostgresSaver.setup()`? also, which postgres version are you running locally? we recommend postgres >= 16 also, i would recommend rerunning a notebook in a fresh virtual environment and see if the error is still there. let me know if you're still running into issues after that
Author
Owner

@naturesh commented on GitHub (Oct 9, 2024):

To be exact, I am using the PostgreSQL api endpoint provided by supabase. I was able to see that setup() automatically creates four tables.

@naturesh commented on GitHub (Oct 9, 2024): To be exact, I am using the PostgreSQL api endpoint provided by supabase. I was able to see that setup() automatically creates four tables.
Author
Owner

@vbarda commented on GitHub (Oct 11, 2024):

@naturesh just researched this error more -- you need to create connection manually with connection kwargs and pass the connection to PostgresSaver (see this section: https://langchain-ai.github.io/langgraph/how-tos/persistence_postgres/#with-a-connection). make sure to include "prepare_threshold": 0 in the kwargs. if that doesn't work, try appending "prepareThreshold=0" to the connection string. let me know if this doesn't help

@vbarda commented on GitHub (Oct 11, 2024): @naturesh just researched this error more -- you need to create connection manually with connection kwargs and pass the connection to `PostgresSaver` (see this section: https://langchain-ai.github.io/langgraph/how-tos/persistence_postgres/#with-a-connection). make sure to include `"prepare_threshold": 0` in the kwargs. if that doesn't work, try appending "prepareThreshold=0" to the connection string. let me know if this doesn't help
Author
Owner

@naturesh commented on GitHub (Oct 13, 2024):

Thank you, and I'll let you know the current situation again after trying.

@naturesh commented on GitHub (Oct 13, 2024): Thank you, and I'll let you know the current situation again after trying.
Author
Owner

@vbarda commented on GitHub (Oct 15, 2024):

@naturesh please let me know if this resolved your issue!

@vbarda commented on GitHub (Oct 15, 2024): @naturesh please let me know if this resolved your issue!
Author
Owner

@naturesh commented on GitHub (Oct 16, 2024):

I tried connection through variable declaration without "with" syntax, but it still doesn't work and returns the same error.

@naturesh commented on GitHub (Oct 16, 2024): I tried connection through variable declaration without "with" syntax, but it still doesn't work and returns the same error.
Author
Owner

@naturesh commented on GitHub (Oct 16, 2024):

Can it be a problem if you don't use with syntax like this when the code structure doesn't allow you to use it? I always ended the session with connection close() at the end of the program without executing the code in duplicate or declaring connection more than once.

@naturesh commented on GitHub (Oct 16, 2024): Can it be a problem if you don't use with syntax like this when the code structure doesn't allow you to use it? I always ended the session with connection close() at the end of the program without executing the code in duplicate or declaring connection more than once.
Author
Owner

@naturesh commented on GitHub (Oct 16, 2024):

I've already run SUPABASE CLIENT twice on the server (svelde backend, python backend), but since it's the Sync client, it doesn't seem to be the same connection as a persistent pool. Could this be a problem?

@naturesh commented on GitHub (Oct 16, 2024): I've already run SUPABASE CLIENT twice on the server (svelde backend, python backend), but since it's the Sync client, it doesn't seem to be the same connection as a persistent pool. Could this be a problem?
Author
Owner

@vbarda commented on GitHub (Oct 16, 2024):

I tried connection through variable declaration without "with" syntax, but it still doesn't work and returns the same error.

Can it be a problem if you don't use with syntax like this when the code structure doesn't allow you to use it? I always ended the session with connection close() at the end of the program without executing the code in duplicate or declaring connection more than once.

this shouldn't be a problem if you're closing connections. how exactly are you passing the connection to PostgresSaver? could you paste an example? are you using the connection kwargs i provided above?

@vbarda commented on GitHub (Oct 16, 2024): > I tried connection through variable declaration without "with" syntax, but it still doesn't work and returns the same error. > > Can it be a problem if you don't use with syntax like this when the code structure doesn't allow you to use it? I always ended the session with connection close() at the end of the program without executing the code in duplicate or declaring connection more than once. > this shouldn't be a problem if you're closing connections. how exactly are you passing the connection to `PostgresSaver`? could you paste an example? are you using the connection kwargs i provided above?
Author
Owner

@naturesh commented on GitHub (Oct 18, 2024):

from psycopg import Connection
from langgraph.checkpoint.postgres import PostgresSaver


connection_kwargs = {
    "autocommit": True,
    "prepare_threshold": 0,
}


conn =  Connection.connect(os.environ['PROJECT_DATABASE_URL'], **connection_kwargs)
checkpointer = PostgresSaver(conn)

graph = builder.compile(checkpointer=checkpointer)




from langchain_core.messages import AIMessageChunk




if __name__ == '__main__': 
    request = [
        {"messages": ("user", 'hi')},
        {
            "configurable": 
                {
                    "thread_id": '~~~',
                }
        },
    ]

 
    import pprint

    pprint.pprint(graph.invoke(*request,stream_mode="values"))





    conn.close()

This is my code. It worked when I ran it again just now, and after running it again, the above error occurs again. I'm testing it using the jupyter notebook.

@naturesh commented on GitHub (Oct 18, 2024): ```python from psycopg import Connection from langgraph.checkpoint.postgres import PostgresSaver connection_kwargs = { "autocommit": True, "prepare_threshold": 0, } conn = Connection.connect(os.environ['PROJECT_DATABASE_URL'], **connection_kwargs) checkpointer = PostgresSaver(conn) graph = builder.compile(checkpointer=checkpointer) from langchain_core.messages import AIMessageChunk if __name__ == '__main__': request = [ {"messages": ("user", 'hi')}, { "configurable": { "thread_id": '~~~', } }, ] import pprint pprint.pprint(graph.invoke(*request,stream_mode="values")) conn.close() ``` This is my code. It worked when I ran it again just now, and after running it again, the above error occurs again. I'm testing it using the jupyter notebook.
Author
Owner

@naturesh commented on GitHub (Oct 18, 2024):

my postgresql url is this format

postgresql://postgres.~~~:~~~@aws-0-ap-northeast-2.pooler.supabase.com:6543/postgres?sslmode=disable

@naturesh commented on GitHub (Oct 18, 2024): my postgresql url is this format > postgresql://postgres.~~~:~~~@aws-0-ap-northeast-2.pooler.supabase.com:6543/postgres?sslmode=disable
Author
Owner

@nfcampos commented on GitHub (Oct 23, 2024):

Can it be a problem if you don't use with syntax like this when the code structure doesn't allow you to use it? I always ended the session with connection close() at the end of the program without executing the code in duplicate or declaring connection more than once.
I've already run SUPABASE CLIENT twice on the server (svelde backend, python backend), but since it's the Sync client, it doesn't seem to be the same connection as a persistent pool. Could this be a problem?

Hi this suggests you're sharing the same connection between different processes? That's not possible, you need to create a new connection for each use. We'd recommend creating a connection pool and passing that to the checkpointer

@nfcampos commented on GitHub (Oct 23, 2024): > Can it be a problem if you don't use with syntax like this when the code structure doesn't allow you to use it? I always ended the session with connection close() at the end of the program without executing the code in duplicate or declaring connection more than once. > I've already run SUPABASE CLIENT twice on the server (svelde backend, python backend), but since it's the Sync client, it doesn't seem to be the same connection as a persistent pool. Could this be a problem? Hi this suggests you're sharing the same connection between different processes? That's not possible, you need to create a new connection for each use. We'd recommend creating a connection pool and passing that to the checkpointer
Author
Owner

@vbarda commented on GitHub (Oct 31, 2024):

Closing due to inactivity, feel free to reopen!

@vbarda commented on GitHub (Oct 31, 2024): Closing due to inactivity, feel free to reopen!
Sign in to join this conversation.
1 Participants
Notifications
Due Date
No due date set.
Dependencies

No dependencies set.

Reference: langchain-ai/langgraph#268