import dspy

question = dspy.InputField()
answer = dspy.OutputField(desc="often between 1 and 5 words")

**Advanced Signatures with Type Hints**:
```python
from typing import List

class DocumentSummary(dspy.Signature):
    """Generate concise document summaries."""

    document: str = dspy.InputField(desc="Full text to summarize")
    key_points: List[str] = dspy.OutputField(desc="3-5 bullet points")
    summary: str = dspy.OutputField(desc="2-3 sentence summary")
    sentiment: str = dspy.OutputField(desc="positive, negative, or neutral")

**Field Descriptions**:
- Short, descriptive phrases (not full sentences)
- Examples: `desc="often between 1 and 5 words"`, `desc="JSON format"`
- Used by optimizers to improve prompt quality

undefined

**Retrieve Module (RAG)**:
```python
class RAGSignature(dspy.Signature):
    """Answer questions using retrieved context."""
    question = dspy.InputField()
    context = dspy.InputField(desc="relevant passages")
    answer = dspy.OutputField(desc="answer based on context")

**ReAct (Reasoning + Acting)**:
```python
class ResearchTask(dspy.Signature):
    """Research a topic using tools."""
    topic = dspy.InputField()
    findings = dspy.OutputField()

**ProgramOfThought**:
```python

**Custom Modules**:
```python
class MultiStepRAG(dspy.Module):
    """Custom module combining retrieval and reasoning."""

    def __init__(self, num_passages=3):
        super().__init__()
        self.retrieve = dspy.Retrieve(k=num_passages)
        self.generate = dspy.ChainOfThought("context, question -> answer")

    def forward(self, question):
        # Retrieve relevant passages
        context = self.retrieve(question).passages

        # Generate answer with context
        prediction = self.generate(context=context, question=question)

        # Return with metadata
        return dspy.Prediction(
            answer=prediction.answer,
            context=context,
            rationale=prediction.rationale
        )

undefined

from dspy.teleprompt import BootstrapFewShot

**How It Works**:
1. Uses your program to generate outputs on training data
2. Filters successful traces using your metric
3. Selects representative examples as demonstrations
4. Returns optimized program with best few-shot examples

from dspy.teleprompt import BootstrapFewShotWithRandomSearch

config = dict(
    max_bootstrapped_demos=4,
    max_labeled_demos=4,
    num_candidate_programs=10,   # Explore 10 candidate programs
    num_threads=4                # Parallel optimization
)

optimizer = BootstrapFewShotWithRandomSearch(
    metric=accuracy_metric,
    **config
)

optimized_program = optimizer.compile(
    qa_module,
    trainset=training_examples,
    valset=validation_examples  # Optional validation set
)

**Advantage**: Explores multiple candidate programs in parallel, selecting best performer via random search.

import dspy
from dspy.teleprompt import MIPROv2

# Weighted composite score
score = (
    correct * 1.0 +
    reasonable_length * 0.2 +
    has_reasoning * 0.3
)
return score / 1.5  # Normalize to [0, 1]

**MIPROv2 Auto-Configuration Modes**:
- **`light`**: Fast optimization, ~20 trials, best for iteration (15-30 min)
- **`medium`**: Balanced optimization, ~50 trials, recommended default (30-60 min)
- **`heavy`**: Exhaustive search, ~100+ trials, highest quality (1-3 hours)

**How MIPROv2 Works**:
1. **Bootstrap Candidates**: Generates few-shot example candidates from training data
2. **Propose Instructions**: Creates instruction variations grounded in task dynamics
3. **Bayesian Optimization**: Uses surrogate model to find optimal instruction + example combinations
4. **Joint Optimization**: Optimizes both components together (not separately) for synergy

**Performance Gains** (2025 Study):
- Prompt Evaluation: +38.5% accuracy (46.2% → 64.0%)
- Guardrail Enforcement: +16.9% accuracy (72.1% → 84.3%)
- Code Generation: +21.9% accuracy (58.4% → 71.2%)
- Hallucination Detection: +20.8% accuracy (65.8% → 79.5%)
- Agent Routing: +18.5% accuracy (69.3% → 82.1%)

from dspy.teleprompt import KNNFewShot

undefined

from dspy.teleprompt import SignatureOptimizer

sig_optimizer = SignatureOptimizer(
    metric=accuracy_metric,
    breadth=10,  # Number of variations to generate
    depth=3      # Optimization iterations
)

optimized_signature = sig_optimizer.compile(
    initial_signature=QuestionAnswer,
    trainset=trainset
)

undefined

undefined

undefined

class CustomTeleprompter:
    """Custom optimization pipeline."""

    def __init__(self, metric):
        self.metric = metric

    def compile(self, student, trainset, valset=None):
        # Stage 1: Bootstrap examples
        bootstrap = BootstrapFewShot(metric=self.metric)
        stage1 = bootstrap.compile(student, trainset=trainset)

        # Stage 2: Optimize instructions
        mipro = MIPROv2(metric=self.metric, auto="light")
        stage2 = mipro.compile(stage1, trainset=trainset)

        # Stage 3: Validate on held-out set
        if valset:
            score = self._evaluate(stage2, valset)
            print(f"Validation score: {score:.2%}")

        return stage2

    def _evaluate(self, program, dataset):
        correct = 0
        for example in dataset:
            prediction = program(**example.inputs())
            if self.metric(example, prediction):
                correct += 1
        return correct / len(dataset)

undefined

def exact_match(example, prediction, trace=None):
    """Exact match metric."""
    return example.answer.lower().strip() == prediction.answer.lower().strip()

from difflib import SequenceMatcher

def fuzzy_match(example, prediction, trace=None):
    """Fuzzy string matching."""
    similarity = SequenceMatcher(
        None,
        example.answer.lower(),
        prediction.answer.lower()
    ).ratio()
    return similarity > 0.8  # 80% similarity threshold

def comprehensive_metric(example, prediction, trace=None):
    """Evaluate on multiple criteria."""
    # Correctness
    correct = example.answer.lower() in prediction.answer.lower()

    # Length appropriateness
    length_ok = 10 < len(prediction.answer) < 200

    # Has reasoning (if CoT)
    has_reasoning = (
        hasattr(prediction, 'rationale') and
        len(prediction.rationale) > 30
    )

    # Citation quality (if RAG)
    has_citations = (
        hasattr(prediction, 'context') and
        len(prediction.context) > 0
    )

    # Composite score
    score = sum([
        correct * 1.0,
        length_ok * 0.2,
        has_reasoning * 0.3,
        has_citations * 0.2
    ]) / 1.7

    return score

def llm_judge_metric(example, prediction, trace=None):
    """Use LLM to evaluate quality."""
    judge_prompt = f"""
    Question: {example.question}
    Expected Answer: {example.answer}
    Predicted Answer: {prediction.answer}

    Evaluate the predicted answer on a scale of 0-10 for:
    1. Correctness
    2. Completeness
    3. Clarity

    Return only a number 0-10.
    """

    judge_lm = dspy.LM('openai/gpt-4o-mini')
    response = judge_lm(judge_prompt)
    score = float(response.strip()) / 10.0

    return score > 0.7  # Pass if score > 7/10

class Evaluator:
    """Comprehensive evaluation system."""

    def __init__(self, program, metrics):
        self.program = program
        self.metrics = metrics

    def evaluate(self, dataset, verbose=True):
        """Evaluate program on dataset."""
        results = {name: [] for name in self.metrics.keys()}

        for example in dataset:
            prediction = self.program(**example.inputs())

            for metric_name, metric_fn in self.metrics.items():
                score = metric_fn(example, prediction)
                results[metric_name].append(score)

        # Aggregate results
        aggregated = {
            name: sum(scores) / len(scores)
            for name, scores in results.items()
        }

        if verbose:
            print("\nEvaluation Results:")
            print("=" * 50)
            for name, score in aggregated.items():
                print(f"{name:20s}: {score:.2%}")

        return aggregated

undefined

import dspy

lm = dspy.LM('openai/gpt-4o', api_key='YOUR_API_KEY')
dspy.configure(lm=lm)

**Anthropic Claude**:
```python
lm = dspy.LM(
    'anthropic/claude-3-5-sonnet-20241022',
    api_key='YOUR_ANTHROPIC_KEY',
    max_tokens=4096
)
dspy.configure(lm=lm)

**Local Models (Ollama)**:
```python

**Multiple Models**:
```python

    # Strong model for reasoning
    with dspy.context(lm=strong_lm):
        self.reason = dspy.ChainOfThought("context, question -> answer")

def forward(self, question):
    context = self.retrieve(question).passages
    return self.reason(context=context, question=question)

undefined

def select_model(task_complexity, budget):
    """Select appropriate model based on task and budget."""
    models = {
        "simple": [
            ("openai/gpt-4o-mini", 0.15),  # (model, cost per 1M tokens)
            ("anthropic/claude-3-haiku-20240307", 0.25),
        ],
        "medium": [
            ("openai/gpt-4o", 2.50),
            ("anthropic/claude-3-5-sonnet-20241022", 3.00),
        ],
        "complex": [
            ("anthropic/claude-3-opus-20240229", 15.00),
            ("openai/o1-preview", 15.00),
        ]
    }

    candidates = models[task_complexity]
    affordable = [m for m, cost in candidates if cost <= budget]

    return affordable[0] if affordable else candidates[0][0]

undefined

class MultiStepPipeline(dspy.Module):
    """Chain multiple reasoning steps."""

    def __init__(self):
        super().__init__()
        self.step1 = dspy.ChainOfThought("question -> subtasks")
        self.step2 = dspy.ChainOfThought("subtask -> solution")
        self.step3 = dspy.ChainOfThought("solutions -> final_answer")

    def forward(self, question):
        # Break down question
        decomposition = self.step1(question=question)

        # Solve each subtask
        solutions = []
        for subtask in decomposition.subtasks.split('\n'):
            if subtask.strip():
                sol = self.step2(subtask=subtask)
                solutions.append(sol.solution)

        # Synthesize final answer
        combined = '\n'.join(solutions)
        final = self.step3(solutions=combined)

        return dspy.Prediction(
            answer=final.final_answer,
            subtasks=decomposition.subtasks,
            solutions=solutions
        )

undefined

class AdaptivePipeline(dspy.Module):
    """Adapt reasoning based on query type."""

    def __init__(self):
        super().__init__()
        self.classifier = dspy.ChainOfThought("question -> category")
        self.math_solver = dspy.ProgramOfThought("problem -> solution")
        self.fact_qa = dspy.ChainOfThought("question -> answer")
        self.creative = dspy.ChainOfThought("prompt -> response")

    def forward(self, question):
        # Classify query type
        category = self.classifier(question=question).category.lower()

        # Route to appropriate module
        if "math" in category or "calculation" in category:
            return self.math_solver(problem=question)
        elif "creative" in category or "story" in category:
            return self.creative(prompt=question)
        else:
            return self.fact_qa(question=question)

undefined

undefined

undefined

import json
from datetime import datetime

class ModelRegistry:
    """Version control for DSPy models."""

    def __init__(self, registry_path="models/registry.json"):
        self.registry_path = registry_path
        self.registry = self._load_registry()

    def register(self, name, version, model_path, metadata=None):
        """Register a model version."""
        model_id = f"{name}:v{version}"

        self.registry[model_id] = {
            "name": name,
            "version": version,
            "path": model_path,
            "created_at": datetime.utcnow().isoformat(),
            "metadata": metadata or {}
        }

        self._save_registry()
        return model_id

    def get_model(self, name, version="latest"):
        """Load model by name and version."""
        if version == "latest":
            versions = [
                v for k, v in self.registry.items()
                if v["name"] == name
            ]
            if not versions:
                raise ValueError(f"No versions found for {name}")

            latest = max(versions, key=lambda x: x["created_at"])
            model_path = latest["path"]
        else:
            model_id = f"{name}:v{version}"
            model_path = self.registry[model_id]["path"]

        # Load model
        module = dspy.ChainOfThought("question -> answer")
        module.load(model_path)
        return module

    def _load_registry(self):
        try:
            with open(self.registry_path, 'r') as f:
                return json.load(f)
        except FileNotFoundError:
            return {}

    def _save_registry(self):
        with open(self.registry_path, 'w') as f:
            json.dump(self.registry, f, indent=2)

undefined

import logging
from datetime import datetime

class DSPyMonitor:
    """Monitor DSPy program execution."""

    def __init__(self, program, log_file="logs/dspy.log"):
        self.program = program
        self.logger = self._setup_logger(log_file)
        self.metrics = []

    def __call__(self, **kwargs):
        """Wrap program execution with monitoring."""
        start_time = datetime.utcnow()

        try:
            # Execute program
            result = self.program(**kwargs)

            # Log success
            duration = (datetime.utcnow() - start_time).total_seconds()
            self._log_execution(
                status="success",
                inputs=kwargs,
                outputs=result,
                duration=duration
            )

            return result

        except Exception as e:
            # Log error
            duration = (datetime.utcnow() - start_time).total_seconds()
            self._log_execution(
                status="error",
                inputs=kwargs,
                error=str(e),
                duration=duration
            )
            raise

    def _log_execution(self, status, inputs, duration, outputs=None, error=None):
        """Log execution details."""
        log_entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "status": status,
            "inputs": inputs,
            "duration_seconds": duration
        }

        if outputs:
            log_entry["outputs"] = str(outputs)
        if error:
            log_entry["error"] = error

        self.logger.info(json.dumps(log_entry))
        self.metrics.append(log_entry)

    def _setup_logger(self, log_file):
        """Setup logging."""
        logger = logging.getLogger("dspy_monitor")
        logger.setLevel(logging.INFO)

        handler = logging.FileHandler(log_file)
        handler.setFormatter(
            logging.Formatter('%(asctime)s - %(message)s')
        )
        logger.addHandler(handler)

        return logger

    def get_stats(self):
        """Get execution statistics."""
        if not self.metrics:
            return {}

        successes = [m for m in self.metrics if m["status"] == "success"]
        errors = [m for m in self.metrics if m["status"] == "error"]

        return {
            "total_calls": len(self.metrics),
            "success_rate": len(successes) / len(self.metrics),
            "error_rate": len(errors) / len(self.metrics),
            "avg_duration": sum(m["duration_seconds"] for m in self.metrics) / len(self.metrics),
            "errors": [m["error"] for m in errors]
        }

undefined

import os
from langsmith import Client
from langsmith.evaluation import evaluate

return {
    "key": "correctness",
    "score": 1.0 if expected.lower() in predicted.lower() else 0.0
}

undefined

class ProductionRAG(dspy.Module):
    """Production-ready RAG system."""

    def __init__(self, k=5):
        super().__init__()
        self.retrieve = dspy.Retrieve(k=k)

        # Multi-stage reasoning
        self.rerank = dspy.ChainOfThought(
            "question, passages -> relevant_passages"
        )
        self.generate = dspy.ChainOfThought(
            "question, context -> answer, citations"
        )

    def forward(self, question):
        # Retrieve candidate passages
        candidates = self.retrieve(question).passages

        # Rerank for relevance
        reranked = self.rerank(
            question=question,
            passages="\n---\n".join(candidates)
        )

        # Generate answer with citations
        result = self.generate(
            question=question,
            context=reranked.relevant_passages
        )

        return dspy.Prediction(
            answer=result.answer,
            citations=result.citations,
            passages=candidates
        )

undefined

class SentimentClassifier(dspy.Module):
    """Multi-class sentiment classification."""

    def __init__(self, classes):
        super().__init__()
        self.classes = classes

        class ClassificationSig(dspy.Signature):
            text = dspy.InputField()
            reasoning = dspy.OutputField(desc="step-by-step reasoning")
            sentiment = dspy.OutputField(desc=f"one of: {', '.join(classes)}")
            confidence = dspy.OutputField(desc="confidence score 0-1")

        self.classify = dspy.ChainOfThought(ClassificationSig)

    def forward(self, text):
        result = self.classify(text=text)

        # Validate output
        if result.sentiment not in self.classes:
            result.sentiment = "neutral"  # Fallback

        return result