langgraph/how-tos/persistence_postgres/

[giscus[bot]]

langgraph/how-tos/persistence_postgres/

Build language agents as graphs

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

[vaibhavp4]

How do I pass config and recursion limit together while streaming the graph? Currently, I'm getting this error - TypeError: Pregel.stream() takes from 2 to 3 positional arguments but 4 were given

[nathan-vo810]

This can be done by adding the recursion_limit key in the config:

config = {"configurable": {"thread_id": "2"}, "recursion_limit": 10}
for event in app.stream(
    {"messages": [input_message]},
    config,
    stream_mode="values",
):
    event["messages"][-1].pretty_print()

[dangchinh25]

Can you also update this guide but for JS version, and maybe a seperate general guide to explain in detail how to create a custom checkpointer and each component a checkpointer should have? I feel like this is a really important feature to bring LangGraph to production and there should be more guidance for it.

[vbarda]

@dangchinh25 we've added a JS how-to guide here https://langchain-ai.github.io/langgraphjs/how-tos/persistence-postgres/

[dangchinh25]

can we update this, the guide seem to have been deleted

[PI-rampi]

Please add a specific section on accessing and modifying the state variables of nodes from outside the node functions.

[hinthornw]

You mean like this? https://langchain-ai.github.io/langgraph/how-tos/human_in_the_loop/edit-graph-state/

[PI-rampi]

William Thanks for the reply, this is not exactly what I am looking for. Consider a simple scenario that an attribute in the state object has to be modified. I have a state object and want to modify a string parameter and set it to a value , for e.g. plan attribute . A simple document to modify state attributes with examples will be useful. class AgentState(TypedDict): task: str plan: str draft: str critique: str content: List[str] revision_number: int max_revisions: int workflow = StateGraph(AgentState) rampi

…

________________________________ From: William FH ***@***.***> Sent: 09 July 2024 03:47 To: langchain-ai/langgraph ***@***.***> Cc: Ramprasad ***@***.***>; Comment ***@***.***> Subject: Re: [langchain-ai/langgraph] langgraph/how-tos/persistence_postgres/ (Discussion #894) You mean like this? https://langchain-ai.github.io/langgraph/how-tos/human_in_the_loop/edit-graph-state/<https://eu-central-1.protection.sophos.com/?d=langchain-ai.github.io&u=aHR0cHM6Ly9sYW5nY2hhaW4tYWkuZ2l0aHViLmlvL2xhbmdncmFwaC9ob3ctdG9zL2h1bWFuX2luX3RoZV9sb29wL2VkaXQtZ3JhcGgtc3RhdGUv&i=NWY1NzY4MDNhOGI2OTAwZWFlOGRmNDI2&t=QkNXVHhueXB5K0FEKzQ2eFNNSEpXelZJTCtEbCs5VGdscllybnBqZThGdz0=&h=afb31f07d9264eedb4c62fc9d0a21266&s=AVNPUEhUT0NFTkNSWVBUSVY6K0T94M0wPyDMzKN9BocAHrXlbSFJ3THbbw-AzQ3TEw> — Reply to this email directly, view it on GitHub<https://eu-central-1.protection.sophos.com/?d=github.com&u=aHR0cHM6Ly9naXRodWIuY29tL2xhbmdjaGFpbi1haS9sYW5nZ3JhcGgvZGlzY3Vzc2lvbnMvODk0I2Rpc2N1c3Npb25jb21tZW50LTk5OTIwOTM=&i=NWY1NzY4MDNhOGI2OTAwZWFlOGRmNDI2&t=V0pKTnpNS3A2clF5NG54MUJYeStraGNLOHoxcVZBRklqU3dQV0xMbGF5RT0=&h=afb31f07d9264eedb4c62fc9d0a21266&s=AVNPUEhUT0NFTkNSWVBUSVY6K0T94M0wPyDMzKN9BocAHrXlbSFJ3THbbw-AzQ3TEw>, or unsubscribe<https://eu-central-1.protection.sophos.com/?d=github.com&u=aHR0cHM6Ly9naXRodWIuY29tL25vdGlmaWNhdGlvbnMvdW5zdWJzY3JpYmUtYXV0aC9CRjVDSlQyVkNRUFJLTFFMWTZDVkdLVFpMTUZZSEFWQ05GU002QUFBQUFCS0U0N1RENlZISTJEU01WUVdJWDNMTVY0M1NSREpPTlJYSzQzVE5GWFc0UTNQTlZXV0szVFVITTRUU09KU0dBNFRH&i=NWY1NzY4MDNhOGI2OTAwZWFlOGRmNDI2&t=SDdWRUxYR1o5RmROWXNnNWhrdktVWGVvZW9rOXY0eDBlbWdsWXdjcVZUST0=&h=afb31f07d9264eedb4c62fc9d0a21266&s=AVNPUEhUT0NFTkNSWVBUSVY6K0T94M0wPyDMzKN9BocAHrXlbSFJ3THbbw-AzQ3TEw>. You are receiving this because you commented.Message ID: ***@***.***> This E-MAIL message is a property of Premjiinvest and is intended for use only by the individual or entity to which it is addressed. The information contained may be confidential and privileged. If this is a forwarded message, the content of this E-MAIL may not have been sent with the authority of the company. If you have received this message by mistake, kindly notify the sender by return E-MAIL and delete this message. Premjiinvest will not be responsible for any viruses or defects or any forwarded attachments emanating either from within or outside. Premjiinvest reserves the right to monitor and review the content of all messages sent to or from Premjiinvest E-MAIL addresses. Messages sent to or from this e-mail address may be stored on the Premjiinvest E-MAIL system or elsewhere.

[nathan-vo810]

You can modify the attribute by using a node and return the new value.

def modify_plan(state):
      current_plan = state['plan']
      new_plan = current_plan + " expired"
      return {'plan': new_plan}

workflow.add_node('modify_plan', modify_plan)
workflow.set_entry_point('modify_plan')
workflow.add_edge('modify_plan', END)

app = workflow.compile()

[merchana]

Hello, How can I limit the memory, avoiding the graph to continuously have more history on its prompt?

E.g., by limiting to the last 10 messages

[DLOVRIC2]

In my usecase I only need the latest 3 questions and answers from the agents, so I have a method something like this:

    def _get_conversation_history(self, checkpoint: Checkpoint | None) -> list[str]:
        """
        Extract and format conversation history from checkpoint.

        Args:
            checkpoint (dict): The checkpoint containing conversation state

        Returns:
            list[str]: Interleaved list of queries and responses, up to MAX_HISTORY_LENGTH
        """
        if not checkpoint:
            return []

        user_queries = checkpoint.get("channel_values", {}).get("query", [])
        agent_responses = checkpoint.get("channel_values", {}).get("response", [])

        # Create interleaved conversation history from recent interactions
        conversation_history = []
        for q, r in zip(user_queries[-self.MAX_HISTORY_LENGTH :], agent_responses[-self.MAX_HISTORY_LENGTH :]):
            conversation_history.extend([q, r])

        logger.info("Current conversation state:")
        logger.info(f"Latest queries: {user_queries[-self.MAX_HISTORY_LENGTH:]}")
        logger.info(f"Latest responses: {agent_responses[-self.MAX_HISTORY_LENGTH:]}")
        logger.info(f"Conversation history: {conversation_history}")

        return conversation_history

Note that in my state I have both 'query' and 'response' args that I use to store the ongoing conversation. Note that input arg is the checkpoint which can be retrieved from Postgres, Redis, MongoDB or any other. Hope that helps

[Brainsoft-Raxat]

I am currently implementing a customer support bot and have been exploring the use of persistent memory to manage user interactions. My initial approach was to load the memory for a specific user from persistent storage, ensuring that all interactions from different users remain isolated. This seemed straightforward for maintaining user-specific contexts across sessions.

However, I am a bit confused about the specific utility of threads in this context. I understand that threads can provide additional granularity by keeping separate interactions within the same user session, but I am struggling to come up with concrete use cases where this granularity would be beneficial.

One example I am considering is within a single chat interaction. Even with persistent memory loaded through a checkpointer, it seems possible to add more granularity by using threads to manage multiple sub-interactions within a single user session. For instance, in a customer support bot scenario, a user might request multiple tasks, and these could be managed as separate threads within the same interaction.

My question is about the practical management of this granularity. How does the agent determine which thread to use at any given time within an ongoing interaction? Are there specific strategies or examples where combining both checkpointers and threads has proven particularly effective? I would greatly appreciate insights from those with more experience in this area.

[lakshaytalkstomachines]

How are you maintining "This seemed straightforward for maintaining user-specific contexts across sessions." without threads? Have you implemented it yourself?

[emarco177]

Any plans to add to the documentation an explanation of the "state' structure once its persisted?

e.g:
b'{"v": 1, "ts": "2024-07-13T05:59:18.599275+00:00", "id": "1ef40dd0-add3-62c2-8000-c6d57b7034eb", "channel_values": {"input": "hello world", "start:step_1": "start"}, "channel_versions": {"start": 2, "input": 2, "start:step_1": 2}, "versions_seen": {"start": {"start": 1}, "step_1": {}, "human_feedback": {}, "step_3": {}}, "pending_sends": []}'

To understand what each of those values here represent :)

[lakshaytalkstomachines]

We need more documentation on "channels" data structure in langraph.

[HoangNguyen689]

Why we need the writes table?

[vbarda]

see below!

[HoangNguyen689]

It should have docs about the way checkpointer work.
What purposes the writes table is used for?

[HoangNguyen689]

cc: @vbarda

[vbarda]

thanks for the feedback -- we'll be updating the documents on this to add more comments / explanations soon!

re: writes table / checkpointer.put_writes method -- it's needed for storing pending writes: if a node fails mid-execution, we store pending checkpoint writes from other successful nodes at that superstep, so that whenever we resume graph execution from that superstep we don't re-run the nodes that completed successfully previously. hope this helps!

[HoangNguyen689]

fyi: #1141

[HoangNguyen689]

@vbarda
Thanks.
Which resources I have to prepare for the writes? Is it the same as checkpoint table?

[vbarda]

thanks for starting the discussion -- we're working on improving the docs / developer experience around checkpointers and will post an update soon.

re: your question -- it really depends on what kind of checkpointer you're implementing, there isn't really a requirement for creating a specific table. the checkpointer just needs to conform to this interface: https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/checkpoint/base.py#L152 and implement .put, .put_writes, .get_tuple, .list or their async versions if you need to run your graph asynchronously

if you use a SQL database, you can just use the same table schemas as the postgres tutorial (although you might have to adjust the field types since the ones in the tutorial are postgres-specific). see more built-in examples here:

https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/checkpoint/sqlite.py#L64
https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/checkpoint/aiosqlite.py#L46

[GaryFail]

I get this error from get_tuple:

CheckpointTuple.new() got an unexpected keyword argument 'pending_writes'

and sure enough it doesn't:

class CheckpointTuple(NamedTuple):
config: RunnableConfig
checkpoint: Checkpoint
metadata: CheckpointMetadata
parent_config: Optional[RunnableConfig] = None

Anybody any ideas?

[GaryFail]

This does now have pending_writes. I've tried to look back to see if it was updated or if it was a mistake on my behalf but I can't find where this has changed and can't find the file on the main branch so perhaps it's also since being refactored. One less thing to worry about though :-)

[vbarda]

Hi folks! We just released a new Postgres checkpointer library -- you can install it with pip install langgraph-checkpoint-postgres. This how-to guide has been updated to use the library directly. This release is part of the larger LangGraph v0.2 release -- see release notes here https://github.com/langchain-ai/langgraph/releases/tag/0.2.0

Please let me know if you run into any issues with the library!

[sataycat]

I just saw @gbaian10's comment and realised that we can't utilize the connection pool unless we compile the graphs on each request invocation :(

[vbarda]

yes, you can use the lifespan to initialize the pool. we'll also look into passing the pool directly into the PostgresSaver

[sataycat]

Here's how I use the lifespan now. I find that compiling the graph per invocation is rather quick in practice, compared to waiting for the first messages to stream out

@asynccontextmanager
async def lifespan(app: FastAPI):
    connection_kwargs = {
        "autocommit": True,
        "prepare_threshold": 0,
        "row_factory": dict_row,
    }
    async with AsyncConnectionPool(settings.DB_URI, kwargs=connection_kwargs) as pool:
        await pool.wait()  # optional
        yield {"pool": pool}


app = FastAPI(lifespan=lifespan)


# elsewhere

@router.post("/message")
async def message(request: Request):
    async with request.state.pool.connection() as conn:
        checkpointer = AsyncPostgresSaver(conn)
        graph = graph_builder.compile(checkpointer=checkpointer)
        async for event in graph.astream_events(input, config=config, version="v2"):
            # do your thing
            pass

[vbarda]

by the way, we added support for passing pool to PostgresSaver directly in langgraph-checkpoint-postgres==1.0.4

[Gontna]

if i want to operate the database to complete some operations,such as view all conversation history. what should i do to get the pool that defined in lifespan

[gbaian10]

from psycopg_pool import AsyncConnectionPool

async with AsyncConnectionPool(
    # Example configuration
    conninfo=DB_URI,
    max_size=20,
    kwargs=connection_kwargs
) as pool, pool.connection() as conn:
    checkpointer = AsyncPostgresSaver(conn)

    # NOTE: you need to call .setup() the first time you're using your checkpointer
    # await checkpointer.setup()

    graph = create_react_agent(model, tools=tools, checkpointer=checkpointer)
    config = {"configurable": {"thread_id": "4"}}
    res = await graph.ainvoke(
        {"messages": [("human", "what's the weather in nyc")]}, config
    )

    checkpoint = await checkpointer.aget(config)

Regarding this piece of pool code, the pool has 20 connections available, but the code only takes one connection from the pool and assigns it to AsyncPostgresSaver.

Does this mean that an AsyncPostgresSaver will always have only one connection?
It seems that the pool's intended effect is not being utilized?

Additionally, I have a similar question to @GaryFail: how exactly should I compile the graph and the checkpointer in the API?

[vbarda]

great question. you can use something like FastAPI lifespan in the code snippet above to initialize the pool. we'll also look into passing the connection pool directly into the PostgresSaver!

[vbarda]

by the way, we added support for passing pool to PostgresSaver directly in langgraph-checkpoint-postgres==1.0.4

[sutharzan-ch]

@vbarda I am trying to figure out how to avoid compiling the graph each time I need to invoke it. Even if I pass the connection pool to PostgresSaver directly, I still have to create a checkpointer using a connection from the pool before compiling the graph. This means I end up compiling the graph for each Postgres DB thread. Is there a way to compile the graph once and reuse it across multiple invocations while still using the connection pool for database operations? Or am I misunderstanding the intended usage here?

[vigneshmj1997]

Any idea on which implementation works the best for the production environment ..?

• is it best to compile the graph on every request
• or is it better to have a complied version and use that object on every request

[GaryFail]

I guess we need to know if the compiled graph object is thread safe.

[vbarda]

yes, compiled graph object is thread safe

[robocanic]

I think the second is better. There are usually several initialize steps before the graph can be complied. These steps may cost a long time. So in order to reduce the whole inference costs, I believe the single pattern is better.

[kvkenyon]

Hey when I run my agent outside of LangGraph studio it's accessing the DB fine. I've even used ngrok to expose to TCP port. But from the langgraph studio docker container it won't connect. Any ideas?

[jhachirag7]

i am getting error on imports
from langgraph.checkpoint.postgres import PostgresSaver

error:
ImportError no pq wrapper available.
Attempts made:

• couldn't import psycopg 'c' implementation: No module named 'psycopg_c'
• couldn't import psycopg 'binary' implementation: No module named 'psycopg_binary'
• couldn't import psycopg 'python' implementation: expected str, bytes or os.PathLike object, not NoneType

[jhachirag7]

got solved after installing pip install "psycopg[binary,pool]"

[DLOVRIC2]

thanks!

[sambhavku]

I am using postgres checkpointer to save STATE and creating THREADS for follow-up questions.

I want to understand if there is a way to filter certain state variables before checkpointers saves them? I want to avoid saving RAG searches in my DB?

[kairobeduti]

Whats the actual implementation for thread checkpoint rollback ?

i mean, if a specific thread gets an exception during execution which leads to a "broken" state i.e. ai_messages with tool calls without tool_messages. Whats the best way to have the last valid checkpoint being the last again ?

[antonioalegria]

How do we select which schema do use?

[antonioalegria]

As in Postgresql schema (namespace) for the created tables

[sidpan1]

We use this code to do it.

        # Create schema if it doesn't exist
        try:
            with conn.connection() as connection:
                with connection.cursor() as cursor:
                    cursor.execute("CREATE SCHEMA IF NOT EXISTS xxx;")
                    cursor.execute("SET search_path TO xxx;")
        except Exception as e:
            conn.close()  # Clean up the connection pool if schema creation fails
            logger.error(f"Failed to create schema: {str(e)}")
            raise

        checkpoint_saver = PostgresSaver(
            conn=conn,
        )

[antonioalegria]

Any plans to support it as a parameter in the saver or via the connection string (by setting search_path, for example)?

[hinthornw]

You'll have to share some code

[itonskie]

Are there any examples that don't use async context managers? The examples are good for simple graphs but creating complex graphs everytime we call a query eats a lot of time. I'm also having connection leaks and errors after running a large amount of queries in a short period

[itonskie]

An example wherein you load the connection parameters into checkpointer and graph and then only call the async context manager when your making a query

[hinthornw]

Not sure if you're asking for examples without it or with it, but typically you'd initialize it once on server startup (as a part of a lifespan event) so you're not constantly re-creating the object

[pedrorjbr]

Why using psycopg3 instead of asyncpg? Is there any motivation behind?

[hinthornw]

pipelining

[my23701]

How can we do memory management with this case, lets say i want the last 3 checkpointers to work as chat_history of the agent, if this cannot be done then how the current implementation works under the hood, Kindly share if someone have some idea?

[arnabsinha99]

You can use techniques like message filtering or (message trimming) in LangGraph. It lets you use the last 3 messages in the chat history as context. Hope this is close to what you are looking for.

[my23701]

Okay, but that is for custom graph right, can we do the same in this prebuilt graph = create_react_agent(model, tools=tools, checkpointer=checkpointer)?

[ishaan812]

Even if I add something to the checkpoint namespace under checkpoint_ns in the configurable it wont get saved to the database. Nor do any of the other fields like metadata and stuff in the PG database.

thread_id = stripped_uuid4()
        config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": "ishaan"}}
        initial_state = WorkflowState(
            messages=[],
            template=1,
            context="",
            product_count=0,
            product_name="",
            nodes=[], edges=[], cards=[], file="file"
        )
        latest_event = None
        for event in self.workflow_instance.stream(initial_state, config, stream_mode="values"):
            latest_event = event
        return thread_id, latest_event

In sample code above, the checkpoint with the corrrect threadId and initialised. But checkpoint_ns is still empty in the database.

[ishaan812]

Update on this,

Tried to debug more, the checkpointer.postgres.init file the put function I presume is what is being used to store it in the database. The configurable object passed to the stream method is not getting passed to there correctly, although threadId is still being passed somehow and being used. Super confused how this is happening.

[ishaan812]

I dont think the stream method in langgraph is using the put at all. Cant figure out how the stream method is calling the checkpointer at all

[cristobalchavez21]

Is there any way we can use sqlalchemy create_engine instead of import ConnectionPool from psycopg_pool?

[hinthornw]

Not today

[tianshangwuyun]

when i used AsyncPostgresStore ,version 2.0.8

@asynccontextmanager
    async def compile_graph(self):
        async with await AsyncConnection.connect(DB_URI, **connection_kwargs) as conn:
            store = AsyncPostgresStore(conn)
            await store.setup()
            memory = CustomPgMemorySaver(conn)
            await memory.setup()
            yield self.app.compile(checkpointer=memory, store=store)

build a graph，and async node中的 await store.age should deadlock ，like this

async def one_node(....,store..):
     ##deadlock
      await store.aget(...)

but if like this no problem

async def one_node(....,store..):
      await asyncio.sleep(1)
      await store.aget(...)

add sleep then no problem
I found that the problem lies in

 async def _batch_get_ops(
        self,
        get_ops: Sequence[tuple[int, GetOp]],
        results: list[Result],
        cur: AsyncCursor[DictRow],
    ) -> None:
        for query, params, namespace, items in self._get_batch_GET_ops_queries(get_ops):
            await cur.execute(query, params)   ## this line

why???

@asynccontextmanager
    async def compile_graph(self):
        async with await AsyncConnection.connect(DB_URI, **connection_kwargs) as conn:
            store = AsyncPostgresStore(conn)
            await store.setup()
            memory = CustomPgMemorySaver(conn)
            await memory.setup()
            yield self.app.compile(checkpointer=memory, store=store)
mybe conn is shared

change pool no problem

[Jeffy-Fung]

How to use Checkpointer if I am using LangGraph Cloud?
I dont see any invoke or steam to trigger the graph when using Cloud

[Jeffy-Fung]

I mean when using LangCloud Server

[hinthornw]

It's automatically included

[sjwi7777]

I keep getting the following error in checkpointer.setup(), what could be the problem?

InvalidSchemaName: no schema has been selected to create in
LINE 1: CREATE TABLE IF NOT EXISTS checkpoint_migrations (

[strangehelix]

Looks like a lot of documentation is missing. For example a table checkpoints should be created in postgres before this can be used. Is there a doc somewhere that describes this?

[strangehelix]

nvm. Figured that checkpointer.setup() takes care of this. Worth mentioning this in the docs though!

[hinthornw]

It's stated 4 times in this doc

[deangenovski]

I am using supabase and postgres sql saver. It seems to work ok. My usecase is sales agent that collects orders. In my UI I would like to present the conversation history, so I am planning to create a view in supabase, do you have any suggestions how to do it, as the structure is not clearly documented?

[hinthornw]

I'd just call get_state?

[headyin]

What’s the best way to do data retention for the Postgres checkpoint save? I want to make sure the data is deleted after 30 days from the Postgres DB.

[wfjt]

pg_cron extension maybe?

[HoangNguyen689]

you can do that with cron job and check the timestamp field