| name | dspy-custom-module-design |
| version | 1.0.0 |
| dspy-compatibility | 3.1.2 |
| description | This skill should be used when the user asks to "create custom DSPy module", "design a DSPy module", "extend dspy.Module", "build reusable DSPy component", mentions "custom module patterns", "module serialization", "stateful modules", "module testing", or needs to design production-quality custom DSPy modules with proper architecture, state management, and testing. |
| allowed-tools | ["Read","Write","Glob","Grep"] |
DSPy Custom Module Design
Goal
Design production-quality custom DSPy modules with proper architecture, state management, serialization, and testing patterns.
When to Use
- Building reusable DSPy components
- Complex logic beyond built-in modules
- Need custom state management
- Sharing modules across projects
- Production deployment requirements
Related Skills
Inputs
| Input | Type | Description |
|---|
task_description | str | What the module should do |
components | list | Sub-modules or predictors |
state | dict | Stateful attributes |
Outputs
| Output | Type | Description |
|---|
custom_module | dspy.Module | Production-ready module |
Workflow
Phase 1: Basic Module Structure
All custom modules inherit from dspy.Module:
import dspy
class BasicQA(dspy.Module):
"""Simple question answering module."""
def __init__(self):
super().__init__()
self.predictor = dspy.Predict("question -> answer")
def forward(self, question):
"""Entry point for module execution."""
return self.predictor(question=question)
dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))
qa = BasicQA()
result = qa(question="What is Python?")
print(result.answer)
Phase 2: Stateful Modules
Modules can maintain state across calls:
import dspy
import logging
logger = logging.getLogger(__name__)
class StatefulRAG(dspy.Module):
"""RAG with query caching."""
def __init__(self, cache_size=100):
super().__init__()
self.retrieve = dspy.Retrieve(k=3)
self.generate = dspy.ChainOfThought("context, question -> answer")
self.cache = {}
self.cache_size = cache_size
def forward(self, question):
if question in self.cache:
return self.cache[question]
passages = self.retrieve(question).passages
result = self.generate(context=passages, question=question)
if len(self.cache) >= self.cache_size:
self.cache.pop(next(iter(self.cache)))
self.cache[question] = result
return result
Phase 3: Error Handling and Validation
Production modules need robust error handling:
import dspy
from typing import Optional
import logging
logger = logging.getLogger(__name__)
class RobustClassifier(dspy.Module):
"""Classifier with validation."""
def __init__(self, valid_labels: list[str]):
super().__init__()
self.valid_labels = set(valid_labels)
self.classify = dspy.Predict("text -> label: str, confidence: float")
def forward(self, text: str) -> dspy.Prediction:
if not text or not text.strip():
return dspy.Prediction(label="unknown", confidence=0.0, error="Empty input")
try:
result = self.classify(text=text)
if result.label not in self.valid_labels:
result.label = "unknown"
result.confidence = 0.0
return result
except Exception as e:
logger.error(f"Classification failed: {e}")
return dspy.Prediction(label="unknown", confidence=, error=(e))
Phase 4: Serialization
Modules support save/load:
import dspy
module = MyCustomModule()
module.save("my_module.json")
loaded = MyCustomModule()
loaded.load("my_module.json")
module.save("./my_module/", save_program=True)
loaded = dspy.load("./my_module/")
Production Example
import dspy
from typing import List, Optional
import logging
logger = logging.getLogger(__name__)
class ProductionRAG(dspy.Module):
"""Production-ready RAG with all best practices."""
def __init__(
self,
retriever_k: int = 5,
cache_enabled: bool = True,
cache_size: int = 1000
):
super().__init__()
self.retriever_k = retriever_k
self.cache_enabled = cache_enabled
self.cache_size = cache_size
self.retrieve = dspy.Retrieve(k=retriever_k)
self.generate = dspy.ChainOfThought("context, question -> answer")
self.cache = {} if cache_enabled else None
self.call_count = 0
def forward(self, question: str) -> dspy.Prediction:
"""Execute RAG pipeline with caching."""
self.call_count += 1
if not question or question.strip():
dspy.Prediction(
answer=,
error=
)
.cache_enabled question .cache:
logger.info()
.cache[question]
:
passages = .retrieve(question).passages
passages:
logger.warning()
dspy.Prediction(
answer=,
passages=[]
)
result = .generate(context=passages, question=question)
result.passages = passages
.cache_enabled:
._update_cache(question, result)
result
Exception e:
logger.error()
dspy.Prediction(
answer=,
error=(e)
)
():
(.cache) >= .cache_size:
.cache.pop(((.cache)))
.cache[key] = value
():
.cache_enabled:
.cache.clear()
Best Practices
- Single responsibility - Each module does one thing well
- Validate inputs - Check for None, empty strings, invalid types
- Handle errors - Return Predictions with error fields, never raise
- Log important events - Cache hits, errors, validation failures
- Test independently - Unit test modules before composition
Limitations
- State increases memory usage (careful with large caches)
- Serialization doesn't automatically save custom state
- Module testing requires mocking LM calls
- Deep module hierarchies can be hard to debug
- Performance overhead from validation in hot paths
Official Documentation
