Issue with Missing Configuration Key __pregel_store When Using langgraph for Multi-User Persistence

[OchOchs]

I am currently working on a web application that utilizes langgraph and enables multi-user persistence. Initially, I attempted to solve this using only a thread ID. The idea was that by providing a thread ID and enabling MemorySaver when compiling the graph, the code could access an "outdated" prompt history even if other prompts were used in the meantime, including memory. However, I found that when accessing this "outdated" prompt history, the information in the state was reset to None instead of being set as it was before or even been overwritten.

After some further research, I discovered a feature called InMemoryStore, which is supposed to allow memory sharing across threads. While this is not something I necessarily need, I thought it might help resolve my issue since it also requires specifying a user ID.

Now, I am facing the problem that when calling graph.update_state with my thread ID and user ID, I encounter the following error:

ValueError: Missing required config key '__pregel_store' for 'initiate_all_interviews'.

This key should ideally be automatically handled by the library and is not something that should need to be manually specified in the thread. I suspect this might be related to the fact that InMemoryStore is relatively new and may still have some unresolved issues that could lead to such errors.

I would appreciate any help or guidance on how this issue could be resolved or whether it might be a bug that the team could address.

Thank you in advance!

[hinthornw]

Could you share example code?

At a quick glance, this looks like the graph you've compiled lacks a store instance, but your node is requesting a store instance.

def my_node(state, store: BaseStore):
    ...

builder.compile() # missing:  compile(..., store=InMemoryStore())
builder.invoke()

Also please do note that the memorysaver/inmemorystore implementations of the BaseCheckpointSaver and BaseStore are both are ephemeral (which is why they are "in memory") and useful for testing & prototyping but not something for use in production. In prod you'd want a DB-backed instance (like the postgressaver and postgresstore, etc.)

[OchOchs]

Sure, this is my graph.update_state method:

THREAD = {"configurable": {"thread_id": session.get("thread_id"), "user_id": session.get("user_id")}}
 GRAPH.update_state(
    THREAD, {"human_analyst_feedback": feedback, "oldAnalysts": analystsData}, as_node="human_feedback"
)

Then this is my Graph:

builder = StateGraph(ResearchGraphState)
 builder.add_node("create_analysts", create_analysts)
 builder.add_node("human_feedback", human_feedback)
 builder.add_node(
     "conduct_interview", conduct_interview
 ) 
 builder.add_node("write_report", write_report)
 builder.add_node("write_introduction", write_introduction)
 builder.add_node("write_conclusion", write_conclusion)
 builder.add_node("finalize_report", finalize_report)

 builder.add_edge(START, "create_analysts")
 builder.add_edge("create_analysts", "human_feedback")
 builder.add_conditional_edges(
     "human_feedback", initiate_all_interviews, ["create_analysts", "conduct_interview"]
 )
 builder.add_edge("conduct_interview", "write_report") 
 builder.add_edge("conduct_interview", "write_introduction")
 builder.add_edge("conduct_interview", "write_conclusion")
 builder.add_edge(
     ["write_conclusion", "write_report", "write_introduction"],
     "finalize_report",
 )
 builder.add_edge("finalize_report", END)
 print("Graph built")

 memory = MemorySaver()
 in_memory_store = InMemoryStore()
 return builder.compile(
     interrupt_before=["human_feedback"], checkpointer=memory, store=in_memory_store
 )

And last but not least this is the function, where the ValueError appears:

def initiate_all_interviews(state: ResearchGraphState, config: RunnableConfig, store: BaseStore):
   print("Initiating all interviews")

   user_id = config["configurable"]["user_id"]
   print("User ID: ", user_id)
   namespace = ("memories", user_id)
   print("namespace: ", namespace)
   memories = store.search(namespace)
   print("Memories: ", memories)

   human_analyst_feedback = state.get("human_analyst_feedback")
   if not human_analyst_feedback == "no":
       return "create_analysts"

   else:
       print
       topic = state["topic"]
       return [
           Send(
               "conduct_interview",
               {
                   "analyst": analyst,
                   "messages": [
                       HumanMessage(
                           content=f"Sie sagten also, Sie würden einen Artikel über {topic} schreiben?"
                       )
                   ],
               },
           )
           for analyst in state["analysts"]
       ]

The Error actually appears when the code trys to run the update_state method. This makes it kinda hard for me to debug that much, because I cant get information from inside the initiate_all_interviews function.
Nonetheless I'd think this should be enough of the code that I used, for you to review it an maybe help me with my issue.

[Manasi-s08]

Did you resolve it? I'm having the same problem

[OchOchs]

Not yet, I have paused working on it for now to wait for an answer here.