LocalGraph

Appearance

Tutorial 05: Reflection ​

This tutorial teaches how to build self-critiquing agents that iteratively improve their outputs through reflection loops.

What You'll Learn ​

  • Reflection loops: Generate → Critique → Revise patterns
  • Self-improvement: Using LLMs to evaluate their own outputs
  • Iteration control: When to stop refining
  • Quality enhancement: Producing better outputs through feedback

Prerequisites ​

What is Reflection? ​

Reflection is a pattern where an LLM:

  1. Generates an initial output
  2. Critiques its own work
  3. Revises based on the critique
  4. Repeats until satisfied

This mirrors how humans improve their work through drafts and revisions.

Why Use Reflection? ​

Single-shot LLM outputs are often good but not great. Through reflection:

  • Errors get caught and corrected
  • Missing information gets added
  • Clarity improves with each revision
  • Quality approaches human-level editing

Use Cases ​

  • Writing: Essays, reports, documentation
  • Code generation: Write, review, refactor
  • Analysis: Initial assessment → deeper analysis → conclusions
  • Problem-solving: Solution → evaluation → refinement

Core Concepts ​

1. The Reflection Loop ​

The graph cycles between generation and critique until approved or max iterations reached.

2. State for Reflection ​

We track more than just messages:

python
class ReflectionState(TypedDict):
    messages: Annotated[list, add_messages]  # History
    task: str           # Original task
    draft: str          # Current draft
    critique: str       # Latest critique
    iteration: int      # Current iteration

3. Stopping Conditions ​

Two common ways to exit the loop:

  1. Approval signal: Critique says "APPROVED" or "No changes needed"
  2. Max iterations: Prevent infinite loops (e.g., max 3 iterations)
python
def should_continue(state):
    if "APPROVED" in state["critique"].upper():
        return "end"
    if state["iteration"] >= MAX_ITERATIONS:
        return "end"
    return "generate"

Building a Reflection Agent ​

Step 1: Define State ​

python
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages

class ReflectionState(TypedDict):
    messages: Annotated[list, add_messages]
    task: str
    draft: str
    critique: str
    iteration: int

Step 2: Create Generator Node ​

python
from langchain_core.messages import HumanMessage, SystemMessage

GENERATOR_PROMPT = """You are a skilled writer.
If this is the first draft, write a complete response.
If you have critique, revise your draft to address the feedback."""

def generate_node(state: ReflectionState) -> dict:
    iteration = state.get("iteration", 0)

    if iteration == 0:
        prompt = f"Write a response: {state['task']}"
    else:
        prompt = f"Revise based on critique:\nDraft: {state['draft']}\nCritique: {state['critique']}"

    response = llm.invoke([
        SystemMessage(content=GENERATOR_PROMPT),
        HumanMessage(content=prompt)
    ])

    return {
        "draft": response.content,
        "iteration": iteration + 1
    }

Step 3: Create Critique Node ​

python
CRITIQUE_PROMPT = """You are a thoughtful editor.
If the draft is excellent, respond with exactly: "APPROVED"
Otherwise, provide specific feedback for improvement."""

def critique_node(state: ReflectionState) -> dict:
    prompt = f"Critique this draft:\n{state['draft']}"

    response = llm.invoke([
        SystemMessage(content=CRITIQUE_PROMPT),
        HumanMessage(content=prompt)
    ])

    return {"critique": response.content}

Step 4: Define Routing ​

python
MAX_ITERATIONS = 3

def should_continue(state: ReflectionState) -> str:
    if "APPROVED" in state.get("critique", "").upper():
        return "end"
    if state["iteration"] >= MAX_ITERATIONS:
        return "end"
    return "generate"

Step 5: Build Graph ​

python
from langgraph.graph import StateGraph, START, END

workflow = StateGraph(ReflectionState)

workflow.add_node("generate", generate_node)
workflow.add_node("critique", critique_node)

workflow.add_edge(START, "generate")
workflow.add_edge("generate", "critique")
workflow.add_conditional_edges(
    "critique",
    should_continue,
    {"generate": "generate", "end": END}
)

graph = workflow.compile()

Step 6: Use It ​

python
result = graph.invoke({
    "task": "Explain LangGraph in 2 sentences.",
    "messages": [],
    "draft": "",
    "critique": "",
    "iteration": 0
})

print(result["draft"])  # Final, refined output

Complete Code ​

python
from typing import Annotated
from typing_extensions import TypedDict
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_ollama import ChatOllama
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph_ollama_local import LocalAgentConfig

# === State ===
class ReflectionState(TypedDict):
    messages: Annotated[list, add_messages]
    task: str
    draft: str
    critique: str
    iteration: int

# === LLM ===
config = LocalAgentConfig()
llm = ChatOllama(
    model=config.ollama.model,
    base_url=config.ollama.base_url,
    temperature=0.7,
)

# === Nodes ===
def generate(state: ReflectionState) -> dict:
    iteration = state.get("iteration", 0)
    if iteration == 0:
        prompt = f"Write a response: {state['task']}"
    else:
        prompt = f"Revise based on critique:\nDraft: {state['draft']}\nCritique: {state['critique']}"

    response = llm.invoke([HumanMessage(content=prompt)])
    return {"draft": response.content, "iteration": iteration + 1}

def critique(state: ReflectionState) -> dict:
    prompt = f"Critique this (say APPROVED if perfect):\n{state['draft']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    return {"critique": response.content}

def should_continue(state: ReflectionState) -> str:
    if "APPROVED" in state.get("critique", "").upper():
        return "end"
    if state["iteration"] >= 3:
        return "end"
    return "generate"

# === Graph ===
workflow = StateGraph(ReflectionState)
workflow.add_node("generate", generate)
workflow.add_node("critique", critique)
workflow.add_edge(START, "generate")
workflow.add_edge("generate", "critique")
workflow.add_conditional_edges("critique", should_continue, {"generate": "generate", "end": END})
graph = workflow.compile()

# === Use ===
result = graph.invoke({
    "task": "Explain recursion in 2 sentences.",
    "messages": [],
    "draft": "",
    "critique": "",
    "iteration": 0
})
print(result["draft"])

Variations ​

Two-Model Reflection ​

Use a stronger model for critique:

python
generator = ChatOllama(model="llama3.2:3b")  # Fast
critic = ChatOllama(model="llama3.1:70b")    # Thorough

Structured Feedback ​

Use JSON for specific improvement areas:

python
CRITIQUE_PROMPT = """Return JSON with:
{
    "approved": true/false,
    "clarity": "feedback on clarity",
    "accuracy": "feedback on accuracy",
    "completeness": "feedback on completeness"
}"""

Common Pitfalls ​

1. Infinite Loops ​

python
# WRONG - no termination condition
def should_continue(state):
    return "generate"  # Always continues!

# CORRECT - multiple exit conditions
MAX_ITERATIONS = 3

def should_continue(state):
    # Exit on approval
    if "APPROVED" in state.get("critique", "").upper():
        return "end"
    # Exit on max iterations
    if state["iteration"] >= MAX_ITERATIONS:
        return "end"
    # Continue refining
    return "generate"

2. Critique Ignoring Instructions ​

python
# WRONG - vague critique prompt
"Give feedback on this draft"

# CORRECT - explicit approval signal
CRITIQUE_PROMPT = """You are a thoughtful editor.

Evaluate this draft against these criteria:
1. Clarity - Is the message clear?
2. Accuracy - Are all claims correct?
3. Completeness - Is anything missing?

If the draft meets all criteria, respond with exactly: "APPROVED"
Otherwise, provide specific, actionable feedback for each issue."""

3. Generator Not Using Critique ​

python
# WRONG - ignoring previous critique
def generate(state):
    prompt = f"Write about: {state['task']}"  # No reference to critique
    ...

# CORRECT - incorporate critique
def generate(state):
    if state["iteration"] == 0:
        prompt = f"Write about: {state['task']}"
    else:
        prompt = f"""Revise this draft to address the critique:

Original task: {state['task']}
Current draft: {state['draft']}
Critique to address: {state['critique']}

Produce an improved version that specifically addresses each point in the critique."""
    ...

Quiz ​

Test your understanding of reflection patterns:

Knowledge Check

What is the primary purpose of a reflection loop in LangGraph?

ATo save computation by caching results
BTo iteratively improve outputs through self-critique
CTo handle errors and retry failed operations
DTo parallelize LLM calls for faster execution

Knowledge Check

Why is MAX_ITERATIONS critical in reflection loops?

ATo limit token usage and reduce costs
BTo prevent infinite loops
CTo ensure consistent output quality
DTo satisfy API rate limits

Knowledge Check

What are the two common ways to exit a reflection loop?

ATimeout or user cancellation
BError thrown or completion flag
CApproval signal or max iterations reached
DQuality threshold or resource limit

Knowledge Check T/F

The generator node should use the critique feedback when producing revisions.

TTrue
FFalse

Knowledge Check Fill In

What field in ReflectionState tracks the current version of the output being refined?

What's Next? ​

Tutorial 06: Plan and Execute - Learn how to break complex tasks into steps, plan before executing, and re-plan based on results.