Skip to content

langgraph.md

Latest commit

 

History

History
214 lines (165 loc) · 6.47 KB

langgraph.md

File metadata and controls

214 lines (165 loc) · 6.47 KB
Raw

LangGraph Integration Guide

Use mcp-memory-service as the persistent memory backend for LangGraph agents and StateGraphs.

Key Differentiator: Cross-Graph Shared Memory

LangGraph's built-in MemorySaver is graph-local and ephemeral — memory is lost between runs and cannot be shared between different StateGraphs.

mcp-memory-service provides persistent shared memory across all graphs, runs, and even separate processes:

Graph A (Researcher)  ──┐
                        ├──→ mcp-memory-service ←──→ All graphs share one store
Graph B (Writer)     ──┘
Graph C (Reviewer)  ──┘

Setup

pip install mcp-memory-service httpx
MCP_ALLOW_ANONYMOUS_ACCESS=true memory server --http

Memory Tools for ReAct Agents

Define memory as @tool functions for use in a ReAct agent:

import httpx
from langchain_core.tools import tool

MEMORY_URL = "http://localhost:8000"

@tool
async def search_memory(query: str) -> str:
    """Search long-term memory for relevant context."""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{MEMORY_URL}/api/memories/search",
            json={"query": query, "limit": 5},
        )
        memories = response.json()["memories"]
        if not memories:
            return "No relevant memories found."
        return "\n".join(f"- {m['content']}" for m in memories)

@tool
async def store_memory(content: str, tags: list[str] = None) -> str:
    """Store a new memory for future retrieval."""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{MEMORY_URL}/api/memories",
            json={"content": content, "tags": tags or []},
        )
        result = response.json()
        return f"Stored memory: {result.get('content_hash', 'unknown')}"

Use in a ReAct agent:

from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-sonnet-4-6")

agent = create_react_agent(
    llm,
    tools=[search_memory, store_memory],
    state_modifier="You have access to long-term memory. Search memory before answering. Store important findings.",
)

Memory Node in StateGraph

Add explicit memory retrieve/store nodes to a StateGraph:

import httpx
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_anthropic import ChatAnthropic

MEMORY_URL = "http://localhost:8000"

class AgentState(TypedDict):
    messages: Annotated[list, "message history"]
    memory_context: str
    agent_id: str


async def retrieve_memory_node(state: AgentState) -> dict:
    """Retrieve relevant memory before calling the LLM."""
    last_message = state["messages"][-1]
    query = last_message.content if hasattr(last_message, "content") else str(last_message)

    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{MEMORY_URL}/api/memories/search",
            json={
                "query": query,
                "limit": 5,
                "tags": [f"agent:{state['agent_id']}"],  # Scope to this agent
            },
        )
        memories = response.json().get("memories", [])

    if memories:
        context = "Relevant memory:\n" + "\n".join(f"- {m['content']}" for m in memories)
    else:
        context = ""

    return {"memory_context": context}


async def llm_node(state: AgentState) -> dict:
    """Call LLM with memory context injected into system message."""
    llm = ChatAnthropic(model="claude-sonnet-4-6")

    system_parts = ["You are a helpful assistant."]
    if state.get("memory_context"):
        system_parts.append(state["memory_context"])

    messages = [SystemMessage(content="\n\n".join(system_parts))] + state["messages"]
    response = await llm.ainvoke(messages)
    return {"messages": state["messages"] + [response]}


async def store_memory_node(state: AgentState) -> dict:
    """Store important information from LLM response."""
    last_response = state["messages"][-1]
    content = last_response.content if hasattr(last_response, "content") else str(last_response)

    async with httpx.AsyncClient() as client:
        await client.post(
            f"{MEMORY_URL}/api/memories",
            json={
                "content": content[:500],  # Store a summary
                "tags": [f"agent:{state['agent_id']}", "llm-response"],
                "memory_type": "observation",
            },
            headers={"X-Agent-ID": state["agent_id"]},
        )

    return {}


# Build the graph
graph = StateGraph(AgentState)
graph.add_node("retrieve_memory", retrieve_memory_node)
graph.add_node("llm", llm_node)
graph.add_node("store_memory", store_memory_node)

graph.set_entry_point("retrieve_memory")
graph.add_edge("retrieve_memory", "llm")
graph.add_edge("llm", "store_memory")
graph.add_edge("store_memory", END)

agent = graph.compile()

Cross-Graph Memory Sharing

Two independent StateGraphs sharing the same memory store:

# Researcher graph — tags memories with agent:researcher
researcher_result = await researcher_agent.ainvoke({
    "messages": [HumanMessage(content="Research API rate limits")],
    "memory_context": "",
    "agent_id": "researcher",
})

# Writer graph — retrieves memories from researcher
async with httpx.AsyncClient() as client:
    response = await client.post(
        f"{MEMORY_URL}/api/memories/search",
        json={
            "query": "API rate limits",
            "tags": ["agent:researcher"],  # Read from researcher's memory
        },
    )
    shared_context = response.json()["memories"]

writer_result = await writer_agent.ainvoke({
    "messages": [HumanMessage(content="Write a summary of API limits")],
    "memory_context": "\n".join(m["content"] for m in shared_context),
    "agent_id": "writer",
})

Incremental Conversation Summaries

Use conversation_id to store incremental summaries without semantic deduplication:

import uuid

conversation_id = str(uuid.uuid4())

async def store_turn_summary(turn: int, summary: str):
    async with httpx.AsyncClient() as client:
        await client.post(
            f"{MEMORY_URL}/api/memories",
            json={
                "content": f"Turn {turn}: {summary}",
                "tags": ["conversation-summary"],
                "conversation_id": conversation_id,  # Bypasses dedup for this convo
                "memory_type": "note",
            },
        )