How to Integrate memU with LangGraph for Agent Workflows

memU provides a native LangGraph adapter that exposes save_memory and search_memory tools through the MemULangGraphTools class, allowing agents to persist and retrieve long-term memories via a MemoryService instance.

The NevaMind-AI/memU repository offers a purpose-built integration layer for LangGraph that transforms its vector memory service into LangChain-compatible tools. When you integrate memU with LangGraph for agent workflows, you gain persistent memory capabilities without writing custom storage logic or tool wrappers.

Architecture Overview

The integration centers on three components working together to provide persistent memory in agent graphs.

MemoryService – The core memU component implemented in src/memu/app/service.py that handles embeddings, vector-store lookups, and metadata management through its memorize and retrieve methods.

MemULangGraphTools – The adapter class defined in src/memu/integrations/langgraph.py that wraps a MemoryService instance and constructs two StructuredTool instances: save_memory and search_memory.

LangGraph ToolNode – The standard LangGraph mechanism for tool execution. The memU tools integrate directly with ToolNode, allowing agents to invoke memory operations as part of their state graph workflow.

Prerequisites and Installation

The LangGraph integration requires optional dependencies beyond memU's core package. Install these using your preferred package manager:

uv add langgraph langchain-core

Or with pip:

pip install langgraph langchain-core

Ensure your environment includes the necessary configuration for memU's core functionality, specifically MEMU_DATABASE_URL for the vector store connection and OPENAI_API_KEY for embeddings generation.

The memU LangGraph Integration Layer

The MemULangGraphTools class creates two specialized tools that bridge memU's storage capabilities with LangGraph's execution model.

save_memory Tool

The save_memory tool persists arbitrary text as a memory item associated with a specific user_id. According to the source code in src/memu/integrations/langgraph.py (lines 76-92), the _save method implements this by:

• Creating a temporary file containing the content
• Calling MemoryService.memorize with the file path and metadata
• Cleaning up the temporary file in a finally block to prevent leaks

This approach allows memU's file-based ingestion pipeline to process the content while maintaining a clean interface for text-based agent interactions.

search_memory Tool

The search_memory tool retrieves relevant memories using natural language queries. Implemented in the _search method (lines 25-55) of src/memu/integrations/langgraph.py, this tool:

• Accepts a query string, user_id, and optional limit parameter
• Builds a LangChain-style query list compatible with MemoryService.retrieve
• Returns the most semantically similar memory items from the vector store

Both tools are defined as async coroutines, requiring await tool.ainvoke({...}) when called within LangGraph nodes.

Step-by-Step Integration Guide

Follow these steps to add memU memory capabilities to your LangGraph agent.

1. Initialize the MemoryService

   Create a service instance that handles the underlying storage and embedding operations:

   from memu.app.service import MemoryService
   
   service = MemoryService()

2. Instantiate the LangGraph Adapter

   Wrap the service with MemULangGraphTools to generate LangChain-compatible tools:

   from memu.integrations.langgraph import MemULangGraphTools
   
   tools_adapter = MemULangGraphTools(service)
   tools = tools_adapter.tools()  # Returns List[BaseTool]
   

3. Register Tools with Your StateGraph

   Add the tools to a ToolNode within your LangGraph workflow:

   from langgraph.graph import StateGraph, ToolNode
   
   workflow = StateGraph(state_schema={"user_id": str, "messages": list})
   workflow.add_node("tools", ToolNode(tools))

4. Invoke Tools in Agent Steps

   Use async invocation to save and retrieve memories during agent execution:

   async def agent_step(state):
       # Save a memory
   
       save_tool = next(t for t in tools if t.name == "save_memory")
       await save_tool.ainvoke({
           "content": "User prefers Python over JavaScript",
           "user_id": state["user_id"],
           "metadata": {"source": "preference_extraction"}
       })
       
       # Retrieve memories
   
       search_tool = next(t for t in tools if t.name == "search_memory")
       result = await search_tool.ainvoke({
           "query": "What programming language does the user prefer?",
           "user_id": state["user_id"],
           "limit": 3
       })
       return {"context": result}

5. Compile and Execute

   Compile your graph and run it as usual; the tools will handle persistence automatically through memU's storage layer.

Complete Working Example

The repository includes a full demonstration in examples/langgraph_demo.py that shows async initialization, tool usage, and proper error handling. This implementation initializes the infrastructure, saves user preferences, and performs semantic retrieval:

import asyncio, logging, os, sys

try:
    import langgraph
    from langchain_core.tools import BaseTool
    from memu.app.service import MemoryService
    from memu.integrations.langgraph import MemULangGraphTools
except ImportError:
    print("Missing dependencies. Run: uv add langgraph langchain-core")
    sys.exit(1)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("langgraph_demo")

async def initialize_infrastructure() -> MemULangGraphTools:
    if not os.getenv("OPENAI_API_KEY"):
        logger.warning("OPENAI_API_KEY not set – embeddings may fail.")
    service = MemoryService()
    return MemULangGraphTools(service)

async def process_conversation(tools: list[BaseTool], user_id: str):
    save_tool = next(t for t in tools if t.name == "save_memory")
    await save_tool.ainvoke({
        "content": "The user prefers dark mode and likes Python programming.",
        "user_id": user_id,
        "metadata": {"source": "demo_script"},
    })
    logger.info("Memory saved.")

async def process_retrieval(tools: list[BaseTool], user_id: str):
    search_tool = next(t for t in tools if t.name == "search_memory")
    result = await search_tool.ainvoke({
        "query": "What are the user's preferences?",
        "user_id": user_id,
        "limit": 3,
    })
    logger.info("Search result:\n%s", result)

async def main():
    logger.info("Starting LangGraph demo.")
    adapter = await initialize_infrastructure()
    tools = adapter.tools()
    user_id = "demo_user_123"
    await process_conversation(tools, user_id)
    await process_retrieval(tools, user_id)
    logger.info("Demo completed.")

if __name__ == "__main__":
    asyncio.run(main())

Using Tools with ToolNode

For standard LangGraph tool-calling patterns, pass the tools list directly to ToolNode. This configuration allows language models to decide when to save or retrieve memories based on the conversation context:

from langgraph.graph import StateGraph, ToolNode
from memu.app.service import MemoryService
from memu.integrations.langgraph import MemULangGraphTools

service = MemoryService()
adapter = MemULangGraphTools(service)
tools = adapter.tools()

graph = StateGraph(state_schema={"user_id": str, "messages": list})
graph.add_node("tools", ToolNode(tools))

# Connect nodes to create agent loop

graph.add_edge("agent", "tools")
graph.add_edge("tools", "agent")

Summary

• MemULangGraphTools in src/memu/integrations/langgraph.py provides the bridge between memU's storage layer and LangGraph's execution environment.
• The adapter exposes two tools: save_memory for persisting content and search_memory for semantic retrieval, both implemented as async coroutines.
• Temporary file handling in the save operation ensures content is processed through memU's pipeline without leaving orphaned files.
• The integration requires langgraph and langchain-core as optional dependencies and operates through standard ToolNode configurations.
• Reference implementation exists in examples/langgraph_demo.py, demonstrating end-to-end async usage with proper initialization patterns.

Frequently Asked Questions

What dependencies are required to integrate memU with LangGraph?

You must install langgraph and langchain-core alongside memU. These are optional dependencies not included in the base package. Install them via uv add langgraph langchain-core or pip install langgraph langchain-core before importing from memu.integrations.langgraph.

How does the save_memory tool handle content storage?

The tool writes content to a temporary file before calling MemoryService.memorize, then deletes the file in a finally block. This approach leverages memU's file-based ingestion pipeline while presenting a simple text interface to the agent. The implementation in src/memu/integrations/langgraph.py lines 76-92 ensures no temporary files persist after the operation completes.

Is the memU LangGraph integration synchronous or asynchronous?

The integration is async-first. Both _save and _search methods are async coroutines, requiring you to use await tool.ainvoke({...}) when calling them inside LangGraph nodes. This design aligns with LangGraph's async execution model and prevents blocking during vector store operations.

Where can I find the source code for the integration?

The core adapter code resides in src/memu/integrations/langgraph.py, containing the MemULangGraphTools class and tool definitions. Additional documentation is available in docs/langgraph_integration.md, and a complete working example is provided in examples/langgraph_demo.py within the NevaMind-AI/memU repository.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}