DSPy

import dspy

from dspy import (
    Module,
    Signature,
    Predict,
    ChainOfThought,
    Example,
    configure,
)

# Prediction modules
from dspy import ReAct, ProgramOfThought, CodeAct, BestOfN

# Optimization
from dspy import BootstrapFewShot, MIPROv2, Evaluate

# Language models
from dspy import LM, Embedder

# Retrieval
from dspy import Retrieve

# Signatures
from dspy import InputField, OutputField

# Datasets
from dspy.datasets import Dataset, DataLoader, HotPotQA, MATH, Colors

# Utilities
from dspy import asyncify, syncify, load, track_usage

import dspy

# Configure with a language model
dspy.configure(lm=dspy.LM('openai/gpt-4o-mini'))

# Create a simple question-answering module
qa = dspy.ChainOfThought("question -> answer")

# Use the module
result = qa(question="What is machine learning?")
print(result.answer)
print(result.reasoning)  # Chain-of-thought reasoning

class RAG(dspy.Module):
    """Retrieval-Augmented Generation system."""

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

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

        # Generate answer from context
        return self.generate(context=context, question=question)

# Configure with retrieval model
dspy.configure(
    lm=dspy.LM('openai/gpt-4o-mini'),
    rm=my_retrieval_model
)

# Use the RAG system
rag = RAG(k=5)
result = rag(question="What is the capital of France?")
print(result.answer)

# Define validation metric
def validate_answer(example, pred, trace=None):
    """Check if prediction matches expected answer."""
    return example.answer.lower() == pred.answer.lower()

# Create training examples
trainset = [
    dspy.Example(question="What is 2+2?", answer="4").with_inputs("question"),
    dspy.Example(question="Capital of Spain?", answer="Madrid").with_inputs("question"),
    # ... more examples
]

# Optimize the program
optimizer = dspy.BootstrapFewShot(
    metric=validate_answer,
    max_bootstrapped_demos=4
)
compiled_rag = optimizer.compile(rag, trainset=trainset)

# The compiled program now has optimized demonstrations
result = compiled_rag(question="What is the capital of Japan?")

class Signature:
    """Base class for defining task signatures."""
    pass

class Module:
    """Base class for all DSPy modules."""

    def __call__(self, *args, **kwargs):
        """Execute the module."""
        pass

    def forward(self, *args, **kwargs):
        """Forward pass (implement in subclasses)."""
        pass

class LM:
    """Language model client."""

    def __init__(self, model: str, **kwargs):
        """
        Initialize language model.

        Args:
            model: Model identifier (e.g., "openai/gpt-4o-mini")
            **kwargs: Additional configuration (temperature, max_tokens, etc.)
        """
        pass

class Teleprompter:
    """Base class for optimizers."""

    def compile(self, student, *, trainset, **kwargs):
        """
        Optimize and compile a program.

        Args:
            student: Module to optimize
            trainset: Training examples
            **kwargs: Additional optimization parameters

        Returns:
            Compiled module
        """
        pass

def configure(
    lm=None,
    adapter=None,
    rm=None,
    trace=None,
    callbacks=None,
    async_max_workers=None,
    track_usage=False,
    **kwargs
):
    """
    Configure global DSPy settings.

    Args:
        lm: Language model instance
        adapter: Adapter for formatting prompts
        rm: Retrieval model
        trace: List to store execution traces
        callbacks: Callback functions
        async_max_workers: Max async workers
        track_usage: Enable token usage tracking
        **kwargs: Additional settings
    """
    pass

def context(**kwargs):
    """
    Create context manager for temporary settings.

    Returns:
        Context manager
    """
    pass

class Signature:
    """Task signature with input/output fields."""

    instructions: str
    input_fields: dict
    output_fields: dict

def InputField(
    desc: str = None,
    prefix: str = None,
    format: callable = None,
    **kwargs
):
    """
    Create input field for signatures.

    Args:
        desc: Field description
        prefix: Display prefix
        format: Formatting function
        **kwargs: Pydantic Field parameters

    Returns:
        Field definition
    """
    pass

def OutputField(desc: str = None, prefix: str = None, **kwargs):
    """
    Create output field for signatures.

    Args:
        desc: Field description
        prefix: Display prefix
        **kwargs: Pydantic Field parameters

    Returns:
        Field definition
    """
    pass

def make_signature(
    signature,
    instructions: str = None,
    custom_types: dict = None
):
    """
    Create signature dynamically from string or dict.

    Args:
        signature: String spec ("input -> output") or field dict
        instructions: Task instructions
        custom_types: Custom type mappings

    Returns:
        Signature class
    """
    pass

class Module:
    """Base class for DSPy modules."""

    def __init__(self, callbacks=None):
        """Initialize module with optional callbacks."""
        pass

    def forward(self, *args, **kwargs):
        """Forward pass (implement in subclasses)."""
        pass

    def named_predictors(self):
        """Get all named Predict instances."""
        pass

    def save(self, path: str, **kwargs):
        """Save module to disk."""
        pass

class Example:
    """Data container for examples and training data."""

    def __init__(self, base=None, **kwargs):
        """
        Create example with key-value pairs.

        Args:
            base: Base data to copy
            **kwargs: Field values
        """
        pass

    def with_inputs(self, *keys):
        """Mark fields as inputs."""
        pass

class Prediction:
    """Prediction result from module execution."""

    completions: object

    def get_lm_usage(self):
        """Get token usage information."""
        pass

class Predict:
    """Basic prediction module mapping inputs to outputs."""

    def __init__(self, signature, callbacks=None, **config):
        """
        Initialize predictor.

        Args:
            signature: Task signature
            callbacks: Optional callbacks
            **config: Default LM parameters
        """
        pass

class ChainOfThought:
    """Chain-of-thought reasoning module."""

    def __init__(self, signature, rationale_field=None, **config):
        """
        Initialize chain-of-thought module.

        Args:
            signature: Task signature
            rationale_field: Field for reasoning
            **config: LM parameters
        """
        pass

class ReAct:
    """Reasoning and Acting agent with tool use."""

    def __init__(self, signature, tools: list, max_iters: int = 10):
        """
        Initialize ReAct agent.

        Args:
            signature: Task signature
            tools: List of callable tools
            max_iters: Maximum iterations
        """
        pass

class ProgramOfThought:
    """Generate and execute Python code to solve problems."""

    def __init__(self, signature, max_iters: int = 3, interpreter=None):
        """
        Initialize program-of-thought module.

        Args:
            signature: Task signature
            max_iters: Max code generation retries
            interpreter: Python interpreter instance
        """
        pass

class CodeAct:
    """Combine code interpretation with tool use."""

    def __init__(self, signature, tools: list, max_iters: int = 5, interpreter=None):
        """
        Initialize CodeAct module.

        Args:
            signature: Task signature
            tools: Available tools
            max_iters: Maximum iterations
            interpreter: Python interpreter
        """
        pass

class BootstrapFewShot:
    """Bootstrap few-shot demonstrations from examples."""

    def __init__(
        self,
        metric=None,
        metric_threshold=None,
        teacher_settings=None,
        max_bootstrapped_demos: int = 4,
        max_labeled_demos: int = 16,
        max_rounds: int = 1,
        max_errors=None
    ):
        """
        Initialize bootstrap optimizer.

        Args:
            metric: Validation metric function
            metric_threshold: Threshold for accepting demos
            teacher_settings: Settings for teacher model
            max_bootstrapped_demos: Max bootstrapped demonstrations
            max_labeled_demos: Max labeled demonstrations
            max_rounds: Bootstrap rounds
            max_errors: Max errors before stopping
        """
        pass

    def compile(self, student, teacher=None, trainset=None):
        """Compile program with bootstrapped demonstrations."""
        pass

class MIPROv2:
    """Multi-prompt Instruction Proposal Optimizer."""

    def __init__(
        self,
        metric,
        prompt_model=None,
        task_model=None,
        auto: str = "light",
        num_candidates=None,
        max_bootstrapped_demos: int = 4,
        **kwargs
    ):
        """
        Initialize MIPROv2 optimizer.

        Args:
            metric: Evaluation metric
            prompt_model: Model for generating prompts
            task_model: Model for task execution
            auto: Auto-config mode ("light", "medium", "heavy")
            num_candidates: Number of candidates to try
            max_bootstrapped_demos: Max demonstrations
            **kwargs: Additional parameters
        """
        pass

class LabeledFewShot:
    """Simple few-shot learning with labeled examples."""

    def __init__(self, k: int = 16):
        """
        Initialize labeled few-shot optimizer.

        Args:
            k: Number of examples per predictor
        """
        pass

class LM:
    """Main language model class."""

    def __init__(
        self,
        model: str,
        model_type: str = "chat",
        temperature: float = None,
        max_tokens: int = None,
        cache: bool = True,
        callbacks=None,
        num_retries: int = 3,
        **kwargs
    ):
        """
        Initialize language model.

        Args:
            model: Model identifier (e.g., "openai/gpt-4o-mini")
            model_type: Type ("chat", "text", "responses")
            temperature: Sampling temperature
            max_tokens: Max tokens per response
            cache: Enable caching
            callbacks: Callback functions
            num_retries: Retry attempts
            **kwargs: Additional parameters (n, rollout_id, etc.)
        """
        pass

    def __call__(self, prompt=None, messages=None, **kwargs):
        """Generate completion."""
        pass

    def inspect_history(self, n: int = 1):
        """Inspect recent calls."""
        pass

class Embedder:
    """Text embedding class."""

    def __init__(self, model, batch_size: int = 200, caching: bool = True, **kwargs):
        """
        Initialize embedder.

        Args:
            model: Model name or custom function
            batch_size: Batch size for processing
            caching: Cache responses
            **kwargs: Additional parameters
        """
        pass

    def __call__(self, texts: list, **kwargs):
        """
        Compute embeddings.

        Args:
            texts: List of texts to embed
            **kwargs: Additional parameters

        Returns:
            numpy array of embeddings
        """
        pass

def inspect_history(lm_or_module, n: int = 1):
    """
    Inspect call history of LM or module.

    Args:
        lm_or_module: LM or Module instance
        n: Number of recent calls
    """
    pass

def configure_cache(
    enable_disk_cache: bool = None,
    enable_memory_cache: bool = None,
    disk_cache_dir: str = None,
    disk_size_limit_bytes: int = None,
    memory_max_entries: int = None
):
    """
    Configure caching system.

    Args:
        enable_disk_cache: Enable disk cache
        enable_memory_cache: Enable memory cache
        disk_cache_dir: Cache directory path
        disk_size_limit_bytes: Disk cache size limit
        memory_max_entries: Max memory cache entries
    """
    pass

class ChatAdapter:
    """Chat-based adapter for conversation models."""

    def __init__(self, callbacks=None, **kwargs):
        """Initialize chat adapter."""
        pass

class JSONAdapter:
    """JSON-based adapter for structured I/O."""

    def __init__(self, callbacks=None, **kwargs):
        """Initialize JSON adapter."""
        pass

class Tool:
    """Wrapper for functions used as tools."""

    def __init__(
        self,
        func,
        name: str = None,
        desc: str = None,
        args: dict = None
    ):
        """
        Create tool from function.

        Args:
            func: Function to wrap
            name: Tool name
            desc: Tool description
            args: Argument schemas
        """
        pass

class Image:
    """Image input type for multimodal models."""
    pass

class Audio:
    """Audio input type."""
    pass

class History:
    """Conversation history type."""
    pass

class Retrieve:
    """Retrieval module for fetching passages."""

    def __init__(self, k: int = 3, callbacks=None):
        """
        Initialize retrieval module.

        Args:
            k: Number of passages to retrieve
            callbacks: Optional callbacks
        """
        pass

    def forward(self, query: str, k: int = None, **kwargs):
        """
        Retrieve passages for query.

        Args:
            query: Search query
            k: Override number of passages
            **kwargs: Additional parameters

        Returns:
            Prediction with 'passages' field
        """
        pass

class Embeddings:
    """Embedding-based retrieval using vector similarity."""

    def __init__(self, embedder, documents: list = None, k: int = 3):
        """Initialize embeddings retrieval."""
        pass

class ColBERTv2:
    """ColBERTv2 retrieval model for dense passage retrieval."""

    def __init__(self, url: str = "http://0.0.0.0", port: int = None, **kwargs):
        """Initialize ColBERT retriever."""
        pass

class WeaviateRM:
    """Weaviate vector database retrieval model."""

    def __init__(
        self,
        weaviate_collection_name: str,
        weaviate_client=None,
        k: int = 3
    ):
        """Initialize Weaviate retrieval model."""
        pass

class DatabricksRM:
    """Databricks Vector Search retrieval model."""

    def __init__(
        self,
        databricks_index_name: str,
        databricks_endpoint: str = None,
        k: int = 3
    ):
        """Initialize Databricks retrieval model."""
        pass

class Evaluate:
    """Evaluate DSPy programs on datasets."""

    def __init__(
        self,
        devset: list,
        metric=None,
        num_threads: int = None,
        display_progress: bool = False,
        display_table: bool = False,
        max_errors: int = None,
        failure_score: float = 0.0,
        **kwargs
    ):
        """
        Initialize evaluator.

        Args:
            devset: Evaluation dataset
            metric: Metric function
            num_threads: Threads for parallel evaluation
            display_progress: Show progress bar
            display_table: Display results table
            max_errors: Max errors before stopping
            failure_score: Score for failed examples
            **kwargs: Additional parameters
        """
        pass

    def __call__(self, program, **kwargs):
        """
        Evaluate program on dataset.

        Args:
            program: Module to evaluate
            **kwargs: Additional parameters

        Returns:
            EvaluationResult with score and details
        """
        pass

def EM(prediction: str, answers_list: list) -> bool:
    """
    Exact match metric after normalization.

    Args:
        prediction: Predicted answer
        answers_list: Reference answers

    Returns:
        True if exact match
    """
    pass

def F1(prediction: str, answers_list: list) -> float:
    """
    Token-level F1 score.

    Args:
        prediction: Predicted answer
        answers_list: Reference answers

    Returns:
        F1 score (0.0 to 1.0)
    """
    pass

def answer_exact_match(example, pred, trace=None, frac: float = 1.0) -> bool:
    """
    Metric function for exact match evaluation.

    Args:
        example: Example with 'answer' field
        pred: Prediction with 'answer' field
        trace: Unused, for compatibility
        frac: Threshold (1.0 = exact match)

    Returns:
        True if match
    """
    pass

def streamify(module):
    """
    Enable streaming for a module.

    Args:
        module: Module to enable streaming for

    Returns:
        StreamListener context manager
    """
    pass

class Dataset:
    """Base class for structured datasets with train/dev/test splits."""

    def __init__(
        self,
        train_seed: int = 0,
        train_size: int = None,
        eval_seed: int = 0,
        dev_size: int = None,
        test_size: int = None,
        input_keys: list = None
    ):
        """Initialize dataset with split configuration."""
        pass

    @property
    def train(self) -> list:
        """Get training split."""
        pass

    @property
    def dev(self) -> list:
        """Get development/validation split."""
        pass

    @property
    def test(self) -> list:
        """Get test split."""
        pass

class DataLoader:
    """Universal data loader for various file formats and sources."""

    def from_huggingface(
        self,
        dataset_name: str,
        *args,
        input_keys: tuple = (),
        fields: tuple = None,
        **kwargs
    ):
        """Load dataset from HuggingFace Hub."""
        pass

    def from_csv(self, file_path: str, fields: list = None, input_keys: tuple = ()) -> list:
        """Load dataset from CSV file."""
        pass

    def from_json(self, file_path: str, fields: list = None, input_keys: tuple = ()) -> list:
        """Load dataset from JSON file."""
        pass

    def from_pandas(self, df, fields: list = None, input_keys: tuple = ()) -> list:
        """Load dataset from pandas DataFrame."""
        pass

class HotPotQA(Dataset):
    """HotPotQA multi-hop question answering dataset."""
    pass

class MATH(Dataset):
    """MATH dataset for mathematical problem solving."""
    pass

class Colors(Dataset):
    """Colors dataset for testing and examples."""
    pass

def asyncify(func):
    """
    Convert sync function to async.

    Args:
        func: Synchronous function

    Returns:
        Async version of function
    """
    pass

def syncify(func):
    """
    Convert async function to sync.

    Args:
        func: Async function

    Returns:
        Sync version of function
    """
    pass

def load(path: str, allow_pickle: bool = False):
    """
    Load saved module from disk.

    Args:
        path: Path to saved module
        allow_pickle: Allow pickle loading (security)

    Returns:
        Module instance
    """
    pass

def track_usage():
    """
    Track token usage in context.

    Returns:
        UsageTracker context manager
    """
    pass

def enable_logging():
    """Enable DSPy logging output."""
    pass

def disable_logging():
    """Disable DSPy logging output."""
    pass