Adding Memory to LangGraph Agents (Beyond Checkpointers)

The Gap Between State and Memory

LangGraph is excellent at managing workflow state. Its checkpointer system lets you save and restore graph execution state — which node ran last, what the current messages are, what variables have been set. If your process crashes, you can resume from the last checkpoint.

But checkpointers are not memory. They are snapshots of execution state at a point in time, tied to a specific thread or run. They do not answer questions like:

• “What did we learn about this customer across all previous conversations?”
• “What approach worked the last time we encountered this type of error?”
• “What does Agent B know that would help Agent A right now?”

This is the difference between state (where am I in this workflow?) and memory (what do I know from all past workflows?). LangGraph handles state. Aegis Memory handles memory.

In this tutorial, we will add persistent, semantic memory to a LangGraph agent using AegisClient directly. There is no LangGraph-specific integration class — instead, you call the Aegis client from within your graph nodes, which gives you full control over when and what to remember.

Prerequisites

You will need:

• Python 3.9+
• Docker running (for the Aegis Memory server)
• An OpenAI API key (for the LLM calls)

Install the packages:

pip install aegis-memory langgraph langchain-openai

Start the Aegis Memory server:

docker-compose up -d

(See the CrewAI tutorial for the full docker-compose.yml if you do not have one yet.)

Understanding the Architecture

Here is how the pieces fit together:

LangGraph Workflow
├── Node: retrieve_context
│   └── AegisClient.query()        ← Pull relevant memories
├── Node: process_request
│   └── LLM call with memory context
├── Node: store_learnings
│   └── AegisClient.add()          ← Store new knowledge
└── Checkpointer
    └── Saves graph state (nodes, edges, messages)

The checkpointer handles the graph execution state. Aegis handles the knowledge that persists across completely separate graph executions. They complement each other — you should use both.

Step 1: Set Up the Aegis Client

Since there is no LangGraph-specific wrapper, you use AegisClient directly. This is actually an advantage: you get full access to every Aegis feature without any abstraction layer in the way.

from aegis_memory.client import AegisClient

client = AegisClient(
    base_url="http://localhost:8741",
    api_key="your-api-key"
)

Step 2: Define Your Graph State

Define a LangGraph state that includes a slot for memory context. This is where retrieved memories will be injected before the LLM processes the user’s request.

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

class AgentState(TypedDict):
    messages: list
    user_id: str
    memory_context: str
    should_store: bool

The memory_context field will hold relevant memories retrieved from Aegis. The should_store flag lets the graph decide whether the current interaction contains knowledge worth persisting.

Step 3: Build Graph Nodes with Memory

Now create the nodes that interact with Aegis Memory. The pattern is straightforward: query before processing, store after processing.

The Memory Retrieval Node

This node runs before the LLM call. It searches Aegis for memories relevant to the user’s latest message and injects them into the state.

def retrieve_memories(state: AgentState) -> AgentState:
    """Query Aegis for relevant memories before the LLM processes the request."""
    messages = state["messages"]
    user_id = state["user_id"]

    # Get the latest user message
    last_message = messages[-1].content if messages else ""

    # Search for relevant memories
    results = client.query(
        query=last_message,
        user_id=user_id,
        agent_id="langgraph-assistant",
        top_k=5
    )

    # Format memories as context
    if results:
        memory_lines = []
        for mem in results:
            memory_lines.append(f"- {mem['content']}")
        memory_context = "Relevant context from previous interactions:\n" + "\n".join(memory_lines)
    else:
        memory_context = ""

    return {**state, "memory_context": memory_context}

The LLM Processing Node

This node calls the language model with the memory context included in the system prompt.

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")

def process_with_llm(state: AgentState) -> AgentState:
    """Call the LLM with memory context injected into the conversation."""
    messages = state["messages"]
    memory_context = state.get("memory_context", "")

    # Build the system message with memory context
    system_content = "You are a helpful assistant with access to memory from past conversations."
    if memory_context:
        system_content += f"\n\n{memory_context}\n\nUse this context to provide more personalized responses."

    # Prepare messages for the LLM
    llm_messages = [SystemMessage(content=system_content)] + messages

    # Call the LLM
    response = llm.invoke(llm_messages)

    # Determine if this interaction contains storable knowledge
    # Simple heuristic: store if the user shared personal info or preferences
    should_store = any(
        keyword in messages[-1].content.lower()
        for keyword in ["i prefer", "i like", "i work", "my name", "i always", "remember that"]
    )

    return {
        **state,
        "messages": messages + [response],
        "should_store": should_store
    }

The Memory Storage Node

This node runs after the LLM response and stores relevant information back to Aegis.

def store_memories(state: AgentState) -> AgentState:
    """Store new knowledge from the interaction into Aegis Memory."""
    if not state.get("should_store", False):
        return state

    messages = state["messages"]
    user_id = state["user_id"]

    # Extract the user message and AI response
    user_msg = messages[-2].content if len(messages) >= 2 else ""
    ai_msg = messages[-1].content if messages else ""

    # Store the interaction as a memory
    client.add(
        content=f"User said: {user_msg}\nAssistant learned: {ai_msg}",
        user_id=user_id,
        agent_id="langgraph-assistant",
        scope="agent-private",
        metadata={
            "type": "conversation-learning",
            "source": "langgraph-workflow"
        }
    )

    return state

Step 4: Assemble the Graph

Wire the nodes into a LangGraph workflow:

# Build the graph
workflow = StateGraph(AgentState)

# Add nodes
workflow.add_node("retrieve_memories", retrieve_memories)
workflow.add_node("process_with_llm", process_with_llm)
workflow.add_node("store_memories", store_memories)

# Define edges
workflow.add_edge(START, "retrieve_memories")
workflow.add_edge("retrieve_memories", "process_with_llm")
workflow.add_edge("process_with_llm", "store_memories")
workflow.add_edge("store_memories", END)

# Compile
app = workflow.compile()

Step 5: Run It

Now run the graph across multiple sessions. The first session establishes knowledge. The second session retrieves it.

# Session 1: User shares preferences
result1 = app.invoke({
    "messages": [HumanMessage(content="I prefer Python over JavaScript, and I work at a fintech startup.")],
    "user_id": "user_42",
    "memory_context": "",
    "should_store": False
})
print("Session 1 response:", result1["messages"][-1].content)

# Session 2: Different conversation, but memories persist
result2 = app.invoke({
    "messages": [HumanMessage(content="Can you suggest a good framework for building an API?")],
    "user_id": "user_42",
    "memory_context": "",
    "should_store": False
})
print("Session 2 response:", result2["messages"][-1].content)
# The response will reference Python and fintech context from Session 1

In Session 2, the retrieve_memories node finds the stored preference for Python and the fintech context. The LLM receives this as additional context and can tailor its recommendation accordingly — suggesting FastAPI or Django REST Framework instead of Express.js.

Advanced: Multi-Agent LangGraph with Shared Memory

LangGraph supports multi-agent architectures where different subgraphs handle different tasks. Aegis Memory’s scoping system maps naturally to this pattern.

# Different agents with different memory scopes
def research_node(state: AgentState) -> AgentState:
    """Research agent stores findings as globally visible memories."""
    findings = "The EU AI Act requires memory systems to support data deletion requests."

    client.add(
        content=findings,
        user_id="system",
        agent_id="researcher",
        scope="global",
        metadata={"type": "research-finding", "topic": "compliance"}
    )
    return state

def compliance_node(state: AgentState) -> AgentState:
    """Compliance agent queries shared memories from all agents."""
    results = client.query(
        query="EU AI Act data deletion requirements",
        user_id="system",
        agent_id="compliance-checker",
        top_k=5
    )

    # Use findings from the research agent
    for mem in results:
        print(f"Found shared memory: {mem['content']}")

    return state

Structured Handoffs Between Agents

When one agent in your graph finishes and another takes over, use Aegis handoffs to transfer context cleanly:

def handoff_node(state: AgentState) -> AgentState:
    """Transfer context from researcher to writer agent."""
    handoff_result = client.handoff(
        source_agent_id="researcher",
        target_agent_id="writer",
        task_context="Research on EU AI Act compliance is complete. "
                     "Key findings stored in global memory. "
                     "Write a summary for the legal team."
    )

    return {**state, "memory_context": str(handoff_result)}

Advanced: Session Tracking for Long Workflows

For LangGraph workflows that run over extended periods (data migrations, multi-step analysis), combine the checkpointer with Aegis session tracking:

# Create a session for a long-running workflow
client.create_session(
    session_id="quarterly-analysis-q1-2026",
    agent_id="analyst"
)

def analysis_step(state: AgentState) -> AgentState:
    """Track progress through a multi-step analysis."""
    current_step = state.get("current_step", "data-collection")

    client.update_session(
        session_id="quarterly-analysis-q1-2026",
        completed_items=["data-collection", "cleaning"],
        in_progress_item="statistical-analysis",
        next_items=["visualization", "report-writing"],
        blocked_items=[],
        summary="Data collected from 3 sources. Cleaned 12,000 records. "
                "Starting correlation analysis.",
        status="in_progress"
    )

    return state

# If the workflow crashes and restarts:
def resume_workflow(state: AgentState) -> AgentState:
    """Resume from where we left off using Aegis session state."""
    session = client.get_session(session_id="quarterly-analysis-q1-2026")

    print(f"Resuming from: {session['in_progress_item']}")
    print(f"Already completed: {session['completed_items']}")
    print(f"Context: {session['summary']}")

    return {**state, "current_step": session["in_progress_item"]}

Checkpointer vs Aegis Memory: When to Use Each

This is the most common question, so let us be precise:

Aspect	LangGraph Checkpointer	Aegis Memory
What it stores	Graph execution state (nodes, edges, messages)	Semantic knowledge (facts, preferences, learnings)
Scope	Single thread/run	Cross-run, cross-agent, cross-workflow
Retrieval	By thread ID (exact match)	By semantic similarity (vector search)
Purpose	Resume interrupted execution	Recall relevant past knowledge
Persistence	Thread-level	User-level, agent-level, or global
Query example	”Resume thread abc-123"	"What do we know about this customer’s preferences?”
Use when	Workflow might crash mid-execution	Agents need knowledge from previous runs

Use both. The checkpointer ensures your graph can resume after crashes. Aegis Memory ensures your agents can recall and apply knowledge from all previous interactions.

Common Patterns

Pattern 1: Memory-Augmented RAG

Combine document retrieval with memory retrieval. The document store has your knowledge base; Aegis has what the agent has learned from interactions.

def augmented_retrieval(state: AgentState) -> AgentState:
    """Combine RAG results with memory results."""
    query = state["messages"][-1].content

    # Standard RAG retrieval (your existing vector store)
    rag_results = your_vector_store.similarity_search(query, k=3)

    # Memory retrieval (what we've learned from past interactions)
    memory_results = client.query(
        query=query,
        user_id=state["user_id"],
        agent_id="assistant",
        top_k=3
    )

    # Combine both into context
    context = "From knowledge base:\n"
    for doc in rag_results:
        context += f"- {doc.page_content}\n"

    context += "\nFrom past interactions:\n"
    for mem in memory_results:
        context += f"- {mem['content']}\n"

    return {**state, "memory_context": context}

Pattern 2: Reflection After Errors

When a graph node encounters an error, store a reflection so future runs can avoid the same mistake.

def handle_error(state: AgentState) -> AgentState:
    """When something goes wrong, record a reflection."""
    error = state.get("last_error", "")

    client.add_reflection(
        content=f"Encountered error during data processing: {error}",
        agent_id="data-processor",
        namespace="etl-pipeline",
        error_pattern=error,
        correct_approach="Validate input schema before processing. "
                        "Check for null values in required fields.",
        applicable_contexts=["data-processing", "etl", "validation"],
        scope="global"
    )

    return state

Pattern 3: Pre-Task Playbook Consultation

Before starting a complex task, query the playbook for relevant lessons.

def consult_playbook(state: AgentState) -> AgentState:
    """Check the playbook before starting work."""
    task_description = state.get("task_description", "")

    playbook = client.query_playbook(
        query=task_description,
        agent_id="analyst",
        min_effectiveness=0.3
    )

    if playbook:
        lessons = "\n".join(
            f"- Avoid: {entry['error_pattern']}. Instead: {entry['correct_approach']}"
            for entry in playbook
        )
        state["memory_context"] = f"Lessons from past runs:\n{lessons}"

    return state

What’s Next

You now have a LangGraph agent with persistent, semantic memory. Here are some directions to explore:

• Self-Improving AI Agents: The ACE Pattern Explained — Understand the full ACE pattern that powers reflections, voting, and playbooks.
• How to Add Persistent Memory to CrewAI — If you are also working with CrewAI, see the dedicated integration.
• Aegis Memory vs Mem0 — Compare approaches to agent memory.

The key insight is that LangGraph checkpointers and Aegis Memory are complementary. Checkpointers handle “where am I?” Aegis handles “what do I know?” Together, they give your agents both reliable execution and genuine learning.