Dspy Chain Of Thought

Step-by-Step Reasoning with dspy.ChainOfThought

Guide the user through using DSPy's ChainOfThought module -- the go-to module for tasks that benefit from intermediate reasoning before producing an answer.

What is ChainOfThought

dspy.ChainOfThought is a drop-in replacement for dspy.Predict that automatically injects a reasoning field before your output fields. Same signature, one-word swap -- the LM reasons step-by-step before answering. The reasoning field is always available on the result even though your signature doesn't declare it.

When CoT helps

ChainOfThought improves accuracy when the task requires the LM to work through intermediate steps before arriving at an answer:

• Multi-step logic -- math, puzzles, conditional reasoning
• Analysis and judgment -- "Is this code buggy?", "Should we approve this loan?"
• Classification with nuance -- when the label depends on weighing multiple factors
• Explanation-heavy tasks -- anything where you want to see why the LM chose its answer
• Complex extraction -- parsing ambiguous data where the LM needs to resolve conflicts

Rule of thumb: If a human would need to think through the problem before answering, use ChainOfThought.

When NOT to use CoT

ChainOfThought adds latency and token cost because the LM generates extra reasoning text. Skip it when:

• Simple lookups -- "What is the capital of France?" No reasoning needed.
• Direct extraction -- pulling a name or date from structured text. Predict is enough.
• Speed-critical paths -- if you need sub-second responses and the task is straightforward, use Predict.
• High-volume, low-complexity -- processing thousands of simple items where reasoning adds cost without improving accuracy.

When in doubt, start with ChainOfThought and switch to Predict later if profiling shows the reasoning is unnecessary.

Passing reasoning downstream

class ReviewDecision(dspy.Module):
    def __init__(self):
        self.analyze = dspy.ChainOfThought("application -> decision: str, risk_level: str")
        self.summarize = dspy.Predict("decision, reasoning -> summary")

    def forward(self, application):
        analysis = self.analyze(application=application)
        # Pass the reasoning to the next step
        summary = self.summarize(
            decision=analysis.decision,
            reasoning=analysis.reasoning,
        )
        return dspy.Prediction(
            decision=analysis.decision,
            risk_level=analysis.risk_level,
            reasoning=analysis.reasoning,
            summary=summary.summary,
        )

Predict vs ChainOfThought

Combining CoT with typed outputs

ChainOfThought works with all the same type constraints as Predict -- Literal, int, float, bool, list[str], and Pydantic models.

import dspy
from pydantic import BaseModel
from typing import Literal

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

class RiskAssessment(BaseModel):
    risk_score: float
    factors: list[str]
    recommendation: Literal["approve", "review", "deny"]

class AssessRisk(dspy.Signature):
    """Assess the risk level of a financial transaction."""
    transaction_details: str = dspy.InputField()
    assessment: RiskAssessment = dspy.OutputField()

assessor = dspy.ChainOfThought(AssessRisk)
result = assessor(
    transaction_details="Wire transfer of $50,000 to a new recipient in a high-risk jurisdiction"
)
print(result.reasoning)                   # detailed risk analysis
print(result.assessment.risk_score)       # 0.85
print(result.assessment.factors)          # ["high amount", "new recipient", "high-risk jurisdiction"]
print(result.assessment.recommendation)   # "review"

The LM reasons through the problem first, then produces the structured output. The reasoning happens before type enforcement, so the LM has space to think before committing to typed fields.

Optimizing CoT with few-shot examples

ChainOfThought benefits significantly from optimization. When you run an optimizer, DSPy discovers high-quality reasoning traces and uses them as few-shot demonstrations:

import dspy
from dspy.evaluate import Evaluate

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

# Your CoT module
classifier = dspy.ChainOfThought("ticket_text -> priority: str, team: str")

# Training data
trainset = [
    dspy.Example(
        ticket_text="Site is down for all users",
        priority="critical",
        team="infrastructure",
    ).with_inputs("ticket_text"),
    dspy.Example(
        ticket_text="Typo on the pricing page",
        priority="low",
        team="content",
    ).with_inputs("ticket_text"),
    # ... more examples
]

# Metric
def ticket_metric(example, prediction, trace=None):
    priority_correct = prediction.priority == example.priority
    team_correct = prediction.team == example.team
    return priority_correct + team_correct

# Optimize -- the optimizer generates and selects good reasoning traces
optimizer = dspy.BootstrapFewShot(metric=ticket_metric, max_bootstrapped_demos=4)
optimized = optimizer.compile(classifier, trainset=trainset)

# The optimized program now includes few-shot examples with reasoning
result = optimized(ticket_text="Users can't upload files larger than 10MB")
print(result.reasoning)  # higher quality reasoning, guided by learned demos
print(result.priority)
print(result.team)

# Save for production
optimized.save("ticket_classifier.json")

What optimization does for ChainOfThought:

• Bootstraps reasoning traces -- the optimizer runs the module on training examples, keeps the traces that led to correct answers, and includes them as few-shot demonstrations
• Improves consistency -- the LM sees examples of good reasoning patterns before generating its own
• Works with all optimizers -- BootstrapFewShot, MIPROv2, BootstrapFewShotWithRandomSearch all support CoT modules

Using CoT inside custom modules

ChainOfThought is a sub-module like any other. Use it in dspy.Module for multi-step pipelines:

import dspy

class CodeReviewer(dspy.Module):
    def __init__(self):
        self.find_issues = dspy.ChainOfThought("code -> issues: list[str], severity: str")
        self.suggest_fix = dspy.ChainOfThought("code, issues -> fixed_code: str")

    def forward(self, code):
        analysis = self.find_issues(code=code)

        if analysis.severity == "none":
            return dspy.Prediction(
                issues=[],
                severity="none",
                fixed_code=code,
                reasoning=analysis.reasoning,
            )

        fix = self.suggest_fix(code=code, issues=analysis.issues)

        return dspy.Prediction(
            issues=analysis.issues,
            severity=analysis.severity,
            fixed_code=fix.fixed_code,
            reasoning=analysis.reasoning,
        )

Both sub-modules use CoT because code review and fix suggestion both benefit from step-by-step thinking. When optimized, DSPy tunes each sub-module's reasoning independently.

Gotchas

• Claude adds `reasoning` as an explicit output field in the signature. DSPy injects it automatically — declaring it yourself creates a duplicate field that confuses the prompt. Just use dspy.ChainOfThought("question -> answer") and access result.reasoning on the output.
• Claude uses ChainOfThought for simple extraction tasks where Predict is sufficient. CoT adds ~100-300 tokens of overhead per call. For tasks like pulling a name from structured text or simple lookups, use dspy.Predict instead — CoT adds cost without improving accuracy on straightforward tasks.
• Claude sets `max_tokens` too low, truncating reasoning before output fields. The reasoning trace is generated before the actual output fields. If max_tokens is tight, the LM runs out of space mid-reasoning and never produces the output fields, causing parse failures. Leave headroom — at least 500 tokens beyond what you expect the output fields to need.
• Claude forgets that `reasoning` is available on the result object. After calling a ChainOfThought module, the reasoning trace is always at result.reasoning even though the signature does not declare it. Claude sometimes re-derives reasoning or asks the LM to explain its answer in a separate call when the trace is already there.
• Claude uses `rationale_field` to rename the reasoning field but does not update downstream references. dspy.ChainOfThought(sig, rationale_field=dspy.OutputField(prefix="Thinking:")) changes the prompt prefix, but the field is still accessed as result.reasoning on the output. Claude sometimes tries to access result.thinking or result.rationale after renaming, which fails silently (returns None).

Additional resources

• dspy.ChainOfThought API docs
• reference.md — constructor parameters, methods, rationale field customization
• examples.md — bug analysis, release decisions, classification with justification

Cross-references

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

• Predict for simple calls without reasoning -- see /dspy-predict
• Signatures for defining input/output contracts -- see /dspy-signatures
• Modules for building multi-step programs with CoT sub-modules -- see /dspy-modules
• Reasoning patterns for broader strategies (decomposition, self-correction) -- see /ai-reasoning
• For worked examples, see examples.md
• 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-chain-of-thought -- Worked Examples

Example 1: Bug analysis with reasoning trace

Analyze a bug report, reason through the likely root cause, and suggest next steps. The reasoning trace makes it possible to audit why the system reached its conclusion.

import dspy
from typing import Literal


class AnalyzeBug(dspy.Signature):
    """Analyze a bug report to identify the likely root cause and suggest debugging steps."""
    title: str = dspy.InputField(desc="Bug report title")
    description: str = dspy.InputField(desc="Bug report description with reproduction steps")
    stack_trace: str = dspy.InputField(desc="Error stack trace, if available")
    root_cause: str = dspy.OutputField(desc="Most likely root cause of the bug")
    component: str = dspy.OutputField(desc="Software component where the bug likely originates")
    debugging_steps: list[str] = dspy.OutputField(desc="Ordered steps to confirm and fix the bug")
    severity: Literal["critical", "high", "medium", "low"] = dspy.OutputField()


# --- Usage ---

lm = dspy.LM("openai/gpt-4o-mini")  # or any LiteLLM-supported provider
dspy.configure(lm=lm)

analyzer = dspy.ChainOfThought(AnalyzeBug)

result = analyzer(
    title="Orders stuck in 'processing' state after payment",
    description=(
        "Since deploying v2.14, ~5% of orders stay in 'processing' after successful "
        "payment. Customers are charged but never receive confirmation. Happens more "
        "during peak hours (2-4pm EST). Rolling back to v2.13 resolves the issue."
    ),
    stack_trace=(
        "TimeoutError: Task timed out after 30s\n"
        "  at OrderService.finalizeOrder(OrderService.java:142)\n"
        "  at PaymentCallback.onSuccess(PaymentCallback.java:67)\n"
        "  at EventLoop.processQueue(EventLoop.java:201)"
    ),
)

# The reasoning trace shows how the LM arrived at its diagnosis
print("=== Reasoning ===")
print(result.reasoning)
# "The bug appeared after v2.14 and resolves on rollback, so the root cause
#  is in v2.14 changes. The TimeoutError at OrderService.finalizeOrder suggests
#  the order finalization step is taking too long. The correlation with peak hours
#  points to a concurrency or resource contention issue..."

print(f"\nRoot cause: {result.root_cause}")
# "Race condition or resource contention in OrderService.finalizeOrder introduced in v2.14"

print(f"Component: {result.component}")
# "OrderService"

print(f"Severity: {result.severity}")
# "critical"

print("\nDebugging steps:")
for i, step in enumerate(result.debugging_steps, 1):
    print(f"  {i}. {step}")
# 1. Diff OrderService.java between v2.13 and v2.14 to identify the change
# 2. Check connection pool and thread pool sizes under load
# 3. Add timing instrumentation to finalizeOrder to find the slow path
# 4. Reproduce under load in staging with v2.14
# 5. Check database locks during peak-hour order finalization


# --- Logging the reasoning for audit trail ---

import json

audit_record = {
    "bug_title": "Orders stuck in 'processing' state after payment",
    "diagnosis": result.root_cause,
    "severity": result.severity,
    "reasoning_trace": result.reasoning,  # keep for auditing
    "steps": result.debugging_steps,
}
print(json.dumps(audit_record, indent=2))

Key points:

• The reasoning field gives a full trace of the LM's diagnostic thought process
• Logging the reasoning creates an audit trail -- useful for post-mortems and quality reviews
• Typed outputs (Literal for severity, list[str] for steps) ensure structured results alongside free-form reasoning

Example 2: Decision-making with visible logic

Make a go/no-go decision on a feature release, with the reasoning visible to stakeholders. The reasoning field serves as the justification document.

import dspy
from typing import Literal
from pydantic import BaseModel


class ReleaseMetrics(BaseModel):
    test_pass_rate: float
    error_rate_delta: float
    p99_latency_ms: float
    rollback_plan: bool
    affected_users_percent: float


class ReleaseDecision(dspy.Signature):
    """Decide whether a feature is safe to release to production based on metrics and context."""
    feature_name: str = dspy.InputField()
    metrics: ReleaseMetrics = dspy.InputField(desc="Current release metrics")
    context: str = dspy.InputField(desc="Additional context about the release")
    decision: Literal["go", "no-go", "conditional"] = dspy.OutputField()
    conditions: list[str] = dspy.OutputField(
        desc="Conditions that must be met before release (empty if decision is 'go')"
    )
    risk_summary: str = dspy.OutputField(desc="One-paragraph risk summary for stakeholders")


# --- Usage ---

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

decider = dspy.ChainOfThought(ReleaseDecision)

metrics = ReleaseMetrics(
    test_pass_rate=0.97,
    error_rate_delta=0.3,
    p99_latency_ms=450,
    rollback_plan=True,
    affected_users_percent=15.0,
)

result = decider(
    feature_name="New checkout flow",
    metrics=metrics,
    context=(
        "Holiday shopping season starts in 3 days. The new checkout flow reduces "
        "cart abandonment by 12% in A/B testing. Error rate increased 0.3% vs baseline "
        "but all errors are non-blocking validation warnings."
    ),
)

# The reasoning shows the decision logic step by step
print("=== Decision Reasoning ===")
print(result.reasoning)
# "Let me evaluate each metric against release criteria:
#  - Test pass rate 97% meets the 95% threshold
#  - Error rate delta of 0.3% is slightly elevated, but context says these are
#    non-blocking validation warnings, not payment failures
#  - P99 latency of 450ms is within the 500ms SLA
#  - Rollback plan exists, which is required
#  - 15% of users affected is significant, especially before holiday season
#  - The 12% reduction in cart abandonment is a strong business case
#  - The timing risk (3 days before holidays) is notable but rollback plan mitigates it..."

print(f"\nDecision: {result.decision}")
# "conditional"

print("\nConditions:")
for condition in result.conditions:
    print(f"  - {condition}")
# - Monitor error rate for 2 hours after staged rollout to first 5% of users
# - Confirm all validation warnings are truly non-blocking in production logs
# - Have on-call engineer available during the first 24 hours

print(f"\nRisk summary: {result.risk_summary}")


# --- Share with stakeholders ---

report = f"""
# Release Decision: {result.feature_name}

**Decision:** {result.decision.upper()}

## Reasoning
{result.reasoning}

## Conditions
{"".join(f"- {c}" + chr(10) for c in result.conditions) if result.conditions else "None -- clear to proceed."}

## Risk Summary
{result.risk_summary}
"""
print(report)

Key points:

• The reasoning field acts as a decision justification that stakeholders can review
• Pydantic input models (ReleaseMetrics) let you pass structured data cleanly
• Literal["go", "no-go", "conditional"] constrains the decision to valid options
• The reasoning is generated before the decision, so the LM weighs all factors before committing

Example 3: Classification with justification

Classify support tickets with a written justification for each classification. The justification helps human reviewers verify the routing and catch misclassifications quickly.

import dspy
from typing import Literal


class ClassifyTicket(dspy.Signature):
    """Classify a customer support ticket and justify the classification.
    Consider the customer's primary intent, not just keywords."""
    ticket_subject: str = dspy.InputField()
    ticket_body: str = dspy.InputField()
    category: Literal[
        "billing", "technical", "account", "feature_request", "other"
    ] = dspy.OutputField()
    priority: Literal["urgent", "high", "normal", "low"] = dspy.OutputField()
    justification: str = dspy.OutputField(
        desc="One-sentence explanation of why this category and priority were chosen"
    )


# --- Usage ---

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

classifier = dspy.ChainOfThought(ClassifyTicket)

result = classifier(
    ticket_subject="Can't access my account since password reset",
    ticket_body=(
        "I reset my password yesterday using the forgot password link. "
        "Since then I get 'invalid credentials' every time I try to log in. "
        "I have a client presentation in 2 hours and all my files are in there. "
        "Please help ASAP."
    ),
)

# reasoning: the LM's internal thought process (for debugging/logging)
print("=== Internal Reasoning ===")
print(result.reasoning)
# "The customer reset their password but can't log in. This is an account access issue,
#  not a billing or feature request. The mention of a presentation in 2 hours and 'ASAP'
#  indicates time pressure. The password reset flow may have a bug (could be technical),
#  but the primary intent is regaining account access..."

# justification: the user-facing explanation (for the support queue)
print(f"\nCategory: {result.category}")
# "account"
print(f"Priority: {result.priority}")
# "urgent"
print(f"Justification: {result.justification}")
# "Account access blocked after password reset with a time-sensitive deadline in 2 hours."


# --- Batch processing with justifications ---

tickets = [
    {
        "subject": "Charge for plan I didn't sign up for",
        "body": "I was charged $49 for a Pro plan but I'm on the free tier. Please refund.",
    },
    {
        "subject": "API returns 500 on large payloads",
        "body": "When sending payloads > 5MB to /api/upload, we get 500 errors. Works fine under 5MB.",
    },
    {
        "subject": "Would love dark mode",
        "body": "Any plans for a dark mode? Would be easier on the eyes for late-night coding sessions.",
    },
]

for ticket in tickets:
    result = classifier(
        ticket_subject=ticket["subject"],
        ticket_body=ticket["body"],
    )
    print(f"\n[{result.priority.upper()}] [{result.category}] {ticket['subject']}")
    print(f"  Justification: {result.justification}")


# --- Optimization with reasoning ---

trainset = [
    dspy.Example(
        ticket_subject="Can't access my account since password reset",
        ticket_body="I reset my password yesterday...",
        category="account",
        priority="urgent",
        justification="Account access blocked after password reset with time-sensitive deadline.",
    ).with_inputs("ticket_subject", "ticket_body"),
    # ... more labeled examples
]

def ticket_metric(example, prediction, trace=None):
    cat_match = prediction.category == example.category
    pri_match = prediction.priority == example.priority
    has_justification = len(prediction.justification.strip()) > 10
    return cat_match + 0.5 * pri_match + 0.25 * has_justification

# optimizer = dspy.BootstrapFewShot(metric=ticket_metric, max_bootstrapped_demos=4)
# optimized = optimizer.compile(classifier, trainset=trainset)
# optimized.save("ticket_classifier.json")

Key points:

• Two levels of explanation: reasoning is the internal trace for developers; justification is a clean, user-facing explanation declared in the signature
• ChainOfThought generates reasoning automatically; justification is an explicit output field you control
• The justification field gives human reviewers a quick way to verify the classification without reading the full ticket
• Optimization teaches the LM to produce better reasoning traces, which improves both the classification accuracy and the quality of justifications

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

dspy.ChainOfThought — API Reference

Constructor

dspy.ChainOfThought(
    signature,                          # str | type[Signature] (required)
    rationale_field=None,               # FieldInfo | None
    rationale_field_type=str,           # type
    **config,                           # dict[str, Any]
)

Inheritance

ChainOfThought extends dspy.Module. It wraps an internal dspy.Predict instance with an augmented signature that prepends a reasoning output field before your declared output fields.

Key methods

Module management methods

Persistence methods

How the reasoning field works

ChainOfThought augments your signature by prepending a reasoning field:

Original:  question -> answer
Augmented: question -> reasoning, answer

• The reasoning field is always accessible as result.reasoning
• It is generated before the output fields, giving the LM space to think
• If rationale_field is set, the prompt prefix changes but the attribute name stays reasoning
• The reasoning field is included in few-shot demos during optimization