Core data structures and base classes for building composable AI programs in DSPy. Modules provide parameter management, execution tracking, and optimization support.
Base class for all DSPy modules. Provides composability, parameter management, and execution tracking.
class Module:
"""
Base class for all DSPy modules.
Modules are composable building blocks that can be nested and combined
to create complex AI pipelines. They handle execution, parameter tracking,
and optimization support.
"""
def __init__(self, callbacks=None):
"""
Initialize module.
Args:
callbacks (list | None): Callback functions for monitoring
"""
pass
def __call__(self, *args, **kwargs):
"""
Execute module synchronously.
Args:
*args: Positional arguments
**kwargs: Keyword arguments
Returns:
Prediction or module-specific result
"""
pass
def acall(self, *args, **kwargs):
"""
Execute module asynchronously.
Args:
*args: Positional arguments
**kwargs: Keyword arguments
Returns:
Awaitable that resolves to Prediction or module result
"""
pass
def forward(self, *args, **kwargs):
"""
Forward pass (must be implemented by subclasses).
Args:
*args: Positional arguments
**kwargs: Keyword arguments
Returns:
Prediction or module-specific result
"""
pass
def aforward(self, *args, **kwargs):
"""
Async forward pass.
Args:
*args: Positional arguments
**kwargs: Keyword arguments
Returns:
Awaitable that resolves to result
"""
pass
def named_predictors(self):
"""
Get all named Predict instances in module.
Returns:
Iterator of (name, predictor) tuples
"""
pass
def predictors(self):
"""
Get all Predict instances in module.
Returns:
List of predictor instances
"""
pass
def named_parameters(self):
"""
Get all named parameters (predictors and sub-modules).
Returns:
Iterator of (name, parameter) tuples
"""
pass
def parameters(self):
"""
Get all parameters.
Returns:
List of parameter instances
"""
pass
def set_lm(self, lm):
"""
Set language model for all predictors in module.
Args:
lm: Language model instance
"""
pass
def get_lm(self):
"""
Get language model (if all predictors use same LM).
Returns:
Language model instance
Raises:
ValueError: If predictors use different LMs
"""
pass
def batch(
self,
examples,
num_threads=None,
max_errors=None,
return_failed_examples=False,
provide_traceback=None,
disable_progress_bar=False
):
"""
Process examples in parallel batches.
Args:
examples (list): List of Example or dict inputs
num_threads (int | None): Number of threads
max_errors (int | None): Max errors before stopping
return_failed_examples (bool): Return failed examples
provide_traceback (bool | None): Include tracebacks
disable_progress_bar (bool): Disable progress display
Returns:
List of Prediction results
"""
pass
def inspect_history(self, n: int = 1):
"""
Pretty-print execution history.
Args:
n (int): Number of recent calls to display (default: 1)
"""
pass
def deepcopy(self):
"""
Create deep copy of module.
Returns:
Deep copied module instance
"""
pass
def reset_copy(self):
"""
Create reset copy of module (clears state).
Returns:
Reset module instance
"""
pass
def save(
self,
path: str,
save_program: bool = False,
save_program_only: bool = False
):
"""
Save module to disk.
Args:
path (str): Directory path to save to
save_program (bool): Save complete program code
save_program_only (bool): Save only program code
"""
pass
@staticmethod
def load(path: str):
"""
Load module from disk.
Args:
path (str): Directory path to load from
Returns:
Loaded module instance
"""
pass
def dump_state(self) -> dict:
"""
Serialize module state to dictionary.
Returns:
Dictionary representation of module state
"""
pass
def load_state(self, state: dict):
"""
Load module state from dictionary.
Args:
state (dict): Serialized module state
"""
passUsage:
import dspy
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):
context = self.retrieve(query=question).passages
return self.generate(context=context, question=question)
# Configure and use
dspy.configure(lm=dspy.LM('openai/gpt-4o-mini'), rm=my_rm)
rag = RAG(k=5)
result = rag(question="What is machine learning?")
print(result.answer)
# Batch processing
questions = [
{"question": "What is ML?"},
{"question": "What is AI?"},
]
results = rag.batch(questions, num_threads=4)
# Save and load
rag.save("my_rag/")
loaded_rag = dspy.Module.load("my_rag/")Lower-level base class providing core module functionality. Module extends BaseModule.
class BaseModule:
"""
Lower-level base class for modules.
Provides core functionality for parameter management and serialization.
Most users should extend Module instead of BaseModule.
"""
def named_parameters(self):
"""
Iterate over named parameters.
Returns:
Iterator of (name, parameter) tuples
"""
pass
def parameters(self):
"""
Get all parameters.
Returns:
List of parameters
"""
pass
def deepcopy(self):
"""
Create deep copy.
Returns:
Deep copied instance
"""
pass
def reset_copy(self):
"""
Create reset copy (clears state).
Returns:
Reset instance
"""
pass
def save(self, path: str, save_program: bool = False):
"""
Save to disk.
Args:
path (str): Directory path
save_program (bool): Save program code
"""
pass
@staticmethod
def load(path: str):
"""
Load from disk.
Args:
path (str): Directory path
Returns:
Loaded instance
"""
passFlexible data container for DSPy examples and training data. Provides dictionary-like access with input/output separation.
class Example:
"""
Data container for examples and training data.
Supports dictionary-like access and separates input fields from
output/label fields for training and evaluation.
"""
def __init__(self, base=None, **kwargs):
"""
Create example with key-value pairs.
Args:
base (Example | dict | None): Base data to copy
**kwargs: Field key-value pairs
"""
pass
def with_inputs(self, *keys):
"""
Mark specified fields as inputs.
Fields not marked as inputs are treated as labels/outputs.
Args:
*keys: Field names to mark as inputs
Returns:
New Example instance with marked inputs
"""
pass
def inputs(self) -> dict:
"""
Get input fields only.
Returns:
Dictionary of input fields
"""
pass
def labels(self) -> dict:
"""
Get output/label fields only.
Returns:
Dictionary of output fields
"""
pass
def copy(self, **kwargs) -> "Example":
"""
Create copy with optional updates.
Args:
**kwargs: Fields to update in copy
Returns:
New Example instance
"""
pass
def without(self, *keys) -> "Example":
"""
Create copy without specified keys.
Args:
*keys: Field names to exclude
Returns:
New Example without specified fields
"""
pass
def toDict(self) -> dict:
"""
Convert to dictionary.
Returns:
Dictionary representation
"""
pass
def keys(self):
"""Get field names."""
pass
def values(self):
"""Get field values."""
pass
def items(self):
"""Get field items as (name, value) pairs."""
pass
def get(self, key, default=None):
"""
Get field value with default.
Args:
key: Field name
default: Default value if key not found
Returns:
Field value or default
"""
pass
def __getattr__(self, key):
"""Get field value by attribute access."""
pass
def __getitem__(self, key):
"""Get field value by index access."""
pass
def __setitem__(self, key, value):
"""Set field value by index access."""
pass
def __contains__(self, key):
"""Check if field exists."""
passUsage:
import dspy
# Create example
example = dspy.Example(
question="What is 2+2?",
answer="4",
explanation="Basic arithmetic"
)
# Mark inputs
example = example.with_inputs("question")
# Access fields
print(example.question) # "What is 2+2?"
print(example["answer"]) # "4"
# Get inputs and labels
inputs = example.inputs() # {"question": "What is 2+2?"}
labels = example.labels() # {"answer": "4", "explanation": "Basic arithmetic"}
# Create variations
example2 = example.copy(answer="2+2=4")
example3 = example.without("explanation")
# Dictionary operations
print(example.keys()) # ["question", "answer", "explanation"]
print(example.get("missing", "default")) # "default"
print("question" in example) # True
# Create training set
trainset = [
dspy.Example(question="What is 2+2?", answer="4").with_inputs("question"),
dspy.Example(question="What is 3+3?", answer="6").with_inputs("question"),
dspy.Example(question="What is 5+5?", answer="10").with_inputs("question"),
]Prediction object containing module outputs. Inherits from Example with additional completion support.
class Prediction(Example):
"""
Prediction result from module execution.
Extends Example with support for multiple completions and token usage tracking.
"""
completions: object
"""Completions object containing all generated completions."""
def get_lm_usage(self) -> dict:
"""
Get token usage information.
Returns:
Dictionary with token usage stats (prompt_tokens, completion_tokens, etc.)
"""
pass
def set_lm_usage(self, value: dict):
"""
Set token usage information.
Args:
value (dict): Token usage stats
"""
pass
@staticmethod
def from_completions(list_or_dict, signature=None):
"""
Create Prediction from completions.
Args:
list_or_dict: List of completions or dict of field completions
signature: Optional signature for type information
Returns:
Prediction instance
"""
pass
def __lt__(self, other):
"""Less than comparison (requires 'score' field)."""
pass
def __gt__(self, other):
"""Greater than comparison (requires 'score' field)."""
pass
def __le__(self, other):
"""Less than or equal comparison (requires 'score' field)."""
pass
def __ge__(self, other):
"""Greater than or equal comparison (requires 'score' field)."""
pass
def __add__(self, other):
"""Addition (requires 'score' field)."""
pass
def __truediv__(self, other):
"""Division (requires 'score' field)."""
passUsage:
import dspy
# Configure
dspy.configure(lm=dspy.LM('openai/gpt-4o-mini'))
# Create predictor
qa = dspy.ChainOfThought("question -> answer")
# Get prediction
pred = qa(question="What is machine learning?")
# Access fields
print(pred.answer) # Answer text
print(pred.reasoning) # Chain of thought
# Access completions (when n > 1)
pred = qa(question="What is ML?", config={"n": 3})
print(pred.completions) # All 3 completions
print(pred.completions[0].answer) # First completion
print(len(pred.completions)) # 3
# Token usage
usage = pred.get_lm_usage()
print(usage["prompt_tokens"])
print(usage["completion_tokens"])
# Comparison (when 'score' field exists)
pred1 = dspy.Prediction(answer="A", score=0.8)
pred2 = dspy.Prediction(answer="B", score=0.9)
print(pred2 > pred1) # True
# Arithmetic on scores
total = pred1 + pred2
average = total / 2Container for multiple completions from a language model.
class Completions:
"""
Container for multiple completions.
Provides access to all generated completions when n > 1 in LM configuration.
"""
def __getitem__(self, key):
"""
Access completion by index or field name.
Args:
key: Integer index or field name
Returns:
Completion at index or list of field values
"""
pass
def __len__(self) -> int:
"""
Get number of completions.
Returns:
Number of completions
"""
pass
def items(self):
"""
Get field items.
Returns:
Iterator of (field_name, values) tuples
"""
passUsage:
import dspy
# Configure with n > 1 for multiple completions
dspy.configure(lm=dspy.LM('openai/gpt-4o-mini'))
qa = dspy.Predict("question -> answer")
pred = qa(question="What is AI?", config={"n": 5})
# Access completions
completions = pred.completions
# By index
print(completions[0].answer) # First answer
print(completions[1].answer) # Second answer
# Length
print(len(completions)) # 5
# Field access
answers = completions["answer"] # List of 5 answers
print(answers[0])
# Iterate
for i, completion in enumerate(completions):
print(f"Completion {i}: {completion.answer}")
# Access all field values
for field_name, values in completions.items():
print(f"{field_name}: {values}")Safe Python code execution environment for ProgramOfThought and CodeAct modules.
class PythonInterpreter:
"""
Safe Python code execution using Deno.
Used by ProgramOfThought and CodeAct modules for executing
generated Python code in a sandboxed environment.
"""
def execute(self, code: str) -> dict:
"""
Execute Python code.
Args:
code (str): Python code to execute
Returns:
Dictionary with 'output', 'error', and 'success' keys
"""
pass
def shutdown(self):
"""Shutdown interpreter and clean up resources."""
passUsage:
import dspy
# Create interpreter
interpreter = dspy.PythonInterpreter()
# Execute code
result = interpreter.execute("""
def fibonacci(n):
if n <= 1:
return n
return fibonacci(n-1) + fibonacci(n-2)
print(fibonacci(10))
""")
print(result["output"]) # "55\n"
print(result["success"]) # True
# With error
result = interpreter.execute("1 / 0")
print(result["error"]) # Error message
print(result["success"]) # False
# Clean up
interpreter.shutdown()
# Use with ProgramOfThought
pot = dspy.ProgramOfThought(
"question -> answer",
interpreter=interpreter
)
result = pot(question="What is 15 factorial?")Basic module with single predictor:
import dspy
class SimpleQA(dspy.Module):
def __init__(self):
super().__init__()
self.qa = dspy.ChainOfThought("question -> answer")
def forward(self, question):
return self.qa(question=question)Module with multiple processing stages:
import dspy
class MultiStageQA(dspy.Module):
def __init__(self):
super().__init__()
self.classify = dspy.Predict("question -> category")
self.retrieve = dspy.Retrieve(k=3)
self.answer = dspy.ChainOfThought("category, context, question -> answer")
def forward(self, question):
# Stage 1: Classify
category = self.classify(question=question).category
# Stage 2: Retrieve
context = self.retrieve(query=question).passages
# Stage 3: Answer
return self.answer(
category=category,
context=context,
question=question
)Modules containing other modules:
import dspy
class QAModule(dspy.Module):
def __init__(self):
super().__init__()
self.qa = dspy.ChainOfThought("context, question -> answer")
def forward(self, context, question):
return self.qa(context=context, question=question)
class RAGSystem(dspy.Module):
def __init__(self):
super().__init__()
self.retrieve = dspy.Retrieve(k=5)
self.qa_module = QAModule() # Nested module
def forward(self, question):
context = self.retrieve(query=question).passages
return self.qa_module(context=context, question=question)Modules with conditional logic:
import dspy
class ConditionalQA(dspy.Module):
def __init__(self):
super().__init__()
self.classify = dspy.Predict("question -> needs_context: bool")
self.retrieve = dspy.Retrieve(k=3)
self.simple_qa = dspy.Predict("question -> answer")
self.rag_qa = dspy.ChainOfThought("context, question -> answer")
def forward(self, question):
# Decide if context is needed
needs_context = self.classify(question=question).needs_context
if needs_context:
# Use RAG for complex questions
context = self.retrieve(query=question).passages
return self.rag_qa(context=context, question=question)
else:
# Direct answer for simple questions
return self.simple_qa(question=question)Modules with loops and iteration:
import dspy
class IterativeReasoning(dspy.Module):
def __init__(self, max_iters=3):
super().__init__()
self.max_iters = max_iters
self.refine = dspy.ChainOfThought("question, previous_answer -> improved_answer")
self.initial = dspy.Predict("question -> answer")
def forward(self, question):
# Initial answer
answer = self.initial(question=question).answer
# Iteratively refine
for i in range(self.max_iters):
result = self.refine(
question=question,
previous_answer=answer
)
answer = result.improved_answer
return dspy.Prediction(answer=answer)