DSPy Integration

DSPy programs are typed signatures over LM calls. Wrap CortexDB as a dspy.Retrieve-compatible module so recall results plug straight into your modules.

Install

pip install cortexdbai[dspy]

Retrieval model

import os
import dspy
from cortexdb import Cortex
from cortexdb.integrations.dspy 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", k=5)
dspy.settings.configure(rm=retriever)

CortexDBMemory(client=client, scope=...) is a dspy.Module with store / recall / forget for write-capable pipelines.

Prefer manual control?

import os
import dspy
from datetime import datetime, timezone
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"


class CortexRetriever(dspy.Retrieve):
    def __init__(self, k: int = 8):
        super().__init__(k=k)

    def forward(self, query: str, k: int | None = None) -> dspy.Prediction:
        pack = client.recall(scope=SCOPE, view="holistic", query=query,
                             include=["facts", "episodes"],
                             budgets={"max_tokens": 4000})
        passages = []
        for fact in pack["layers"].get("facts", [])[: k or self.k]:
            passages.append(f"{fact['subject']['name']} {fact['predicate']} {fact['object']['value']}")
        return dspy.Prediction(passages=passages)


dspy.settings.configure(rm=CortexRetriever())

Use as any other DSPy retriever inside ChainOfThought, ReAct, etc.

Capture the final DSPy answer back into CortexDB so future runs can recall it:

def capture_dspy_run(question: str, answer: str) -> None:
    client.experience(
        scope=SCOPE,
        text=f"Question: {question}\nDSPy answer: {answer}",
        role="assistant",
        observed_at=datetime.now(timezone.utc).isoformat(),
        idempotency_key=f"dspy-{hash(question + answer)}",
    )

See also

• Python SDK
• Layers