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

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

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

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

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
• Intent Classification