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

AsyncConnectionPool AsyncPostgresSaver cannot send pipeline when not in pipeline mode #419

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

Originally created by @lam-dm on GitHub (Jan 24, 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

#main.py
 @asynccontextmanager
    async def lifespan(self, app: FastAPI):
        try:
            async with AsyncConnectionPool(
                conninfo=DB_URI, #connection information    
                kwargs={
                    "autocommit": True,
                    "prepare_threshold": 0,
                    "row_factory": dict_row
                },
                min_size=5, #minimum size of the pool
                max_size=15, #maximum size of the pool
            ) as pool, pool.connection() as conn:
                await AsyncPostgresSaver(conn).setup()
                yield {"conn": conn}
        except Exception as e:
            logger.error(f"Error setting up connection pool: {e}")
            raise
        finally:
            logger.info(f"End of lifespan")

#chat.py
# Use the connection directly from request.state.pool
            async with request.state.conn as conn:  # Use 'conn' directly
                checkpointer = AsyncPostgresSaver(conn)

                tools = [
                    GetInformationTool(
                        metadata={"information": only_structure}),
                    ValidateInformationTool(
                        metadata={"information": only_structure}),
                    RegisterInformationTool(metadata={
                        "information": only_structure,
                        "user_id": user_id,
                        "family_id": family_id,
                        "organization_id": organization_id,
                        "uuid": uuid,
                        "session_id": session_id
                    })
                ]

                graph_builder = RegistrationGraphBuilder(model=model, prompt=registration_prompt, system_prompt=system_message,
                                                            member_name=member_name, language="Japanese", tools=tools, checkpointer=checkpointer)
                graph = graph_builder.build()

                config = {"configurable": {"thread_id": thread_id}}
                result = await graph.ainvoke(
                    {
                        "messages": [HumanMessage(content=user_message)]
                    },
                    config
                )

                content = self.handle_result(result)
                return ResponseHandler.ok(message=VALID_RESPONSE_MESSAGE, body=content)

Error Message and Stack Trace (if applicable)

OperationalError('sending prepared query failed: cannot send pipeline when not in pipeline mode\nanother command is already in progress\ncannot exit pipeline mode while busy\ncannot enter pipeline mode, connection not idle\ncannot enter pipeline mode, connection not idle\ncannot enter pipeline mode, connection not idle\nanother command is already in progress\nanother command is already in progress')Traceback (most recent call last):


  File "/usr/local/lib/python3.10/site-packages/langgraph/pregel/__init__.py", line 1836, in astream
    async with AsyncPregelLoop(


  File "/usr/local/lib/python3.10/site-packages/langgraph/pregel/loop.py", line 988, in __aenter__
    saved = await self.checkpointer.aget_tuple(self.checkpoint_config)


  File "/usr/local/lib/python3.10/site-packages/langgraph/checkpoint/postgres/aio.py", line 186, in aget_tuple
    await cur.execute(


  File "/usr/local/lib/python3.10/site-packages/psycopg/cursor_async.py", line 97, in execute
    raise ex.with_traceback(None)


psycopg.OperationalError: sending prepared query failed: cannot send pipeline when not in pipeline mode
another command is already in progress
cannot exit pipeline mode while busy
cannot enter pipeline mode, connection not idle
cannot enter pipeline mode, connection not idle
cannot enter pipeline mode, connection not idle
another command is already in progress
another command is already in progress

Description

I'm developing a chatbot with FastAPI and using LangGraph with prebuilt create_react_agent
I'm using AsyncConnectionPool and AsyncPostgresSaver

I have a problem with managing the connection pool of Postgres with LangGraph, I saw this error: cannot send pipeline when not in pipeline mode

Could you help me to explain what happened and if I missed something, many thanks.
I appreciate that.

System Info

fastapi==0.115.6
langchain==0.3.14
langgraph==0.2.62
langgraph-checkpoint-postgres==2.0.10
psycopg==3.2.3
psycopg-pool==3.2.4

Originally created by @lam-dm on GitHub (Jan 24, 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 #main.py @asynccontextmanager async def lifespan(self, app: FastAPI): try: async with AsyncConnectionPool( conninfo=DB_URI, #connection information kwargs={ "autocommit": True, "prepare_threshold": 0, "row_factory": dict_row }, min_size=5, #minimum size of the pool max_size=15, #maximum size of the pool ) as pool, pool.connection() as conn: await AsyncPostgresSaver(conn).setup() yield {"conn": conn} except Exception as e: logger.error(f"Error setting up connection pool: {e}") raise finally: logger.info(f"End of lifespan") #chat.py # Use the connection directly from request.state.pool async with request.state.conn as conn: # Use 'conn' directly checkpointer = AsyncPostgresSaver(conn) tools = [ GetInformationTool( metadata={"information": only_structure}), ValidateInformationTool( metadata={"information": only_structure}), RegisterInformationTool(metadata={ "information": only_structure, "user_id": user_id, "family_id": family_id, "organization_id": organization_id, "uuid": uuid, "session_id": session_id }) ] graph_builder = RegistrationGraphBuilder(model=model, prompt=registration_prompt, system_prompt=system_message, member_name=member_name, language="Japanese", tools=tools, checkpointer=checkpointer) graph = graph_builder.build() config = {"configurable": {"thread_id": thread_id}} result = await graph.ainvoke( { "messages": [HumanMessage(content=user_message)] }, config ) content = self.handle_result(result) return ResponseHandler.ok(message=VALID_RESPONSE_MESSAGE, body=content) ``` ### Error Message and Stack Trace (if applicable) ```shell OperationalError('sending prepared query failed: cannot send pipeline when not in pipeline mode\nanother command is already in progress\ncannot exit pipeline mode while busy\ncannot enter pipeline mode, connection not idle\ncannot enter pipeline mode, connection not idle\ncannot enter pipeline mode, connection not idle\nanother command is already in progress\nanother command is already in progress')Traceback (most recent call last): File "/usr/local/lib/python3.10/site-packages/langgraph/pregel/__init__.py", line 1836, in astream async with AsyncPregelLoop( File "/usr/local/lib/python3.10/site-packages/langgraph/pregel/loop.py", line 988, in __aenter__ saved = await self.checkpointer.aget_tuple(self.checkpoint_config) File "/usr/local/lib/python3.10/site-packages/langgraph/checkpoint/postgres/aio.py", line 186, in aget_tuple await cur.execute( File "/usr/local/lib/python3.10/site-packages/psycopg/cursor_async.py", line 97, in execute raise ex.with_traceback(None) psycopg.OperationalError: sending prepared query failed: cannot send pipeline when not in pipeline mode another command is already in progress cannot exit pipeline mode while busy cannot enter pipeline mode, connection not idle cannot enter pipeline mode, connection not idle cannot enter pipeline mode, connection not idle another command is already in progress another command is already in progress ``` ### Description I'm developing a chatbot with FastAPI and using LangGraph with prebuilt create_react_agent I'm using AsyncConnectionPool and AsyncPostgresSaver I have a problem with managing the connection pool of Postgres with LangGraph, I saw this error: **cannot send pipeline when not in pipeline mode** Could you help me to explain what happened and if I missed something, many thanks. I appreciate that. ### System Info fastapi==0.115.6 langchain==0.3.14 langgraph==0.2.62 langgraph-checkpoint-postgres==2.0.10 psycopg==3.2.3 psycopg-pool==3.2.4
yindo closed this issue 2026-02-20 17:40:01 -05:00
yindo commented 2026-02-20 17:40:02 -05:00
Author
Owner
Copy Link
Copy Source

@vbarda commented on GitHub (Jan 24, 2025):

I think the issue might be in reusing the same connection in your lifespan -- i would recommend propagating the pool from your lifespan and then doing something like

async with pool.connection() as conn
    checkpointer = AsyncPostgresSaver(conn)
    ...

let me know if this resolves your issues!

@vbarda commented on GitHub (Jan 24, 2025): I think the issue might be in reusing the same connection in your lifespan -- i would recommend propagating the pool from your `lifespan` and then doing something like ```python async with pool.connection() as conn checkpointer = AsyncPostgresSaver(conn) ... ``` let me know if this resolves your issues!
yindo commented 2026-02-20 17:40:02 -05:00
Author
Owner
Copy Link
Copy Source

@lam-dm commented on GitHub (Jan 24, 2025):

I resolved it, I will close this issue. Thank you

@lam-dm commented on GitHub (Jan 24, 2025): I resolved it, I will close this issue. Thank you
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#419
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?