Dspy Rlm

Iterative Self-Refinement with dspy.RLM

Guide the user through using DSPy's RLM (Recursive Language Model) module. RLM lets the LM explore data programmatically in a sandboxed Python REPL, writing code to examine inputs, querying sub-LMs for semantic analysis, and iterating until it produces a final answer.

Experimental. RLM is marked as experimental in DSPy. The API may change in future releases.

Step 1: Gather context

Before building with RLM, clarify:

1. What data will the LM explore? Large text corpus, log files, structured data dumps, multi-document collections? 2. What kind of answer do you need? Free-text summary, structured extraction (counts, lists), or a specific value? 3. Does the task need external tools? If the LM needs to call APIs or databases during exploration, you can pass custom tool functions. 4. What LM are you using? RLM works best with strong reasoning models (GPT-4o, Claude Sonnet) as the main LM; cheaper models can handle sub-queries.

What is RLM

dspy.RLM implements the Recursive Language Models approach (Zhang, Kraska, Khattab 2025). Instead of feeding the full input context into the LM's prompt, RLM:

1. Shows metadata only -- the LM receives type, length, and a preview of each input, not the full content. 2. Lets the LM write code -- the LM generates Python in a sandboxed REPL to search, filter, aggregate, or transform the data. 3. Executes in a sandbox -- code runs in a WASM-based Python interpreter (Pyodide via Deno) for safety. 4. Supports sub-LM queries -- the LM can call llm_query(prompt) to do semantic analysis on slices of the data. 5. Iterates -- the LM loops through code-execute-observe cycles until it calls SUBMIT(output) with a final answer.

This makes RLM ideal for tasks where the input is too large for the context window, or where the LM needs to programmatically explore the data to find the answer.

When to use RLM

When RLM is not the right fit:

• Simple input-output tasks -- use dspy.Predict or dspy.ChainOfThought
• Tasks that need external tool use (APIs, databases) -- use dspy.ReAct
• Quick classification or extraction -- overhead of the REPL loop is unnecessary

Prerequisites

RLM's default sandbox requires Deno for the Pyodide WASM interpreter:

curl -fsSL https://deno.land/install.sh | sh

Basic usage

import dspy

dspy.configure(lm=dspy.LM("openai/gpt-4o"))  # or "anthropic/claude-sonnet-4-5-20250929", etc.

rlm = dspy.RLM("context, query -> answer")
result = rlm(
    context="...very long document or dataset...",
    query="What is the total revenue for Q3?",
)
print(result.answer)

The LM will: 1. See a preview of context (type, length, first/last chars). 2. Write Python code to search and parse the content. 3. Optionally call llm_query() for semantic questions about slices. 4. Call SUBMIT(answer) when ready.

Constructor parameters

dspy.RLM(
    signature,              # str | Signature -- required, defines inputs/outputs
    max_iterations=20,      # max REPL interaction loops
    max_llm_calls=50,       # max sub-LM query calls per execution
    max_output_chars=10_000,# max chars from REPL output per step
    verbose=False,          # enable detailed execution logging
    tools=None,             # list[Callable] -- custom tool functions
    sub_lm=None,            # dspy.LM -- separate (cheaper) LM for sub-queries
    interpreter=None,       # custom CodeInterpreter (defaults to PythonInterpreter)
)

Built-in tools available inside the REPL

When the LM writes code in the sandbox, these functions are available:

Using a cheaper sub-LM

Route expensive reasoning to a strong model while using a cheap model for sub-queries:

main_lm = dspy.LM("openai/gpt-4o")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
cheap_lm = dspy.LM("openai/gpt-4o-mini")  # or any cheaper model

dspy.configure(lm=main_lm)

rlm = dspy.RLM("data, query -> summary", sub_lm=cheap_lm)
result = rlm(data=large_dataset, query="Summarize the key trends")

Typed outputs

RLM supports DSPy's typed output fields, just like other modules:

rlm = dspy.RLM("logs -> error_count: int, critical_errors: list[str]")
result = rlm(logs=server_logs)
print(result.error_count)        # int
print(result.critical_errors)    # list[str]

Custom tools

Pass additional Python functions that the LM can call inside the sandbox:

def fetch_metadata(doc_id: str) -> str:
    """Look up metadata for a document by ID."""
    return database.get_metadata(doc_id)

rlm = dspy.RLM("documents, query -> answer", tools=[fetch_metadata])
result = rlm(documents=docs, query="Which document has the latest revision?")

Inspecting the trajectory

After execution, inspect the code-execute-observe steps the LM took:

result = rlm(context=data, query="Find the outlier values")

for step in result.trajectory:
    print(f"Code:\n{step['code']}")
    print(f"Output:\n{step['output']}\n")

This is useful for debugging, understanding the LM's exploration strategy, and building trust in the result.

Async execution

async def process():
    result = await rlm.aforward(context=data, query="Summarize findings")
    return result.answer

How RLM differs from other refinement approaches

Key difference: RLM gives the LM a code execution environment to actively explore the input, rather than just re-prompting with feedback. The LM decides its own exploration strategy.

Thread safety

RLM instances with custom interpreters are not thread-safe. For concurrent usage, create separate instances or use the default PythonInterpreter.

Gotchas

• Forgetting Deno installation. RLM's default PythonInterpreter uses a Pyodide WASM sandbox that requires Deno. If Deno is not installed, you get a cryptic subprocess error. Always check deno --version before running RLM code.
• Using RLM for tasks that fit in context. Claude defaults to RLM when asked about "iterative refinement" even for short inputs. RLM adds significant overhead (multiple REPL loops, sub-LM calls). For inputs under ~50K chars, use dspy.ChainOfThought or dspy.Predict instead.
• Forgetting `print()` in REPL code. The LM must print() values to see REPL output -- assignments alone produce no visible result. If the LM's exploration seems stuck, check the trajectory for code that computes but never prints.
• Setting `max_iterations` too low for complex tasks. Claude tends to set max_iterations=5 for brevity. RLM defaults to 20 for a reason -- complex data exploration often needs 10-15 iterations. Only lower it for simple lookups.
• Not using `sub_lm` for cost control. Every llm_query() call inside the REPL uses the main LM by default. For large-context tasks with many sub-queries, this gets expensive fast. Always set sub_lm to a cheaper model for semantic analysis calls.

Additional resources

• dspy.RLM API docs
• For API details, see reference.md
• For worked examples, see examples.md

Cross-references

Install any skill: npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill <name>

• Refine for reward-function-based retry loops -- see /dspy-refine
• Best-of-N for generating and scoring multiple candidates -- see /dspy-best-of-n
• Improving accuracy with optimizers and evaluation -- see /ai-improving-accuracy
• Building pipelines with multi-step module composition -- see /ai-building-pipelines
• Install `/ai-do` if you do not have it — it routes any AI problem to the right skill and is the fastest way to work: npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill ai-do

dspy-rlm Examples

Example 1: Quality-guided text generation with reward function

Analyze a large corpus of customer reviews to generate a summary, using RLM's code exploration to ensure the summary covers all major themes.

import dspy

# Configure LMs
main_lm = dspy.LM("openai/gpt-4o")
cheap_lm = dspy.LM("openai/gpt-4o-mini")
dspy.configure(lm=main_lm)

# Suppose we have thousands of customer reviews as a single large string
reviews_text = open("reviews_export.txt").read()  # e.g. 500K chars

# RLM will explore the reviews programmatically rather than
# trying to fit them all in a single prompt
rlm = dspy.RLM(
    "reviews, query -> summary, top_complaints: list[str], sentiment_breakdown: str",
    sub_lm=cheap_lm,
    max_iterations=15,
    verbose=True,
)

result = rlm(
    reviews=reviews_text,
    query="Summarize the key themes, list the top 5 complaints, and give a sentiment breakdown",
)

print("Summary:", result.summary)
print("Top complaints:", result.top_complaints)
print("Sentiment:", result.sentiment_breakdown)

# Inspect how the LM explored the data
for i, step in enumerate(result.trajectory):
    print(f"\n--- Step {i + 1} ---")
    print(f"Code:\n{step['code']}")
    print(f"Output:\n{step['output'][:200]}")

What RLM does internally: 1. The LM sees metadata about reviews (type: str, length: 500000, preview of first/last chars). 2. It writes code to split reviews by delimiter and count them. 3. It samples batches of reviews and calls llm_query_batched() to classify sentiment and extract themes. 4. It aggregates results with Python (counters, sorting). 5. It calls SUBMIT() with the structured summary.

Adding a quality check wrapper

You can wrap RLM in a custom module that validates output quality:

class QualityGuidedAnalysis(dspy.Module):
    def __init__(self):
        self.analyze = dspy.RLM(
            "reviews, query -> summary, top_complaints: list[str]",
            sub_lm=dspy.LM("openai/gpt-4o-mini"),
            max_iterations=15,
        )
        self.judge = dspy.ChainOfThought(
            "query, summary, top_complaints -> quality_score: float, feedback"
        )

    def forward(self, reviews, query):
        result = self.analyze(reviews=reviews, query=query)

        # Score the output
        evaluation = self.judge(
            query=query,
            summary=result.summary,
            top_complaints=result.top_complaints,
        )

        return dspy.Prediction(
            summary=result.summary,
            top_complaints=result.top_complaints,
            quality_score=evaluation.quality_score,
        )

def quality_reward(args, pred):
    """Reward analysis that meets a minimum quality threshold."""
    judge = dspy.ChainOfThought(
        "query, summary, top_complaints -> quality_score: float, feedback"
    )
    evaluation = judge(
        query=args["query"],
        summary=pred.summary,
        top_complaints=pred.top_complaints,
    )
    return float(evaluation.quality_score)

qa = dspy.Refine(
    module=QualityGuidedAnalysis(),
    N=3,
    reward_fn=quality_reward,
    threshold=0.7,
)
result = qa(reviews=reviews_text, query="Summarize customer feedback")

---

Example 2: Constrained generation with reward-based refinement

Analyze server log files to find error patterns, with constraints on output format and completeness.

import dspy

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

# Custom tool that the LM can call inside the sandbox
def validate_timestamp(ts: str) -> bool:
    """Check if a timestamp string matches ISO 8601 format."""
    from datetime import datetime
    try:
        datetime.fromisoformat(ts)
        return True
    except ValueError:
        return False

# Large log file content
logs = open("/var/log/app/server.log").read()  # e.g. 2M chars

rlm = dspy.RLM(
    "logs, query -> error_patterns: list[str], timeline: str, root_cause: str",
    tools=[validate_timestamp],
    max_iterations=20,
    max_llm_calls=30,
    verbose=True,
)

result = rlm(
    logs=logs,
    query="Identify recurring error patterns, build a timeline of incidents, and suggest root causes",
)

print("Error patterns:", result.error_patterns)
print("Timeline:", result.timeline)
print("Root cause:", result.root_cause)

Wrapping with constraints for production use

Combine RLM with a reward function to enforce output requirements:

class ConstrainedLogAnalysis(dspy.Module):
    def __init__(self):
        self.analyze = dspy.RLM(
            "logs, query -> error_patterns: list[str], timeline: str, root_cause: str",
            max_iterations=20,
            max_llm_calls=30,
        )

    def forward(self, logs, query):
        return self.analyze(logs=logs, query=query)


def log_analysis_reward(args, pred):
    """Hard constraints return 0.0 on failure; soft constraints subtract a small penalty."""
    # Hard constraints - must have patterns and substantive root cause
    if len(pred.error_patterns) == 0:
        return 0.0
    if len(pred.root_cause) < 20:
        return 0.0
    # Soft constraint - prefer focused pattern lists
    score = 1.0
    if len(pred.error_patterns) > 10:
        score -= 0.1
    return score


analyzer = dspy.Refine(
    module=ConstrainedLogAnalysis(),
    N=3,
    reward_fn=log_analysis_reward,
    threshold=0.9,
)


# Use with evaluation
def completeness_metric(example, pred, trace=None):
    """Score based on whether all required fields are populated and useful."""
    has_patterns = len(pred.error_patterns) > 0
    has_timeline = len(pred.timeline) > 50
    has_root_cause = len(pred.root_cause) > 20
    return (has_patterns + has_timeline + has_root_cause) / 3.0


result = analyzer(
    logs=logs,
    query="Find error patterns and root causes from the last 24 hours",
)

Using RLM with an optimizer

RLM modules can be optimized like any other DSPy module:

from dspy.evaluate import Evaluate

# Prepare labeled examples
trainset = [
    dspy.Example(
        logs=open(f"logs/sample_{i}.txt").read(),
        query="Identify error patterns and root causes",
        error_patterns=expected_patterns[i],
        root_cause=expected_causes[i],
    ).with_inputs("logs", "query")
    for i in range(50)
]

# Evaluate baseline
evaluator = Evaluate(devset=trainset[:10], metric=completeness_metric, num_threads=2)
baseline_score = evaluator(analyzer)
print(f"Baseline: {baseline_score}")

# Optimize
optimizer = dspy.BootstrapFewShot(metric=completeness_metric, max_bootstrapped_demos=3)
optimized = optimizer.compile(analyzer, trainset=trainset)

optimized_score = evaluator(optimized)
print(f"Optimized: {optimized_score}")

optimized.save("optimized_log_analyzer.json")

RLM API Reference

Condensed from dspy.ai/api/modules/RLM. Verify against upstream for latest.

Constructor

dspy.RLM(
    signature,                # str | Signature -- required
    max_iterations=20,        # max REPL interaction loops
    max_llm_calls=50,         # max sub-LM queries per execution
    max_output_chars=10_000,  # max chars from REPL output per step
    verbose=False,            # detailed execution logging
    tools=None,               # list[Callable] -- custom tool functions
    sub_lm=None,              # dspy.LM -- cheaper LM for sub-queries
    interpreter=None,         # CodeInterpreter (defaults to PythonInterpreter)
)

Key Methods

• forward(**inputs) -> dspy.Prediction -- run RLM with provided inputs
• aforward(**inputs) -> dspy.Prediction -- async variant
• batch(examples, num_threads, max_errors, ...) -- parallel processing

Built-in REPL Functions