Pre-built DSPy modules for common AI patterns including basic prediction, reasoning, tool use, code generation, and result refinement. These modules provide ready-to-use implementations of standard prompting techniques.
Core prediction module that maps inputs to outputs using a language model.
class Predict:
"""
Basic DSPy module for input-to-output prediction.
Maps input fields to output fields using an LM based on a signature.
Supports configuration, callbacks, and state management.
"""
def __init__(self, signature, callbacks=None, **config):
"""
Initialize predictor.
Args:
signature (str | type): Input/output signature describing the task
- str: "input1, input2 -> output1, output2"
- type: Signature class
callbacks (list | None): Optional callbacks for instrumentation
**config: Default keyword arguments forwarded to LM (temperature, max_tokens, etc.)
"""
pass
def __call__(self, **kwargs):
"""
Execute prediction with keyword arguments.
Args:
**kwargs: Input field values and optional config overrides
Returns:
Prediction with output fields
"""
pass
def acall(self, **kwargs):
"""
Async version of __call__.
Args:
**kwargs: Input field values and optional config overrides
Returns:
Awaitable resolving to Prediction
"""
pass
def forward(self, **kwargs):
"""
Internal forward pass.
Args:
**kwargs: Input field values
Returns:
Prediction with output fields
"""
pass
def aforward(self, **kwargs):
"""
Async forward pass.
Args:
**kwargs: Input field values
Returns:
Awaitable resolving to Prediction
"""
pass
def dump_state(self, json_mode: bool = True) -> dict:
"""
Serialize state to dictionary.
Args:
json_mode (bool): Use JSON-compatible format
Returns:
Dictionary representation
"""
pass
def load_state(self, state: dict):
"""
Load serialized state.
Args:
state (dict): Serialized state dictionary
"""
pass
def update_config(self, **kwargs):
"""
Update default configuration.
Args:
**kwargs: Configuration parameters to update
"""
pass
def get_config(self) -> dict:
"""
Get current configuration.
Returns:
Configuration dictionary
"""
passUsage:
import dspy
dspy.configure(lm=dspy.LM('openai/gpt-4o-mini'))
# Basic usage
predict = dspy.Predict("question -> answer")
result = predict(question="What is 2+2?")
print(result.answer)
# With configuration
predict = dspy.Predict(
"question -> answer",
temperature=0.7,
max_tokens=500
)
# Override config per call
result = predict(
question="Complex question?",
config={"temperature": 0.0, "n": 3}
)
# Multiple outputs
predict = dspy.Predict("text -> summary, sentiment, keywords")
result = predict(text="Long article text...")
print(result.summary)
print(result.sentiment)
print(result.keywords)Module that reasons step-by-step before producing outputs.
class ChainOfThought:
"""
Chain-of-thought reasoning module.
Extends Predict with intermediate reasoning step that explains
the thought process before generating final outputs.
"""
def __init__(
self,
signature,
rationale_field=None,
rationale_field_type=None,
**config
):
"""
Initialize chain-of-thought module.
Args:
signature (str | type): Task signature
rationale_field (FieldInfo | None): Custom field for reasoning
rationale_field_type (type | None): Type of rationale field (default: str)
**config: Configuration passed to underlying Predict
"""
pass
def forward(self, **kwargs):
"""
Execute with chain-of-thought reasoning.
Args:
**kwargs: Input field values
Returns:
Prediction with reasoning and output fields
"""
pass
def aforward(self, **kwargs):
"""
Async forward pass with reasoning.
Args:
**kwargs: Input field values
Returns:
Awaitable resolving to Prediction
"""
passUsage:
import dspy
dspy.configure(lm=dspy.LM('openai/gpt-4o-mini'))
# Basic chain-of-thought
cot = dspy.ChainOfThought("question -> answer")
result = cot(question="Why is the sky blue?")
print(result.reasoning) # Step-by-step explanation
print(result.answer) # Final answer
# Custom rationale field
cot = dspy.ChainOfThought(
"context, question -> answer",
rationale_field=dspy.OutputField(desc="Detailed analysis")
)
result = cot(
context="Physics information...",
question="Explain Rayleigh scattering"
)Reasoning and Acting paradigm for building tool-using agents.
class ReAct:
"""
Reasoning and Acting agent with tool use.
Iteratively reasons about the task and calls tools until
reaching a final answer. Implements the ReAct pattern.
"""
def __init__(self, signature, tools: list, max_iters: int = 10):
"""
Initialize ReAct agent.
Args:
signature (type): Input/output signature (must be Signature class, not string)
tools (list): List of callable tools or Tool instances
max_iters (int): Maximum iterations (default: 10)
"""
pass
def forward(self, **kwargs):
"""
Execute ReAct loop.
Args:
**kwargs: Input field values
Returns:
Prediction with final answer and trajectory
"""
pass
def aforward(self, **kwargs):
"""
Async execution of ReAct loop.
Args:
**kwargs: Input field values
Returns:
Awaitable resolving to Prediction
"""
pass
def truncate_trajectory(self, trajectory: list):
"""
Truncate trajectory to fit context window.
Args:
trajectory (list): List of (thought, action, observation) tuples
Returns:
Truncated trajectory
"""
passUsage:
import dspy
# Define tools
def get_weather(city: str) -> str:
"""Get current weather for a city."""
return f"The weather in {city} is sunny, 72°F."
def search_web(query: str) -> str:
"""Search the web for information."""
return f"Search results for '{query}': ..."
# Create signature
class QASignature(dspy.Signature):
question: str = dspy.InputField()
answer: str = dspy.OutputField()
# Initialize ReAct
dspy.configure(lm=dspy.LM('openai/gpt-4o-mini'))
react = dspy.ReAct(
QASignature,
tools=[get_weather, search_web],
max_iters=5
)
# Use agent
result = react(question="What's the weather like in Tokyo today?")
print(result.answer)
# With Tool wrapper for better control
weather_tool = dspy.Tool(
func=get_weather,
name="weather_lookup",
desc="Look up current weather conditions"
)
react = dspy.ReAct(QASignature, tools=[weather_tool])Generate and execute Python code to solve problems.
class ProgramOfThought:
"""
Program-of-thought module.
Generates Python code to solve problems and executes it in a
sandboxed environment. Requires Deno to be installed.
"""
def __init__(
self,
signature,
max_iters: int = 3,
interpreter=None
):
"""
Initialize program-of-thought module.
Args:
signature (str | type): Task signature
max_iters (int): Max code generation retries on errors (default: 3)
interpreter (PythonInterpreter | None): Python interpreter instance
"""
pass
def forward(self, **kwargs):
"""
Generate and execute code.
Args:
**kwargs: Input field values
Returns:
Prediction with code, execution results, and final answer
"""
passUsage:
import dspy
dspy.configure(lm=dspy.LM('openai/gpt-4o-mini'))
# Basic usage
pot = dspy.ProgramOfThought("question -> answer")
result = pot(question="What is 15 factorial?")
print(result.answer) # "1307674368000"
print(result.code) # Generated Python code
# With custom interpreter
interpreter = dspy.PythonInterpreter()
pot = dspy.ProgramOfThought(
"problem -> solution",
max_iters=5,
interpreter=interpreter
)
result = pot(problem="Calculate the 20th Fibonacci number")
interpreter.shutdown()Combine code interpretation with predefined tools.
class CodeAct:
"""
CodeAct module combining code and tools.
Extends both ReAct and ProgramOfThought to support both
code generation/execution and tool calling in one agent.
"""
def __init__(
self,
signature,
tools: list,
max_iters: int = 5,
interpreter=None
):
"""
Initialize CodeAct module.
Args:
signature (str | type): Task signature
tools (list): List of tool functions or Tool instances
max_iters (int): Maximum iterations (default: 5)
interpreter (PythonInterpreter | None): Python interpreter instance
"""
pass
def forward(self, **kwargs):
"""
Execute CodeAct loop.
Args:
**kwargs: Input field values
Returns:
Prediction with final answer
"""
passUsage:
import dspy
# Define tools
def factorial(n: int) -> int:
"""Calculate factorial of n."""
result = 1
for i in range(1, n + 1):
result *= i
return result
def search_database(query: str) -> str:
"""Search internal database."""
return f"Database results for: {query}"
# Initialize CodeAct
dspy.configure(lm=dspy.LM('openai/gpt-4o-mini'))
act = dspy.CodeAct(
"task -> result",
tools=[factorial, search_database],
max_iters=5
)
# Agent can use both code and tools
result = act(task="Calculate 10! and search for factorial history")
print(result.result)Wrapper for functions to be used as tools in agents.
class Tool:
"""
Tool wrapper for agent functions.
Wraps functions with metadata for better tool use in ReAct and CodeAct.
"""
def __init__(
self,
func: callable,
name: str = None,
desc: str = None,
args: dict = None,
arg_types: dict = None,
arg_desc: dict = None
):
"""
Create tool from function.
Args:
func (callable): The function to wrap
name (str | None): Tool name (auto-inferred from function)
desc (str | None): Tool description (auto-inferred from docstring)
args (dict | None): Argument schemas
arg_types (dict | None): Argument type annotations
arg_desc (dict | None): Argument descriptions
"""
pass
def __call__(self, **kwargs):
"""
Execute the tool.
Args:
**kwargs: Tool arguments
Returns:
Tool result
"""
pass
def acall(self, **kwargs):
"""
Async execution.
Args:
**kwargs: Tool arguments
Returns:
Awaitable resolving to tool result
"""
passUsage:
import dspy
# Basic tool
def get_time() -> str:
"""Get current time."""
import datetime
return datetime.datetime.now().isoformat()
tool = dspy.Tool(get_time)
# Tool with custom metadata
def calculate_distance(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
"""Calculate distance between two coordinates."""
# Implementation...
return 0.0
tool = dspy.Tool(
calculate_distance,
name="distance_calculator",
desc="Calculate great circle distance between GPS coordinates",
arg_desc={
"lat1": "Latitude of first point",
"lon1": "Longitude of first point",
"lat2": "Latitude of second point",
"lon2": "Longitude of second point"
}
)Run a module multiple times and return the best result.
class BestOfN:
"""
Best-of-N sampling module.
Runs a module up to N times and returns the prediction with
the highest reward according to a reward function.
"""
def __init__(
self,
module,
N: int,
reward_fn: callable,
threshold: float,
fail_count: int = None
):
"""
Initialize best-of-N sampler.
Args:
module: The module to run multiple times
N (int): Number of attempts
reward_fn (callable): Function(args: dict, pred: Prediction) -> float
threshold (float): Reward threshold for early stopping
fail_count (int | None): Max failures before raising error
"""
pass
def forward(self, **kwargs):
"""
Execute best-of-N sampling.
Args:
**kwargs: Input arguments for module
Returns:
Prediction with highest reward
"""
passUsage:
import dspy
dspy.configure(lm=dspy.LM('openai/gpt-4o-mini'))
# Create module
qa = dspy.ChainOfThought("question -> answer")
# Define reward function
def one_word_answer(args, pred):
"""Reward single-word answers."""
word_count = len(pred.answer.split())
return 1.0 if word_count == 1 else 1.0 / word_count
# Best-of-N wrapper
best_of_5 = dspy.BestOfN(
module=qa,
N=5,
reward_fn=one_word_answer,
threshold=1.0
)
result = best_of_5(question="What is the capital of Belgium?")
print(result.answer) # Most likely "Brussels" (one word)Refine outputs by running multiple attempts with automatic feedback.
class Refine:
"""
Output refinement module.
Iteratively refines module outputs by generating feedback
and running refinement attempts until threshold is met.
"""
def __init__(
self,
module,
N: int,
reward_fn: callable,
threshold: float,
fail_count: int = None
):
"""
Initialize refinement module.
Args:
module: Module to refine
N (int): Number of refinement attempts
reward_fn (callable): Reward function for evaluating quality
threshold (float): Target reward threshold
fail_count (int | None): Max failures before error
"""
pass
def forward(self, **kwargs):
"""
Execute refinement loop.
Args:
**kwargs: Input arguments
Returns:
Refined prediction meeting threshold
"""
passUsage:
import dspy
dspy.configure(lm=dspy.LM('openai/gpt-4o-mini'))
# Module to refine
summarize = dspy.Predict("article -> summary")
# Reward function (prefer shorter summaries)
def brevity_reward(args, pred):
length = len(pred.summary.split())
return 1.0 if length <= 50 else 50.0 / length
# Refinement
refiner = dspy.Refine(
module=summarize,
N=3,
reward_fn=brevity_reward,
threshold=0.9
)
result = refiner(article="Long article text...")
print(result.summary) # Refined to meet length targetCompare multiple chain-of-thought attempts and synthesize the best answer.
class MultiChainComparison:
"""
Multi-chain comparison module.
Generates multiple chain-of-thought reasoning attempts,
compares them, and synthesizes the best final answer.
"""
def __init__(
self,
signature,
M: int = 3,
temperature: float = 0.7,
**config
):
"""
Initialize multi-chain comparison.
Args:
signature: Task signature
M (int): Number of reasoning chains to compare (default: 3)
temperature (float): Sampling temperature (default: 0.7)
**config: Additional configuration
"""
pass
def forward(self, completions=None, **kwargs):
"""
Compare completions and synthesize answer.
Args:
completions: Optional pre-generated completions
**kwargs: Input field values
Returns:
Prediction with synthesized answer
"""
passUsage:
import dspy
dspy.configure(lm=dspy.LM('openai/gpt-4o-mini'))
# Multi-chain comparison
mcc = dspy.MultiChainComparison(
"question -> answer",
M=5, # Generate 5 reasoning chains
temperature=0.8
)
result = mcc(question="What are the main causes of climate change?")
print(result.answer) # Synthesized from 5 different reasoning pathsRetrieve similar examples from a training set using embeddings.
class KNN:
"""
K-nearest neighbors retriever.
Finds k most similar examples from a training set based on
embedding similarity. Used for dynamic few-shot learning.
"""
def __init__(self, k: int, trainset: list, vectorizer):
"""
Initialize KNN retriever.
Args:
k (int): Number of nearest neighbors to retrieve
trainset (list): List of Example instances
vectorizer (Embedder): Embedder for computing embeddings
"""
pass
def __call__(self, **kwargs):
"""
Retrieve k nearest neighbors.
Args:
**kwargs: Input fields to match against
Returns:
List of k most similar Examples
"""
passUsage:
import dspy
from sentence_transformers import SentenceTransformer
# Create training set
trainset = [
dspy.Example(input="hello", output="greeting"),
dspy.Example(input="goodbye", output="farewell"),
dspy.Example(input="thanks", output="gratitude"),
# ... more examples
]
# Set up embedder
model = SentenceTransformer("all-MiniLM-L6-v2")
embedder = dspy.Embedder(model.encode)
# Create KNN retriever
knn = dspy.KNN(k=3, trainset=trainset, vectorizer=embedder)
# Retrieve similar examples
similar = knn(input="hi there")
for ex in similar:
print(f"{ex.input} -> {ex.output}")Execute module calls in parallel across multiple examples.
class Parallel:
"""
Parallel execution module.
Processes multiple examples in parallel using thread pool,
with error handling and progress tracking.
"""
def __init__(
self,
num_threads: int = None,
max_errors: int = None,
access_examples: bool = True,
return_failed_examples: bool = False,
provide_traceback: bool = None,
disable_progress_bar: bool = False
):
"""
Initialize parallel executor.
Args:
num_threads (int | None): Number of threads (default: settings.num_threads)
max_errors (int | None): Max errors before stopping
access_examples (bool): Whether to pass examples to module
return_failed_examples (bool): Include failed examples in results
provide_traceback (bool | None): Include error tracebacks
disable_progress_bar (bool): Disable progress display
"""
pass
def forward(self, exec_pairs: list, num_threads: int = None):
"""
Execute in parallel.
Args:
exec_pairs (list): List of (module, inputs) tuples
num_threads (int | None): Override thread count
Returns:
List of results (or (result, example) if return_failed_examples=True)
"""
passUsage:
import dspy
dspy.configure(lm=dspy.LM('openai/gpt-4o-mini'))
# Create module
qa = dspy.ChainOfThought("question -> answer")
# Prepare examples
questions = [
{"question": "What is ML?"},
{"question": "What is AI?"},
{"question": "What is DL?"},
]
# Parallel execution
parallel = dspy.Parallel(num_threads=4, disable_progress_bar=False)
exec_pairs = [(qa, q) for q in questions]
results = parallel(exec_pairs)
for result in results:
print(result.answer)Return the most common completion across multiple predictions.
def majority(
prediction_or_completions,
normalize: callable = None,
field: str = None
):
"""
Return most common completion for a field.
Args:
prediction_or_completions: Prediction with multiple completions or Completions object
normalize (callable | None): Function to normalize values before comparison
field (str | None): Field name to aggregate (required if multiple output fields)
Returns:
Prediction with majority value
"""
passUsage:
import dspy
dspy.configure(lm=dspy.LM('openai/gpt-4o-mini'))
# Get multiple completions
qa = dspy.Predict("question -> answer")
pred = qa(question="Capital of France?", config={"n": 10})
# Majority voting
result = dspy.majority(pred, field="answer")
print(result.answer) # Most common answer across 10 completions
# With normalization
def normalize_answer(text):
return text.lower().strip()
result = dspy.majority(pred, field="answer", normalize=normalize_answer)