LangGraph Agent Memory Integration
Build stateful, personalized multi-agent graphs with LangGraph using the local-first, zero-knowledge sovseal memory layer.
Integrate LangGraph with sovseal to build complex, multi-agent state machines that maintain memory across sessions. By storing agent memories locally in a zero-knowledge LanceDB store, you eliminate network latency and keep agent context private.
LangGraph operates by passing an active state dict through graph nodes. We can hook the local memory node into the starting node (to inject context) and the ending node (to save learnings).
Installation
Install the required packages:
pip install langgraph langchain-openai mcp-client-cli python-dotenvState Graph Integration
Below is a complete, runnable Python example demonstrating how to load relevant memories at the start of a LangGraph chatbot node, and persist new conversation snippets at the end:
import os
from typing import Annotated, TypedDict, List
from dotenv import load_dotenv
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from mcp_client import StdioMCPClient
load_dotenv()
# 1. Define State structure
class AgentGraphState(TypedDict):
messages: Annotated[List[HumanMessage | AIMessage], add_messages]
user_id: str
# 2. Build the chatbot node with memory injection
async def query_chatbot(state: AgentGraphState):
messages = state["messages"]
user_id = state["user_id"]
last_query = messages[-1].content
# Connect to the local sovseal MCP server
async with StdioMCPClient(command="npx", args=["-y", "@sovseal/mcp-server"]) as mcp:
# 1. Recall historical context
recall_response = await mcp.call_tool(
"recall_memory",
{"query": last_query, "topK": 3}
)
memories = recall_response.get("content", [])
context_string = "\n".join([m.get("text", "") for m in memories])
# 2. Construct prompt incorporating retrieved context
system_prompt = f"""You are a personalized assistant. Use the following context from previous interactions:
{context_string}"""
llm = ChatOpenAI(model="gpt-4o")
full_messages = [SystemMessage(content=system_prompt)] + messages
response = await llm.ainvoke(full_messages)
# 3. Store the current interaction asynchronously
await mcp.call_tool(
"store_memory",
{"content": f"User asked: '{last_query}' — Assistant responded: '{response.content}'"}
)
return {"messages": [response]}
# 3. Compile the Graph
workflow = StateGraph(AgentGraphState)
workflow.add_node("chatbot", query_chatbot)
workflow.add_edge(START, "chatbot")
workflow.add_edge("chatbot", END)
app = workflow.compile()
# Example usage
async def run():
inputs = {
"messages": [HumanMessage(content="I am developing a state continuity system.")],
"user_id": "alice_dev"
}
config = {"configurable": {"thread_id": "alice_dev"}}
async for event in app.astream(inputs, config):
for value in event.values():
print("Chatbot:", value["messages"][-1].content)
if __name__ == "__main__":
import asyncio
asyncio.run(run())