dspy-custom-module-design

Files (2)

• SKILL.md
• example.py

SKILL.md (first lines)

# 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

- Module composition: [dspy-advanced-module-composition](../dspy-advanced-module-composition/SKILL.md)
- Signature design: [dspy-signature-designer](../dspy-signature-designer/SKILL.md)
- Optimization: [dspy-miprov2-optimizer](../dspy-miprov2-optimizer/SKILL.md)

## 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`:

```python
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)

# Usage
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:

```python
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):
        # Check cache
        if question in self.cache:
            return self.cache[question]

        # Retrieve and generate
        passages = self.retrieve(question).passages
        result = self.generate(context=passages, question=question)

        # Update cache with size limit
        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:

```python
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():

From OmidZamani/dspy-skills, MIT.

More from OmidZamani/dspy-skills