Dspy Modules

Build Composable AI Programs with dspy.Module

Guide the user through structuring DSPy programs as reusable, composable modules. A dspy.Module is the building block for all DSPy programs -- like PyTorch's nn.Module but for language model pipelines.

What is dspy.Module

dspy.Module is the building block for multi-step DSPy programs. Declare sub-modules in __init__ as self. attributes, wire them together with Python logic in forward(). DSPy optimizers automatically discover and tune all sub-modules in the tree.

Composing modules -- nesting modules within modules

Modules are composable. A module can use other custom modules as sub-modules:

class Summarizer(dspy.Module):
    def __init__(self):
        self.summarize = dspy.ChainOfThought("text -> summary")

    def forward(self, text):
        return self.summarize(text=text)


class AnalyzeAndSummarize(dspy.Module):
    def __init__(self):
        self.classify = dspy.Predict("text -> category")
        self.summarizer = Summarizer()  # nested custom module
        self.respond = dspy.ChainOfThought("category, summary -> response")

    def forward(self, text):
        category = self.classify(text=text).category
        summary = self.summarizer(text=text).summary
        return self.respond(category=category, summary=summary)

DSPy optimizers traverse the full module tree. When you optimize AnalyzeAndSummarize, the inner Summarizer's prompts get optimized too.

Printing module structure

Use print() to inspect all sub-modules and their signatures:

pipeline = AnalyzeAndSummarize()
print(pipeline)

Output shows the module tree:

AnalyzeAndSummarize(
  classify = Predict(text -> category)
  summarizer = Summarizer(
    summarize = ChainOfThought(text -> summary)
  )
  respond = ChainOfThought(category, summary -> response)
)

This is useful for verifying your module hierarchy and debugging which sub-modules exist.

Module state -- save and load

After optimization, save the learned state (few-shot demos, instructions) and reload it later:

# Save after optimization
optimized_program = optimizer.compile(my_program, trainset=trainset)
optimized_program.save("my_program.json")

# Load into a fresh instance
loaded = MyProgram()
loaded.load("my_program.json")

# Use the loaded program -- it has the optimized prompts
result = loaded(question="What is DSPy?")

What gets saved:

• Few-shot demonstrations discovered by optimizers
• Optimized instructions (from MIPROv2, GEPA)
• Any state that DSPy's Predict modules track

What does not get saved:

• Python logic in forward() -- that's your code
• Model weights (unless you used BootstrapFinetune)
• The LM configuration -- you must call dspy.configure() before loading

Validated outputs with Refine

Use dspy.Refine to enforce quality constraints on outputs through a reward function. This replaces the older dspy.Assert/dspy.Suggest pattern:

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

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


def answer_reward(args, pred):
    """Score answer quality. Returns float between 0.0 and 1.0."""
    score = 0.0

    # Hard requirement -- must provide a substantive answer
    if pred.answer.strip() and pred.answer != "I don't know":
        score += 0.6

    # Quality preference -- at least 10 words
    if len(pred.answer.split()) >= 10:
        score += 0.4

    return score


# Wrap with Refine to retry until quality threshold is met
validated_qa = dspy.Refine(
    module=SafeQA(),
    N=3,
    reward_fn=answer_reward,
    threshold=0.6,  # must at least pass the hard requirement
)

• `dspy.Refine` -- wraps a module, scores each attempt with a reward function, and retries until the threshold is met (up to N attempts). Use for requirements that must be met.
• Graduated scores -- return partial scores (0.0 to 1.0) to let Refine pick the best near-miss when no attempt fully succeeds.
• `dspy.BestOfN` -- similar to Refine but without cross-attempt feedback; use when attempts are independent.

For detailed Refine patterns and examples, see `/dspy-refine` and `/dspy-best-of-n`.

Common patterns

Conditional logic in forward()

Route to different sub-modules based on intermediate results:

class ConditionalPipeline(dspy.Module):
    def __init__(self):
        self.classify = dspy.Predict("text -> category")
        self.simple_handler = dspy.Predict("text -> response")
        self.complex_handler = dspy.ChainOfThought("text -> response")

    def forward(self, text):
        category = self.classify(text=text).category

        if category in ("simple", "faq"):
            return self.simple_handler(text=text)
        else:
            return self.complex_handler(text=text)

Loops in forward()

Process a list of items or iterate until a condition is met:

class BatchProcessor(dspy.Module):
    def __init__(self):
        self.process_item = dspy.ChainOfThought("item -> result")

    def forward(self, items: list[str]):
        results = []
        for item in items:
            result = self.process_item(item=item)
            results.append(result.result)
        return dspy.Prediction(results=results)

Iterative refinement

Keep improving until quality is sufficient:

class Refiner(dspy.Module):
    def __init__(self, max_rounds=3):
        self.draft = dspy.ChainOfThought("task -> output")
        self.critique = dspy.ChainOfThought("task, output -> feedback, is_good: bool")
        self.revise = dspy.ChainOfThought("task, output, feedback -> output")
        self.max_rounds = max_rounds

    def forward(self, task):
        result = self.draft(task=task)

        for _ in range(self.max_rounds):
            check = self.critique(task=task, output=result.output)
            if check.is_good:
                break
            result = self.revise(
                task=task,
                output=result.output,
                feedback=check.feedback,
            )

        return result

Error handling

Wrap sub-module calls to handle failures gracefully:

class ResilientModule(dspy.Module):
    def __init__(self):
        self.primary = dspy.ChainOfThought("question -> answer")
        self.fallback = dspy.Predict("question -> answer")

    def forward(self, question):
        try:
            return self.primary(question=question)
        except Exception:
            return self.fallback(question=question)

Returning custom predictions

Use dspy.Prediction to return structured results from forward():

class MultiOutput(dspy.Module):
    def __init__(self):
        self.analyze = dspy.ChainOfThought("text -> sentiment, topics: list[str]")
        self.summarize = dspy.ChainOfThought("text -> summary")

    def forward(self, text):
        analysis = self.analyze(text=text)
        summary = self.summarize(text=text)

        return dspy.Prediction(
            sentiment=analysis.sentiment,
            topics=analysis.topics,
            summary=summary.summary,
        )

Setting different LMs per sub-module

Assign cheaper models to simpler steps:

expensive_lm = dspy.LM("openai/gpt-4o")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
cheap_lm = dspy.LM("openai/gpt-4o-mini")  # or any smaller model

pipeline = MyProgram()
pipeline.classify.set_lm(cheap_lm)
pipeline.generate.set_lm(expensive_lm)

Batch processing

Use batch() to process multiple examples in parallel:

pipeline = MyProgram()
examples = [dspy.Example(question=q).with_inputs("question") for q in questions]
results = pipeline.batch(examples, num_threads=4, timeout=120)

Gotchas

1. Claude stores sub-modules in a plain list instead of as `self.` attributes. Optimizers discover sub-modules by traversing self. attributes in __init__. A Predict stored in a local variable or a plain list is invisible to optimization. Use a dict assigned to self. — DSPy traverses dicts for parameters. 2. Claude puts `dspy.configure()` inside `forward()`. Configure once at startup. Calling it per-forward adds overhead and causes unexpected behavior during optimization. 3. Claude names `forward()` args differently from training example fields. When an optimizer traces your module, it passes inputs from training examples to forward(). Mismatched argument names cause silent failures. Use the same field names as your dspy.Example inputs. 4. Claude creates a module with no `forward()` method. Every dspy.Module subclass must implement forward(). Without it, calling the module raises an error.

Cross-references

Install any skill: npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill <name>

• Signatures define inputs and outputs for each sub-module -- see /dspy-signatures
• Predict is the simplest sub-module for direct LM calls -- see /dspy-predict
• ChainOfThought adds step-by-step reasoning -- see /dspy-chain-of-thought
• Multi-step pipelines with real-world patterns -- see /ai-building-pipelines
• Optimizing modules to improve accuracy -- see /ai-improving-accuracy
• Install `/ai-do` if you do not have it — it routes any AI problem to the right skill and is the fastest way to work: npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill ai-do

Additional resources

• dspy.Module API docs
• For API details, see reference.md
• For worked examples, see examples.md

dspy-modules -- Worked Examples

Example 1: Simple QA module with pre/post processing

A question-answering module that normalizes input, generates an answer, and formats the output.

import dspy


class CleanQA(dspy.Module):
    """QA module with input normalization and output formatting."""

    def __init__(self):
        self.answer = dspy.ChainOfThought("question -> answer")

    def forward(self, question: str):
        # Pre-processing: normalize the question
        cleaned = question.strip().rstrip("?").strip() + "?"
        cleaned = cleaned[0].upper() + cleaned[1:]

        # Generate answer
        result = self.answer(question=cleaned)

        # Post-processing: ensure answer is complete sentences
        answer_text = result.answer.strip()
        if answer_text and not answer_text.endswith((".", "!", "?")):
            answer_text += "."

        return dspy.Prediction(
            answer=answer_text,
            reasoning=result.reasoning,
        )


# --- Usage ---

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

qa = CleanQA()
result = qa(question="  what is dspy  ")
print(result.answer)
print(result.reasoning)

Key points:

• forward() is just Python -- do whatever pre/post processing you need
• dspy.Prediction(...) lets you return a clean result with named fields
• The ChainOfThought sub-module is declared in __init__ so optimizers can find it

Example 2: RAG pipeline module

A retrieval-augmented generation module that searches for context, then generates a grounded answer with source citations.

import dspy
from typing import Literal


class AnswerWithSources(dspy.Signature):
    """Answer the question using only the provided context. Cite your sources."""
    context: list[str] = dspy.InputField(desc="Retrieved passages")
    question: str = dspy.InputField()
    answer: str = dspy.OutputField(desc="Answer grounded in the context")
    confidence: Literal["high", "medium", "low"] = dspy.OutputField(
        desc="How confident the answer is based on available context"
    )


class RAGPipeline(dspy.Module):
    """Retrieve relevant passages, then generate a grounded answer."""

    def __init__(self, k=3):
        self.retrieve = dspy.Retrieve(k=k)
        self.generate = dspy.ChainOfThought(AnswerWithSources)

    def forward(self, question: str):
        # Stage 1: Retrieve relevant passages
        retrieval_result = self.retrieve(question)
        passages = retrieval_result.passages

        # Guard: if no passages found, say so
        if not passages:
            return dspy.Prediction(
                answer="No relevant information found.",
                confidence="low",
                passages=[],
            )

        # Stage 2: Generate grounded answer
        result = self.generate(context=passages, question=question)

        return dspy.Prediction(
            answer=result.answer,
            confidence=result.confidence,
            passages=passages,
        )


def rag_confidence_reward(args, pred):
    """Prefer answers with high or medium confidence."""
    if pred.confidence == "low":
        return 0.5
    return 1.0


# --- Usage ---

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

# Note: dspy.Retrieve requires a retrieval model to be configured.
# See DSPy docs for ColBERTv2 or custom retriever setup.
rag = dspy.Refine(module=RAGPipeline(k=5), N=3, reward_fn=rag_confidence_reward, threshold=1.0)
result = rag(question="How does DSPy optimize prompts?")
print(result.answer)
print(f"Confidence: {result.confidence}")
print(f"Sources: {len(result.passages)} passages retrieved")

Key points:

• Each stage has a clear signature with typed fields
• forward() handles edge cases (no passages) with plain Python
• dspy.Refine wraps the module and retries when confidence is low, without hard-failing
• The module returns a Prediction that bundles the answer, confidence, and source passages

Example 3: Multi-stage analysis -- classify, route, generate

A module that classifies incoming text, routes it to a specialized handler, and generates a tailored response. This pattern is common in support systems, content moderation, and document processing.

import dspy
from typing import Literal


# --- Signatures ---

class ClassifyIntent(dspy.Signature):
    """Classify the user's message into a category."""
    message: str = dspy.InputField(desc="The user's message")
    category: Literal["question", "complaint", "feedback", "request"] = dspy.OutputField()


class AnswerQuestion(dspy.Signature):
    """Answer a factual question helpfully and concisely."""
    message: str = dspy.InputField()
    answer: str = dspy.OutputField(desc="A direct, helpful answer")


class HandleComplaint(dspy.Signature):
    """Respond to a complaint with empathy and a resolution plan."""
    message: str = dspy.InputField()
    response: str = dspy.OutputField(desc="Empathetic response with next steps")
    escalate: bool = dspy.OutputField(desc="Whether this needs human review")


class HandleFeedback(dspy.Signature):
    """Acknowledge feedback and summarize the key points."""
    message: str = dspy.InputField()
    response: str = dspy.OutputField(desc="Acknowledgment and summary")


class HandleRequest(dspy.Signature):
    """Process a request and explain what will happen next."""
    message: str = dspy.InputField()
    response: str = dspy.OutputField(desc="Confirmation and next steps")


# --- Module ---

class SmartRouter(dspy.Module):
    """Classify a message, then route to a specialized handler."""

    def __init__(self):
        self.classify = dspy.Predict(ClassifyIntent)
        self.handlers = {
            "question": dspy.ChainOfThought(AnswerQuestion),
            "complaint": dspy.ChainOfThought(HandleComplaint),
            "feedback": dspy.Predict(HandleFeedback),
            "request": dspy.Predict(HandleRequest),
        }

    def forward(self, message: str):
        # Stage 1: Classify
        classification = self.classify(message=message)
        category = classification.category

        # Stage 2: Route to the right handler
        handler = self.handlers.get(category, self.handlers["question"])
        result = handler(message=message)

        # Stage 3: Build unified response
        response_text = result.answer if hasattr(result, "answer") else result.response
        escalate = result.escalate if hasattr(result, "escalate") else False

        return dspy.Prediction(
            category=category,
            response=response_text,
            escalate=escalate,
        )


# --- Usage ---

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

router = SmartRouter()

# Print module structure to verify
print(router)

# Test with different message types
messages = [
    "What are your business hours?",
    "I've been waiting 3 weeks for my order and nobody responds!",
    "Love the new dashboard design, much easier to navigate.",
    "Can you change my subscription to the annual plan?",
]

for msg in messages:
    result = router(message=msg)
    print(f"\n[{result.category.upper()}] {msg}")
    print(f"Response: {result.response}")
    if result.escalate:
        print("** ESCALATE TO HUMAN **")


# --- Optimization ---

def router_metric(example, prediction, trace=None):
    """Score based on correct category and response quality."""
    category_correct = prediction.category == example.category
    # Check response is non-empty and reasonable length
    has_response = len(prediction.response.strip()) > 20
    return category_correct + 0.5 * has_response

# trainset = [dspy.Example(message=m, category=c).with_inputs("message") for m, c in data]
# optimizer = dspy.BootstrapFewShot(metric=router_metric, max_bootstrapped_demos=4)
# optimized_router = optimizer.compile(router, trainset=trainset)
# optimized_router.save("smart_router.json")

Key points:

• Classify then route is one of the most useful patterns -- cheap classification directs traffic to specialized handlers
• Handlers stored in a dict make the module easy to extend (add a new category = add a new handler)
• dspy.Predict (no reasoning) is used for simple tasks; dspy.ChainOfThought for ones that benefit from step-by-step thinking
• The unified dspy.Prediction return normalizes different handler output shapes
• When optimized, DSPy tunes the classifier and every handler together to maximize the end-to-end metric

Condensed from dspy.ai/api/modules/Module/. Verify against upstream for latest.

dspy.Module — API Reference

Constructor

dspy.Module(callbacks=None)

Subclass dspy.Module and implement forward() to define your program logic.

Key methods

Execution

State management

Introspection

Language model management

Utilities

batch() details

module.batch(
    examples,                      # list[dspy.Example]
    num_threads=None,              # parallel threads (default: dspy.settings.num_threads)
    max_errors=None,               # max errors before stopping
    return_failed_examples=False,  # if True, returns (results, failed, exceptions)
    timeout=120,                   # per-example timeout in seconds
    straggler_limit=3,             # max straggler threads to wait for
)

save() / load() details

# Save optimized state (few-shot demos, instructions)
optimized.save("my_program.json")

# Save full program including code (experimental)
optimized.save("my_program_dir/", save_program=True)

# Load into a fresh instance
program = MyProgram()
program.load("my_program.json")

What gets saved: few-shot demonstrations, optimized instructions, Predict module state. What does NOT get saved: Python logic in forward(), model weights, LM configuration.