Add persistent memory to LangGraph agents and workflows.

LangGraph Integration

LangGraph handles short-term state inside a single graph run; CortexDB handles long-term memory across runs. Wire them with two nodes: recall at the entry, capture at the exit.

Install

pip install cortexdbai[langgraph]

Checkpointer

import os
from cortexdb import Cortex
from cortexdb.integrations.langgraph import CortexDBCheckpointer
from langgraph.graph import StateGraph

client = Cortex(
    api_url="https://api-v1.cortexdb.ai",
    actor="user:alice",
    bearer=os.environ["CORTEX_TOKEN"],
)

checkpointer = CortexDBCheckpointer(client=client, scope="org:acme/user:alice")

# graph = StateGraph(MyState); ... build graph ...
app = graph.compile(checkpointer=checkpointer)

cortexdb_memory_node is also exported — a graph node that injects recalled long-term context into running state.

Prefer manual control?

import os
from datetime import datetime, timezone
from uuid import uuid4
from langgraph.graph import StateGraph, MessagesState, END
from langchain_core.messages import AIMessage, HumanMessage
from cortexdb.v1 import V1Client

client = V1Client(
    api_url="https://api-v1.cortexdb.ai",
    actor="user:alice",
    bearer=os.environ["CORTEX_TOKEN"],
)
SCOPE = "org:acme/user:alice"


def recall_memory(state: MessagesState) -> dict:
    last_user = next(m.content for m in reversed(state["messages"]) if isinstance(m, HumanMessage))
    pack = client.recall(
        scope=SCOPE,
        view="holistic",
        query=last_user,
        # "events" matters: the context_block prose is rendered from event
        # content — excluding events can leave it empty on enrichment-heavy
        # scopes even when facts/beliefs return.
        include=["events", "beliefs", "facts", "episodes"],
        budgets={"max_tokens": 3000},
    )
    # .get(): servers < v0.8.15 omit context_block entirely when it is empty.
    context = pack.get("context_block", "")
    if context:
        return {"messages": [HumanMessage(content=f"<recall>{context}</recall>")]}
    return {}


def capture_memory(state: MessagesState) -> dict:
    last_user = next((m for m in reversed(state["messages"]) if isinstance(m, HumanMessage)), None)
    last_ai   = next((m for m in reversed(state["messages"]) if isinstance(m, AIMessage)), None)
    now = datetime.now(timezone.utc).isoformat()

    if last_user:
        client.experience(scope=SCOPE, text=last_user.content, role="user",
                          observed_at=now, idempotency_key=f"u-{uuid4()}")
    if last_ai:
        client.experience(scope=SCOPE, text=last_ai.content, role="assistant",
                          observed_at=now, idempotency_key=f"a-{uuid4()}")
    return {}


builder = StateGraph(MessagesState)
builder.add_node("recall_memory",  recall_memory)
builder.add_node("agent",          agent_node)         # your existing agent node
builder.add_node("capture_memory", capture_memory)
builder.set_entry_point("recall_memory")
builder.add_edge("recall_memory",  "agent")
builder.add_edge("agent",          "capture_memory")
builder.add_edge("capture_memory", END)

graph = builder.compile()

Tips

  • Scope-per-thread. For multi-user systems, derive the scope from config["configurable"]["thread_id"] rather than a constant.
  • Cancel a turn's memory. Use raw HTTP against POST /v1/lifecycle/memory-event/{event_id}/cancel if a turn should be excluded from future recall.
  • Streaming recall. Use /v1/recall/stream (raw HTTP) to render context incrementally in your UI.

See also