# Pipeline Routing

> Intent routing and classification patterns for DSPy

- Skill: `j33bs/pipeline-routing` (Agent Skill)
- Install (CLI): `npx skillmds@latest add j33bs/pipeline-routing`
- Raw SKILL.md: https://api.skillmd.com/api/skills/j33bs/pipeline-routing/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: DevOps & Infra
- Author: j33bs (https://skillmd.com/u/j33bs)
- Updated: 2026-09-22
- Page: https://skillmd.com/skills/j33bs/pipeline-routing

---


# Pipeline Routing

## 🎯 Trigger Conditions
Use when asked about routing DSPy pipelines, intent classification, or multi-path execution.

## 📚 Prerequisites
- `dspy` package installed
- Routing criteria defined
- Multiple pipeline paths available

## 🛠️ Routing Patterns

### 1. Simple Intent Routing
```python
class IntentRouter(dspy.Module):
    def __init__(self):
        self.classifier = dspy.Predict("question -> intent")
        self.handlers = {
            "factual": FactualQA(),
            "opinion": OpinionAnalysis(),
            "code": CodeGeneration()
        }
    
    def forward(self, question):
        intent = self.classifier(question=question).intent
        handler = self.handlers[intent]
        return handler(question=question)
```

### 2. Confidence-Based Routing
```python
class ConfidenceRouter(dspy.Module):
    def __init__(self, threshold=0.8):
        self.threshold = threshold
        self.classifier = dspy.Predict("question -> intent, confidence")
        self.expert = ExpertSystem()
        self.fallback = FallbackHandler()
    
    def forward(self, question):
        result = self.classifier(question=question)
        intent = result.intent
        confidence = float(result.confidence)
        
        if confidence >= self.threshold:
            return self.handlers[intent](question=question)
        else:
            return self.fallback(question=question)
```

### 3. Multi-Stage Routing
```python
class MultiStageRouter(dspy.Module):
    def __init__(self):
        self.stage1 = Stage1Router()
        self.stage2 = Stage2Router()
        self.specialists = {
            "math": MathSolver(),
            "science": ScienceExpert(),
            "general": GeneralQA()
        }
    
    def forward(self, question):
        # Stage 1: Broad classification
        broad_intent = self.stage1.classify(question)
        
        # Stage 2: Specific classification
        specific_intent = self.stage2.refine(broad_intent, question)
        
        # Route to specialist
        specialist = self.specialists[specific_intent]
        return specialist(question=question)
```

### 4. Dynamic Routing
```python
class DynamicRouter(dspy.Module):
    def __init__(self):
        self.router = dspy.Predict("question, history -> next_step")
        self.steps = {
            "search": SearchStep(),
            "reason": ReasoningStep(),
            "generate": GenerationStep(),
            "verify": VerificationStep()
        }
    
    def forward(self, question, history=None):
        next_step = self.router(
            question=question,
            history=history
        ).next_step
        
        return self.steps[next_step](
            question=question,
            history=history
        )
```

## ⚠️ Pitfalls
- **Over-routing**: Too many routes add complexity
- **Misclassification**: Wrong routing leads to poor results
- **Latency**: Routing adds processing time
- **Maintenance**: More routes = more maintenance

## 📖 References
- [Routing Patterns](https://dspy-docs.vercel.app/docs/deep-dive/pipelines/routing)
- [Intent Classification](https://arxiv.org/abs/2010.02756)