LangChain
Use CortexDB as a long-term memory provider for LangChain agents.
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 — the only model
it invokes internally is the one behind POST /v1/answer / POST /v1/understanding/synthesize
(claude-opus-4-6 is the cloud default; self-hosted it's whatever CORTEX_ANSWER_* configures, see
Self-hosting defaults). Your llm= choice is unrelated.
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, ororg:<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"]gives[fact|event|belief]:idmarkers for your UI.