dspy-rag-pipeline

Files (2)

• SKILL.md
• example.py

SKILL.md (first lines)

# DSPy RAG Pipeline

## Goal

Build retrieval-augmented generation pipelines with ColBERTv2 that can be systematically optimized.

## When to Use

- Questions require external knowledge
- You have a document corpus to search
- Need grounded, factual responses
- Want to optimize retrieval + generation jointly

## Related Skills

- Optimize this pipeline: [dspy-miprov2-optimizer](../dspy-miprov2-optimizer/SKILL.md), [dspy-bootstrap-fewshot](../dspy-bootstrap-fewshot/SKILL.md)
- Build local semantic retrieval: [dspy-embedding-retrieval](../dspy-embedding-retrieval/SKILL.md)
- Evaluate results: [dspy-evaluation-suite](../dspy-evaluation-suite/SKILL.md)
- Design signatures: [dspy-signature-designer](../dspy-signature-designer/SKILL.md)

## Inputs

| Input | Type | Description |
|-------|------|-------------|
| `question` | `str` | User query |
| `k` | `int` | Number of passages to retrieve |
| `rm` | `dspy.Retrieve` | Retrieval model (ColBERTv2) |

## Outputs

| Output | Type | Description |
|--------|------|-------------|
| `context` | `list[str]` | Retrieved passages |
| `answer` | `str` | Generated response |

## Workflow

### Phase 1: Configure Retrieval

```python
import dspy

# Configure LM and retriever
colbert = dspy.ColBERTv2(url='http://20.102.90.50:2017/wiki17_abstracts')
dspy.configure(
    lm=dspy.LM("openai/gpt-4o-mini"),
    rm=colbert
)
```

### Phase 2: Define Signature

```python
class GenerateAnswer(dspy.Signature):
    """Answer questions with short factoid answers."""
    context: list[str] = dspy.InputField(desc="May contain relevant facts")
    question: str = dspy.InputField()
    answer: str = dspy.OutputField(desc="Often between 1 and 5 words")
```

### Phase 3: Build RAG Module

```python
class RAG(dspy.Module):
    def __init__(self, num_passages=3):
        super().__init__()
        self.retrieve = dspy.Retrieve(k=num_passages)
        self.generate = dspy.ChainOfThought(GenerateAnswer)
    
    def forward(self, question):
        context = self.retrieve(question).passages
        pred = self.generate(context=context, question=question)
        return dspy.Prediction(context=context, answer=pred.answer)
```

### Phase 4: Use

```python
rag = RAG(num_passages=3)
result = rag(question="What is the capital of France?")
print(result.answer)  # Paris
```

## Production Example

```python
import dspy
from dspy.teleprompt import BootstrapFewShot
from dspy.evaluate import Evaluate
import logging

logger = logging.getLogger(__name__)

class GenerateAnswer(dspy.Signature):
    """Answer questions using the provided context."""
    context: list[str] = dspy.InputField(desc="Retrieved passages")
    question: str = dspy.InputField()
    answer: str = dspy.OutputField(desc="Concise factual answer")

class ProductionRAG(dspy.Module):
    def __init__(self, num_passages=5):
        super().__init__()
        self.num_passages = num_passages
        self.retrieve = dspy.Retrieve(k=num_passages)
        self.generate = dspy.ChainOfThought(GenerateAnswer)
    
    def forward(self, question: str):
        try:
            # Retrieve
            retrieval_result = self.retrieve(question)
            context = retrieval_result.passages
            
            if not context:
                logger.warning(f"No passages retrieved for: {question}")
                return dspy.Prediction(
                    context=[],
                    answer="I couldn't find relevant information."
                )

From OmidZamani/dspy-skills, MIT.

More from OmidZamani/dspy-skills