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

Recall + capture nodes

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,
        include=["beliefs", "facts", "episodes"],
        budgets={"max_tokens": 3000},
    )
    if pack["context_block"]:
        return {"messages": [HumanMessage(content=f"<recall>{pack['context_block']}</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