langgraph/how-tos/persistence_mongodb/

[giscus[bot]]

langgraph/how-tos/persistence_mongodb/

Build language agents as graphs

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

[HoangNguyen689]

Sometimes, I see that you use checkpoint_id for the thread_ts, sometimes you use checkpoint_ts for the thread_ts. Which value should be use?

[vbarda]

@HoangNguyen689 you should use checkpoint_id, thread_ts is a legacy name

[HoangNguyen689]

@vbarda

We are moving from AgentExecutor to LangGraph.
In AgentExecutor, we use DynamoDB for persistent memory through chat_history content.
We pass the chat_history to the prompt as a placeholder.
When using LangGraph, we defined a DynamoDBSaver to implement the checkpointer.
That checkpointer go well, when printing it, i can see the chat history in the channel_values["messages"].
However, the LLM agent doesn't have any knowledge about conversation history.
We already followed the above instruction and also read the blog about PostreSQL and MongoDB.
Is the conversation history in the checkpointer consumed automatically by the agent node?

[HoangNguyen689]

figure it out. Just passing all the state["messages"] that should be fine

[AlcebiadesFilho]

@HoangNguyen689

you can show how you do that ? Im stay with the same problem

[HoangNguyen689]

@AlcebiadesFilho
Sorry for late response. Which problem did you face?
In my case, State store the messages variable, which store all the conversation history.

class State(TypedDict):
    today_date: str

    messages: Annotated[list[AnyMessage], add_messages]

After that, you can define the model/assistant like below:

    def call_assistant(self, state: State) -> dict[str, list[BaseMessage]]:
        model_runnable = self.prompt | self.llm.bind_tools(
            self.tools, parallel_tool_calls=False
        )

        response = model_runnable.invoke(
            {
                "messages": state["messages"],
            }
        )

        return {"messages": [response]}

And finally, you just need to pass the user input

res = app.invoke(
                {
                    "messages": user_input,
                    "today_date": today_date,
                },
            )

[roboticsocialism]

I cannot find the module in the latest current version 0.1.9

[vbarda]

@roboticsocialism there is no module for this in the library, this is just a reference implementation

[guilledt]

Hi! How can I limit the number of previous messages in the memory, so that not all messages are remembered!

[Mr-Frk]

I got same issue. Does anyone know the solution?

[mihailtd]

Any solution to this? just change the thread ID?

[HoangNguyen689]

I believe that this example is outdated.
As I see, the langgraph also has put_writes method.
What purposes the writes is used for?

[vbarda]

We just updated the example to have all of the necessary methods. writes refer to intermediate writes from successful nodes, so that if one of the nodes running in parallel with others fails and we restart execution from the failure point, we don't re-run the ones that already succeeded. hope this helps!

[xuejuhui]

when using Asynchronous implementation of mongodb, it error out in the new version of langgraph, works in 0.1.9

AsyncBackgroundExecutor.exit(self, exc_type, exc_value, traceback)
152 for task in self.tasks:
153 try:
--> 154 task.result()
155 except asyncio.CancelledError:
156 pass

NotImplementedError: This method was added in langgraph 0.1.7. Please update your checkpoint saver to implement it.

[vbarda]

We just updated the guide to have all of the necessary methods

[LordO54]

how do I stream this?

config = {"configurable": {"thread_id": "2"}}
inputs = {"question": "me gustaria saber que consideraciones nutricionales necesito tener a la hora de bajar de peso","sessionid":"006", "userdata":userdata,}
async for event in app_memo.astream_events(input=inputs, version="v2", config=config):
kind = event["event"]
tags = event.get("tags", [])
if kind == "on_chat_model_stream" and "tool_llm" in tags:
data = event["data"]["chunk"].content
if data:
# 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, end="")

File c:\Users\PC\anaconda3\envs\myenv\Lib\site-packages\langgraph\pregel_init_.py:1098, in Pregel.astream(self, input, config, stream_mode, output_keys, input_keys, interrupt_before, interrupt_after, debug)
1095 processes = {**self.nodes}
1096 # get checkpoint from saver, or create an empty one
...
(...)
141 ),
142 )

TypeError: 'async for' requires an object with aiter method, got Cursor

[RichmondAlake]

Replace: langgraph.serde.jsonplus import JsonPlusSerializer
With: from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer

[vbarda]

We just updated the guide, thanks for flagging!

[vbarda]

Hi all! The guide has been updated with the correct implementtions of .put_writes and .aput_writes

[parokshsaxena]

Thanks for providing this implementation! This is super helpful.

One problem I am facing is even if I am setting the checkpoint_ns, I am always getting checkpoint_ns as '' while debugging in aput function.

My config is
config = {"configurable": {"thread_id": "2", "checkpoint_ns": "test"}}

Am I missing anything?

[parokshsaxena]

One observation. while debugging config received in aput function, I see in metadata it has correct value of checkpoint_ns but in configurable it's empty. Although while calling I am setting it in configurable parameter as specified in this doc

[Taha0229]

Hi,
I have implemented a graph with mongodb checkpointer in my FastAPI, but following this example, every time a request comes in, a new context is created for the checkpointer and a graph object is also instantiated which compiles the graph with the checkpointer. Following this the graph is invoked. Is this the only way to do it? I cannot compile a graph beforehand, because mongodb context is required which is passed as the checkpointer and within that context I need to invoke the graph. So to invoke the graph on every incoming request I first create a context then compile the graph by making an object (of my graph class) and then invoking it.

[theofficialvedantjoshi]

You do not need to initiate a new graph object everytime.
Create a class Workflow and inside it define a stategraph and a checkpointer object of type MongoDBSaver.
Create a self.app = stategraph.compile(checkpointer=self.checkpointer).
In your code create a workflow object and invoke workflow.app with specific state data and config.
The db keeps updating in the back and is not reset.
If this doesnt help you elaborate on your issue.

[my23701]

I am getting this error does anyone faced the same problem?

psycopg.Not SupportedError: the feature 'Connection.pipeline()" is not available: the client libpq version (imported from system libraries) is 13.11; the feature requires libpq version 14.0 or newer

[my23701]

Sorry, this was wrong blog you guys can ignore it, its for Postgres implementation.

[cris-m]

I am having issue using MongoDBSaver in langgraph. I am getting AttributeError: '_GeneratorContextManager' object has no attribute 'get_next_version'

checkpointer = MongoDBSaver.from_conn_info(
    host="localhost", port=27017, db_name="checkpoints"
)
graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=[],
    interrupt_after=[],
)
graph.name = "Agent"

[bhat-abhishek]

Even I am getting the same error

[vbarda]

that's because you need to use it as a context manager

with MongoDB.from_conn_info(...) as checkpointer:
    graph = builder.compile(
        checkpointer=checkpointer,
        interrupt_before=[],
        interrupt_after=[],
    )
    graph.invoke(...)

[gokpm]

What indexes should be created in MongoDB to support the queries this driver would use?

[gokpm]

Also, which field to add the TTL index in case older records should be removed?

[pranaykoppula]

Although I'm trying to use thread level persistence, the model seems to remember details from conversation in different threads. Am I doing it wring?

class CustomState(TypedDict):
user_context: str
humanReadableDate:str
messages: Annotated[list[BaseMessage], add_messages]
is_last_step: IsLastStep

memory = MongoDBSaver(client=mongo_client,db_name="user_conversations")
agent = create_react_agent(llm, tools, state_schema=CustomState, state_modifier=prompt, checkpointer=memory)
response=agent.invoke(inputs,config={'configurable':{'thread_id':session_id}})