tessl/pypi-dspy

A declarative framework for programming foundation models through compositional Python code, enabling modular AI systems with automated optimization algorithms that synthesize examples, generate instructions, and finetune weights based on user-defined metrics.

Overview

Eval results

Files

Configuration & Settings

Name: tessl/pypi-dspy
Author: tessl

Global configuration system for DSPy that manages language models, adapters, retrieval models, and execution settings. The configuration system is thread-safe and supports context-based temporary overrides.

Capabilities

Global Configuration

Configure DSPy's global settings for language models, adapters, and execution parameters.

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

    Args:
        lm (BaseLM | None): Language model instance to use globally
        adapter (Adapter | None): Adapter for formatting prompts and parsing outputs
        rm (Any | None): Retrieval model for document retrieval
        trace (list | None): List to store execution traces for optimization
        callbacks (list | None): Global callback functions for monitoring
        async_max_workers (int | None): Maximum number of async workers
        track_usage (bool): Enable automatic token usage tracking (default: False)
        num_threads (int | None): Number of threads for parallel processing
        max_errors (int | None): Maximum errors before halting execution
        **kwargs: Additional configuration options
    """
    pass

Usage:

import dspy

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

# Configuration with multiple settings
dspy.configure(
    lm=dspy.LM('openai/gpt-4o-mini', temperature=0.7),
    rm=my_retrieval_model,
    num_threads=4,
    track_usage=True
)

# Configuration with custom adapter
dspy.configure(
    lm=dspy.LM('openai/gpt-4o-mini'),
    adapter=dspy.JSONAdapter()
)

Context Manager

Create temporary configuration overrides that automatically revert when exiting the context. Thread-safe and propagates to child threads.

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

    Args:
        **kwargs: Settings to override (same as configure parameters)

    Returns:
        Context manager that restores previous settings on exit
    """
    pass

Usage:

import dspy

# Configure default LM
dspy.configure(lm=dspy.LM('openai/gpt-4o-mini'))

# Temporarily use different LM
with dspy.context(lm=dspy.LM('anthropic/claude-3-5-sonnet-20241022')):
    result = module(question="Question")
    # Uses Claude inside this context

# Back to using GPT-4o-mini
result = module(question="Another question")

# Nested contexts are supported
with dspy.context(temperature=0.0):
    with dspy.context(max_tokens=100):
        # Both temperature and max_tokens overrides are active
        result = module(question="Question")

Settings Object

Access current configuration through the global settings object. Provides read-only access to active settings.

class Settings:
    """Global settings object providing access to current configuration."""

    lm: object
    """Current language model instance."""

    adapter: object
    """Current adapter instance."""

    rm: object
    """Current retrieval model instance."""

    trace: list
    """Execution trace list for optimization."""

    callbacks: list
    """Global callback functions."""

    num_threads: int
    """Number of threads for parallel processing."""

    max_errors: int
    """Maximum errors before halting."""

    track_usage: bool
    """Whether token usage tracking is enabled."""

    async_max_workers: int
    """Maximum async workers."""

    disable_history: bool
    """Whether to disable history tracking."""

Usage:

import dspy

# Access current settings
current_lm = dspy.settings.lm
current_adapter = dspy.settings.adapter

# Check if usage tracking is enabled
if dspy.settings.track_usage:
    print("Usage tracking is active")

# Access trace for debugging
trace_list = dspy.settings.trace

ColBERT Retrieval

ColBERTv2 retrieval model integration for dense passage retrieval. Used for document retrieval in RAG systems.

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.

        Args:
            url (str): ColBERT server URL (default: "http://0.0.0.0")
            port (int | str): Server port
            **kwargs: Additional ColBERT-specific parameters
        """
        pass

    def __call__(self, query: str, k: int = 10, **kwargs):
        """
        Retrieve documents for query.

        Args:
            query (str): Search query
            k (int): Number of documents to retrieve
            **kwargs: Additional parameters

        Returns:
            List of retrieved passages
        """
        pass

Usage:

import dspy

# Initialize ColBERT retriever
colbert = dspy.ColBERTv2(url="http://localhost", port=8893)

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

# Use with Retrieve module
retrieve = dspy.Retrieve(k=5)
result = retrieve(query="What is machine learning?")
print(result.passages)

Cache Configuration

Configure the caching system for language model responses. Supports both disk and memory caching.

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 DSPy caching system.

    Args:
        enable_disk_cache (bool | None): Enable disk cache (default: True)
        enable_memory_cache (bool | None): Enable memory cache (default: True)
        disk_cache_dir (str | None): Directory path for disk cache
        disk_size_limit_bytes (int | None): Maximum disk cache size in bytes
        memory_max_entries (int | None): Maximum number of memory cache entries
    """
    pass

Usage:

import dspy

# Configure cache settings
dspy.configure_cache(
    enable_disk_cache=True,
    enable_memory_cache=True,
    disk_cache_dir="/tmp/dspy_cache",
    disk_size_limit_bytes=1_000_000_000,  # 1GB
    memory_max_entries=1000
)

# Disable caching
dspy.configure_cache(
    enable_disk_cache=False,
    enable_memory_cache=False
)

Cache Object

Global cache instance for storing and retrieving language model responses.

class Cache:
    """Global cache for LM responses."""

    def clear(self):
        """Clear all cached responses."""
        pass

    def get_stats(self):
        """Get cache statistics."""
        pass

Usage:

import dspy

# Clear cache
dspy.cache.clear()

# Get cache statistics
stats = dspy.cache.get_stats()
print(f"Cache hits: {stats['hits']}, misses: {stats['misses']}")

Logging Configuration

Control logging output from DSPy and LiteLLM.

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

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

def enable_litellm_logging():
    """Enable LiteLLM logging output."""
    pass

def disable_litellm_logging():
    """Disable LiteLLM logging output."""
    pass

def configure_dspy_loggers(name: str):
    """
    Configure DSPy's logging system.

    Args:
        name (str): Logger name
    """
    pass

Usage:

import dspy

# Enable logging for debugging
dspy.enable_logging()

# Disable verbose LiteLLM logs
dspy.disable_litellm_logging()

# Configure specific logger
dspy.configure_dspy_loggers("dspy.predict")

Configuration Patterns

Multi-Model Setup

Use different models for different purposes:

import dspy

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

# Use stronger model in specific context
with dspy.context(lm=dspy.LM('openai/gpt-4o')):
    # Complex reasoning with GPT-4o
    result = complex_module(input=data)

# Use cheaper model for simple tasks
with dspy.context(lm=dspy.LM('openai/gpt-3.5-turbo')):
    # Simple classification with GPT-3.5
    result = simple_classifier(text=text)

Development vs Production

Different configurations for different environments:

import dspy
import os

if os.getenv("ENV") == "production":
    # Production: caching enabled, parallel processing
    dspy.configure(
        lm=dspy.LM('openai/gpt-4o-mini', temperature=0.0),
        num_threads=8,
        track_usage=True
    )
    dspy.configure_cache(enable_disk_cache=True)
else:
    # Development: no caching, verbose logging
    dspy.configure(
        lm=dspy.LM('openai/gpt-4o-mini', temperature=0.7),
        num_threads=1
    )
    dspy.configure_cache(enable_disk_cache=False)
    dspy.enable_logging()

Optimization Setup

Configuration for program optimization:

import dspy

# Configure teacher model for optimization
teacher_lm = dspy.LM('openai/gpt-4o', temperature=0.8)

# Configure task model
task_lm = dspy.LM('openai/gpt-4o-mini', temperature=0.0)

# Set up optimizer with teacher
dspy.configure(lm=task_lm)

optimizer = dspy.BootstrapFewShot(
    metric=my_metric,
    teacher_settings={"lm": teacher_lm}
)

# Compile program
compiled = optimizer.compile(program, trainset=trainset)

Thread Safety

The configuration system is thread-safe:

Each thread can have its own context overrides
Context overrides propagate to child threads
The global settings object uses thread-local storage

import dspy
import threading

# Configure globally
dspy.configure(lm=dspy.LM('openai/gpt-4o-mini'))

def worker_function():
    # This thread has its own context
    with dspy.context(temperature=0.5):
        result = module(input=data)
        # Uses temperature=0.5 in this thread

# Main thread uses default settings
result = module(input=data)

# Worker thread has independent context
thread = threading.Thread(target=worker_function)
thread.start()

Install with Tessl CLI