Module 5: Multi-Step Reasoning and Planning

8. Project: Research Agent v2 — Planning and Reflection

Project Overview

In Module 4 you built Research Agent v1: 4 nodes (planning, research, analysis, synthesis) connected by a StateGraph with conditional routing. It worked — it took a query, decomposed it into sub-questions, searched for information, evaluated quality, and iterated until the quality_score passed the threshold or the iterations ran out. But it had a limitation you probably already spotted: the planning was static and the quality evaluation was about the collected data, not about the final answer. If the agent researched in the wrong direction, it kept researching in that direction until it exhausted its iterations.

In this project you turn that agent into Research Agent v2 by adding three capabilities that fundamentally change how it reasons:

Deep planning: The planner generates a structured plan with priorities and success criteria. It isn't "search X, then Y, then Z" — it's "to answer the main question, I need to resolve A (high priority) and B (medium priority), where B depends on what I find in A."
Reflection: After synthesizing, a reflection node evaluates the final answer with specific critique prompts — did it answer the original question? are there claims with no source? are there contradictions? If the quality doesn't pass the threshold, it generates concrete notes about what's missing.
Re-planning: When the reflection detects gaps, the agent generates an alternative plan that accounts for what's already been researched and the reflection notes. It's the difference between "I repeat the search" and "I search from another angle."

Why it matters: An agent that doesn't reflect delivers the first answer it generates. An agent that reflects but doesn't re-plan detects problems but can't fix them. An agent with planning + reflection + re-planning detects that its research was insufficient, identifies what's missing, generates a new plan to fill those gaps, and re-executes only what's needed.

Estimated time: 60-90 minutes.

Project Goal

Extend Research Agent v1 (M4) with structured planning, reflection with critique prompts, conditional re-planning, and complete reasoning traces.

By the end you'll be able to:

Implement a planner that generates plans with prioritized sub-questions and success criteria
Build a reflection node with specific critique prompts and actionable feedback
Implement conditional re-planning that generates alternative plans when quality is low
Capture reasoning traces in every node for debugging and explainability
Extend an existing StateGraph without rewriting the base
Measure the impact of planning + reflection: compare quality, iterations, and cost between v1 and v2

What Changes vs v1 (M4)

Research Agent v1's foundation stays — you don't rewrite from scratch. But three things change fundamentally:

New state fields

Field	Type	Who writes it	What for
`reflection_notes`	`list[str]`	reflection → re-planning	Feedback on what's missing or wrong
`reasoning_trace`	`list[dict]`	every node	A record of decisions for debugging
`plan_revisions`	`int`	re-planning → routing	How many times it re-planned
`max_plan_revisions`	`int`	initial state → routing	A safety net against loops
`plan.priority_order`	`list[int]`	planning, re-planning	The priority order of the sub-questions
`plan.success_criteria`	`str`	planning	The definition of success for this research

New nodes

Node	Position	What it does
`reflection_node`	After synthesis	Evaluates the final answer with critique prompts
`replan_node`	After reflection (conditional)	An alternative plan based on the reflection notes

Changes to the flow

v1: planning → research ⇄ analysis → synthesis → END

v2: planning → research ⇄ analysis → synthesis → reflection ─┐
                ▲                                              │
                │            quality < threshold               │
                │            & revisions < max                 │
                └──── replan ◄─────────────────────────────────┘
                                    │
                              quality >= threshold
                                    │
                                    ▼
                                   END

The key difference: in v1, the quality gate was in analysis (it evaluated data). In v2, the quality gate is in reflection (it evaluates the final answer). You can have excellent data but a poor synthesis, and v2 catches that.

Technical Specifications

Stack

Technology	Version	Use
Python	3.11+	Runtime
langchain	v1.2+	init_chat_model, prompts
langchain-openai	latest	OpenAI provider
langgraph	v1.0+	StateGraph, conditional edges
tavily-python	latest	Web search
python-dotenv	any	Environment variables

pip install langchain langchain-openai langgraph tavily-python python-dotenv

You need the same .env from M4:

OPENAI_API_KEY=sk-proj-your-api-key-here
TAVILY_API_KEY=tvly-your-api-key-here

v2 Architecture

                                         quality >= threshold
                                    ┌──────────────────────────────┐
                                    │                              ▼
[START] → [PLANNING] → [RESEARCH] ⇄ [ANALYSIS] → [SYNTHESIS] → [REFLECTION] → [END]
              ▲            ▲  score<0.7   │                        │
              │            │  & iter<max  │                        │ quality < threshold
              │            └──────────────┘                        │ & revisions < max
              │                                                    │
              └──────────────── [RE-PLAN] ◄────────────────────────┘

The detailed flow

START → planning: A structured plan with prioritized sub-questions.
planning → research: Searches the highest-priority pending sub-question.
research → analysis: Evaluates whether the data is sufficient.
analysis → research (conditional): quality_score < 0.7 and iteration_count < max_iterations.
analysis → synthesis (conditional): Sufficient quality or iterations exhausted.
synthesis → reflection: Evaluates the final answer against 5 criteria.
reflection → END (conditional): quality_score >= 0.7.
reflection → replan (conditional): Low quality and plan_revisions < max.
replan → research: A new plan → execute from research.

Three safety nets

Max iterations (3): The limit on the research ↔ analysis cycle.
Max plan revisions (2): The limit on re-planning.
Quality threshold (0.7): The gate in reflection that decides between delivering and re-planning.

Step 1: Extend the State

from typing import Annotated, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages


class ResearchPlan(TypedDict):
    """Research plan v2 — with priorities and criteria."""
    main_query: str
    sub_questions: list[str]
    completed_questions: list[str]
    priority_order: list[int]
    success_criteria: str


class AgentState(TypedDict):
    # v1 fields
    messages: Annotated[list[BaseMessage], add_messages]
    plan: ResearchPlan
    iteration_count: int
    max_iterations: int
    research_data: list[str]
    quality_score: float
    final_answer: str
    metadata: dict

    # v2: Reflection + Re-planning
    reflection_notes: list[str]
    plan_revisions: int
    max_plan_revisions: int

    # v2: Reasoning traces
    reasoning_trace: list[dict]

What changed vs v1

New field	Default	Purpose
`reflection_notes`	`[]`	Reflection notes: what's missing, what to improve
`plan_revisions`	`0`	A counter of re-plans
`max_plan_revisions`	`2`	The maximum number of re-plans
`reasoning_trace`	`[]`	A reasoning trace from every node
`plan.priority_order`	`[]`	Indices of sub-questions by priority
`plan.success_criteria`	`""`	The success criterion defined by the planner

Each trace entry is a dict with node, timestamp, decision, reasoning, and metadata relevant to the node. The flexible format lets each node capture what it needs without a rigid schema. In M9, these traces will be the basis of trajectory evaluation tests.

Step 2: Implement the Planning Node

v2's planning generates a structured plan with priorities. If it receives reflection_notes (on a re-run), it incorporates them.

from dotenv import load_dotenv
load_dotenv()

import json
from datetime import datetime, timezone
from langchain.chat_models import init_chat_model
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage

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


def _parse_json(text: str) -> dict:
    """Extract JSON from an LLM response (tolerant of markdown)."""
    text = text.strip()
    if text.startswith("```"):
        text = "\n".join(text.split("\n")[1:-1])
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        import re
        match = re.search(r'\{[^{}]*\}', text, re.DOTALL)
        if match:
            try:
                return json.loads(match.group())
            except json.JSONDecodeError:
                pass
    return {}


def _trace(state, node, decision, reasoning, **extra):
    """Create a trace entry and return the extended list."""
    entry = {
        "node": node,
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "decision": decision,
        "reasoning": reasoning,
        **extra,
    }
    return state.get("reasoning_trace", []) + [entry]


def planning_node(state: dict) -> dict:
    """Generate a structured research plan.

    If there are reflection_notes, it incorporates them (re-planning aware).
    """
    query = ""
    for msg in reversed(state["messages"]):
        if isinstance(msg, HumanMessage):
            query = msg.content
            break

    reflection_notes = state.get("reflection_notes", [])

    context = ""
    if reflection_notes:
        context = (
            f"\nRE-PLANNING CONTEXT:\n"
            f"Notes: {json.dumps(reflection_notes, ensure_ascii=False)}\n"
            f"Previous data: {len(state.get('research_data', []))} fragments.\n"
            "Generate DIFFERENT sub-questions that cover the gaps."
        )

    response = model.invoke([
        SystemMessage(content=(
            "You are an expert researcher. Break the question into "
            "2-5 specific sub-questions with priorities.\n"
            f"{context}\n\n"
            "Answer with ONLY JSON:\n"
            '{"sub_questions": ["q1","q2"], "priority_order": [0,1], '
            '"success_criteria": "...", "reasoning": "..."}\n\n'
            "priority_order: indices (0-based) sorted by priority."
        )),
        HumanMessage(content=query),
    ])

    parsed = _parse_json(response.content)
    sub_qs = parsed.get("sub_questions", [query])
    prio = parsed.get("priority_order", list(range(len(sub_qs))))
    criteria = parsed.get("success_criteria", "A complete answer")
    reason = parsed.get("reasoning", "")

    plan = {
        "main_query": query,
        "sub_questions": sub_qs,
        "completed_questions": state.get("plan", {}).get(
            "completed_questions", []
        ),
        "priority_order": prio,
        "success_criteria": criteria,
    }

    text = f"Plan: {query}\nCriterion: {criteria}\n"
    for idx in prio:
        if idx < len(sub_qs):
            text += f"  [P{prio.index(idx)+1}] {sub_qs[idx]}\n"
    if reflection_notes:
        text += f"\nRe-planning based on: {reflection_notes[0]}"

    return {
        "plan": plan,
        "messages": [AIMessage(content=text)],
        "reasoning_trace": _trace(
            state, "planning",
            "re-plan" if reflection_notes else "initial_plan",
            reason, sub_questions_count=len(sub_qs),
        ),
    }

What changed vs v1

In v1, the planning was blind — it didn't know whether this was the first time or a re-plan. In v2, it checks reflection_notes. If there are notes, the prompt changes: "you already researched this, now research what's missing." That avoids repeating the same searches.

The priority_order lets research search the most important sub-questions first. If the iterations run out, at least it answered the high-priority ones.

Step 3: Implement the Reflection Node

This is the completely new node. It evaluates the final answer with specific critique prompts — not a generic "is it fine?", but 5 concrete criteria.

def reflection_node(state: dict) -> dict:
    """Evaluate the final answer with specific critique prompts."""
    answer = state.get("final_answer", "")
    plan = state.get("plan", {})
    data = state.get("research_data", [])
    prev_notes = state.get("reflection_notes", [])
    data_summary = "\n".join(data[:8])[:2000]

    user_text = (
        f"QUESTION: {plan.get('main_query', '')}\n"
        f"CRITERION: {plan.get('success_criteria', '')}\n"
        f"DATA ({len(data)} fragments):\n{data_summary}\n\n"
        f"ANSWER:\n{answer[:3000]}"
    )
    if prev_notes:
        user_text += f"\n\nPREVIOUS NOTES:\n{json.dumps(prev_notes, ensure_ascii=False)}"

    response = model.invoke([
        SystemMessage(content=(
            "Evaluate the answer against SPECIFIC criteria:\n"
            "1. COMPLETENESS: Does it cover every aspect?\n"
            "2. ACCURACY: Are the claims backed by data?\n"
            "3. COHERENCE: Is it internally consistent?\n"
            "4. SOURCES: Are specific sources cited?\n"
            "5. GAPS: Which aspects weren't covered?\n\n"
            "Score 0.7+: most criteria are good.\n"
            "Score 0.5-0.7: significant gaps.\n"
            "Score <0.5: insufficient.\n\n"
            "Be STRICT. If coverage of ONE important aspect is missing, "
            "the score must NOT exceed 0.6.\n\n"
            "Answer with ONLY JSON:\n"
            '{"quality_score": 0.0-1.0, "completeness": 0.0-1.0, '
            '"accuracy": 0.0-1.0, "coherence": 0.0-1.0, '
            '"source_quality": 0.0-1.0, "reasoning": "...", '
            '"missing_aspects": ["..."], "specific_feedback": "..."}'
        )),
        HumanMessage(content=user_text),
    ])

    parsed = _parse_json(response.content)
    score = max(0.0, min(1.0, float(parsed.get("quality_score", 0.5))))
    missing = parsed.get("missing_aspects", [])
    feedback = parsed.get("specific_feedback", "")
    reasoning = parsed.get("reasoning", "")

    new_notes = []
    if missing:
        new_notes.append(f"Missing aspects: {', '.join(missing)}")
    if feedback:
        new_notes.append(f"Feedback: {feedback}")

    text = f"Reflection: score={score:.2f}"
    if missing:
        text += f"\nGaps: {', '.join(missing)}"

    return {
        "quality_score": score,
        "reflection_notes": prev_notes + new_notes,
        "messages": [AIMessage(content=text)],
        "reasoning_trace": _trace(
            state, "reflection",
            "approve" if score >= 0.7 else "reject",
            reasoning, quality_score=score, missing_aspects=missing,
        ),
    }

Why 5 criteria instead of a generic score

A generic score ("rate it 0 to 10") isn't actionable. If the score is 5, you don't know whether it's because the answer is incomplete, imprecise, or incoherent. With sub-criteria, you can diagnose:

Low completeness, high accuracy: Good research but it doesn't cover every aspect → re-plan focused on the gaps.
High completeness, low accuracy: It covers everything but with vague data → re-research with more specific queries.
Low source quality: The data has no concrete sources → the re-plan prioritizes official sources.

This differential diagnosis is what makes the re-planning smart instead of repetitive.

Step 4: Implement Re-planning

It takes the reflection notes and generates a new plan that addresses the gaps. It doesn't repeat the original plan.

def replan_node(state: dict) -> dict:
    """Generate an alternative plan based on the reflection notes."""
    plan = state.get("plan", {})
    notes = state.get("reflection_notes", [])
    completed = plan.get("completed_questions", [])
    revisions = state.get("plan_revisions", 0)
    main_query = plan.get("main_query", "")

    response = model.invoke([
        SystemMessage(content=(
            "Re-plan the research. Do NOT repeat what's already been researched.\n"
            "Focus on the gaps from the feedback.\n"
            "Maximum 3 new sub-questions.\n\n"
            "Answer with ONLY JSON:\n"
            '{"sub_questions": ["q1","q2"], "priority_order": [0,1], '
            '"success_criteria": "...", "reasoning": "..."}'
        )),
        HumanMessage(content=(
            f"QUESTION: {main_query}\n"
            f"ALREADY RESEARCHED: {json.dumps(completed, ensure_ascii=False)}\n"
            f"DATA: {len(state.get('research_data', []))} fragments\n"
            f"FEEDBACK:\n{json.dumps(notes, ensure_ascii=False)}"
        )),
    ])

    parsed = _parse_json(response.content)
    new_qs = parsed.get("sub_questions", [f"Go deeper: {main_query}"])
    prio = parsed.get("priority_order", list(range(len(new_qs))))
    criteria = parsed.get("success_criteria",
                           plan.get("success_criteria", ""))
    reason = parsed.get("reasoning", "")

    new_plan = {
        "main_query": main_query,
        "sub_questions": new_qs,
        "completed_questions": [],
        "priority_order": prio,
        "success_criteria": criteria,
    }

    text = (f"Re-plan #{revisions + 1}: {reason}\n"
            + "".join(f"  {i}. {q}\n" for i, q in enumerate(new_qs, 1)))

    return {
        "plan": new_plan,
        "plan_revisions": revisions + 1,
        "iteration_count": 0,
        "messages": [AIMessage(content=text)],
        "reasoning_trace": _trace(
            state, "replan", "new_plan", reason,
            revision_number=revisions + 1,
        ),
    }

Key decisions

iteration_count resets to 0. Every re-plan starts a new research ↔ analysis cycle. Without resetting, the agent could exhaust its iterations before executing the new plan.

completed_questions gets cleared. The new plan has different sub-questions. But research_data is NOT cleared — the previous data is still useful for the synthesis.

Maximum 3 new sub-questions. Re-planning is for filling gaps, not for re-researching everything.

Step 5: Add Reasoning Traces

The planning, reflection, and replan nodes already capture traces with the _trace() function. Now we update research and analysis (inherited from v1) to capture them too.

Research node with traces and priorities

from langchain_community.tools.tavily_search import TavilySearchResults

search_tool = TavilySearchResults(max_results=3)


def research_node(state: dict) -> dict:
    """Search the highest-priority pending sub-question."""
    plan = state.get("plan", {})
    completed = set(plan.get("completed_questions", []))
    sub_qs = plan.get("sub_questions", [])
    prio = plan.get("priority_order", list(range(len(sub_qs))))

    pending = [sub_qs[i] for i in prio
               if i < len(sub_qs) and sub_qs[i] not in completed]

    if not pending:
        return {"messages": [AIMessage(content="All sub-questions researched.")]}

    current = pending[0]

    try:
        results = search_tool.invoke(current)
        success = True
    except Exception as e:
        results = [{"content": f"Error: {e}"}]
        success = False

    gathered = [r["content"] if isinstance(r, dict) and "content" in r
                else str(r) for r in results]

    return {
        "plan": {**plan, "completed_questions": list(completed) + [current]},
        "research_data": state.get("research_data", []) + gathered,
        "messages": [AIMessage(content=f"Researching: {current}\nSources: {len(gathered)}")],
        "reasoning_trace": _trace(
            state, "research", "search", f"Searching: {current}",
            search_success=success, results_count=len(gathered),
        ),
    }

Analysis node with traces

def analysis_node(state: dict) -> dict:
    """Evaluate the quality of the collected data."""
    plan = state.get("plan", {})
    data = state.get("research_data", [])
    iteration = state.get("iteration_count", 0)

    if not data:
        return {
            "quality_score": 0.0, "iteration_count": iteration + 1,
            "messages": [AIMessage(content="No data. Score: 0.0")],
            "reasoning_trace": _trace(state, "analysis", "no_data", "No data", quality_score=0.0),
        }

    completed = plan.get("completed_questions", [])
    total = len(plan.get("sub_questions", []))
    summary = "\n\n".join(data[:10])[:3000]

    response = model.invoke([
        SystemMessage(content=(
            "Evaluate the research quality: coverage, depth, relevance.\n\n"
            'Answer with ONLY JSON:\n{"quality_score": 0.0-1.0, "reasoning": "...", "missing_aspects": []}'
        )),
        HumanMessage(content=f"Question: {plan.get('main_query', '')}\nCompleted: {len(completed)}/{total}\nData:\n{summary}"),
    ])

    parsed = _parse_json(response.content)
    score = max(0.0, min(1.0, float(parsed.get("quality_score", 0.5))))
    reasoning = parsed.get("reasoning", "Parsing failed")

    return {
        "quality_score": score, "iteration_count": iteration + 1,
        "messages": [AIMessage(content=f"Analysis (iter {iteration+1}): score={score:.2f} — {reasoning}")],
        "reasoning_trace": _trace(
            state, "analysis", "sufficient" if score >= 0.7 else "insufficient",
            reasoning, quality_score=score, iteration=iteration + 1,
        ),
    }

Synthesis node with traces

def synthesis_node(state: dict) -> dict:
    """Combine all the research into a final answer."""
    plan = state.get("plan", {})
    data = state.get("research_data", [])
    quality = state.get("quality_score", 0.0)
    iterations = state.get("iteration_count", 0)
    completed = plan.get("completed_questions", [])
    combined = "\n---\n".join(data[:15])[:5000]

    response = model.invoke([
        SystemMessage(content=(
            "Synthesize the findings in English.\n\nFormat:\n"
            "## Summary\n(2-3 sentences)\n\n"
            "## Detailed Findings\n(Points with evidence)\n\n"
            "## Limitations\n(What couldn't be determined)"
        )),
        HumanMessage(content=(
            f"Question: {plan.get('main_query', '')}\n"
            f"Researched: {json.dumps(completed, ensure_ascii=False)}\n"
            f"Score: {quality:.2f}\nData:\n{combined}"
        )),
    ])

    return {
        "final_answer": response.content,
        "messages": [AIMessage(content=f"Synthesis complete. Iter: {iterations}, Score: {quality:.2f}")],
        "metadata": {
            **state.get("metadata", {}),
            "total_iterations": iterations,
            "final_quality_score": quality,
            "data_fragments": len(data),
        },
        "reasoning_trace": _trace(
            state, "synthesis", "synthesize",
            f"Synthesizing {len(data)} fragments", data_fragments=len(data),
        ),
    }

A utility to inspect traces

def print_reasoning_trace(trace: list[dict]) -> None:
    """Print the reasoning trace in a readable way."""
    print(f"\n{'='*60}")
    print("  REASONING TRACE")
    print(f"{'='*60}")
    for i, entry in enumerate(trace, 1):
        node = entry.get("node", "?").upper()
        decision = entry.get("decision", "?")
        reasoning = entry.get("reasoning", "")
        score = entry.get("quality_score")

        print(f"\n  [{i}] {node}")
        print(f"      Decision: {decision}")
        if score is not None:
            print(f"      Score: {score}")
        if reasoning:
            print(f"      Reasoning: {reasoning[:120]}")
        if entry.get("missing_aspects"):
            print(f"      Missing: {entry['missing_aspects']}")

    print(f"\n{'='*60}")
    print(f"  Total steps: {len(trace)}")
    print(f"{'='*60}")

This function is your main debugging tool. When the agent produces an unexpected answer, the trace tells you exactly what each node decided and why.

Step 6: Update the Graph

Two conditional edges: one in analysis (from v1, unchanged) and a new one in reflection.

from langgraph.graph import StateGraph, START, END


def should_continue_research(state: dict) -> str:
    """analysis → research or synthesis. Same logic as v1."""
    if state.get("quality_score", 0.0) >= 0.7:
        return "synthesis"
    if state.get("iteration_count", 0) >= state.get("max_iterations", 3):
        return "synthesis"
    return "research"


def should_replan_or_finish(state: dict) -> str:
    """reflection → END or replan.

    1. quality >= 0.7 → END
    2. quality < 0.7 AND revisions < max → replan
    3. quality < 0.7 AND revisions >= max → END (safety net)
    """
    if state.get("quality_score", 0.0) >= 0.7:
        return "end"
    if state.get("plan_revisions", 0) < state.get("max_plan_revisions", 2):
        return "replan"
    return "end"


def build_research_agent_v2():
    graph = StateGraph(AgentState)

    graph.add_node("planning", planning_node)
    graph.add_node("research", research_node)
    graph.add_node("analysis", analysis_node)
    graph.add_node("synthesis", synthesis_node)
    graph.add_node("reflection", reflection_node)
    graph.add_node("replan", replan_node)

    graph.add_edge(START, "planning")
    graph.add_edge("planning", "research")
    graph.add_edge("research", "analysis")
    graph.add_edge("synthesis", "reflection")
    graph.add_edge("replan", "research")

    graph.add_conditional_edges(
        "analysis", should_continue_research,
        {"research": "research", "synthesis": "synthesis"},
    )
    graph.add_conditional_edges(
        "reflection", should_replan_or_finish,
        {"end": END, "replan": "replan"},
    )

    return graph


agent_v2 = build_research_agent_v2().compile()

Anatomy of the v2 graph

Four types of connections:

Regular edges (→). START → planning → research → analysis, synthesis → reflection, replan → research.
Conditional edge 1 (◇). analysis ◇→ research|synthesis. The data's quality.
Conditional edge 2 (◇). reflection ◇→ end|replan. The final answer's quality.
Two cycles. The inner loop (research ↔ analysis) and the outer loop (reflection → replan → research → ... → synthesis → reflection).

The should_replan_or_finish function has three possibilities in order: sufficient quality → END; insufficient quality + revisions available → replan; insufficient quality + no revisions left → END (the safety net). Without the third case, a quality_score that never reaches 0.7 creates an infinite loop.

The Complete Research Agent v2

The code from steps 1-6 is everything you need. To run the agent end-to-end, add this execution function:

def run_research_v2(
    query: str,
    max_iterations: int = 3,
    max_plan_revisions: int = 2,
    verbose: bool = True,
    show_trace: bool = True,
) -> dict:
    """Run Research Agent v2."""
    initial_state = {
        "messages": [HumanMessage(content=query)],
        "plan": {
            "main_query": "", "sub_questions": [],
            "completed_questions": [], "priority_order": [],
            "success_criteria": "",
        },
        "iteration_count": 0,
        "max_iterations": max_iterations,
        "research_data": [],
        "quality_score": 0.0,
        "final_answer": "",
        "metadata": {},
        "reflection_notes": [],
        "plan_revisions": 0,
        "max_plan_revisions": max_plan_revisions,
        "reasoning_trace": [],
    }

    if verbose:
        print(f"\n{'='*60}")
        print(f"  Research Agent v2")
        print(f"  Query: {query}")
        print(f"  Max iterations: {max_iterations}, Max re-plans: {max_plan_revisions}")
        print(f"{'='*60}")

    result = agent_v2.invoke(initial_state)

    if verbose:
        revs = result.get("plan_revisions", 0)
        iters = result.get("iteration_count", 0)
        score = result.get("quality_score", 0.0)
        frags = len(result.get("research_data", []))
        print(f"\n{'='*60}")
        print(f"  RESULT")
        print(f"  Iterations: {iters}, Re-plans: {revs}")
        print(f"  Quality score: {score:.2f}, Data: {frags} fragments")
        print(f"{'='*60}")

    if show_trace:
        print_reasoning_trace(result.get("reasoning_trace", []))

    return result


if __name__ == "__main__":
    result = run_research_v2(
        "What are the current trends in AI agents "
        "and how do LangGraph and CrewAI compare?"
    )
    print(f"\n{result.get('final_answer', 'No answer')}")

Expected output (schematic)

============================================================
  Research Agent v2
  Query: What are the current trends in AI agents...?
============================================================

Plan: 3 sub-questions (trends, LangGraph, CrewAI)
  [P1] The main trends in AI agents in 2026
  [P2] LangGraph: advantages and limitations
  [P3] CrewAI: differences from LangGraph

Research → Analysis (iter 1): score=0.40
Research → Analysis (iter 2): score=0.60
Research → Analysis (iter 3): score=0.80 → Synthesis

Reflection: score=0.65 — Gaps: pricing comparison, benchmarks

Re-plan #1: pricing and benchmarks
Research → Analysis (iter 1): score=0.75 → Synthesis

Reflection: score=0.78 → APPROVED

  REASONING TRACE (14 steps)
  [1] PLANNING → initial_plan     [8] SYNTHESIS → synthesize
  [2-7] RESEARCH/ANALYSIS cycle   [9] REFLECTION → reject (0.65)
                                  [10] REPLAN → new_plan
                                  [11-12] RESEARCH/ANALYSIS
                                  [13] SYNTHESIS → synthesize
                                  [14] REFLECTION → approve (0.78)
============================================================

v1 would have delivered at step 8. v2 detected gaps, re-planned, researched what was missing, and delivered a more complete answer. 14 steps vs 8 — more expensive, but measurably better quality.

Recommended Tests

Query 1: Broad topic — expect re-planning

result = run_research_v2(
    "How does the Model Context Protocol (MCP) work and what "
    "are its implications for the AI agent ecosystem?",
    max_iterations=3, max_plan_revisions=2,
)

What it validates: The agent should plan with sub-questions about MCP (what it is, how it works, the ecosystem, the implications). The reflection will likely find that depth on implications is missing and trigger a re-plan.

What to watch in the trace: Did the re-plan generate different sub-questions? Does the new data complement (not repeat) the previous data?

Query 2: Specific topic — expect no re-planning

result = run_research_v2(
    "What is FastAPI and what are its advantages over Flask?",
    max_iterations=3, max_plan_revisions=2,
)

What it validates: A well-defined topic should pass reflection without re-planning. If it re-plans, your reflection prompt is too strict.

Query 3: Niche topic — stress test

result = run_research_v2(
    "What are the technical differences between Pydantic AI "
    "and LangGraph for building agents in production?",
    max_iterations=2, max_plan_revisions=2,
)

What it validates: A topic that requires a detailed technical comparison. How many re-plans? Did the quality_score improve between reflections?

Query 4: Max re-plans = 0 — v1 mode

result = run_research_v2(
    "What are the trends in AI agents?",
    max_iterations=3, max_plan_revisions=0,
)

What it validates: With no re-planning, the agent behaves like v1 with reflection. Compare the final quality_score against a run with max_plan_revisions=2 to measure the impact of re-planning.

Success Criteria

The agent plans before researching. The reasoning trace shows a planning step with sub-questions and priorities before any search.
The agent reflects and improves. Reflection generates a quality_score and, when it's low, actionable feedback. Compare max_plan_revisions=0 vs max_plan_revisions=2 — the second should be more complete.
Failed steps trigger re-planning. When reflection rejects, the agent generates a plan with new sub-questions that address the gaps. It doesn't repeat the same searches.
A complete reasoning trace. Every node generates a trace entry with node, decision, reasoning. The trace lets you reconstruct the whole chain of decisions.
The safety nets work. max_iterations stops the inner loop. max_plan_revisions stops the outer loop. Tested with both at low values.

Checklist

Common Errors

Error 1: The outer loop (reflection → replan) enters an infinite loop

Symptom: The agent re-plans indefinitely without improving.

Cause: max_plan_revisions isn't in the initial state, or should_replan_or_finish doesn't check it.

Solution: Make sure the initial state includes max_plan_revisions and that the routing function uses it as a safety net:

initial_state = {
    ...
    "plan_revisions": 0,
    "max_plan_revisions": 2,  # REQUIRED
}

Error 2: The re-plan repeats the same sub-questions

Symptom: After re-planning, it searches for exactly the same thing. The quality_score doesn't improve.

Cause: The replan_node's prompt doesn't include the already-completed sub-questions or the reflection notes.

Solution: The prompt must explicitly include what's already been researched and the feedback:

user_content = (
    f"ALREADY RESEARCHED: {json.dumps(completed)}\n"
    f"FEEDBACK: {json.dumps(reflection_notes)}\n"
    "Generate DIFFERENT sub-questions."
)

Error 3: `reasoning_trace` gets replaced instead of accumulating

Symptom: The trace only shows the last node that ran.

Cause: The node returns {"reasoning_trace": [new_entry]} without combining it with the existing entries.

Solution: Always combine it with the existing trace (just like research_data):

# BAD
return {"reasoning_trace": [trace_entry]}

# GOOD
existing = state.get("reasoning_trace", [])
return {"reasoning_trace": existing + [trace_entry]}

Error 4: The reflection always approves (score >= 0.7)

Symptom: The agent never re-plans because reflection always gives a high score.

Cause: The LLM tends to be "nice" with generic evaluations.

Solution: Add severity instructions to the prompt:

"Be STRICT. If coverage of ONE important aspect is missing,
the score must NOT exceed 0.6. We'd rather re-plan
than deliver an incomplete answer."

Error 5: `iteration_count` doesn't reset on re-plan

Symptom: After re-planning, it goes straight to synthesis because iteration_count >= max_iterations.

Cause: The replan_node doesn't reset iteration_count.

Solution: The replan_node must return "iteration_count": 0 so the new plan gets fresh iterations.

Error 6: JSON parsing fails repeatedly

Symptom: The reasoning trace shows "Fallback" in multiple nodes.

Cause: The LLM wraps the JSON in markdown code blocks or adds text.

Solution: Use _parse_json, which strips the markdown and extracts JSON with a regex fallback:

def _parse_json(text: str) -> dict:
    text = text.strip()
    if text.startswith("```"):
        text = "\n".join(text.split("\n")[1:-1])
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        import re
        match = re.search(r'\{[^{}]*\}', text, re.DOTALL)
        if match:
            try:
                return json.loads(match.group())
            except json.JSONDecodeError:
                pass
    return {}

Error 7: `research_data` grows without limit across multiple re-plans

Symptom: After 2 re-plans with 3 iterations, research_data has 27+ fragments. Slow and confused synthesis.

Cause: Each cycle appends data without clearing (intentional design — the previous data is useful).

Solution: Don't clear the data, but limit how much gets passed to the LLM with truncation:

# synthesis: maximum 15 fragments, 5000 chars
combined = "\n---\n".join(data[:15])[:5000]

# reflection: maximum 8 fragments, 2000 chars
summary = "\n".join(data[:8])[:2000]

Connection to M6

Your Research Agent v2 plans, researches, reflects, and re-plans. But it has a limitation you hit fast in real use: it remembers nothing between runs. If you ask "What are the trends in AI agents?" and then "And how does that compare to 2 years ago?", it has no context from the first question.

In Module 6 (Memory Systems) you solve this:

Checkpointing: The agent can pause mid-research and resume where it left off. Without checkpointing, a failure at step 7 loses steps 1-6.
Conversational memory: The agent remembers previous conversations across sessions.
MemorySaver + PostgresSaver: Added in one line: graph.compile(checkpointer=MemorySaver()).

The transition: "Your agent thinks and reflects (M5) → now make it remember (M6)." The reasoning_trace you captured here is especially useful with checkpointing — you can inspect a past run's trace to understand why the agent made certain decisions.

Resources

Plan-and-Execute Agents (LangGraph) — The official plan-and-execute pattern
Reflexion Paper — The original paper on self-reflection in LLM agents
ReAct Paper — Reasoning + Acting: the base pattern v2 extends
LangGraph Conditional Edges — Documentation on conditional routing
LangSmith Tracing — Tracing for debugging reasoning traces in production
LangGraph StateGraph Reference — The complete StateGraph API

← Previous7. Comparing Reasoning Patterns Next →1. Introduction: Agents with memory

8. Project: Research Agent v2 — Planning and Reflection

Project Overview

Project Goal

By the end you'll be able to:

What Changes vs v1 (M4)

New state fields

New nodes

Changes to the flow

Technical Specifications

Stack

v2 Architecture

The detailed flow

Three safety nets

Step 1: Extend the State

What changed vs v1

Step 2: Implement the Planning Node

What changed vs v1

Step 3: Implement the Reflection Node

Why 5 criteria instead of a generic score

Step 4: Implement Re-planning

Key decisions

Step 5: Add Reasoning Traces

Research node with traces and priorities

Analysis node with traces

Synthesis node with traces

A utility to inspect traces

Step 6: Update the Graph

Anatomy of the v2 graph

The Complete Research Agent v2

Expected output (schematic)

Recommended Tests

Query 1: Broad topic — expect re-planning

Query 2: Specific topic — expect no re-planning

Query 3: Niche topic — stress test

Query 4: Max re-plans = 0 — v1 mode

Success Criteria

Checklist

Common Errors

Error 1: The outer loop (reflection → replan) enters an infinite loop

Error 2: The re-plan repeats the same sub-questions

Error 3: reasoning_trace gets replaced instead of accumulating

Error 4: The reflection always approves (score >= 0.7)

Error 5: iteration_count doesn't reset on re-plan

Error 6: JSON parsing fails repeatedly

Error 7: research_data grows without limit across multiple re-plans

Connection to M6

Resources

Error 3: `reasoning_trace` gets replaced instead of accumulating

Error 5: `iteration_count` doesn't reset on re-plan

Error 7: `research_data` grows without limit across multiple re-plans