Ai Reasoning

Build AI That Reasons Through Hard Problems

Guide the user through making AI solve problems that need more than a simple answer. When a task requires planning, multi-step logic, or choosing the right approach, basic prompting fails. DSPy gives you composable reasoning strategies.

Step 1: Does the task need advanced reasoning?

Use this decision tree:

If the user isn't sure, start with `ChainOfThought` — it's the right default for most tasks.

Step 2: Basic reasoning patterns

ChainOfThought — think step by step

The workhorse. Adds intermediate reasoning before the final answer:

import dspy

class AnalyzeBug(dspy.Signature):
    """Analyze the bug report and determine root cause."""
    bug_report: str = dspy.InputField(desc="The bug report with error details")
    root_cause: str = dspy.OutputField(desc="The most likely root cause")
    fix_suggestion: str = dspy.OutputField(desc="Suggested fix")

analyzer = dspy.ChainOfThought(AnalyzeBug)
result = analyzer(bug_report="Users see 500 errors after deploying v2.3...")
print(result.reasoning)  # shows step-by-step thinking
print(result.root_cause)

ProgramOfThought — write code to compute the answer

When the answer requires calculation, let the AI write and execute code:

class CalculateMetrics(dspy.Signature):
    """Calculate business metrics from the provided data."""
    data_description: str = dspy.InputField(desc="Description of the data and what to calculate")
    result: str = dspy.OutputField(desc="The calculated result")

calculator = dspy.ProgramOfThought(CalculateMetrics)
result = calculator(data_description="Revenue was $50k in Jan, $63k in Feb, $58k in March. What's the average monthly growth rate?")

ProgramOfThought generates Python code, runs it in a sandbox, and returns the output. Use this for anything involving math, dates, data manipulation, or counting.

MultiChainComparison — generate multiple answers, pick the best

When quality matters more than speed, reason multiple ways and compare:

class RecommendApproach(dspy.Signature):
    """Recommend the best technical approach for this problem."""
    problem: str = dspy.InputField()
    recommendation: str = dspy.OutputField()

recommender = dspy.MultiChainComparison(RecommendApproach)
result = recommender(problem="We need to add real-time notifications to our app")
# Internally generates multiple chains of thought, then picks the best

When to use each

class SmartReasoner(dspy.Module):
    """Route to the best reasoning strategy based on the task."""
    def __init__(self):
        self.classify = dspy.Predict("question -> task_type: str")
        self.cot = dspy.ChainOfThought("question -> answer")
        self.pot = dspy.ProgramOfThought("question -> answer")
        self.mcc = dspy.MultiChainComparison("question -> answer")

    def forward(self, question):
        task_type = self.classify(question=question).task_type.lower()

        if "math" in task_type or "calcul" in task_type or "count" in task_type:
            return self.pot(question=question)
        elif "compare" in task_type or "recommend" in task_type or "best" in task_type:
            return self.mcc(question=question)
        else:
            return self.cot(question=question)

Step 3: Self-Discovery pattern

For genuinely hard problems where the AI needs to figure out how to think, not just think harder. Inspired by Self-Discover prompting research.

The 4-stage pipeline:

1. Select — pick relevant reasoning strategies from a library 2. Adapt — tailor those strategies to the specific task 3. Plan — create a structured reasoning plan 4. Execute — follow the plan to produce the answer

from pydantic import BaseModel, Field

# Reasoning strategy library
REASONING_STRATEGIES = [
    "Break the problem into smaller sub-problems",
    "Think about edge cases and exceptions",
    "Work backwards from the desired outcome",
    "Consider analogies to simpler problems",
    "Identify constraints and requirements first",
    "Generate multiple hypotheses and evaluate each",
    "Think about what information is missing",
    "Check if the problem has been solved before in a different context",
    "Separate facts from assumptions",
    "Consider the problem from different stakeholder perspectives",
]

class SelectStrategies(dspy.Signature):
    """Select the most relevant reasoning strategies for this task."""
    task: str = dspy.InputField(desc="The problem to solve")
    available_strategies: list[str] = dspy.InputField()
    selected_strategies: list[str] = dspy.OutputField(
        desc="2-4 most relevant strategies for this task"
    )

class AdaptStrategies(dspy.Signature):
    """Adapt the selected strategies to this specific task."""
    task: str = dspy.InputField()
    strategies: list[str] = dspy.InputField(desc="Selected reasoning strategies")
    adapted_strategies: list[str] = dspy.OutputField(
        desc="Strategies rewritten for this specific problem"
    )

class ReasoningStep(BaseModel):
    step_number: int
    strategy: str = Field(description="Which reasoning strategy this step uses")
    description: str = Field(description="What to do in this step")

class CreatePlan(dspy.Signature):
    """Create a structured step-by-step reasoning plan."""
    task: str = dspy.InputField()
    adapted_strategies: list[str] = dspy.InputField()
    plan: list[ReasoningStep] = dspy.OutputField(desc="Ordered reasoning steps")

class ExecutePlan(dspy.Signature):
    """Execute the reasoning plan to solve the task."""
    task: str = dspy.InputField()
    plan: list[ReasoningStep] = dspy.InputField()
    step_results: list[str] = dspy.OutputField(desc="Result of each reasoning step")
    final_answer: str = dspy.OutputField(desc="The final answer based on all reasoning")

class SelfDiscoveryReasoner(dspy.Module):
    def __init__(self):
        self.select = dspy.ChainOfThought(SelectStrategies)
        self.adapt = dspy.ChainOfThought(AdaptStrategies)
        self.plan = dspy.ChainOfThought(CreatePlan)
        self.execute = dspy.ChainOfThought(ExecutePlan)

    def forward(self, task):
        # Stage 1: Select relevant strategies
        selected = self.select(
            task=task,
            available_strategies=REASONING_STRATEGIES,
        ).selected_strategies

        # Stage 2: Adapt to this task
        adapted = self.adapt(
            task=task,
            strategies=selected,
        ).adapted_strategies

        # Stage 3: Create reasoning plan
        plan = self.plan(
            task=task,
            adapted_strategies=adapted,
        ).plan

        # Stage 4: Execute the plan
        result = self.execute(task=task, plan=plan)

        return dspy.Prediction(
            strategies=selected,
            plan=plan,
            step_results=result.step_results,
            answer=result.final_answer,
        )

Step 4: Structured reasoning plans

For complex tasks, force the AI to show its work in a structured format:

class ReasoningTrace(BaseModel):
    step: str = Field(description="What this reasoning step does")
    observation: str = Field(description="What was observed or concluded")
    confidence: float = Field(description="0.0-1.0 confidence in this step")

class StructuredReasoner(dspy.Module):
    def __init__(self):
        self.reason = dspy.ChainOfThought(ReasonWithTrace)

    def forward(self, question):
        result = self.reason(question=question)

        return result

class ReasonWithTrace(dspy.Signature):
    """Solve the problem step by step, showing reasoning at each stage."""
    question: str = dspy.InputField()
    trace: list[ReasoningTrace] = dspy.OutputField(desc="Step-by-step reasoning trace")
    answer: str = dspy.OutputField(desc="Final answer based on the reasoning trace")

Step 5: Evaluate reasoning quality

Don't just check the final answer — evaluate the reasoning process:

Judge intermediate steps

class JudgeReasoning(dspy.Signature):
    """Judge whether the reasoning process is sound."""
    question: str = dspy.InputField()
    reasoning_steps: list[str] = dspy.InputField(desc="The steps taken to reach the answer")
    answer: str = dspy.InputField()
    steps_are_logical: bool = dspy.OutputField(desc="Each step follows from the previous")
    no_logical_leaps: bool = dspy.OutputField(desc="No unjustified jumps in reasoning")
    answer_follows: bool = dspy.OutputField(desc="The answer follows from the reasoning")

def reasoning_quality_metric(example, prediction, trace=None):
    # Check final answer correctness
    correct = prediction.answer.strip().lower() == example.answer.strip().lower()

    # Also check reasoning quality
    judge = dspy.Predict(JudgeReasoning)
    quality = judge(
        question=example.question,
        reasoning_steps=prediction.step_results if hasattr(prediction, 'step_results') else [prediction.reasoning],
        answer=prediction.answer,
    )

    reasoning_score = (
        quality.steps_are_logical + quality.no_logical_leaps + quality.answer_follows
    ) / 3

    # Weight: 60% correct answer, 40% good reasoning
    return (0.6 * correct) + (0.4 * reasoning_score)

Compare reasoning approaches

Test which reasoning strategy works best for your task:

from dspy.evaluate import Evaluate

evaluator = Evaluate(devset=devset, metric=reasoning_quality_metric, num_threads=4)

# Test different approaches
cot = dspy.ChainOfThought("question -> answer")
pot = dspy.ProgramOfThought("question -> answer")
self_disc = SelfDiscoveryReasoner()

print("ChainOfThought:", evaluator(cot))
print("ProgramOfThought:", evaluator(pot))
print("SelfDiscovery:", evaluator(self_disc))

Step 6: Optimize reasoning

BootstrapFewShot per stage

For multi-stage reasoning (like Self-Discovery), optimize each stage. Typical improvement: 15-30% on reasoning quality metrics (e.g., a ChainOfThought module going from 62% to 81% on a multi-step QA task after 4 bootstrapped demos):

optimizer = dspy.BootstrapFewShot(
    metric=reasoning_quality_metric,
    max_bootstrapped_demos=4,
)
optimized = optimizer.compile(SelfDiscoveryReasoner(), trainset=trainset)

MIPROv2 for instruction tuning

Automatically discover better instructions for the reasoning prompts:

optimizer = dspy.MIPROv2(metric=reasoning_quality_metric, auto="medium")
optimized = optimizer.compile(SelfDiscoveryReasoner(), trainset=trainset)

GEPA for reflective analysis

GEPA analyzes traces of successful and failed attempts to generate better instructions:

optimizer = dspy.GEPA(metric=reasoning_quality_metric)
optimized = optimizer.compile(SelfDiscoveryReasoner(), trainset=trainset)

Key patterns

• Default to ChainOfThought — it's the right choice for most tasks that need reasoning
• ProgramOfThought for computation — let the AI write code for math, dates, counting
• MultiChainComparison for high stakes — generate multiple answers and pick the best
• Self-Discovery for novel problems — dynamically select how to think, not just what to think
• Evaluate the reasoning, not just the answer — good reasoning produces reliably correct answers
• Structured traces — JSON reasoning steps make debugging and optimization easier

Other reasoning-capable modules

Gotchas

• Adding a `reasoning` field to your signature when using ChainOfThought. DSPy injects the reasoning field automatically. Adding your own creates a duplicate that confuses the LM and produces garbled output. Just define your task-specific input/output fields and let dspy.ChainOfThought handle the rest.
• Using ProgramOfThought for everything involving numbers. ProgramOfThought generates and executes Python code, which requires a sandbox and adds latency. For simple numeric comparisons or estimates that do not need exact computation, ChainOfThought is faster and sufficient. Reserve ProgramOfThought for actual arithmetic, date math, or data manipulation.
• Forgetting that MultiChainComparison makes N separate LM calls. Each chain is an independent call, so cost and latency scale linearly. For latency-sensitive paths, consider using a single ChainOfThought wrapped with dspy.Refine instead of MultiChainComparison with 3-5 chains.
• Building a Self-Discovery pipeline without optimizing each stage separately. When you call BootstrapFewShot on a multi-stage module, it optimizes end-to-end but the intermediate stages (select, adapt, plan) often get weak demos. Evaluate intermediate outputs during development to catch silent degradation in early stages.
• Using string matching to route between reasoning strategies. The SmartReasoner pattern with if "math" in task_type is brittle — LMs produce unpredictable classification labels. Use dspy.Predict with Literal types for routing, or better yet, let the optimizer discover which strategy works best via dspy.Evaluate comparisons.

Cross-references

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

• ChainOfThought for the core reasoning module — see /dspy-chain-of-thought
• ProgramOfThought for code-generating computation — see /dspy-program-of-thought
• MultiChainComparison for multi-path reasoning — see /dspy-multi-chain-comparison
• Signatures for defining input/output contracts — see /dspy-signatures
• Refine for constraining reasoning quality with reward functions — see /dspy-refine
• Simple calls without reasoning — see /dspy-predict
• Need AI to call APIs and use tools? See /ai-taking-actions
• Need multi-step pipelines with predetermined stages? See /ai-building-pipelines
• Measure and improve your reasoning system — see /ai-improving-accuracy
• Not sure which skill to use? Try /ai-do
• 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

Additional resources

• dspy.ChainOfThought API docs
• dspy.ProgramOfThought API docs
• dspy.MultiChainComparison API docs
• For worked examples (complex questions, data analysis, planning), see examples.md
• For condensed API reference, see reference.md

AI Reasoning — Worked Examples

Example 1: Complex customer question solver

Handle nuanced customer questions that need multi-step reasoning to answer correctly.

Setup

import dspy
from pydantic import BaseModel, Field

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

Signatures and module

class AnalyzeQuestion(dspy.Signature):
    """Break down a complex customer question into sub-questions."""
    question: str = dspy.InputField(desc="Customer's question")
    context: str = dspy.InputField(desc="Relevant product/policy information")
    sub_questions: list[str] = dspy.OutputField(desc="Sub-questions to answer first")

class AnswerSubQuestion(dspy.Signature):
    """Answer one specific sub-question using the context."""
    sub_question: str = dspy.InputField()
    context: str = dspy.InputField()
    answer: str = dspy.OutputField()

class SynthesizeAnswer(dspy.Signature):
    """Combine sub-answers into a complete, helpful response."""
    question: str = dspy.InputField(desc="Original customer question")
    sub_answers: list[str] = dspy.InputField(desc="Answers to each sub-question")
    response: str = dspy.OutputField(desc="Complete answer addressing all parts")

class ComplexQuestionSolver(dspy.Module):
    def __init__(self):
        self.analyze = dspy.ChainOfThought(AnalyzeQuestion)
        self.answer_sub = dspy.ChainOfThought(AnswerSubQuestion)
        self.synthesize = dspy.ChainOfThought(SynthesizeAnswer)

    def forward(self, question, context):
        # Break down the question
        analysis = self.analyze(question=question, context=context)

        # Answer each sub-question
        sub_answers = []
        for sq in analysis.sub_questions:
            result = self.answer_sub(sub_question=sq, context=context)
            sub_answers.append(f"{sq}: {result.answer}")

        # Synthesize into a complete response
        final = self.synthesize(question=question, sub_answers=sub_answers)

        return dspy.Prediction(
            sub_questions=analysis.sub_questions,
            sub_answers=sub_answers,
            response=final.response,
        )

Usage

solver = ComplexQuestionSolver()

result = solver(
    question="If I upgrade from the Pro plan to Enterprise mid-billing-cycle, do I get a prorated refund for Pro, and does the Enterprise trial period still apply?",
    context="""
    Pricing policy:
    - Pro plan: $49/month, billed monthly
    - Enterprise plan: $199/month, billed annually
    - Mid-cycle upgrades: prorated credit applied to new plan
    - Enterprise trial: 14-day free trial for new customers only
    - Existing customers upgrading: no trial, immediate billing
    """,
)

print(result.sub_questions)
# ["Does a mid-cycle upgrade get prorated?", "Does the Enterprise trial apply to upgrades?", "How is the billing difference calculated?"]
print(result.response)
# Clear answer covering proration, no trial for upgrades, and billing details

Metric

class JudgeCustomerResponse(dspy.Signature):
    """Judge if the response correctly and completely answers the customer question."""
    question: str = dspy.InputField()
    context: str = dspy.InputField()
    response: str = dspy.InputField()
    gold_answer: str = dspy.InputField()
    is_correct: bool = dspy.OutputField(desc="Factually correct given the context")
    is_complete: bool = dspy.OutputField(desc="Addresses all parts of the question")
    is_clear: bool = dspy.OutputField(desc="Easy for a customer to understand")

def customer_qa_metric(example, prediction, trace=None):
    judge = dspy.Predict(JudgeCustomerResponse)
    result = judge(
        question=example.question,
        context=example.context,
        response=prediction.response,
        gold_answer=example.response,
    )
    return (result.is_correct + result.is_complete + result.is_clear) / 3

---

Example 2: Multi-step data analysis

Analyze data by breaking the problem into computation steps.

Signatures and module

class PlanAnalysis(dspy.Signature):
    """Plan the steps needed to analyze this data question."""
    question: str = dspy.InputField(desc="The analysis question")
    data_description: str = dspy.InputField(desc="Description of available data")
    steps: list[str] = dspy.OutputField(desc="Ordered computation steps")

class ComputeStep(dspy.Signature):
    """Perform one computation step of the analysis."""
    step_description: str = dspy.InputField()
    data_description: str = dspy.InputField()
    prior_results: list[str] = dspy.InputField(desc="Results from previous steps")
    result: str = dspy.OutputField(desc="The computed result for this step")

class DataAnalyzer(dspy.Module):
    def __init__(self):
        self.plan = dspy.ChainOfThought(PlanAnalysis)
        self.compute = dspy.ProgramOfThought(ComputeStep)

    def forward(self, question, data_description):
        # Plan the analysis
        analysis_plan = self.plan(
            question=question,
            data_description=data_description,
        )

        # Execute each step
        prior_results = []
        for step in analysis_plan.steps:
            result = self.compute(
                step_description=step,
                data_description=data_description,
                prior_results=prior_results,
            )
            prior_results.append(f"{step}: {result.result}")

        return dspy.Prediction(
            plan=analysis_plan.steps,
            step_results=prior_results,
            answer=prior_results[-1] if prior_results else "No result",
        )

Usage

analyzer = DataAnalyzer()

result = analyzer(
    question="Which product category had the highest month-over-month growth in Q4?",
    data_description="""
    Monthly revenue by category:
    Electronics: Oct=$120k, Nov=$135k, Dec=$180k
    Clothing: Oct=$80k, Nov=$95k, Dec=$110k
    Home: Oct=$60k, Nov=$55k, Dec=$70k
    """,
)

print(result.plan)
# ["Calculate MoM growth rates for each category", "Compare growth rates", "Identify the highest"]
print(result.answer)
# "Electronics had the highest MoM growth at 33.3% (Nov→Dec)"

---

Example 3: Planning and scheduling assistant

Use Self-Discovery reasoning to plan complex tasks with constraints.

Signatures and module

PLANNING_STRATEGIES = [
    "Identify all constraints and hard deadlines",
    "Find dependencies — what must happen before what",
    "Estimate effort for each task",
    "Look for tasks that can happen in parallel",
    "Identify the critical path (longest chain of dependencies)",
    "Build in buffer time for unknowns",
    "Consider resource availability",
]

class SelectPlanningStrategies(dspy.Signature):
    """Select the most relevant planning strategies for this scenario."""
    scenario: str = dspy.InputField()
    strategies: list[str] = dspy.InputField()
    selected: list[str] = dspy.OutputField(desc="2-4 most relevant strategies")

class TaskItem(BaseModel):
    name: str
    duration: str = Field(description="Estimated duration, e.g. '2 days'")
    depends_on: list[str] = Field(description="Names of tasks that must complete first")
    assigned_to: str = Field(description="Who should do this, or 'unassigned'")

class CreatePlan(dspy.Signature):
    """Create a project plan based on the reasoning strategies."""
    scenario: str = dspy.InputField()
    strategies: list[str] = dspy.InputField(desc="Planning strategies to apply")
    tasks: list[TaskItem] = dspy.OutputField(desc="Ordered list of tasks")
    critical_path: list[str] = dspy.OutputField(desc="Tasks on the critical path")
    estimated_total: str = dspy.OutputField(desc="Total estimated time")
    risks: list[str] = dspy.OutputField(desc="Key risks to watch")

class PlanningAssistant(dspy.Module):
    def __init__(self):
        self.select = dspy.ChainOfThought(SelectPlanningStrategies)
        self.plan = dspy.ChainOfThought(CreatePlan)

    def forward(self, scenario):
        # Select relevant strategies
        selected = self.select(
            scenario=scenario,
            strategies=PLANNING_STRATEGIES,
        ).selected

        # Create the plan
        return self.plan(
            scenario=scenario,
            strategies=selected,
        )


def planning_completeness_reward(args, pred):
    """Soft reward encouraging realistic, risk-aware plans."""
    score = 1.0
    if len(pred.tasks) < 3:
        score -= 0.2  # soft: a real plan should have at least 3 tasks
    if len(pred.risks) < 1:
        score -= 0.2  # soft: every plan has at least one risk
    return score

planner = dspy.Refine(
    module=PlanningAssistant(), N=3, reward_fn=planning_completeness_reward, threshold=0.8
)

Usage

result = planner(scenario="""
We need to migrate our database from PostgreSQL to a new managed service.
Constraints: zero downtime for the API, 3 engineers available, must complete
in 2 weeks, 50GB of data, the API handles 1000 req/sec during peak hours.
""")

for task in result.tasks:
    deps = f" (after: {', '.join(task.depends_on)})" if task.depends_on else ""
    print(f"  [{task.duration}] {task.name}{deps} — {task.assigned_to}")

print(f"\nCritical path: {' → '.join(result.critical_path)}")
print(f"Total estimate: {result.estimated_total}")
print(f"Risks: {result.risks}")

Metric

class JudgePlan(dspy.Signature):
    """Judge whether the plan is realistic and complete."""
    scenario: str = dspy.InputField()
    tasks: list[str] = dspy.InputField(desc="Task names from the plan")
    risks: list[str] = dspy.InputField()
    addresses_constraints: bool = dspy.OutputField(desc="Plan accounts for all stated constraints")
    dependencies_make_sense: bool = dspy.OutputField(desc="Task ordering is logical")
    is_actionable: bool = dspy.OutputField(desc="Someone could follow this plan and execute it")

def planning_metric(example, prediction, trace=None):
    judge = dspy.Predict(JudgePlan)
    result = judge(
        scenario=example.scenario,
        tasks=[t.name for t in prediction.tasks],
        risks=prediction.risks,
    )
    return (result.addresses_constraints + result.dependencies_make_sense + result.is_actionable) / 3

optimizer = dspy.BootstrapFewShot(metric=planning_metric, max_bootstrapped_demos=4)
optimized = optimizer.compile(planner, trainset=trainset)

AI Reasoning — API Reference

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

dspy.ChainOfThought

Adds step-by-step reasoning before producing the final answer.

cot = dspy.ChainOfThought(signature, rationale_field=None, rationale_field_type=str, **config)

Key methods:

Output: Returns a Prediction with a .reasoning field (the step-by-step trace) plus all signature output fields.

---

dspy.ProgramOfThought

Generates and executes Python code to compute the answer.

pot = dspy.ProgramOfThought(signature, max_iters=3, interpreter=None)

Key methods: Same as ChainOfThought (forward, aforward, batch, set_lm, save, load).

How it works: Generates Python code, executes it in a sandboxed interpreter, and returns the output. Retries up to max_iters times if execution fails.

---

dspy.MultiChainComparison

Generates multiple reasoning chains and selects the best answer.

mcc = dspy.MultiChainComparison(signature, M=3, temperature=0.7, **config)

Key methods:

How it works: Internally generates M chains of thought at higher temperature, then passes all rationales to a comparison step that selects the best answer.

---

dspy.BestOfN

Generate N completions and return the highest-scoring one.

bon = dspy.BestOfN(module, N=3, reward_fn=my_reward, threshold=1.0)

Simpler than MultiChainComparison when you have a programmatic metric. No internal comparison LM call — just runs the metric on each completion.

---

dspy.Refine

Iteratively improve an answer using feedback.

refine = dspy.Refine(module, N=3, reward_fn=my_reward, threshold=0.7)

Runs the module up to N times, scores each output with reward_fn(args, pred) -> float, and returns the first output meeting threshold (or the best seen). Useful for tasks where first-draft quality is acceptable but refinement improves results (writing, code generation).

---

dspy.RLM

Reasoning Language Model — test-time compute scaling for verified reasoning.

rlm = dspy.RLM(signature)

Uses extended generation with verification steps. Best for tasks with verifiable answers (math proofs, code that must pass tests). Higher latency but stronger correctness guarantees.

---

Common patterns

Accessing reasoning traces

result = cot(question="Why did the deploy fail?")
print(result.reasoning)  # Step-by-step trace (auto-injected by ChainOfThought)
print(result.answer)     # Final answer from your signature

Composing reasoning modules

class Pipeline(dspy.Module):
    def __init__(self):
        self.plan = dspy.ChainOfThought("task -> steps: list[str]")
        self.execute = dspy.ProgramOfThought("steps, data -> result")

    def forward(self, task, data):
        steps = self.plan(task=task).steps
        return self.execute(steps=steps, data=data)

Optimization

All reasoning modules support the standard DSPy optimization flow:

optimizer = dspy.BootstrapFewShot(metric=my_metric, max_bootstrapped_demos=4)
optimized = optimizer.compile(my_module, trainset=trainset)

# Or for instruction tuning:
optimizer = dspy.MIPROv2(metric=my_metric, auto="medium")
optimized = optimizer.compile(my_module, trainset=trainset)