Use CortexDB as a long-term memory provider for LangChain agents.

LangChain Integration

CortexDB plugs into LangChain as a persistent, hybrid-retrieval memory layer. The pattern is the same as any custom LangChain memory — implement BaseChatMemory.load_memory_variables to recall from CortexDB and save_context to capture experiences.

Install

pip install cortexdbai langchain langchain-openai

Prereqs. The example below uses ChatOpenAI and therefore requires an OPENAI_API_KEY. CortexDB itself is LLM-provider-agnostic — the only model CortexDB invokes internally is the one used by POST /v1/answer and POST /v1/understanding/synthesize (Claude Opus 4.6 by default, configurable per deployment). Your LangChain llm= choice is independent of that. Swap ChatOpenAI for ChatAnthropic, ChatBedrock, ChatGoogleGenerativeAI, or any local model wrapper — CortexMemory does not care.

Memory class

from typing import Any, Dict, List
from langchain.memory.chat_memory import BaseChatMemory
from langchain_core.messages import HumanMessage, AIMessage
from cortexdb.v1 import V1Client

class CortexMemory(BaseChatMemory):
    """LangChain memory backed by CortexDB v1."""

    def __init__(self, client: V1Client, scope: str, **kwargs):
        super().__init__(**kwargs)
        self._client = client
        self._scope = scope

    @property
    def memory_variables(self) -> List[str]:
        return ["history"]

    def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
        pack = self._client.recall(
            scope=self._scope,
            view="holistic",
            query=inputs.get("input", ""),
            include=["beliefs", "facts", "episodes"],
            budgets={"max_tokens": 3000},
        )
        return {"history": pack["context_block"]}

    def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None:
        # Capture the user turn
        self._client.experience(
            scope=self._scope,
            text=inputs["input"],
            role="user",
            observed_at=_now(),
            idempotency_key=_idem("user", inputs["input"]),
        )
        # Capture the assistant turn
        self._client.experience(
            scope=self._scope,
            text=outputs["output"],
            role="assistant",
            observed_at=_now(),
            idempotency_key=_idem("assistant", outputs["output"]),
        )

_now() and _idem() are small helpers — use datetime.utcnow().isoformat() + "Z" and an MD5 or f"chat-{uuid4()}" respectively.

Use it

import os
from langchain_openai import ChatOpenAI
from langchain.chains import ConversationChain

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

memory = CortexMemory(client=client, scope="org:acme/user:alice")

chain = ConversationChain(
    llm=ChatOpenAI(model="gpt-4o"),
    memory=memory,
)

chain.invoke("Remember that I prefer Python over JavaScript.")
# ... later, even in a fresh process ...
chain.invoke("What programming language do I prefer?")
# → "Based on what we know, you prefer Python over JavaScript."

Retriever variant

For RetrievalQA-style chains, expose CortexDB's recall as a BaseRetriever:

from langchain_core.retrievers import BaseRetriever
from langchain_core.documents import Document

class CortexRetriever(BaseRetriever):
    def __init__(self, client: V1Client, scope: str, k: int = 8):
        super().__init__()
        self._client = client
        self._scope = scope
        self._k = k

    def _get_relevant_documents(self, query: str) -> List[Document]:
        pack = self._client.recall(
            scope=self._scope,
            view="holistic",
            query=query,
            include=["facts", "episodes"],
            budgets={"max_tokens": 4000},
        )
        docs = []
        for fact in pack["layers"].get("facts", []):
            docs.append(Document(
                page_content=f"{fact['subject']['name']} {fact['predicate']} {fact['object']['value']}",
                metadata={"layer": "fact", "id": fact["fact_id"], "confidence": fact["confidence"]},
            ))
        return docs[: self._k]

Tips

  • Scope per user. A common shape is org:<your-org>/user:<id> for end-user memory, or org:<your-org>/agent:<id> for agent-scoped memory.
  • Use wait="indexed" if you need read-after-write within the same chain invocation — typically only needed for the first turn of a fresh session.
  • Citations. pack["provenance"]["citations"] gives you [fact|event|belief]:id markers you can surface in your UI.

See also