LangChain

CortexDB plugs into LangChain as a persistent, hybrid-retrieval memory layer. The pattern is the same as any custom LangChain memory — recall from CortexDB to build context, and capture experiences on each turn.

Install

pip install cortexdbai[langchain]

Your LangChain model is independent of CortexDB's

The example uses ChatOpenAI (needs OPENAI_API_KEY). CortexDB is provider-agnostic: its own LLM calls (entity extraction and enrichment on the server's CORTEX_LLM_* / CORTEX_ENRICHMENT_* lanes, answers on CORTEX_ANSWER_*; see LLM & Answer Generation) are configured on the server and are unrelated to your llm= choice.

Retriever

import os
from cortexdb import Cortex
from cortexdb.integrations.langchain import CortexDBRetriever

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

retriever = CortexDBRetriever(client=client, scope="org:acme/user:alice")
docs = retriever.invoke("What did we decide about caching?")

The adapter also exports CortexDBChatMessageHistory(client=client, scope=...) for chat memory and CortexDBSearchTool / CortexDBStoreTool / CortexDBForgetTool (each (client=client, scope=...)) as agent tools.

Prefer manual control?

from typing import Any, Dict, List
from langchain.memory.chat_memory import BaseChatMemory
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=["events", "beliefs", "facts", "episodes"],
            budgets={"max_tokens": 3000},
        )
        return {"history": pack.get("context_block", "")}

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

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

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, self._scope, self._k = client, scope, 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']['id']} {fact['predicate']} {fact['object']['value']}",
                metadata={"layer": "fact", "id": fact["id"], "confidence": fact["confidence"]},
            ))
        return docs[: self._k]

Fact fields: id and subject.id

Live facts are { id, subject: { type, id }, object: { type, datatype, value }, … } — there is no fact_id and subject has no name. Use fact["id"] and fact["subject"]["id"] (older snippets showed fact["fact_id"] / fact["subject"]["name"], which raise KeyError). Facts require enrichment on a self-host — see Self-hosting defaults.

Tips

Scope per user. A common shape is org:<org>/user:<id> for end-user memory, or org:<org>/agent:<id> for agent-scoped memory.
Use wait="indexed" if you need read-after-write within the same chain invocation.
Citations. pack["provenance"]["citations"] maps each derived record id to the events it came from (empty until enrichment is on). For UI citation markers, use the citations list returned by client.answer().