DSPy (Declarative Self-improving Python)

Overview

RLM (Recursive Language Model) is a DSPy module that allows LLMs to programmatically explore large contexts using a sandboxed Python REPL. Instead of passing massive amounts of text directly into a prompt (which can lead to "context rot"), RLM treats the context as external data that the LLM accesses via code execution and recursive sub-LLM calls.

When to use RLM

• Large Contexts: When the data is too large to fit effectively in the LLM's context window.
• Programmatic Exploration: When the task requires searching, filtering, aggregating, or chunking data.
• Autonomous Decomposition: When you want the LLM to decide how to break down a problem rather than defining the steps yourself.

How it works

RLM operates in an iterative loop:

1. Metadata Access: The LLM sees metadata (type, length, preview) rather than the full context.
2. Code Execution: The LLM writes Python code to explore the data (e.g., print(), re.findall()).
3. Sandboxed Execution: Code runs in a secure WASM sandbox (via Deno/Pyodide).
4. Sub-LLM Calls: The LLM can use llm_query(prompt) to perform semantic analysis on specific snippets.
5. Submission: The LLM calls SUBMIT(output) to return the final answer.

import dspy

dspy.configure(lm=dspy.LM("openai/gpt-5"))

# Create an RLM module
rlm = dspy.RLM("context, query -> answer")

# Call it like any other module
result = rlm(
    context="...very long document or data...",
    query="What is the total revenue mentioned?"
)
print(result.answer)

In DSPy, a metric is a Python function used by optimizers to determine what "better" means for a program. Optimizers use metrics to run a loop: executing your program many times and keeping the version that achieves the highest score.

Key characteristics of a metric:

• Input Signature: A metric function must accept two arguments: the original example (containing inputs and potentially gold labels) and the program's prediction.
• Return Value: It should return a single float. By convention, scores range from 0.0 to 1.0, where higher values indicate better performance.
• Focus: Since DSPy Signatures handle structural validation (like JSON formatting), metrics should focus on task-specific quality, such as whether an answer is correct, whether a tone is appropriate, or whether specific rules were followed.

Common metric patterns include:

1. Labeled data comparisons: Comparing a prediction against a "gold" output (human-labeled ground truth).
2. Rule-based checks: Using code to verify properties (e.g., syllable counts, presence of specific keywords).
3. LLM judges: Using a larger, more capable model to evaluate the quality of a smaller model's output.

def my_metric(example, prediction) -> float:
    # logic to compare prediction against example or rules
    return 1.0  # or 0.0

Concept

dspy.Flex is a specialized DSPy module where the implementation is optimizable code rather than a fixed prompt. While standard modules use fixed instructions, a Flex module allows an optimizer like dspy.GEPA to rewrite its entire source code. This means the optimizer can split tasks into multiple predictors, convert deterministic logic (like math or parsing) into plain Python, and author helper functions.

Use dspy.Flex when:

• The optimal program structure (decomposition) is unknown and you want the optimizer to discover it.
• You want to move deterministic tasks (arithmetic, normalization, lookups) out of the LLM and into Python code to save costs.
• You want to optimize the trade-off between accuracy and execution cost.

import dspy

dspy.configure(lm=dspy.LM("openai/gpt-5"))

# Construct Flex from a signature
solve = dspy.Flex("invoice: str -> total_cents: int")

# Runs the baseline (a single dspy.Predict)
result = solve(invoice="2 widgets @ $3.50, shipping $1.00")
print(result.total_cents)

You can build a memory-enhanced agent by defining a dspy.Signature that instructs the model to use its memory tools, and then wrapping dspy.ReAct within a dspy.Module.

1. Define a Signature: Include instructions in the docstring telling the agent to store information in memory for future use.
2. Define a Module: In the __init__ of your module, instantiate your memory tools and pass them as a list to dspy.ReAct.
3. Implement Forward: The forward method calls the ReAct agent with the user input.

class MemoryQA(dspy.Signature):
    """
    You're a helpful assistant and have access to memory method. 
    Whenever you answer a user's input, remember to store the information in memory 
    so that you can use it later.
    """
    user_input: str = dspy.InputField()
    response: str = dspy.OutputField()

class MemoryReActAgent(dspy.Module):
    def __init__(self, memory: Memory):
        super().__init__()
        self.memory_tools = MemoryTools(memory)
        self.tools = [
            self.memory_tools.store_memory,
            self.memory_tools.search_memories,
            # ... other tools
        ]
        self.react = dspy.ReAct(
            signature=MemoryQA,
            tools=self.tools,
            max_iters=6
        )

    def forward(self, user_input: str):
        return self.react(user_input=user_input)

When using the GEPA optimizer, your metric function can return more than just a numerical score. You can return a dspy.Prediction object containing a feedback string. This text feedback is passed to the reflection_lm, allowing it to understand why a prediction failed (e.g., "Don't reference the input season verbatim") and use that information to write better instructions.

Example metric signature:

def my_metric(example, prediction, trace=None, pred_name=None, pred_trace=None):
    # ... logic ...
    return dspy.Prediction(score=0.0, feedback="Detailed explanation of failure")

def haiku_score_gepa(example, prediction, trace=None, pred_name=None, pred_trace=None):
    """
    Penalize verbatim use of the input season string.
    A haiku should evoke the season through imagery, not name it
    directly.
    """
    text = prediction.haiku.lower()
    if example.season.strip().lower() in text:
        return dspy.Prediction(
            score=0.0,
            feedback="Don't reference the input season verbatim."
        )
    return dspy.Prediction(score=1.0, feedback=None)

In DSPy, you define the structure of your tasks by subclassing dspy.Signature. You use dspy.InputField to define inputs and dspy.OutputField to define the expected outputs. This allows the AI to understand the schema of the data it needs to process and generate.

Example of a StoryGenerator signature:

class StoryGenerator(dspy.Signature):
    """Generate dynamic story content based on current game state."""
    location: str = dspy.InputField(desc="Current location")
    player_info: str = dspy.InputField(desc="Player information and stats")
    story_progress: int = dspy.InputField(desc="Current story progress level")
    recent_actions: str = dspy.InputField(desc="Player's recent actions")
    
    scene_description: str = dspy.OutputField(desc="Vivid description of current scene")
    available_actions: list[str] = dspy.OutputField(desc="List of possible player actions")
    npcs_present: list[str] = dspy.OutputField(desc="NPCs present in this location")
    items_available: list[str] = dspy.OutputField(desc="Items that can be found or interacted with")

A dspy.Module encapsulates a pipeline of operations. You can compose multiple sub-modules (like dspy.ChainOfThought) within the __init__ method and define the data flow in the forward method.

In a complex pipeline, the output of one module (e.g., classification.email_type) can be passed as an input to the next module (e.g., entity_extractor). The final result is typically returned as a dspy.Prediction object containing all structured fields.

class EmailProcessor(dspy.Module):
    def __init__(self):
        super().__init__()
        self.classifier = dspy.ChainOfThought(ClassifyEmail)
        self.entity_extractor = dspy.ChainOfThought(ExtractEntities)
        # ... other components

    def forward(self, email_subject: str, email_body: str, sender: str = ""):
        classification = self.classifier(email_subject=email_subject, email_body=email_body, sender=sender)
        entities = self.entity_extractor(email_content=..., email_type=classification.email_type)
        # ... logic
        return dspy.Prediction(email_type=classification.email_type, ...)