DSPy: The Framework for Programming LLMs

import dspy

# Configure LLM
lm = dspy.LM("openai/gpt-4o-mini")
dspy.configure(lm=lm)

# Define a simple signature
class Summarize(dspy.Signature):
    """Summarize the document in 2-3 sentences."""
    document = dspy.InputField()
    summary = dspy.OutputField()

# Create a module
summarizer = dspy.Predict(Summarize)

# Use it
result = summarizer(document="DSPy is a framework for programming...")
print(result.summary)

import dspy

# Chain of Thought adds step-by-step reasoning
class MathProblem(dspy.Signature):
    """Solve the math problem step by step."""
    problem = dspy.InputField()
    reasoning = dspy.OutputField(desc="step-by-step solution")
    answer = dspy.OutputField(desc="final numerical answer")

# CoT automatically elicits reasoning
solver = dspy.ChainOfThought(MathProblem)

result = solver(problem="If a train travels 120 miles in 2 hours, what is its average speed?")
print(f"Reasoning: {result.reasoning}")
print(f"Answer: {result.answer}")

import dspy
from dspy.retrieve.chromadb_rm import ChromadbRM

# Configure retriever and LLM
retriever = ChromadbRM(collection_name="docs", persist_directory="./chroma")
lm = dspy.LM("openai/gpt-4o")
dspy.configure(lm=lm, rm=retriever)

class RAG(dspy.Module):
    def __init__(self, num_passages=3):
        self.retrieve = dspy.Retrieve(k=num_passages)
        self.generate = dspy.ChainOfThought(
            "context, question -> answer"
        )

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

# Use it
rag = RAG()
result = rag("What is the refund policy?")
print(result.answer)

import dspy

class GroundedQA(dspy.Module):
    def __init__(self):
        self.generate = dspy.ChainOfThought("context, question -> answer")

    def forward(self, context, question):
        result = self.generate(context=context, question=question)

        # Assertion: answer must be grounded in context
        dspy.Assert(
            self._is_grounded(result.answer, context),
            f"Answer must be supported by context. Got: {result.answer}"
        )

        return result

    def _is_grounded(self, answer, context):
        # Check if key claims in answer appear in context
        # (Simplified - use NLI model in production)
        answer_words = set(answer.lower().split())
        context_words = set(context.lower().split())
        overlap = len(answer_words & context_words) / len(answer_words)
        return overlap > 0.3

# Assertions trigger retries with backtracking
with dspy.context(backtrack_to=self.generate):
    result = grounded_qa(context="...", question="...")

import dspy

# Configure multiple models
gpt4 = dspy.LM("openai/gpt-4o")
claude = dspy.LM("anthropic/claude-3-5-sonnet-20241022")
local = dspy.LM("ollama/llama3.1:8b")

# Use different models for different tasks
class Pipeline(dspy.Module):
    def __init__(self):
        # Cheap model for classification
        self.classify = dspy.Predict("text -> category")
        # Powerful model for generation
        self.generate = dspy.ChainOfThought("category, text -> response")

    def forward(self, text):
        with dspy.context(lm=local):  # Use local for classification
            category = self.classify(text=text).category

        with dspy.context(lm=gpt4):  # Use GPT-4 for generation
            response = self.generate(category=category, text=text)

        return response

# Or cascade: try cheap first, fall back to expensive
class CascadePipeline(dspy.Module):
    def forward(self, question):
        # Try cheap model first
        with dspy.context(lm=local):
            result = self.answer(question=question)
            if self._is_confident(result):
                return result

        # Fall back to expensive model
        with dspy.context(lm=gpt4):
            return self.answer(question=question)

import dspy

# Define tools
def search_web(query: str) -> str:
    """Search the web for current information."""
    # Implement web search
    return f"Results for: {query}"

def calculate(expression: str) -> str:
    """Evaluate a mathematical expression."""
    return str(eval(expression))

def lookup_database(entity: str) -> str:
    """Look up entity in knowledge base."""
    # Implement database lookup
    return f"Info about: {entity}"

# ReAct agent with tools
class ResearchAgent(dspy.Module):
    def __init__(self):
        self.react = dspy.ReAct(
            "question -> answer",
            tools=[search_web, calculate, lookup_database],
            max_iters=5
        )

    def forward(self, question):
        return self.react(question=question)

agent = ResearchAgent()
result = agent("What is the population of Tokyo times 2?")

import dspy
# Enable persistent caching
dspy.configure(
    lm=dspy.LM("openai/gpt-4o"),
    cache_dir="./dspy_cache"  # Persists across runs
)

import dspy
import asyncio

# Enable async mode
dspy.configure(lm=dspy.LM("openai/gpt-4o"), async_mode=True)

async def process_batch(questions):
    module = dspy.ChainOfThought("question -> answer")
    tasks = [module(question=q) for q in questions]
    return await asyncio.gather(*tasks)

# After optimization
optimized_module.save("./model/optimized_rag.json")

# In production
loaded_module = RAG()
loaded_module.load("./model/optimized_rag.json")

import dspy
from dspy.utils import inspect_history

# Enable detailed logging
dspy.configure(lm=dspy.LM("openai/gpt-4o"), trace=[])

# After running
result = module(question="...")

# Inspect what happened
inspect_history(n=3)  # Show last 3 LLM calls