DSPy

CortexDB integrates with DSPy as both a retrieval model and a persistent memory module, enabling DSPy programs to pull context from and store knowledge into CortexDB's long-term memory.

Installation

pip install cortexdbai[dspy]

Quick Start

import dspy
from cortexdb import Cortex
from cortexdb_dspy import CortexDBRetriever, CortexDBMemory

client = Cortex(base_url="https://api.cortexdb.ai")

# Use as a retrieval model
retriever = CortexDBRetriever(client=client, tenant_id="my-app", k=5)
dspy.settings.configure(rm=retriever)

# Use as a persistent memory module
memory = CortexDBMemory(client=client, tenant_id="my-app")

As a Retrieval Model

CortexDBRetriever implements dspy.Retrieve, making CortexDB a drop-in retrieval backend for any DSPy program that uses retrieval-augmented generation:

import dspy
from cortexdb import Cortex
from cortexdb_dspy import CortexDBRetriever

client = Cortex(base_url="https://api.cortexdb.ai")
retriever = CortexDBRetriever(client=client, tenant_id="my-app", k=5)

# Configure DSPy globally
dspy.settings.configure(rm=retriever)

# Use in a RAG program
class RAGProgram(dspy.Module):
    def __init__(self):
        self.retrieve = dspy.Retrieve(k=5)
        self.generate = dspy.ChainOfThought("context, question -> answer")

    def forward(self, question):
        context = self.retrieve(question).passages
        return self.generate(context=context, question=question)

rag = RAGProgram()
result = rag(question="What was the deployment decision?")

Direct Retrieval

You can also use the retriever directly:

# Single query
result = retriever.forward("deployment schedule")
print(result.passages)

# Multiple queries (results are deduplicated)
result = retriever.forward(["deployment", "release process"], k=10)
print(result.passages)

As a Persistent Memory Module

CortexDBMemory is a dspy.Module that provides store, recall, and forget operations for building self-improving pipelines:

from cortexdb import Cortex
from cortexdb_dspy import CortexDBMemory

client = Cortex(base_url="https://api.cortexdb.ai")
memory = CortexDBMemory(client=client, tenant_id="my-pipeline")

# Store learnings from pipeline runs
memory.store("Batch size 32 gave the best F1 score on this dataset.")
memory.store("Temperature 0.7 produces the most coherent outputs.")

# Recall relevant context (returns a single result with .context, .confidence, .latency_ms)
result = memory.recall("optimal hyperparameters")

# Use as a DSPy module in a pipeline
prediction = memory.forward("hyperparameter recommendations")
print(prediction.passages)

# Forget outdated information
memory.forget("old batch size", reason="Superseded by new experiment results")

Self-Improving Pipeline

import dspy
from cortexdb import Cortex
from cortexdb_dspy import CortexDBMemory, CortexDBRetriever

client = Cortex(base_url="https://api.cortexdb.ai")

class LearningPipeline(dspy.Module):
    def __init__(self):
        self.memory = CortexDBMemory(client=client, tenant_id="learner")
        self.generate = dspy.ChainOfThought("context, question -> answer")

    def forward(self, question):
        # Recall past learnings
        context = self.memory.forward(question).passages

        # Generate answer with context
        result = self.generate(context=context, question=question)

        # Store the new answer for future use
        self.memory.store(f"Q: {question}\nA: {result.answer}")

        return result

Under the Hood

The integration wraps CortexDB's REST API. Here are the equivalent calls:

# remember() — store a learning
curl -X POST https://api.cortexdb.ai/v1/remember \
  -H "Authorization: Bearer your-api-key" \
  -H "Content-Type: application/json" \
  -d '{
    "content": "Batch size 32 gave the best F1 score.",
    "tenant_id": "my-pipeline"
  }'
# Returns: { "event_id": "019d6359-d3cc-7671-9e4c-9151011fa016" }

# recall() — retrieve relevant context
curl -X POST https://api.cortexdb.ai/v1/recall \
  -H "Authorization: Bearer your-api-key" \
  -H "Content-Type: application/json" \
  -d '{
    "query": "optimal hyperparameters",
    "tenant_id": "my-pipeline"
  }'
# Returns: { "context": "...", "confidence": 0.88, "latency_ms": 14 }

Configuration

| Parameter | Default | Description | |---|---|---| | client | Required | Initialized Cortex client instance | | tenant_id | "default" | Tenant identifier for multi-tenant isolation | | k | 5 | Default number of passages to retrieve (retriever) |

Complete Example

import dspy
from cortexdb import Cortex
from cortexdb_dspy import CortexDBRetriever, CortexDBMemory

client = Cortex(base_url="https://api.cortexdb.ai")

# Set up retrieval-augmented generation with persistent memory
retriever = CortexDBRetriever(client=client, tenant_id="support-bot", k=5)
memory = CortexDBMemory(client=client, tenant_id="support-bot")

dspy.settings.configure(rm=retriever)

class SupportBot(dspy.Module):
    def __init__(self):
        self.retrieve = dspy.Retrieve(k=5)
        self.memory = memory
        self.answer = dspy.ChainOfThought("context, history, question -> answer")

    def forward(self, question):
        # Get knowledge base context
        kb_context = self.retrieve(question).passages

        # Get conversation history from memory
        history_result = self.memory.recall(question)
        history = history_result.context

        # Generate answer
        result = self.answer(
            context=kb_context,
            history=history,
            question=question,
        )

        # Remember this interaction
        self.memory.store(f"User asked: {question}\nAnswer: {result.answer}")

        return result

bot = SupportBot()
answer = bot(question="How do I reset my password?")