3. Plan-and-Execute Agents

Description

In the previous capsule you opened up the ReAct loop: Thought→Action→Observation on every iteration, the agent deciding what to do right now based on what it just observed. It works well for short tasks — search for something, use a tool, answer. But when the task takes 8 coordinated steps, ReAct has a fundamental problem: there's no plan. The agent improvises step by step, and if step 6 of 8 fails, it has no way to reorganize the remaining steps because it never knew there were remaining steps.

Plan-and-Execute solves this by separating two cognitive phases that ReAct mixes together: planning (decomposing the task into steps before acting) and execution (running each step of the plan with tools). The separation looks obvious, but it has a deep implication: the agent can reason about the entire task before executing any step. And when something fails — an API down, unexpected data, an impossible step — it can re-plan the remaining steps without losing the progress of the completed ones.

Connection with the module: This capsule is the second reasoning pattern you study (after ReAct). In capsule 04 you'll see task decomposition (how to break tasks into sub-tasks with dependencies), which is the advanced version of the planner you'll build here. In capsule 05, reflection will complement the re-planner: not only re-planning when something fails, but when the quality of the result isn't good enough.

The Plan-and-Execute Pattern

The intuition

Think about how a senior developer approaches a complex task. They don't open the editor and start writing code immediately. First they think: "I need to create the model, then the endpoint, then the tests, and finally the documentation." They have a mental plan. If the tests fail, they don't start from scratch — they adjust the endpoint code and re-run only the tests.

Plan-and-Execute is exactly that, but for agents. Three distinct components:

┌─────────────┐     ┌──────────────────┐     ┌───────────────┐
│   PLANNER   │────▶│    EXECUTOR      │────▶│  SYNTHESIZER  │
│ (LLM call)  │     │ (loop with tools)│     │   (answer)    │
└─────────────┘     └──────────────────┘     └───────────────┘
       ▲                     │
       │    re-plan if       │
       └─────it fails────────┘

1. Planner: Takes the original task and generates an ordered list of steps. It uses an LLM to decompose the task. It executes nothing — it only plans.
2. Executor: Takes each step of the plan and executes it using tools. Typically it's a ReAct sub-agent for each step.
3. Re-planner: When a step fails or produces unexpected results, it generates a new plan for the remaining steps, taking into account what was already completed and what failed.

Why separate planning from execution

The main trade-off: Plan-and-Execute adds up-front latency (the LLM needs to generate the plan before doing anything) and assumes the task is complex enough to justify that overhead.

Implementing the Planner

The agent's State

First you need a state that supports planning. Extend LangGraph's basic state:

from typing import TypedDict, Annotated
from langchain_core.messages import AnyMessage
from langgraph.graph.message import add_messages
from pydantic import BaseModel, Field


class Step(BaseModel):
    """A single step of the plan."""
    id: int = Field(description="Step number (1-indexed)")
    description: str = Field(description="What to do in this step")
    tool_hint: str = Field(
        default="",
        description="Suggested tool for this step (optional)"
    )


class Plan(BaseModel):
    """Complete plan generated by the planner."""
    goal: str = Field(description="Original goal of the task")
    steps: list[Step] = Field(description="Ordered list of steps")
    reasoning: str = Field(
        description="Why the planner chose these steps"
    )


class PlanExecuteState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    plan: Plan | None
    current_step_index: int
    step_results: list[dict]
    completed: bool

Plan as a Pydantic model for structured output (no manual parsing), step_results as a list of dicts (the re-planner needs to know what happened in each step), and current_step_index for progress tracking.

The Planner Node

from langchain.chat_models import init_chat_model

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

PLANNER_SYSTEM_PROMPT = """You are an expert planner. Your job is to decompose complex 
tasks into concrete, executable steps.

Rules:
- Each step must be a concrete action an agent with tools can execute
- The steps must be in logical order
- Maximum 8 steps (if you need more, the task is too complex)
- Include tool_hint when you know which tool is appropriate
- The last step must always be to synthesize/compile the results"""


def plan_node(state: PlanExecuteState) -> dict:
    """Generate a structured plan for the task."""
    task = state["messages"][-1].content
    planner_with_structure = model.with_structured_output(Plan)

    plan = planner_with_structure.invoke([
        {"role": "system", "content": PLANNER_SYSTEM_PROMPT},
        {"role": "user", "content": f"Task: {task}"}
    ])

    return {
        "plan": plan,
        "current_step_index": 0,
        "step_results": [],
    }

The with_structured_output(Plan) tells the LLM that its response must conform to the Plan schema. You get a Plan object validated by Pydantic, not a string you have to parse with regex.

What the planner produces

For "Research best practices for caching in distributed systems":

Plan(
    goal="Research best practices for caching in distributed systems",
    reasoning="I split it into: fundamentals, patterns, real-world cases, and synthesis.",
    steps=[
        Step(id=1, description="Search for the fundamentals of distributed caching",
             tool_hint="tavily_search"),
        Step(id=2, description="Search for patterns: write-through, cache-aside",
             tool_hint="tavily_search"),
        Step(id=3, description="Search for real cases: Netflix, Amazon, Twitter",
             tool_hint="tavily_search"),
        Step(id=4, description="Search for benchmarks and metrics",
             tool_hint="tavily_search"),
        Step(id=5, description="Synthesize into a technical report",
             tool_hint=""),
    ]
)

Implementing the Executor

The concept: one ReAct per step

The executor is a ReAct sub-agent that takes a plan step as its instruction and uses tools to complete it. Unlike a normal ReAct, the executor has a limited scope: it only needs to solve one step, not the whole task.

from langchain_core.tools import tool
from langchain_core.messages import HumanMessage
from langgraph.prebuilt import create_react_agent


@tool
def tavily_search(query: str) -> str:
    """Search the web for current information."""
    from tavily import TavilyClient
    client = TavilyClient()
    results = client.search(query, max_results=3)
    formatted = []
    for r in results["results"]:
        formatted.append(f"**{r['title']}**\n{r['content']}\nURL: {r['url']}")
    return "\n\n---\n\n".join(formatted)


executor_agent = create_react_agent(
    model=init_chat_model("openai:gpt-4.1-mini"),
    tools=[tavily_search],
)

The Execute Step Node (with error handling)

MAX_EXECUTOR_STEPS = 10


def execute_step_node(state: PlanExecuteState) -> dict:
    """Execute the current step of the plan using a ReAct sub-agent."""
    plan = state["plan"]
    step_index = state["current_step_index"]
    step = plan.steps[step_index]

    context_from_previous = ""
    if state["step_results"]:
        previous = state["step_results"][-1]
        context_from_previous = (
            f"\n\nResult of the previous step: {previous['output'][:500]}"
        )

    step_instruction = (
        f"Execute this step of a research plan:\n\n"
        f"Step {step.id}: {step.description}\n"
        f"{'Suggested tool: ' + step.tool_hint if step.tool_hint else ''}"
        f"{context_from_previous}\n\n"
        f"Reply with the concrete result of the step. Be specific."
    )

    try:
        result = executor_agent.invoke(
            {"messages": [HumanMessage(content=step_instruction)]},
            config={"recursion_limit": MAX_EXECUTOR_STEPS},
        )
        step_output = result["messages"][-1].content

        if len(step_output.strip()) < 20:
            raise ValueError("Result too short — possible failure")

        step_result = {
            "step_id": step.id,
            "description": step.description,
            "output": step_output,
            "success": True,
        }
    except Exception as e:
        step_result = {
            "step_id": step.id,
            "description": step.description,
            "output": f"ERROR: {str(e)}",
            "success": False,
            "error": str(e),
        }

    return {
        "step_results": state["step_results"] + [step_result],
        "current_step_index": step_index + 1,
    }

Three decisions: the executor passes the previous step's result as context (the "synthesize" step needs to know what was found); recursion_limit prevents infinite loops in the sub-agent; and the minimum length check (< 20 chars) treats empty responses as failures so the re-planner kicks in.

Re-planning: When the Plan Fails

Why re-planning is the critical feature

Without re-planning, Plan-and-Execute is worse than ReAct. A static plan that executes steps linearly breaks at the first error and has no way to recover. ReAct at least improvises — if a tool fails, it tries something else on the next iteration.

Re-planning is what makes the pattern useful: the agent detects that a step failed, analyzes why it failed, and generates alternative steps. Already-completed steps are preserved.

Original plan: [1 ✓] [2 ✓] [3 ✗] [4] [5]
                                ↓
Re-plan:       [1 ✓] [2 ✓] [3a: alternative] [4a: adjusted] [5]

Implementing the Re-planner

REPLANNER_SYSTEM_PROMPT = """You are a re-planner. A research plan had a step that 
failed. Your job is to generate alternative steps to complete the task.

You will receive:
- The original goal
- The steps completed successfully and their results
- The step that failed and the error
- The pending steps of the original plan

Generate a new plan that:
1. Does NOT repeat the steps already completed
2. Offers an alternative for the failed step
3. Adjusts the pending steps if necessary
4. Keeps the original goal"""


class ReplanOutput(BaseModel):
    """Output of the re-planner."""
    analysis: str = Field(description="Why the step failed")
    new_steps: list[Step] = Field(description="New steps")
    should_abort: bool = Field(
        default=False,
        description="True if the task is impossible to complete"
    )


MAX_REPLANS = 2


def replan_node(state: PlanExecuteState) -> dict:
    """Re-plan when a step fails."""
    plan = state["plan"]
    results = state["step_results"]
    current_index = state["current_step_index"]

    replan_count = sum(1 for r in results if not r["success"])
    if replan_count >= MAX_REPLANS:
        return {
            "completed": True,
            "messages": [{
                "role": "assistant",
                "content": "Re-planning attempts exhausted. "
                           "Delivering partial results."
            }],
        }

    failed_step = results[-1]
    completed_steps = [r for r in results if r["success"]]
    remaining_original = plan.steps[current_index:]

    completed_summary = "\n".join(
        f"  Step {r['step_id']}: {r['description']} → {r['output'][:200]}"
        for r in completed_steps
    )
    remaining_summary = "\n".join(
        f"  Step {s.id}: {s.description}" for s in remaining_original
    )

    replan_prompt = f"""Original goal: {plan.goal}

Steps completed successfully:
{completed_summary}

Step that failed:
  Step {failed_step['step_id']}: {failed_step['description']}
  Error: {failed_step['output']}

Pending steps of the original plan:
{remaining_summary}

Generate a new plan for the remaining steps."""

    replanner = model.with_structured_output(ReplanOutput)
    replan_result = replanner.invoke([
        {"role": "system", "content": REPLANNER_SYSTEM_PROMPT},
        {"role": "user", "content": replan_prompt},
    ])

    if replan_result.should_abort:
        return {
            "completed": True,
            "messages": [{
                "role": "assistant",
                "content": f"I couldn't complete the task. "
                           f"Analysis: {replan_result.analysis}"
            }],
        }

    new_plan = Plan(
        goal=plan.goal,
        steps=replan_result.new_steps,
        reasoning=f"Re-plan after failure in step "
                  f"{failed_step['step_id']}: {replan_result.analysis}",
    )

    return {
        "plan": new_plan,
        "current_step_index": 0,
    }

How re-planning works

Four key mechanics: (1) the re-planner receives full context — what was achieved, what failed, what was still pending; (2) if the task is impossible, should_abort=True ends things gracefully; (3) current_step_index resets to 0 because the new plan has its own steps; (4) the accumulated step_results aren't wiped — the synthesizer has access to the results of all the plans.

StateGraph for Plan-and-Execute

The complete graph

from langgraph.graph import StateGraph, START, END


def check_progress(state: PlanExecuteState) -> str:
    """Decide the next step based on current progress."""
    if state.get("completed"):
        return "synthesize"

    results = state["step_results"]
    current_index = state["current_step_index"]

    if results and not results[-1]["success"]:
        replan_count = sum(1 for r in results if not r["success"])
        if replan_count >= MAX_REPLANS:
            return "synthesize"
        return "replan"

    if current_index >= len(state["plan"].steps):
        return "synthesize"

    return "execute_step"


def synthesize_node(state: PlanExecuteState) -> dict:
    """Synthesize all the results into a final answer."""
    plan = state["plan"]
    successful_results = [r for r in state["step_results"] if r["success"]]

    results_text = "\n\n".join(
        f"### Step {r['step_id']}: {r['description']}\n{r['output']}"
        for r in successful_results
    )

    synthesis_prompt = f"""Goal: {plan.goal}

Research results:
{results_text}

Synthesize this into a complete, coherent, well-structured answer. 
Include sources and specific data. If there are gaps, mention them."""

    response = model.invoke([
        {"role": "system", "content": "You are an expert synthesizer."},
        {"role": "user", "content": synthesis_prompt},
    ])

    return {
        "completed": True,
        "messages": [{"role": "assistant", "content": response.content}],
    }


graph = StateGraph(PlanExecuteState)

graph.add_node("plan", plan_node)
graph.add_node("execute_step", execute_step_node)
graph.add_node("replan", replan_node)
graph.add_node("synthesize", synthesize_node)

graph.add_edge(START, "plan")
graph.add_edge("plan", "execute_step")
graph.add_conditional_edges(
    "execute_step",
    check_progress,
    {
        "execute_step": "execute_step",
        "replan": "replan",
        "synthesize": "synthesize",
    },
)
graph.add_edge("replan", "execute_step")
graph.add_edge("synthesize", END)

plan_execute_agent = graph.compile()

Visualizing the flow

START
  │
  ▼
┌──────────┐
│   plan   │  ← LLM generates a structured plan
└────┬─────┘
     │
     ▼
┌──────────────┐
│ execute_step │  ← ReAct sub-agent executes the current step
└──────┬───────┘
       │
       ▼
  check_progress ─────────────────────────┐
       │              │                    │
       │ (next step)  │ (step failed)      │ (all done)
       ▼              ▼                    ▼
  execute_step    ┌────────┐         ┌────────────┐
       ▲          │ replan │         │ synthesize │
       │          └───┬────┘         └─────┬──────┘
       │              │                    │
       └──────────────┘                    ▼
                                          END

Execution and streaming

result = plan_execute_agent.invoke({
    "messages": [{"role": "user", "content": "Research best practices "
                  "for caching in distributed systems"}],
    "plan": None,
    "current_step_index": 0,
    "step_results": [],
    "completed": False,
})

print(result["messages"][-1].content)

One advantage of Plan-and-Execute: you can show progress step by step with streaming:

for event in plan_execute_agent.stream({
    "messages": [{"role": "user", "content": "Research distributed caching"}],
    "plan": None, "current_step_index": 0,
    "step_results": [], "completed": False,
}):
    for node_name, node_output in event.items():
        if node_name == "plan":
            plan = node_output["plan"]
            print(f"📋 Plan generated ({len(plan.steps)} steps):")
            for step in plan.steps:
                print(f"  {step.id}. {step.description}")
        elif node_name == "execute_step":
            latest = node_output["step_results"][-1]
            status = "✅" if latest["success"] else "❌"
            print(f"{status} Step {latest['step_id']}: {latest['description']}")
        elif node_name == "replan" and node_output.get("plan"):
            print(f"🔄 Re-planning ({len(node_output['plan'].steps)} steps)")
        elif node_name == "synthesize":
            print("📝 Synthesizing the final answer...")

When to Use Plan-and-Execute vs ReAct

Don't use Plan-and-Execute for everything. The planner's overhead isn't justified on simple tasks.

Concrete scenarios

Use ReAct: "What's the current price of Bitcoin?" (one tool call). "Summarize this article" (one LLM call).

Use Plan-and-Execute: "Research the pros and cons of microservices vs monoliths and generate a report" (multiple searches + synthesis). "Compare the 3 best database options for high write volume with benchmarks" (multi-source + comparison).

Comparative metrics

Pure ReAct:          3-12 tool calls | 8-25s  | ~$0.03-0.08 | Quality: 6/10
Plan-and-Execute:    6-8 tool calls  | 15-30s | ~$0.05-0.10 | Quality: 8/10
With re-planning:    6-12 tool calls | 20-45s | ~$0.07-0.15 | Quality: 8/10

Plan-and-Execute costs ~50% more in latency and tokens, but produces more complete and predictable results. On tasks of 4+ steps, that trade-off is worth it.

Connection with the Project

The M4 Research Agent (a StateGraph with basic nodes) gains three new nodes: research_planner (decomposes the question into sub-questions), step_executor (executes each step with ReAct + tools), and research_replanner (adjusts steps when a source fails). New state fields: research_plan, plan_step_results, replan_count.

Capsule 04 (Task Decomposition) will take the planner further with dependency graphs and parallel execution. Capsule 05 (Reflection) adds a quality gate after the synthesizer.

Troubleshooting

Problem 1: The planner generates too many steps

Symptom: A 15-20 step plan for a simple task. Solution: A limit in the prompt ("Maximum 8 steps") + post-generation truncation with plan.steps = plan.steps[:7] + [Step(id=8, description="Synthesize")].

Problem 2: The re-planner enters a loop

Symptom: It re-plans indefinitely. Solution: MAX_REPLANS = 2 as a guardrail. Also, pass the full failure history to the re-planner so it doesn't repeat the same strategy that already failed.

Problem 3: The executor sub-agent gets stuck in a loop

Symptom: A step takes minutes. Solution: recursion_limit in the sub-agent (already included). If it persists, add a timeout with asyncio.wait_for.

Problem 4: Results from previous steps get lost

Symptom: The synthesis step has no access to previous results. Solution: The executor must return state["step_results"] + [step_result] (accumulate), not [step_result] (replace).

Problem 5: The planner's structured output fails

Symptom: with_structured_output(Plan) raises ValidationError. Solution: Validation in Pydantic (Field(ge=1, le=20), min_length=10) + try/except with a fallback plan of 2 generic steps.

Exercises

Exercise 1: A plan for comparison tasks

Write a PLANNER_SYSTEM_PROMPT specialized for comparison tasks (e.g. "Compare React vs Vue vs Svelte"). The prompt must produce a plan that searches for information about each option separately before comparing.

COMPARISON_PLANNER_PROMPT = """You are a planner specialized in comparisons.

When you receive a task that compares N options, your plan MUST follow this structure:
1. One search step for EACH individual option (N steps)
2. One search step for existing direct comparisons
3. One synthesis step with a criteria table and a recommendation

Rules:
- Use the SAME criteria for all options (a fair comparison)
- Maximum 8 steps
- The synthesis step ALWAYS includes a comparison table"""

planner = model.with_structured_output(Plan)
plan = planner.invoke([
    {"role": "system", "content": COMPARISON_PLANNER_PROMPT},
    {"role": "user", "content": "Compare React vs Vue vs Svelte for an "
                                 "enterprise dashboard"},
])
for step in plan.steps:
    print(f"  {step.id}. {step.description}")

Exercise 2: A re-planner with a strategy change

Modify replan_node so that, when a search step fails, the re-planner doesn't simply retry — it reformulates the question or changes the source. Add a strategy_change field to ReplanOutput.

class ReplanOutput(BaseModel):
    analysis: str
    strategy_change: str = Field(
        description="How the strategy changes. E.g.: 'I reformulated the search "
                    "from academic papers to technical blog posts'"
    )
    new_steps: list[Step]
    should_abort: bool = False

REPLANNER_SMART_PROMPT = """You are a strategic re-planner. When a step fails, 
you do NOT simply retry the same thing. You change the STRATEGY:

- If a technical search fails → search blogs/tutorials instead of papers
- If a source is down → use a different alternative source
- If the data doesn't exist → reformulate to get proxy data

Explain in strategy_change what change you're making and why."""

def replan_node(state: PlanExecuteState) -> dict:
    replanner = model.with_structured_output(ReplanOutput)
    result = replanner.invoke([
        {"role": "system", "content": REPLANNER_SMART_PROMPT},
        {"role": "user", "content": build_replan_prompt(state)},
    ])
    print(f"🔄 Strategy change: {result.strategy_change}")
    
    if result.should_abort:
        return {"completed": True, "messages": [...]}
    
    new_plan = Plan(
        goal=state["plan"].goal,
        steps=result.new_steps,
        reasoning=f"Re-plan: {result.strategy_change}",
    )
    return {"plan": new_plan, "current_step_index": 0}

Exercise 3: Per-step timeout with asyncio

Implement a 30-second timeout for each executor step. If a step exceeds the time, mark it as failed and trigger the re-planner. Use asyncio.wait_for with asyncio.to_thread to wrap the synchronous invocation.

import asyncio

STEP_TIMEOUT_SECONDS = 30

async def execute_step_with_timeout(state: PlanExecuteState) -> dict:
    step = state["plan"].steps[state["current_step_index"]]
    try:
        result = await asyncio.wait_for(
            asyncio.to_thread(
                executor_agent.invoke,
                {"messages": [HumanMessage(content=f"Step {step.id}: {step.description}")]}
            ),
            timeout=STEP_TIMEOUT_SECONDS,
        )
        step_result = {"step_id": step.id, "description": step.description,
                       "output": result["messages"][-1].content, "success": True}
    except asyncio.TimeoutError:
        step_result = {"step_id": step.id, "description": step.description,
                       "output": f"ERROR: Timeout ({STEP_TIMEOUT_SECONDS}s)",
                       "success": False, "error": "timeout"}

    return {"step_results": state["step_results"] + [step_result],
            "current_step_index": state["current_step_index"] + 1}

Exercise 4: Conditional steps

Extend Step with a condition: str | None field (e.g. "previous_step_contains:benchmark"). Implement evaluate_condition, which parses the condition and decides whether to execute or skip the step.

class Step(BaseModel):
    id: int = Field(ge=1, le=20)
    description: str = Field(min_length=10)
    tool_hint: str = Field(default="")
    condition: str | None = Field(default=None,
        description="E.g.: 'previous_step_contains:benchmark'")

def evaluate_condition(condition: str | None, prev: dict) -> bool:
    if condition is None:
        return True
    if condition.startswith("previous_step_contains:"):
        keyword = condition.split(":", 1)[1].strip().lower()
        return keyword in prev.get("output", "").lower()
    if condition == "previous_step_success":
        return prev.get("success", False)
    return True

def execute_step_node(state: PlanExecuteState) -> dict:
    step = state["plan"].steps[state["current_step_index"]]
    if step.condition and state["step_results"]:
        if not evaluate_condition(step.condition, state["step_results"][-1]):
            return {
                "step_results": state["step_results"] + [{"step_id": step.id,
                    "description": step.description,
                    "output": "SKIPPED: condition not met", "success": True}],
                "current_step_index": state["current_step_index"] + 1,
            }
    # ... normal execution of the step

Exercise 5: Progress dashboard

Build print_plan_dashboard(state) that prints the plan with icons: ✅ completed, ❌ failed, 🔄 running, ⬚ pending. Include a progress count and whether there was a re-plan.

def print_plan_dashboard(state: PlanExecuteState) -> None:
    plan = state["plan"]
    if not plan:
        print("⏳ Waiting for the plan..."); return

    print(f"\n{'='*60}")
    print(f"🎯 {plan.goal}  |  📋 {len(plan.steps)} steps")
    print(f"{'='*60}")

    result_map = {r["step_id"]: r for r in state["step_results"]}
    for step in plan.steps:
        r = result_map.get(step.id)
        if r and r["success"]:
            icon, detail = "✅", r["output"][:60] + "..."
        elif r and not r["success"]:
            icon, detail = "❌", r.get("error", "Error")
        elif step.id - 1 == state["current_step_index"]:
            icon, detail = "🔄", "Running..."
        else:
            icon, detail = "⬚ ", "Pending"
        print(f"  {icon} Step {step.id}: {step.description}")
        print(f"       └─ {detail}")

    ok = sum(1 for r in state["step_results"] if r.get("success"))
    fail = sum(1 for r in state["step_results"] if not r.get("success"))
    print(f"\n📊 {ok}/{len(plan.steps)} completed, {fail} failures")
    if "Re-plan" in (plan.reasoning or ""):
        print(f"🔄 {plan.reasoning}")
    print(f"{'='*60}\n")

Summary

• Plan-and-Execute separates planning from execution: the planner generates a structured plan, the executor runs each step with a ReAct sub-agent.
• Structured output (Pydantic + with_structured_output) is key so the plan is a typed object, not free text you have to parse.
• Re-planning is the feature that justifies the pattern: when a step fails, the re-planner generates alternative steps while preserving the progress.
• Mandatory guardrails: MAX_REPLANS, recursion_limit, plan validation. Without them, infinite loops and 50-step plans.
• Trade-off: +50% latency/cost vs more complete and predictable results. Justified for tasks of 4+ steps.
• The StateGraph ties it all together: plan → execute_step → check_progress → (replan | next_step | synthesize).

Additional Resources

1. Plan-and-Execute Paper (Wang et al.) — The original paper proposing the separation of planning and execution
2. LangGraph Plan-and-Execute Tutorial — Official tutorial with a reference implementation
3. Structured Output in LangChain — Documentation for with_structured_output
4. ReAct Paper — To compare with the previous capsule's pattern
5. LangGraph StateGraph API — Reference for the graph API