Phase 4: Multi-Agent Systems · 60 min · Python · Anthropic SDK

Reflection & Self-Critique Agents

Single-pass outputs are a first draft, not a final answer.

Real-world application: Reflection loops are used in production at companies running AI writing assistants, code review tools, and automated analysis pipelines. The ability to implement reliable self-critique with proper termination conditions and cost awareness is a sign of engineering maturity — not just pattern-following.

What you will learn

Implement a reflection loop: generate → critique → revise → repeat until passing
Design a critique prompt that produces actionable, specific feedback
Apply self-consistency: run the task multiple times and vote on the best output

The Problem

Single-pass generation is reliable for simple tasks, but as complexity grows, the first output is rarely the best. The model doesn't know what it doesn't know. A reflection loop gives the model a second perspective on its own work: generate, step back and critique, revise based on specific feedback, repeat until the output meets a quality threshold.

The insight from the Reflexion paper (Shinn et al., 2023): verbal feedback from a critic agent is a surprisingly effective form of learning signal. The model uses the critique to reason about its mistakes and generate better outputs.

The reflection pattern

The loop has three roles, each with a distinct purpose:

Generator: produces the initial output. Can be the same model with a "generate" system prompt, or a separate model.

Critic: evaluates the output against specific quality dimensions. Must produce structured, actionable feedback — not vague praise or complaints.

Reviser: takes the original output + the critic's feedback and produces an improved version. Uses the critique as an explicit set of instructions.

The critic's passes field is the termination condition. When the critic says passes: true, the loop exits. The hard revision limit (max_revisions=3) is the safety net.

Your reflection loop runs the critic 5 times but the output never "passes." What safeguard must you have?

Reflection loops without a revision limit can run indefinitely, consuming tokens and time. In production, set max_revisions=3. After 3 cycles, return the output at its current quality state — potentially with a flag that human review is recommended. The goal is 'better than single-pass,' not 'perfect.'

Designing the critique prompt

The critique prompt is the most important engineering decision in a reflection loop. A vague critique produces useless revisions. A structured critique produces targeted improvements.

Bad critique prompt:

"Review this report and tell me how to improve it."

Good critique prompt:

Review this output on exactly 4 dimensions:
1. Factual accuracy: are all claims verifiable?
2. Completeness: are any key aspects missing?
3. Tone: does it match the specified audience?
4. Clarity: are there any confusing sentences?

Output format (required):
{
  "overall_score": <1-10>,
  "issues": [
    {
      "dimension": "factual_accuracy",
      "severity": "high|medium|low",
      "description": "<specific issue>",
      "fix": "<specific action to fix it>"
    }
  ],
  "passes": <true if score >= 7 and no high-severity issues>
}

The passes field is boolean — deterministic termination. The fix field in each issue gives the reviser explicit instructions — no interpretation required.

Self-consistency and voting

A different quality improvement strategy: instead of revising one output, generate N independent outputs and vote on the best.

When to use self-consistency:

Tasks with discrete outputs (classification, multiple-choice, code correctness)
Tasks where you want to confirm model confidence
Budget: N× cost and latency in exchange for more robust outputs

Implementation:

Run the generator N times (typically 3–5) with temperature > 0
Extract the answer from each output
Take the majority answer (most common)
Optionally: report confidence as (majority_count / N)

# 5 runs: [POSITIVE, POSITIVE, NEGATIVE, POSITIVE, POSITIVE]
# Majority: POSITIVE (4/5)
# Confidence: 0.80

You're using self-consistency for a classification task. You run the model 5 times and get: [POSITIVE, POSITIVE, NEGATIVE, POSITIVE, POSITIVE]. What's the output?

Self-consistency takes the majority vote: POSITIVE appears 4 times vs. NEGATIVE 1 time. The output is POSITIVE with 80% confidence. Requiring unanimity would make self-consistency impractical and would likely never terminate on nuanced tasks. Majority voting is a well-established statistical principle for combining noisy estimates.

Build

Build it

"""
lesson_04_06_reflection.py

Reflection loop: generator → critic → reviser, with max revision limit.
Self-consistency voter: N runs → majority vote.
"""

import os
import json
from collections import Counter
from dataclasses import dataclass, field
from typing import Optional
import anthropic

ANTHROPIC_KEY = os.getenv("ANTHROPIC_API_KEY")
client = anthropic.Anthropic(api_key=ANTHROPIC_KEY) if ANTHROPIC_KEY else None
MAX_REVISIONS = 3

@dataclass
class CritiqueIssue:
    dimension: str
    severity: str     # "high" | "medium" | "low"
    description: str
    fix: str

@dataclass
class CritiqueResult:
    overall_score: int     # 1–10
    issues: list[CritiqueIssue]
    passes: bool
    raw_critique: str = ""

@dataclass
class ReflectionTrace:
    task: str
    final_output: str
    revisions_used: int
    final_score: int
    passed: bool
    history: list[dict] = field(default_factory=list)

def _llm(prompt: str, system: str = "", max_tokens: int = 600) -> str:
    if not client:
        return f"[Mock output for: {prompt[:60]}...]"
    r = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=max_tokens,
        system=system or "You are a helpful assistant.",
        messages=[{"role": "user", "content": prompt}],
    )
    return r.content[0].text

class ReflectionLoop:
    """
    Generator → Critic → Reviser loop with max revision cap.
    Terminates when critic says passes=True or max_revisions is reached.
    """

    CRITIQUE_SYSTEM = (
        "You are an editorial critic. Evaluate the provided text on exactly these 4 dimensions: "
        "factual_accuracy, completeness, tone, clarity. "
        "Return a JSON object with this structure:\n"
        '{"overall_score": <1-10>, "issues": [{"dimension": "...", "severity": "high|medium|low", '
        '"description": "...", "fix": "..."}], "passes": <true if score >= 7 and no high-severity issues>}'
        "\nReturn ONLY the JSON object — no extra text."
    )

    def __init__(self, max_revisions: int = MAX_REVISIONS):
        self.max_revisions = max_revisions

    def generate(self, task: str) -> str:
        return _llm(
            f"Write a 150-word report on: {task}",
            system="You are a technical writer. Be clear, factual, and well-structured.",
        )

    def critique(self, output: str, task: str) -> CritiqueResult:
        raw = _llm(
            f"Task: {task}\n\nEvaluate this output:\n{output}",
            system=self.CRITIQUE_SYSTEM,
        )
        try:
            # Strip markdown code blocks if present
            clean = raw.strip()
            if clean.startswith("\"\"\""):
                clean = "\n".join(clean.split("\n")[1:-1])
            data = json.loads(clean)
            issues = [
                CritiqueIssue(**i) for i in data.get("issues", [])
            ]
            return CritiqueResult(
                overall_score=data.get("overall_score", 5),
                issues=issues,
                passes=data.get("passes", False),
                raw_critique=raw,
            )
        except (json.JSONDecodeError, TypeError):
            # Fallback: if JSON parsing fails, treat as passing to avoid infinite loop
            return CritiqueResult(
                overall_score=6, issues=[], passes=True, raw_critique=raw
            )

    def revise(self, original: str, critique: CritiqueResult, task: str) -> str:
        issue_instructions = "\n".join(
            f"- [{i.severity.upper()}] {i.dimension}: {i.description} → Fix: {i.fix}"
            for i in critique.issues
        )
        return _llm(
            f"Task: {task}\n\nOriginal output:\n{original}\n\n"
            f"Required fixes (address ALL of these):\n{issue_instructions}",
            system="You are a technical writer revising based on specific editorial feedback.",
            max_tokens=600,
        )

    def run(self, task: str) -> ReflectionTrace:
        output = self.generate(task)
        trace = ReflectionTrace(task=task, final_output=output, revisions_used=0,
                                final_score=0, passed=False)
        trace.history.append({"revision": 0, "output": output[:200], "critique": None})

        for rev in range(self.max_revisions):
            critique = self.critique(output, task)
            trace.history[-1]["critique"] = {
                "score": critique.overall_score,
                "passes": critique.passes,
                "issues": len(critique.issues),
            }
            print(
                f"  [reflection] revision={rev}  score={critique.overall_score}  "
                f"issues={len(critique.issues)}  passes={critique.passes}"
            )

            if critique.passes:
                trace.final_output = output
                trace.final_score = critique.overall_score
                trace.passed = True
                trace.revisions_used = rev
                return trace

            output = self.revise(output, critique, task)
            trace.revisions_used = rev + 1
            trace.history.append({"revision": rev + 1, "output": output[:200], "critique": None})

        # Max revisions reached — return best effort
        final_critique = self.critique(output, task)
        trace.final_output = output
        trace.final_score = final_critique.overall_score
        trace.passed = False
        trace.history[-1]["critique"] = {
            "score": final_critique.overall_score,
            "passes": final_critique.passes,
            "issues": len(final_critique.issues),
            "note": "max_revisions_reached",
        }
        return trace

class SelfConsistencyVoter:
    """
    Run a task N times, extract answers, take majority vote.
    Best for tasks with discrete outputs.
    """

    def __init__(self, n_samples: int = 5):
        self.n_samples = n_samples

    def run(self, task: str, answer_key: str = "classification") -> dict:
        answers = []
        for i in range(self.n_samples):
            response = _llm(
                task,
                system="Answer with a single label from the allowed options. No explanation.",
            )
            answer = response.strip().split()[0].upper()
            answers.append(answer)
            print(f"  [sc] sample {i+1}: {answer}")

        counts = Counter(answers)
        majority = counts.most_common(1)[0][0]
        confidence = counts[majority] / self.n_samples

        return {
            "majority_answer": majority,
            "confidence": confidence,
            "vote_distribution": dict(counts),
            "n_samples": self.n_samples,
        }

if __name__ == "__main__":
    print("=" * 60)
    print("DEMO 1: Reflection loop — report generation")
    print("=" * 60)
    loop = ReflectionLoop(max_revisions=3)
    trace = loop.run("The environmental impact of large language model training")
    print(f"\nFinal: score={trace.final_score}  passed={trace.passed}  "
          f"revisions={trace.revisions_used}")
    print(f"Output:\n{trace.final_output[:500]}...")

    print("\n\n" + "=" * 60)
    print("DEMO 2: Self-consistency voting — classification")
    print("=" * 60)
    voter = SelfConsistencyVoter(n_samples=5)
    result = voter.run(
        "Classify the sentiment: 'The model is fast but the accuracy is disappointing.' "
        "Options: POSITIVE, NEGATIVE, MIXED. Answer with one word."
    )
    print(f"\nMajority answer: {result['majority_answer']}")
    print(f"Confidence:      {result['confidence']:.0%}")
    print(f"Distribution:    {result['vote_distribution']}")

Unlock the full lesson

You've read the first 2 sections. The rest of this lesson covers Check Yourself — plus a hands-on lab, quiz, and project artifact.

Create a free account — the opening phases of 29 of 35 courses are free, no credit card.

Key terms in this lesson

Self-Consistency