Dspy Refine

Iterative Self-Improvement with dspy.Refine

Guide the user through using dspy.Refine to build pipelines that automatically retry and improve outputs until they meet a quality threshold.

What is dspy.Refine

dspy.Refine is a DSPy module wrapper that runs another module up to N times, scoring each attempt with a reward function. It returns the first output that meets a threshold -- or the best output if none do. When an attempt fails to meet the threshold, Refine generates feedback that gets fed into the next attempt, enabling genuine iterative improvement rather than just random retries.

Key properties:

• Wraps any DSPy module -- ChainOfThought, Predict, ReAct, or your custom modules
• Scores each attempt with a reward function you define
• Generates feedback when an attempt falls short, improving subsequent tries
• Returns early as soon as an output meets the threshold (saves LM calls)
• Falls back gracefully -- returns the best attempt even if none hit the threshold

When to use Refine

Use dspy.Refine when:

• Outputs must meet measurable quality criteria (format, length, accuracy)
• You can write a function that scores output quality as a number
• You want the LM to learn from its mistakes within a single request
• Quality is worth the extra LM calls (2-5x cost for N attempts)

Do not use Refine when:

• You have no clear way to score outputs -- use dspy.ChainOfThought instead
• You need human-in-the-loop feedback -- build a custom module with dspy.Suggest
• Speed matters more than quality -- use a single dspy.Predict call
• You just want multiple independent attempts without feedback -- use dspy.BestOfN (see comparison below)

Basic usage

Three things are needed: a module to wrap, a reward function, and a threshold.

import dspy

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

# 1. Define the module to refine
qa = dspy.ChainOfThought("question -> answer")

# 2. Define a reward function
# Takes (args_dict, prediction) -> float
def concise_answer(args, pred):
    """Reward one-word answers."""
    return 1.0 if len(pred.answer.split()) == 1 else 0.0

# 3. Wrap with Refine
refined_qa = dspy.Refine(
    module=qa,
    N=3,
    reward_fn=concise_answer,
    threshold=1.0,
)

# Use it -- same interface as the wrapped module
result = refined_qa(question="What is the capital of Belgium?")
print(result.answer)  # "Brussels"

Constructor parameters

dspy.Refine(
    module,       # The DSPy module to refine (required)
    N,            # Max number of attempts (required, int)
    reward_fn,    # Callable(args_dict, prediction) -> float (required)
    threshold,    # Target reward score to accept an output (required, float)
    fail_count,   # Max failures before raising an error (optional, defaults to N)
)

Writing reward functions

The reward function is the core of Refine. It receives two arguments:

1. `args` -- a dict of the inputs passed to the module (e.g., {"question": "What is..."}) 2. `pred` -- the module's prediction object (access fields like pred.answer, pred.reasoning)

It must return a float. Higher is better.

Simple binary reward

def valid_json(args, pred):
    """Accept only valid JSON outputs."""
    import json
    try:
        json.loads(pred.output)
        return 1.0
    except (json.JSONDecodeError, TypeError):
        return 0.0

Graduated reward

Return partial scores to help Refine pick the best attempt even when none fully succeed:

def quality_score(args, pred):
    """Score answer quality on multiple criteria."""
    score = 0.0
    answer = pred.answer

    # Criterion 1: not empty
    if answer.strip():
        score += 0.3

    # Criterion 2: reasonable length (20-200 words)
    word_count = len(answer.split())
    if 20 <= word_count <= 200:
        score += 0.4

    # Criterion 3: addresses the question
    if args["question"].split()[0].lower() in answer.lower():
        score += 0.3

    return score

Using external validation

import re

def valid_email_extraction(args, pred):
    """Reward valid email addresses extracted from text."""
    emails = pred.emails if isinstance(pred.emails, list) else []
    if not emails:
        return 0.0
    email_pattern = r'^[\w.-]+@[\w.-]+\.\w+$'
    valid_count = sum(1 for e in emails if re.match(email_pattern, e))
    return valid_count / len(emails)

How iteration count (N) works

Each attempt runs the wrapped module at temperature=1.0 with a different rollout ID, producing diverse outputs. Refine's selection logic:

1. Run the module and score the output with reward_fn 2. If the score meets or exceeds threshold, return immediately 3. If not, generate feedback from the failure and try again 4. After N attempts, return the attempt with the highest reward score

Choosing N:

The sweet spot for most use cases is N=3 to N=5. Beyond 5, diminishing returns are common unless the reward function is very specific.

The feedback mechanism

What makes Refine different from random retries is feedback generation. When an attempt fails to meet the threshold:

1. Refine examines why the attempt scored below the threshold 2. It generates natural-language feedback describing the shortcoming 3. This feedback is included in the prompt for the next attempt 4. The LM uses this feedback to produce a better output

This means later attempts are informed by earlier failures. Attempt 3 knows what went wrong in attempts 1 and 2.

You do not write the feedback logic -- Refine handles it automatically based on your reward function's scores.

Refine vs BestOfN -- when to use which

Both modules run a wrapped module multiple times and select the best output, but they work differently:

Use Refine when the LM can improve with feedback -- writing tasks, format compliance, multi-criteria quality.

Use BestOfN when attempts are independent and feedback would not help -- creative generation, sampling diverse options, simple binary checks.

Wrapping custom modules

Refine works with any dspy.Module, not just built-in ones:

class Summarizer(dspy.Module):
    def __init__(self):
        self.summarize = dspy.ChainOfThought("article -> summary")

    def forward(self, article):
        return self.summarize(article=article)


def good_summary(args, pred):
    """Score summary quality."""
    summary = pred.summary
    article = args["article"]
    score = 0.0

    # Shorter than original
    if len(summary) < len(article) * 0.3:
        score += 0.5

    # At least 2 sentences
    if summary.count('.') >= 2:
        score += 0.5

    return score


refined_summarizer = dspy.Refine(
    module=Summarizer(),
    N=3,
    reward_fn=good_summary,
    threshold=0.8,
)

result = refined_summarizer(article="Long article text here...")
print(result.summary)

Tips

• Start with N=3 and increase only if outputs consistently miss the threshold
• Use graduated rewards (0.0 to 1.0) rather than binary (0 or 1) so Refine can pick the best near-miss
• Keep reward functions fast -- they run on every attempt, so avoid expensive operations like LM calls inside them
• Set threshold realistically -- if your reward function rarely returns 1.0, set the threshold to 0.8 or similar
• Use `fail_count` to limit retries on genuinely impossible inputs rather than burning through all N attempts

Cross-references

• Chain of thought reasoning as the inner module -- see /dspy-chain-of-thought
• Checking and validating outputs with assertions -- see /ai-checking-outputs
• Improving accuracy with optimization -- see /ai-improving-accuracy
• For worked examples, see examples.md

dspy.Refine Examples

Example 1: Iterative text improvement with quality criteria

A content writing pipeline that refines blog post introductions until they meet multiple quality criteria: appropriate length, includes a hook, and avoids filler phrases.

import dspy

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

# Define the writing module
class WriteIntro(dspy.Module):
    def __init__(self):
        self.write = dspy.ChainOfThought(
            "topic, audience -> introduction"
        )

    def forward(self, topic, audience):
        return self.write(topic=topic, audience=audience)


# Define quality criteria as a graduated reward function
FILLER_PHRASES = [
    "in today's world",
    "it's no secret that",
    "in this article",
    "have you ever wondered",
    "let's dive in",
]

def intro_quality(args, pred):
    """Score introduction quality on multiple criteria (0.0 to 1.0)."""
    intro = pred.introduction
    score = 0.0

    # Criterion 1: Length between 50-150 words (0.3 points)
    word_count = len(intro.split())
    if 50 <= word_count <= 150:
        score += 0.3
    elif 30 <= word_count <= 200:
        score += 0.15  # partial credit

    # Criterion 2: No filler phrases (0.3 points)
    has_filler = any(phrase in intro.lower() for phrase in FILLER_PHRASES)
    if not has_filler:
        score += 0.3

    # Criterion 3: Mentions the target audience (0.2 points)
    audience = args["audience"].lower()
    if audience in intro.lower() or any(
        word in intro.lower() for word in audience.split()
    ):
        score += 0.2

    # Criterion 4: Has at least 2 sentences (0.2 points)
    sentence_count = intro.count('.') + intro.count('!') + intro.count('?')
    if sentence_count >= 2:
        score += 0.2

    return score


# Wrap with Refine -- up to 4 attempts, accept at 0.8+
refined_writer = dspy.Refine(
    module=WriteIntro(),
    N=4,
    reward_fn=intro_quality,
    threshold=0.8,
)

# Use it
result = refined_writer(
    topic="Using type hints in Python",
    audience="backend developers",
)
print(result.introduction)
print(f"Score: {intro_quality({'topic': 'Using type hints in Python', 'audience': 'backend developers'}, result)}")

What this demonstrates:

• Graduated reward function with four weighted criteria -- Refine picks the best attempt even if none score perfectly
• Wrapping a custom module -- WriteIntro is a standard dspy.Module with forward()
• Multiple input fields -- both topic and audience are available in the args dict
• Practical quality checks -- length bounds, filler phrase detection, audience relevance, sentence structure

Example 2: Self-correcting code generation

A code generation pipeline that writes a Python function and validates it by parsing and running basic checks. Refine retries with feedback when the generated code has syntax errors or fails validation.

import ast
import dspy

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

# Signature with structured output expectations
class GenerateFunction(dspy.Signature):
    """Generate a Python function that solves the given task."""
    task_description: str = dspy.InputField(desc="What the function should do")
    function_name: str = dspy.InputField(desc="Name for the generated function")
    code: str = dspy.OutputField(desc="Complete Python function code, no markdown fences")
    explanation: str = dspy.OutputField(desc="Brief explanation of the approach")


class CodeGenerator(dspy.Module):
    def __init__(self):
        self.generate = dspy.ChainOfThought(GenerateFunction)

    def forward(self, task_description, function_name):
        return self.generate(
            task_description=task_description,
            function_name=function_name,
        )


def valid_python_function(args, pred):
    """Score generated code on syntax, structure, and basic quality."""
    code = pred.code
    expected_name = args["function_name"]
    score = 0.0

    # Strip markdown fences if present
    code = code.strip()
    if code.startswith("```"):
        lines = code.split("\n")
        code = "\n".join(lines[1:-1]) if lines[-1].strip() == "```" else "\n".join(lines[1:])

    # Criterion 1: Valid Python syntax (0.4 points)
    try:
        tree = ast.parse(code)
        score += 0.4
    except SyntaxError:
        return 0.0  # no point checking further

    # Criterion 2: Contains a function definition (0.2 points)
    functions = [
        node for node in ast.walk(tree)
        if isinstance(node, ast.FunctionDef)
    ]
    if functions:
        score += 0.2

    # Criterion 3: Function has the expected name (0.2 points)
    func_names = [f.name for f in functions]
    if expected_name in func_names:
        score += 0.2

    # Criterion 4: Has a docstring (0.1 points)
    for func in functions:
        if (
            func.body
            and isinstance(func.body[0], ast.Expr)
            and isinstance(func.body[0].value, ast.Constant)
            and isinstance(func.body[0].value.value, str)
        ):
            score += 0.1
            break

    # Criterion 5: Has a return statement (0.1 points)
    for func in functions:
        for node in ast.walk(func):
            if isinstance(node, ast.Return):
                score += 0.1
                break
        break  # only check first matching function

    return score


# Wrap with Refine -- up to 5 attempts for code generation
refined_coder = dspy.Refine(
    module=CodeGenerator(),
    N=5,
    reward_fn=valid_python_function,
    threshold=0.9,
)

# Generate a function
result = refined_coder(
    task_description="Calculate the nth Fibonacci number using memoization. Handle negative inputs by raising ValueError.",
    function_name="fibonacci",
)

print("Generated code:")
print(result.code)
print(f"\nExplanation: {result.explanation}")

# Verify the score
score = valid_python_function(
    {"task_description": "...", "function_name": "fibonacci"},
    result,
)
print(f"Quality score: {score}")

What this demonstrates:

• AST-based validation -- uses Python's ast module for reliable syntax and structure checking, no regex heuristics
• Class-based signature -- GenerateFunction with typed fields and descriptions gives the LM clear expectations
• Early exit in reward -- returns 0.0 immediately on syntax error since no other criteria matter
• Higher N (5) -- code generation benefits from more attempts because valid code is harder to produce
• Practical code checks -- correct function name, docstring presence, return statement -- criteria you would check in a real code review
• Feedback loop -- when attempt 1 has a syntax error, Refine tells the LM what went wrong so attempt 2 can fix it