5. Reflection and Self-Correction

Overview

In the previous capsule you decomposed complex tasks into manageable sub-tasks with dependency graphs. But there's a problem we didn't address: what happens when the agent finishes executing its plan and the result is bad? Without reflection, the agent delivers whatever it generated — with factual errors, incomplete information, internal contradictions, or answers that don't even address the original question. It's like a developer who pushes without a code review: sometimes it works, sometimes it burns production.

Reflection is the agent's ability to evaluate its own output before delivering it. The agent generates an answer, analyzes it with specific critique prompts, assigns a quality score, and makes a decision: if the quality is good enough, it delivers. If not, it corrects itself — re-researches, rewrites, adds sources, removes contradictions — and evaluates again. This generate → critique → improve loop is what separates a reactive agent from a deliberative one.

The catch is that superficial reflection is useless. A generic critique prompt like "is your answer good?" produces useless evaluations where the model says "yes, it's fine" 90% of the time. Effective reflection requires specific and verifiable questions: "Does the answer address the original question?", "Are there claims without a source?", "Is there a contradiction between paragraph 2 and paragraph 5?" — questions that force the model to inspect its output rigorously.

The Reflection Pattern

Generate → Evaluate → Decide

The pattern is conceptually simple:

┌──────────────────────────────────────────────────────────┐
│                                                          │
│   ┌──────────┐    ┌──────────┐    ┌──────────────────┐  │
│   │ Generate │───→│ Critique │───→│ Quality Gate     │  │
│   │ output   │    │ output   │    │ score >= thresh? │  │
│   └──────────┘    └──────────┘    └────────┬─────────┘  │
│        ▲                                   │             │
│        │              NO                   │             │
│        └───────────────────────────────────┘             │
│                                    │ YES                 │
│                                    ▼                     │
│                              ┌──────────┐                │
│                              │ Deliver  │                │
│                              └──────────┘                │
└──────────────────────────────────────────────────────────┘

Three phases:

1. Generate: The agent produces its answer (research, synthesis, code, whatever).
2. Critique: A second LLM call (or the same model with a different prompt) evaluates the output's quality against specific criteria.
3. Quality Gate: If the score passes the threshold, the output ships. If not, the agent gets the critique's feedback and regenerates.

Why it works

Reflection exploits a fundamental asymmetry of LLMs: it's easier to evaluate than to generate. Asking a model to write a perfect analysis from scratch is hard. Asking it to find errors in an existing analysis is significantly easier. The critique prompt turns "generate perfect content" into "find problems in this content" — a task where LLMs are more reliable.

Reflection isn't magic

There are real limits. If the model doesn't have the knowledge to generate a good answer, it won't generate a good critique either. Reflection improves the output 20-40% on tasks where the model "knows" the answer but didn't articulate it well. It doesn't save answers on topics the model knows nothing about.

Effective Critique Prompts

The mistake: a generic critique

This is the critique prompt that does not work:

BAD_CRITIQUE = """Evaluate this answer. Is it good? Does it have errors?

Answer: {output}"""

Why doesn't it work? Because "good" and "errors" are vague terms. The model will tend to answer "yes, it's fine" without actually inspecting the content. It's like asking a reviewer "is the PR okay?" — the answer is always "yes" unless something is obviously terrible.

Critique with specific questions

An effective critique breaks "quality" into verifiable dimensions:

SPECIFIC_CRITIQUE = """Evaluate this research answer against these SPECIFIC criteria.
For each criterion, answer YES or NO with evidence.

ANSWER TO EVALUATE:
{output}

ORIGINAL QUESTION:
{original_question}

CRITERIA:
1. COMPLETENESS: Does the answer address ALL aspects of the original question?
   - List which aspects were covered and which are missing.

2. ACCURACY: Are there factual claims with no source or evidence?
   - List every claim that looks factual but has no support.

3. COHERENCE: Are there internal contradictions between different parts of the answer?
   - If paragraph 2 says X and paragraph 5 says not-X, point it out.

4. RECENCY: Are the data and references from 2024-2025, or is some information outdated?

5. ACTIONABILITY: Does the answer give concrete recommendations or does it only describe the general landscape?

RESPONSE FORMAT:
SCORE: [1-10]
ISSUES: [list of problems found]
IMPROVEMENTS: [specific actions to improve]"""

Every criterion forces the model to run a concrete check. "List which aspects were covered and which are missing" can't be answered with "yes, it's fine" — it requires inspecting the content.

Critique by domain

The criteria change with the domain. You don't evaluate a research analysis the same way you evaluate a code snippet:

RESEARCH_CRITIQUE = """Evaluate this research answer:
1. Does it cite specific sources (papers, reports, articles)?
2. Does it distinguish between verifiable facts and opinions?
3. Does it cover multiple perspectives or only one?
4. Do the numbers and statistics have context (date, source, methodology)?"""

CODE_CRITIQUE = """Evaluate this generated code:
1. Does it have obvious syntax errors?
2. Does it handle edge cases (empty input, None, wrong types)?
3. Are the variable/function names descriptive?
4. Are there any missing imports?
5. Could it cause runtime errors (division by zero, key errors)?"""

SUMMARY_CRITIQUE = """Evaluate this summary:
1. Does it capture the original text's main points?
2. Does it introduce information that was NOT in the original?
3. Is it shorter than the original (at least 50% reduction)?
4. Does it preserve the author's tone and conclusions?"""

Scoring with a rubric

Instead of asking for a free-form score (where the model tends to give 7/10 to everything), define an explicit rubric:

from pydantic import BaseModel, Field
from typing import Literal

class CritiqueResult(BaseModel):
    completeness: int = Field(
        description="1-3: key aspects missing. 4-6: covers the main points. 7-10: complete."
    )
    accuracy: int = Field(
        description="1-3: serious errors. 4-6: minor inaccuracies. 7-10: accurate."
    )
    coherence: int = Field(
        description="1-3: contradictions. 4-6: uneven flow. 7-10: coherent."
    )
    overall_score: int = Field(description="Weighted average 1-10")
    issues: list[str] = Field(description="Specific problems found")
    improvements: list[str] = Field(description="Concrete actions to improve")
    verdict: Literal["deliver", "improve"] = Field(
        description="deliver if overall_score >= 7, improve if < 7"
    )

With structured output, the model can't get away with "it's fine". It has to assign numbers to each dimension and list concrete problems.

Implementing the critiquer

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")
critique_model = model.with_structured_output(CritiqueResult)

CRITIQUE_PROMPT = """You are a rigorous evaluator. Evaluate this answer with the provided rubric.
Be STRICT: only give high scores if it truly deserves them.

ORIGINAL QUESTION: {question}
ANSWER TO EVALUATE: {output}

Criteria:
- completeness: Does it cover every aspect of the question?
- accuracy: Are the claims verifiable and correct?
- coherence: Is it internally consistent and well structured?
- overall_score: weighted average (accuracy counts double)
- verdict: "deliver" if overall >= 7, "improve" if < 7"""

def critique_output(question: str, output: str) -> CritiqueResult:
    """Evaluate an output with a structured critique."""
    result = critique_model.invoke(
        CRITIQUE_PROMPT.format(question=question, output=output)
    )
    return result

Implementation with LangGraph

State for reflection

You need extra fields in the state to track the reflection cycle:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage
from pydantic import BaseModel, Field

model = init_chat_model("openai:gpt-4.1-mini")

class ReflectionState(TypedDict):
    messages: Annotated[list, add_messages]
    question: str
    draft: str
    critique: dict
    quality_score: int
    reflection_count: int
    max_reflections: int
    reflection_history: list[dict]

The key fields:

• draft: the agent's current output (updated on each iteration)
• critique: the result of the last critique (issues, improvements, score)
• quality_score: the numeric score from the last critique
• reflection_count: how many times it has reflected (for guardrails)
• max_reflections: the iteration limit (default 3)
• reflection_history: the history of critiques for tracking

The generation node

def generate_node(state: ReflectionState) -> dict:
    """Generate or regenerate the draft based on previous feedback."""
    question = state["question"]
    
    if state.get("reflection_count", 0) == 0:
        prompt = (
            f"Answer this question completely and with good structure.\n\n"
            f"Question: {question}"
        )
    else:
        critique = state.get("critique", {})
        issues = critique.get("issues", [])
        improvements = critique.get("improvements", [])
        
        prompt = (
            f"Your previous answer had these problems:\n"
            f"- Issues: {', '.join(issues)}\n"
            f"- Suggested improvements: {', '.join(improvements)}\n\n"
            f"PREVIOUS DRAFT:\n{state['draft']}\n\n"
            f"ORIGINAL QUESTION: {question}\n\n"
            f"Generate an improved version that fixes ALL the listed problems."
        )
    
    response = model.invoke([HumanMessage(content=prompt)])
    
    return {
        "draft": response.content,
        "messages": [SystemMessage(
            content=f"Draft generated (iteration {state.get('reflection_count', 0) + 1})"
        )],
    }

The node behaves differently on the first iteration (generating from scratch) vs the ones after (regenerating with the critique's feedback). The previous draft and the specific issues go in the prompt so the model knows exactly what to fix.

The reflection node

class CritiqueOutput(BaseModel):
    completeness: int = Field(description="1-10: coverage of the question")
    accuracy: int = Field(description="1-10: factual accuracy")
    coherence: int = Field(description="1-10: internal consistency")
    overall_score: int = Field(description="Weighted average 1-10")
    issues: list[str] = Field(description="Problems found")
    improvements: list[str] = Field(description="Actions to improve")
    verdict: Literal["deliver", "improve"] = Field(
        description="deliver if >= 7, improve if < 7"
    )

critique_model = model.with_structured_output(CritiqueOutput)

REFLECTION_PROMPT = """Evaluate this answer rigorously. Be STRICT.

ORIGINAL QUESTION: {question}

ANSWER:
{draft}

Evaluate:
1. completeness (1-10): Does it address EVERY aspect of the question?
2. accuracy (1-10): Are the claims precise and verifiable?
3. coherence (1-10): Are there contradictions or logical jumps?
4. overall_score: average (accuracy counts double)
5. issues: a SPECIFIC list of problems (not generic ones)
6. improvements: CONCRETE actions (not "improve the answer")
7. verdict: "deliver" if overall >= 7, "improve" if < 7"""

def reflect_node(state: ReflectionState) -> dict:
    """Evaluate the current draft with a structured critique."""
    result = critique_model.invoke(
        REFLECTION_PROMPT.format(
            question=state["question"],
            draft=state["draft"],
        )
    )
    
    history = state.get("reflection_history", []).copy()
    history.append({
        "iteration": state.get("reflection_count", 0) + 1,
        "score": result.overall_score,
        "issues": result.issues,
    })
    
    return {
        "critique": result.model_dump(),
        "quality_score": result.overall_score,
        "reflection_count": state.get("reflection_count", 0) + 1,
        "reflection_history": history,
        "messages": [SystemMessage(
            content=f"Reflection #{state.get('reflection_count', 0) + 1}: "
                    f"score={result.overall_score}/10, "
                    f"issues={len(result.issues)}, "
                    f"verdict={result.verdict}"
        )],
    }

The quality gate (conditional edge)

def quality_gate(state: ReflectionState) -> str:
    """Decide whether to deliver or improve."""
    if state.get("reflection_count", 0) >= state.get("max_reflections", 3):
        return "deliver"
    
    critique = state.get("critique", {})
    if critique.get("verdict") == "deliver":
        return "deliver"
    
    return "improve"

Wiring the graph

graph = StateGraph(ReflectionState)
graph.add_node("generate", generate_node)
graph.add_node("reflect", reflect_node)

graph.add_edge(START, "generate")
graph.add_edge("generate", "reflect")

graph.add_conditional_edges("reflect", quality_gate, {
    "deliver": END,
    "improve": "generate",
})

agent = graph.compile()

The flow: START → generate → reflect → (deliver → END | improve → generate → reflect → ...). The loop repeats until the quality gate says "deliver" or it hits max_reflections.

Execution

result = agent.invoke({
    "messages": [],
    "question": "What are the 3 main frameworks for building AI agents in 2025 and when should you use each one?",
    "draft": "",
    "critique": {},
    "quality_score": 0,
    "reflection_count": 0,
    "max_reflections": 3,
    "reflection_history": [],
})

print("=== Final Draft ===")
print(result["draft"][:500])
print(f"\n=== Reflection Stats ===")
print(f"Iterations: {result['reflection_count']}")
print(f"Final score: {result['quality_score']}/10")
for r in result["reflection_history"]:
    print(f"  Iter {r['iteration']}: score={r['score']}, issues={len(r['issues'])}")

Visualization

from IPython.display import Image, display
display(Image(agent.get_graph().draw_mermaid_png()))

The diagram shows: START → generate → reflect → (END | generate). The edge back to generate is the self-correction loop.

Self-Correction: From Critique to Action

The problem: feedback with no action

A critique that says "sources are missing" is useless if the agent can't search for sources. Self-correction isn't just identifying problems — it's acting on them. The type of action depends on the type of problem.

Mapping issues to actions

CORRECTION_STRATEGIES = {
    "missing_sources": {
        "action": "re_research",
        "description": "Find additional sources for unsupported claims",
        "requires_tools": True,
    },
    "incomplete_coverage": {
        "action": "expand",
        "description": "Research the missing aspects of the question",
        "requires_tools": True,
    },
    "contradictions": {
        "action": "rewrite",
        "description": "Rewrite contradictory sections while keeping coherence",
        "requires_tools": False,
    },
    "outdated_info": {
        "action": "re_research",
        "description": "Find up-to-date data to replace stale info",
        "requires_tools": True,
    },
    "poor_structure": {
        "action": "rewrite",
        "description": "Reorganize the answer with a better structure",
        "requires_tools": False,
    },
    "vague_recommendations": {
        "action": "specify",
        "description": "Turn generic recommendations into concrete actions",
        "requires_tools": False,
    },
}

There are two categories of correction:

1. Re-research: The agent needs to find more information (sources, updated data, missing aspects). It requires tools.
2. Rewrite: The agent has the information but articulated it poorly (contradictions, poor structure, vagueness). It only needs to regenerate.

A smart correction node

Instead of always regenerating from scratch, a correction node can decide which type of action to take:

class CorrectionPlan(BaseModel):
    strategy: Literal["rewrite", "re_research", "expand"] = Field(
        description="Type of correction needed"
    )
    specific_fixes: list[str] = Field(
        description="What exactly is going to be fixed"
    )
    sections_to_keep: list[str] = Field(
        description="Parts of the draft that are fine and shouldn't change"
    )

correction_model = model.with_structured_output(CorrectionPlan)

def plan_correction(critique: dict, draft: str) -> CorrectionPlan:
    """Decide which type of correction to apply."""
    prompt = (
        f"Given this critique of a draft, decide on a correction strategy.\n\n"
        f"ISSUES: {critique['issues']}\n"
        f"IMPROVEMENTS: {critique['improvements']}\n\n"
        f"DRAFT:\n{draft[:1000]}\n\n"
        f"Decide:\n"
        f"- 'rewrite' if the problems are structural/editorial\n"
        f"- 're_research' if data or sources are missing\n"
        f"- 'expand' if the answer is incomplete\n"
        f"Identify which parts of the draft are FINE and shouldn't change."
    )
    return correction_model.invoke(prompt)

This prevents the "throw it all out and start over" problem. If 80% of the draft is fine and only a source is missing, the agent finds the source and patches it — it doesn't rewrite everything.

Correction-aware generate

def generate_with_correction(state: ReflectionState) -> dict:
    """Generate or correct based on the correction plan."""
    if state.get("reflection_count", 0) == 0:
        prompt = f"Answer this completely:\n\n{state['question']}"
    else:
        critique = state.get("critique", {})
        plan = plan_correction(critique, state["draft"])
        
        if plan.strategy == "rewrite":
            prompt = (
                f"Rewrite this answer, fixing ONLY these problems:\n"
                f"{plan.specific_fixes}\n\n"
                f"KEEP these parts that are fine: {plan.sections_to_keep}\n\n"
                f"CURRENT DRAFT:\n{state['draft']}\n\n"
                f"QUESTION: {state['question']}"
            )
        elif plan.strategy == "re_research":
            prompt = (
                f"The answer needs more data. Look up information for:\n"
                f"{plan.specific_fixes}\n\n"
                f"Integrate the new findings into the existing draft.\n\n"
                f"CURRENT DRAFT:\n{state['draft']}\n\n"
                f"QUESTION: {state['question']}"
            )
        else:  # expand
            prompt = (
                f"The answer is incomplete. Expand it to cover:\n"
                f"{plan.specific_fixes}\n\n"
                f"CURRENT DRAFT:\n{state['draft']}\n\n"
                f"QUESTION: {state['question']}"
            )
    
    response = model.invoke([HumanMessage(content=prompt)])
    return {"draft": response.content}

Guardrails for Reflection

Guardrail 1: Max iterations

The most basic and most important guardrail. Without a limit, an agent can enter an infinite loop where the critique always finds something to improve (because perfection doesn't exist):

MAX_REFLECTIONS = 3

def quality_gate(state: ReflectionState) -> str:
    if state.get("reflection_count", 0) >= state.get("max_reflections", MAX_REFLECTIONS):
        return "deliver"
    
    if state.get("critique", {}).get("verdict") == "deliver":
        return "deliver"
    
    return "improve"

Why 3? It's a pragmatic number based on diminishing returns. The first reflection catches coarse errors (completeness, contradictions). The second catches fine errors (sources, precision). The third rarely improves things significantly. After the third, the extra cost (tokens, latency) almost never justifies the marginal gain.

Guardrail 2: Diminishing returns detection

If the score doesn't improve between iterations, stop — you're burning tokens with no progress:

def check_diminishing_returns(history: list[dict], min_improvement: int = 1) -> bool:
    """Return True if it should stop for diminishing returns."""
    if len(history) < 2:
        return False
    
    last_score = history[-1]["score"]
    prev_score = history[-2]["score"]
    
    improvement = last_score - prev_score
    return improvement < min_improvement

def quality_gate_with_diminishing(state: ReflectionState) -> str:
    """Quality gate with diminishing returns detection."""
    if state.get("reflection_count", 0) >= state.get("max_reflections", 3):
        return "deliver"
    
    if state.get("critique", {}).get("verdict") == "deliver":
        return "deliver"
    
    history = state.get("reflection_history", [])
    if check_diminishing_returns(history):
        return "deliver"
    
    return "improve"

If iteration 1 gave score 5, iteration 2 gave score 7, and iteration 3 gave score 7 — the improvement between 2 and 3 is 0. Stop. It won't improve on iteration 4.

Guardrail 3: Cost tracking

Every reflection costs tokens. In production, you need to know how much you're spending:

from langchain_core.callbacks import BaseCallbackHandler

class CostTracker(BaseCallbackHandler):
    def __init__(self):
        self.total_tokens = 0
        self.total_calls = 0
    
    def on_llm_end(self, response, **kwargs):
        usage = response.llm_output.get("token_usage", {})
        self.total_tokens += usage.get("total_tokens", 0)
        self.total_calls += 1

def estimate_reflection_cost(
    base_tokens: int,
    num_reflections: int,
    cost_per_1k_tokens: float = 0.002,
) -> dict:
    """Estimate the cost of N reflections."""
    generate_tokens = base_tokens
    critique_tokens = base_tokens * 0.3
    
    total_tokens = generate_tokens + num_reflections * (critique_tokens + generate_tokens)
    total_cost = (total_tokens / 1000) * cost_per_1k_tokens
    
    return {
        "total_tokens": int(total_tokens),
        "total_cost_usd": round(total_cost, 4),
        "calls": 1 + (num_reflections * 2),
        "overhead_vs_no_reflection": f"{num_reflections * 2}x more calls",
    }

print(estimate_reflection_cost(base_tokens=2000, num_reflections=3))

Guardrail 4: Score regression prevention

Sometimes an "improvement" makes the draft worse. If the score drops, revert to the previous draft:

def generate_with_rollback(state: ReflectionState) -> dict:
    """Generate an improvement, but revert if the score drops."""
    prev_draft = state["draft"]
    prev_score = state.get("quality_score", 0)
    
    new_draft = generate_improved_draft(state)
    
    new_critique = critique_model.invoke(
        REFLECTION_PROMPT.format(
            question=state["question"],
            draft=new_draft,
        )
    )
    
    if new_critique.overall_score < prev_score:
        return {
            "draft": prev_draft,
            "messages": [SystemMessage(
                content=f"Rollback: new score ({new_critique.overall_score}) < "
                        f"prev score ({prev_score}). Keeping the previous draft."
            )],
        }
    
    return {"draft": new_draft}

When to Use Reflection

Not every task justifies reflection's overhead. Use this table as a guide:

Rule of thumb: If the cost of an error in the output is greater than the cost of 2-3 extra LLM calls, use reflection. If speed matters more than perfection, don't.

Quantified trade-offs

Without reflection:
  1 LLM call  │  ~2s latency  │  ~2000 tokens  │  ~$0.004
  Quality: baseline

With 1 reflection:
  3 LLM calls │  ~6s latency  │  ~5000 tokens  │  ~$0.010
  Quality: +20-30% in completeness and coherence

With 3 reflections:
  7 LLM calls │  ~14s latency │  ~12000 tokens │  ~$0.024
  Quality: +30-40% vs baseline (diminishing returns after 2)

Connection to the Project

In this module's project (capsule 08, Research Agent with Planning and Reflection):

• The reflection step gets added after the project's synthesize node. The agent generates its research report, evaluates it with a research-specific critique (did it answer the question? are the sources trustworthy? are there gaps?), and decides whether to re-iterate or deliver.
• The quality gate is a conditional edge in the StateGraph: reflect → (improve → generate → reflect | deliver → END). The state includes quality_score, reflection_count, and reflection_history.
• The guardrails (max 3 reflections, diminishing returns detection) prevent infinite loops. The reflection_history lets the supervisor (M8) see how many iterations the agent needed — a confidence indicator.

In later modules:

• M6 (Memory): The reflection history gets saved. If the agent learns that certain questions always require re-research (a low score on the first iteration), it can jump straight to deeper research next time.
• M8 (Multi-Agent): In multi-agent systems, one agent can critique another agent's output — cross-reflection. The researcher agent generates, the editor agent critiques, the researcher corrects. More effective than self-reflection because it removes the bias of evaluating your own work.
• M9 (Testing): Critique prompts are testable: given an output with known errors, does the critique catch them? Golden datasets of outputs with labeled issues validate that the reflection works.

Troubleshooting

Problem 1: The critique always gives a high score (> 8) regardless of quality

Symptom: The model gives 9/10 to obviously incomplete answers. The quality gate always says "deliver" on the first iteration.

Cause: The critique prompt is too soft, or the model has a bias toward positive evaluations (self-serving bias — it tends to rate what it generated itself favorably).

Solution: Use a different model for the critique (the one that generates isn't the one that evaluates). If you can't, add calibration instructions:

STRICT_CRITIQUE = """You are a STRICT evaluator. Your job is to find problems.
A score of 8+ means the answer is publishable with no editing.
Most answers deserve 5-7.
If you don't find at least 2 problems, you're being too generous.

{normal_critique_prompt}"""

Problem 2: The reflection loop doesn't converge (the score oscillates without improving)

Symptom: Iteration 1: score 5. Iteration 2: score 7. Iteration 3: score 5. Iteration 4: score 6. The score doesn't converge.

Cause: The agent "fixes" one problem but introduces another. Typical when the rewrite is from scratch instead of incremental.

Solution: Implement incremental correction (patch, don't rewrite) and diminishing returns detection:

if len(history) >= 3:
    scores = [h["score"] for h in history[-3:]]
    if max(scores) - min(scores) <= 1:
        return "deliver"

Problem 3: The correction loses information from the original draft

Symptom: The original draft had 5 good points and 2 bad ones. After the correction, it has 3 good points (different ones) and 0 bad ones. It lost information.

Cause: The regeneration prompt doesn't tell the model to preserve what's good.

Solution: Explicitly include the sections to keep in the correction prompt (see CorrectionPlan.sections_to_keep in the previous section). Alternatively, use append instead of replace:

INCREMENTAL_PROMPT = """Do NOT rewrite the whole answer.
ONLY modify the parts with problems. Keep everything else EXACTLY the same.

Problems to fix:
{issues}

Current draft (keep everything except the parts with problems):
{draft}"""

Problem 4: Reflection on simple tasks burns tokens with no benefit

Symptom: Simple questions like "What is Python?" go through 3 reflection iterations, spending 12,000 tokens to improve an answer that was already good.

Solution: Add a gate up front that decides whether the task deserves reflection:

class ReflectionDecision(BaseModel):
    needs_reflection: bool = Field(
        description="True if the answer could have significant errors"
    )
    reason: str

def should_reflect(question: str, draft: str) -> bool:
    """Decide whether the draft needs reflection."""
    decision_model = model.with_structured_output(ReflectionDecision)
    result = decision_model.invoke(
        f"Could this answer have significant errors that "
        f"justify a detailed evaluation?\n\n"
        f"Question: {question}\n"
        f"Answer: {draft[:500]}"
    )
    return result.needs_reflection

Problem 5: The critique's structured output fails (parsing errors)

Symptom: ValidationError when parsing CritiqueOutput. The model doesn't follow the schema.

Cause: The critique prompt is too complex for with_structured_output with certain models.

Solution: Simplify the schema or use a fallback:

def safe_critique(question: str, draft: str) -> dict:
    """Critique with a fallback to a simple format."""
    try:
        return critique_model.invoke(
            REFLECTION_PROMPT.format(question=question, draft=draft)
        ).model_dump()
    except Exception:
        response = model.invoke([HumanMessage(
            content=f"Rate this 1-10 and list 2 problems:\n\n{draft[:1000]}"
        )])
        return {
            "overall_score": 5,
            "issues": [response.content],
            "improvements": ["Re-evaluate manually"],
            "verdict": "improve",
        }

Exercises

Exercise 1: Critique prompt by domain (Easy)

Write a critique prompt specifically for evaluating a professional email generated by an agent. Define at least 4 verifiable criteria (not generic ones).

EMAIL_CRITIQUE = """Evaluate this professional email against specific criteria.

EMAIL:
{email}

CONTEXT: {context}

CRITERIA (answer YES/NO with evidence for each):

1. TONE: Is it appropriate for the recipient? (formal for executives, 
   semi-formal for colleagues, etc.)
   
2. CLARITY: Is the email's purpose clear in the first 2 sentences?
   Does the recipient know what's expected of them?

3. LENGTH: Is it concise? An ideal professional email is 5-15 sentences.
   Are there paragraphs that could be cut without losing information?

4. CALL TO ACTION: Does it end with a clear action? ("Can we schedule a 
   call on Thursday?", not "Looking forward to your reply")

5. ERRORS: Are there grammatical errors, wrong names, or inconsistent 
   data?

SCORE: [1-10]
VERDICT: deliver if >= 7, improve if < 7"""

Exercise 2: Implement diminishing returns detection (Medium)

Implement a function that takes the reflection_history and decides whether another iteration is worth it. Criteria: (a) if the score didn't improve by >= 1 point in the last iteration, stop; (b) if the score has oscillated (up/down/up) over the last 3 iterations, stop.

history = [
    {"iteration": 1, "score": 4, "issues": ["incomplete", "no sources"]},
    {"iteration": 2, "score": 7, "issues": ["no sources"]},
    {"iteration": 3, "score": 7, "issues": ["minor formatting"]},
]
# Should it continue?

def should_continue_reflecting(history: list[dict], min_improvement: int = 1) -> bool:
    """Return True if another iteration is worth it."""
    if len(history) < 2:
        return True
    
    last = history[-1]["score"]
    prev = history[-2]["score"]
    if last - prev < min_improvement:
        return False
    
    if len(history) >= 3:
        scores = [h["score"] for h in history[-3:]]
        diffs = [scores[i+1] - scores[i] for i in range(len(scores)-1)]
        is_oscillating = any(d > 0 for d in diffs) and any(d < 0 for d in diffs)
        if is_oscillating:
            return False
    
    return True

# Test with the example history:
result = should_continue_reflecting(history)
print(result)  # False — the improvement between iter 2 (score 7) and iter 3 (score 7) is 0

# Test with a history that improves:
improving = [
    {"iteration": 1, "score": 3, "issues": []},
    {"iteration": 2, "score": 6, "issues": []},
]
print(should_continue_reflecting(improving))  # True — improved by 3 points

# Test with oscillation:
oscillating = [
    {"iteration": 1, "score": 5, "issues": []},
    {"iteration": 2, "score": 7, "issues": []},
    {"iteration": 3, "score": 6, "issues": []},
]
print(should_continue_reflecting(oscillating))  # False — it oscillates (5→7→6)

Exercise 3: A complete reflection graph with LangGraph (Medium)

Implement a StateGraph with three nodes (generate, reflect, and a final deliver node) and a conditional edge that decides between improving and delivering. Include max_reflections = 3 as a guardrail. Test it with a research question.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage
from pydantic import BaseModel, Field

model = init_chat_model("openai:gpt-4.1-mini")

class CritiqueSchema(BaseModel):
    score: int = Field(description="1-10")
    issues: list[str] = Field(description="Problems found")
    improvements: list[str] = Field(description="Suggested improvements")
    verdict: Literal["deliver", "improve"] = Field(
        description="deliver if score >= 7"
    )

critique_llm = model.with_structured_output(CritiqueSchema)

class RState(TypedDict):
    messages: Annotated[list, add_messages]
    question: str
    draft: str
    score: int
    issues: list[str]
    count: int
    max_count: int

def generate(state: RState) -> dict:
    if state.get("count", 0) == 0:
        prompt = f"Answer this completely:\n{state['question']}"
    else:
        prompt = (
            f"Improve this answer. Problems: {state['issues']}\n\n"
            f"Draft: {state['draft']}\n\nQuestion: {state['question']}"
        )
    resp = model.invoke([HumanMessage(content=prompt)])
    return {"draft": resp.content}

def reflect(state: RState) -> dict:
    result = critique_llm.invoke(
        f"Evaluate strictly (score >= 7 = deliver):\n\n"
        f"Question: {state['question']}\n\n"
        f"Answer:\n{state['draft']}"
    )
    return {
        "score": result.score,
        "issues": result.issues,
        "count": state.get("count", 0) + 1,
        "messages": [SystemMessage(
            content=f"Reflection #{state.get('count', 0) + 1}: "
                    f"score={result.score}, verdict={result.verdict}"
        )],
    }

def route(state: RState) -> str:
    if state.get("count", 0) >= state.get("max_count", 3):
        return "end"
    if state.get("score", 0) >= 7:
        return "end"
    return "improve"

g = StateGraph(RState)
g.add_node("generate", generate)
g.add_node("reflect", reflect)

g.add_edge(START, "generate")
g.add_edge("generate", "reflect")
g.add_conditional_edges("reflect", route, {
    "end": END,
    "improve": "generate",
})

agent = g.compile()

result = agent.invoke({
    "messages": [],
    "question": "What are the key differences between LangGraph and CrewAI for multi-agent systems?",
    "draft": "",
    "score": 0,
    "issues": [],
    "count": 0,
    "max_count": 3,
})

print(f"Final score: {result['score']}/10")
print(f"Iterations: {result['count']}")
print(f"Final draft:\n{result['draft'][:400]}")

Exercise 4: Cross-model reflection (Hard)

Implement a system where one model generates and a different model does the critique. Use gpt-4.1-mini to generate and gpt-4.1-nano to critique. Compare the scores with self-reflection (the same model for both). Is cross-model stricter?

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from pydantic import BaseModel, Field
from typing import Literal

generator = init_chat_model("openai:gpt-4.1-mini")
critic = init_chat_model("openai:gpt-4.1-nano")

class Critique(BaseModel):
    score: int = Field(description="1-10")
    issues: list[str]
    verdict: Literal["deliver", "improve"]

critique_self = generator.with_structured_output(Critique)
critique_cross = critic.with_structured_output(Critique)

CRITIQUE_PROMPT = """Evaluate this answer strictly (1-10).
Score 8+ = excellent. Most deserve 5-7.

Question: {q}
Answer: {a}"""

def compare_reflection(question: str) -> dict:
    """Compare self-reflection vs cross-model reflection."""
    draft = generator.invoke(
        [HumanMessage(content=f"Answer this:\n{question}")]
    ).content
    
    self_result = critique_self.invoke(
        CRITIQUE_PROMPT.format(q=question, a=draft)
    )
    
    cross_result = critique_cross.invoke(
        CRITIQUE_PROMPT.format(q=question, a=draft)
    )
    
    return {
        "draft_preview": draft[:200],
        "self_reflection": {
            "score": self_result.score,
            "issues": self_result.issues,
            "verdict": self_result.verdict,
        },
        "cross_reflection": {
            "score": cross_result.score,
            "issues": cross_result.issues,
            "verdict": cross_result.verdict,
        },
        "cross_is_stricter": cross_result.score < self_result.score,
    }

result = compare_reflection(
    "Explain the advantages and disadvantages of the Reflection pattern in AI agents"
)

print(f"Self-reflection score: {result['self_reflection']['score']}")
print(f"Cross-reflection score: {result['cross_reflection']['score']}")
print(f"Cross is stricter: {result['cross_is_stricter']}")
print(f"\nSelf issues: {result['self_reflection']['issues']}")
print(f"Cross issues: {result['cross_reflection']['issues']}")

Exercise 5: Reflection with automatic rollback (Hard)

Implement a reflection system where, if the "improvement" produces a lower score than the previous draft, it automatically rolls back to the previous draft and delivers. Include cost tracking (token count) for each iteration.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage
from pydantic import BaseModel, Field

model = init_chat_model("openai:gpt-4.1-mini")

class CritiqueOut(BaseModel):
    score: int = Field(description="1-10")
    issues: list[str]
    verdict: Literal["deliver", "improve"]

critique_model = model.with_structured_output(CritiqueOut)

class RBState(TypedDict):
    messages: Annotated[list, add_messages]
    question: str
    draft: str
    best_draft: str
    best_score: int
    current_score: int
    issues: list[str]
    count: int
    max_count: int
    history: list[dict]
    rolled_back: bool

def generate(state: RBState) -> dict:
    if state.get("count", 0) == 0:
        prompt = f"Answer this completely:\n{state['question']}"
    else:
        prompt = (
            f"Fix these problems:\n{state['issues']}\n\n"
            f"Current draft:\n{state['draft']}\n\n"
            f"Question: {state['question']}"
        )
    resp = model.invoke([HumanMessage(content=prompt)])
    return {"draft": resp.content}

def reflect(state: RBState) -> dict:
    result = critique_model.invoke(
        f"Evaluate this (1-10, strict):\n\nQuestion: {state['question']}\n\n"
        f"Answer:\n{state['draft']}"
    )
    
    best_draft = state.get("best_draft", state["draft"])
    best_score = state.get("best_score", 0)
    rolled_back = False
    
    if result.score > best_score:
        best_draft = state["draft"]
        best_score = result.score
    elif result.score < state.get("current_score", 0):
        rolled_back = True
    
    history = state.get("history", []).copy()
    history.append({
        "iteration": state.get("count", 0) + 1,
        "score": result.score,
        "issues_count": len(result.issues),
        "rolled_back": rolled_back,
    })
    
    return {
        "current_score": result.score,
        "best_draft": best_draft,
        "best_score": best_score,
        "issues": result.issues,
        "count": state.get("count", 0) + 1,
        "history": history,
        "rolled_back": rolled_back,
        "messages": [SystemMessage(
            content=f"Iter {state.get('count', 0) + 1}: score={result.score}, "
                    f"best={best_score}"
                    f"{' (ROLLBACK)' if rolled_back else ''}"
        )],
    }

def finalize(state: RBState) -> dict:
    """Deliver the best draft (not necessarily the last one)."""
    return {
        "draft": state.get("best_draft", state["draft"]),
        "messages": [SystemMessage(
            content=f"Delivering the best draft (score={state['best_score']}). "
                    f"Iterations: {state['count']}. "
                    f"Rollbacks: {sum(1 for h in state.get('history', []) if h.get('rolled_back'))}"
        )],
    }

def route(state: RBState) -> str:
    if state.get("count", 0) >= state.get("max_count", 3):
        return "finalize"
    if state.get("best_score", 0) >= 8:
        return "finalize"
    if state.get("rolled_back", False):
        return "finalize"
    return "improve"

g = StateGraph(RBState)
g.add_node("generate", generate)
g.add_node("reflect", reflect)
g.add_node("finalize", finalize)

g.add_edge(START, "generate")
g.add_edge("generate", "reflect")
g.add_conditional_edges("reflect", route, {
    "finalize": "finalize",
    "improve": "generate",
})
g.add_edge("finalize", END)

agent = g.compile()

result = agent.invoke({
    "messages": [],
    "question": "Compare Redis vs Memcached for caching in Python applications",
    "draft": "", "best_draft": "", "best_score": 0,
    "current_score": 0, "issues": [],
    "count": 0, "max_count": 3,
    "history": [], "rolled_back": False,
})

print(f"Final score: {result['best_score']}/10")
print(f"Iterations: {result['count']}")
print(f"Rollbacks: {sum(1 for h in result['history'] if h.get('rolled_back'))}")
print(f"\nHistory:")
for h in result["history"]:
    rb = " ← ROLLBACK" if h.get("rolled_back") else ""
    print(f"  Iter {h['iteration']}: score={h['score']}, issues={h['issues_count']}{rb}")

Summary

In this capsule you learned:

• Reflection is evaluating before delivering. The agent generates output, critiques it with specific prompts, and decides whether to improve or deliver. It's the difference between an agent that ships its first draft and one that does a self-review.
• Generic critique prompts don't work. "Is it fine?" produces useless evaluations. An effective critique uses specific, verifiable questions: "Are there claims with no source?", "Does it address every aspect of the question?", scoring with rubrics, and structured output.
• Self-correction goes beyond the critique. Identifying problems is only half of it. The other half is acting: re-research for missing sources, rewrite for contradictions, expand for incomplete coverage. The type of correction depends on the type of problem.
• Guardrails are mandatory. Max iterations (3), diminishing returns detection (the score stops improving), cost tracking (tokens per iteration), and score regression prevention (rollback if it gets worse). Without guardrails, reflection can enter infinite loops or burn tokens with no benefit.
• Not every task needs reflection. Short answers, conversational chat, FAQ questions — the overhead isn't justified. Reflection shines in research reports, complex code generation, multi-source synthesis, and sensitive domains where the cost of an error is high.
• Implementation in LangGraph. Two nodes (generate, reflect) with a conditional edge (the quality gate). The loop is controlled with reflection_count and max_reflections. The reflection_history enables diagnosis and optimization.

Next capsule: Reasoning Traces and Explainability — making the agent's thinking visible for debugging, observability, and user trust. If reflection is "the agent evaluates itself", reasoning traces is "the agent shows how it thinks."

Additional Resources

1. Reflexion Paper — Self-reflection agents: verbal reinforcement learning so agents improve iteratively
2. Self-Refine Paper — Iterative refinement with self-feedback: generate, critique, refine with no human supervision
3. Constitutional AI — Anthropic's principles of self-evaluation and correction
4. LangGraph Reflection Tutorial — The official reflection implementation in LangGraph
5. CRITIC Paper — LLMs that validate and correct their own output with external tools