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 cortexdb-dspy

Quick Start

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

client = Cortex(base_url="http://localhost:3141")

# 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="http://localhost:3141")
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="http://localhost:3141")
memory = CortexDBMemory(client=client, tenant_id="my-pipeline", top_k=5)

# 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
passages = 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="http://localhost:3141")

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

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) | | top_k | 5 | Default number of results for recall (memory) |

Complete Example

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

client = Cortex(base_url="http://localhost:3141")

# 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 = self.memory.recall(question, top_k=3)

        # 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?")