⚠️ Before You Start

Risk: Medium

Duration: 1-2 days

Last Validated: Not yet validated (draft)

Kind: How-to

Risk Note: Unbounded replanning loops drive exponential token costs and request timeouts

Implement plan-and-execute with bounded replanning in LangGraph¶

Draft outline: not validated

Drafted with AI from official sources on 16 Oct 2026. Verified and expanded under AGT-03-T05 on 23 Oct 2026. Do not use on a live system until it is written, dry-run by someone other than the author and signed off (planned 2026-12-29). Author: Sanjay Sarvesh · dry-run: Yasar · sign-off: James. Work items: AGT-03-T05, AGT-03-T06, AGT-03-T07, AGT-03-T08.

1. Summary¶

This how-to guides the implementation of a Plan-and-Execute agent architecture with bounded replanning using LangGraph StateGraph. It establishes a dedicated planner node that generates a structured multi-step task list, an execution node that dispatches tasks against registered tools, a replanner node that adjusts the remaining plan based on intermediate observations, and strict replan limits to eliminate runaway loops. The outcome is a deterministic, observable planning workflow that completes complex multi-step user goals with predictable cost and bounded latency.

2. When to use and when not to¶

Use this how-to when:

Tasks require solving multi-step, complex objectives where deciding all steps upfront reduces tool-calling errors and context drift compared to a single ReAct loop.
The plan must be visible, inspectable, or editable by human operators before or during execution.
Tasks occasionally require dynamic course-correction (replanning) based on tool failures or unexpected findings, but replanning must be strictly capped to prevent unbounded token expenditure.
The planner pattern decision guide (PB-AGT-03-01) selects Plan-and-Execute over pure ReAct or static workflows.

Do not use it when:

The task is a simple, deterministic multi-step sequence with fixed tools: use a static prompt-chaining workflow (START -> step_1 -> step_2 -> END) without planner LLM overhead.
Tasks are open-ended conversational exchanges or reactive single-turn tool calls: use a standard ReAct task agent (PB-AGT-01-02).
Tasks require autonomous reflection and self-critique over generated artifacts: use the Reflexion pattern instead.
Sub-tasks can execute completely independently in parallel: use the LangGraph Send API (Map-Reduce pattern) or an orchestrator-worker topology without sequential replanning.

3. What you must know first¶

Separation of planning, execution, and replanning. In Plan-and-Execute, task decomposition is decoupled from execution. The Planner generates an initial ordered step list (Plan(steps=[...])), the Executor runs the head task against available tools, and the Replanner reviews the original objective, remaining steps, and past step observations ((task, result)) to either output a final user Response or an updated Plan.
Pydantic schemas with structured output (Union[Response, Plan]). The replanner must return structured output mapped to an envelope model (class Act(BaseModel): action: Union[Response, Plan]). This forces the model to choose explicitly between terminating with a final response or outputting refined remaining steps, avoiding ambiguous free-text routing.
Bounded replanning (soft stop) vs engine recursion limit (hard stop). A state counter replans: int counts only plan changes (not continuations) and bounds them (state["replans"] >= MAX_REPLANS). When reached, the router forcibly bypasses the replanner, synthesizing the best possible answer from accumulated observations and routing to END. The LangGraph engine's recursion_limit in the invocation config provides a hard safety net that prevents runaway execution in case of routing bugs.
Context preservation across steps. The state must maintain past_steps: Annotated[list[tuple[str, str]], operator.add] storing each executed task and its tool output. Passing past execution results back to the replanner is required to prevent "groundhog day" loops where the planner continually re-emits already completed tasks.
Parallel fan-out via Send API. When the planner outputs independent, non-overlapping steps (e.g. searching 3 distinct entities), the orchestrator can fan out tasks concurrently using LangGraph's Send("executor", {"step": step}) instead of executing sequentially.
Plans shown to users must be readable: short, plain-language steps that a user or operator can review before or during execution.

4. Prerequisites and access¶

Item	Detail
Model Deployment	Azure OpenAI or OpenAI model deployment supporting structured outputs (`with_structured_output(Plan)` and `with_structured_output(Act)`), e.g. `gpt-4o` or `gpt-4o-mini`.
Development Tools	Python 3.10+ with `langgraph>=0.2.0`, `langchain-core`, `langchain-openai`, and `pydantic>=2.0`.
Tool Registry	Registered task tools (e.g. web search, database query, arithmetic) with typed Pydantic schemas (PB-AGT-02-01).
Template Repository	Planner agent template cloned.
Evaluation Dataset	Tasks that need zero, one and too many replans.
Observability	LangSmith API key with `LANGSMITH_TRACING=true` to capture planning decompositions, executor step runs, and replan events.
Configuration	Maximum replan limit (`MAX_REPLANS`, value agreed with the agent owner), a maximum plan length enforced on both `Plan` outputs, and an explicit `recursion_limit` in the run config (the default varies by LangGraph version; never rely on it).

Set up the environment variables:

export OPENAI_API_KEY="${OPENAI_API_KEY}"
export LANGSMITH_TRACING="true"
export LANGSMITH_API_KEY="${LANGSMITH_API_KEY}"
export LANGSMITH_PROJECT="plan-and-execute"

5. Steps¶

Step 1: Define graph state with goal, plan, past observations, and replan counter¶

Define a typed dictionary representing the graph state. Include the original user input, current remaining plan steps, an additive list of completed (task, result) tuples, the final response text, and the replan counter.

import operator
from typing import Annotated, List, Tuple, TypedDict

class PlanExecuteState(TypedDict):
    input: str
    plan: List[str]
    past_steps: Annotated[List[Tuple[str, str]], operator.add]
    response: str
    replans: int

Check: Instantiating PlanExecuteState with empty plan and replans=0 succeeds without type errors.

Step 2: Define structured planning and replanning schemas with Pydantic¶

Define Pydantic models for the initial plan, final response, and the replanner union envelope (Act).

from typing import List, Union
from pydantic import BaseModel, Field

class Plan(BaseModel):
    """Plan to follow in the future."""
    steps: List[str] = Field(description="Ordered list of actionable tasks to complete the objective.")

class Response(BaseModel):
    """Final response to the user."""
    response: str = Field(description="Final comprehensive answer fulfilling the user objective.")

class Act(BaseModel):
    """Action to perform: either a final response or a revised plan."""
    action: Union[Response, Plan] = Field(
        description="Return Response if objective is fully met; return Plan if additional steps are needed."
    )

Check: Instantiating Act(action=Plan(steps=["step 1"])) and Act(action=Response(response="done")) validate successfully.

Step 3: Implement the planner node for initial plan decomposition¶

Implement the planner function. Use an LLM with structured output bound to Plan. Prompt the model to break down the user request into clear, actionable, tool-executable steps.

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

planner_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are an expert planner. Break down the user objective into 3 to 5 clear, sequential, actionable steps. Do not attempt to answer the request directly."),
    ("user", "{input}")
])

planner = planner_prompt | ChatOpenAI(model="gpt-4o", temperature=0).with_structured_output(Plan)

def plan_node(state: PlanExecuteState) -> dict:
    plan = planner.invoke({"input": state["input"]})
    return {"plan": plan.steps, "replans": 0}

Check: Invoking plan_node({"input": "Summarise Q3 sales trends"}) returns a dictionary containing a non-empty list of steps.

Step 4: Implement the executor node to run the current step¶

Implement the execution node. Extract the head step (state["plan"][0]), dispatch it to the registered tools or a ReAct task agent, and append the (task, result) tuple to past_steps while advancing the remaining plan.

from langchain_openai import ChatOpenAI

executor_llm = ChatOpenAI(model="gpt-4o", temperature=0)

def execute_node(state: PlanExecuteState) -> dict:
    current_task = state["plan"][0]
    # In production, dispatch to tools or create_react_agent
    prompt = f"Objective: {state['input']}\nTask: {current_task}\nExecute this task and provide concise findings."
    result = executor_llm.invoke(prompt)
    return {
        "past_steps": [(current_task, result.content)],
        "plan": state["plan"][1:]
    }

Check: Invoking execute_node advances plan by one item and appends one entry to past_steps.

Step 5: Implement the replanner node with bounded replanning logic¶

Implement the replanner function. First check if state["replans"] >= MAX_REPLANS. If the replan threshold is reached, synthesize a final answer immediately without calling the planning LLM. Otherwise, prompt the model with the objective, remaining plan, and past step results using structured output Act.

Increment replans only when the replanner changes the remaining plan (the returned steps differ from state["plan"]). Returning the unchanged remaining steps is a continuation, not a replan.

MAX_REPLANS = 2

replanner_prompt = ChatPromptTemplate.from_messages([
    ("system", (
        "You are an evaluator and replanner. Review the user objective, existing plan, and completed steps with findings.\n"
        "If the objective is satisfied, return a Response object with the final answer.\n"
        "If more steps are required, return an updated Plan with only the remaining steps.\n"
        "Do not repeat tasks that have already been completed."
    )),
    ("user", "Objective: {input}\nRemaining Plan: {plan}\nCompleted Steps: {past_steps}")
])

replanner = replanner_prompt | ChatOpenAI(model="gpt-4o", temperature=0).with_structured_output(Act)

def replan_node(state: PlanExecuteState) -> dict:
    if state.get("replans", 0) >= MAX_REPLANS:
        # Soft stop: force conclusion when replan budget is exhausted
        summary_prompt = f"Objective: {state['input']}\nCompleted steps: {state['past_steps']}\nSummarise the best possible answer from completed findings."
        final_ans = executor_llm.invoke(summary_prompt).content
        return {"response": f"[Replan limit reached] {final_ans}", "plan": []}

    act = replanner.invoke({
        "input": state["input"],
        "plan": state["plan"],
        "past_steps": state["past_steps"]
    })

    if isinstance(act.action, Response):
        return {"response": act.action.response, "plan": []}

    is_replan = act.action.steps != state.get("plan", [])
    return {
        "plan": act.action.steps,
        "replans": state.get("replans", 0) + (1 if is_replan else 0)
    }

Check: Calling replan_node with replans=2 returns a final response without invoking the replanner LLM, and a replanner call that returns the unchanged remaining plan leaves replans unchanged.

Step 6: Assemble the StateGraph with conditional edges and terminal routing¶

Construct the graph using StateGraph. Add planner, executor, and replanner nodes. Route from START to planner, planner to executor, and executor to replan_node. Use a conditional edge from replan_node to decide whether to loop back to executor or terminate at END.

from langgraph.graph import END, START, StateGraph

def should_end(state: PlanExecuteState) -> str:
    if state.get("response") or not state.get("plan"):
        return END
    return "executor"

builder = StateGraph(PlanExecuteState)
builder.add_node("planner", plan_node)
builder.add_node("executor", execute_node)
builder.add_node("replan_node", replan_node)

builder.add_edge(START, "planner")
builder.add_edge("planner", "executor")
builder.add_edge("executor", "replan_node")
builder.add_conditional_edges("replan_node", should_end, ["executor", END])

graph = builder.compile()

Check: Inspecting graph.get_graph().nodes contains "planner", "executor", and "replan_node".

Step 7: Configure compilation with checkpointer and engine recursion limit¶

Compile the graph with a checkpointer for state persistence and pass a top-level recursion_limit in the invocation config. Each executed step uses two super-steps (executor and replanner) and the planner uses one, so with a maximum plan length P set recursion_limit above 1 + 2 × P × (MAX_REPLANS + 1). Enforce P on the planner and replanner output (for example a maximum list length on Plan.steps).

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
app = builder.compile(checkpointer=checkpointer)

config = {
    "configurable": {"thread_id": "plan-session-001"},
    "recursion_limit": 50
}

Check: The config dictionary contains recursion_limit at the top level and thread_id inside configurable.

Step 8: Test zero-replan, single-replan, and replan-limit exhaustion paths¶

Execute validation runs covering:

Direct execution: Initial plan of 3 steps completes sequentially without replan modifications, terminating at END.
Dynamic replanning: Step 1 observation causes the replanner to inject an additional step; replan counter increments to 1, then finishes.
Bounded limit enforcement: Mocking task failures to verify execution halts at MAX_REPLANS with exit message [Replan limit reached] and zero GraphRecursionError exceptions.

Check: All test runs produce expected state transitions and terminate within configured bounds.

Step 9: Emit plan and replan events to traces¶

Record the initial plan and every replan (previous plan, new plan, replan count) as trace events or run metadata.

from langchain_core.callbacks.manager import dispatch_custom_event

# Emit plan and replan events to LangSmith trace run tree
dispatch_custom_event(
    "replan",
    {
        "previous_plan": state.get("plan", []),
        "new_plan": act.action.steps,
        "replans": state.get("replans", 0)
    }
)

Check: The trace for the replanning test shows the initial plan and each revised plan.

Snippets to add¶

snippets/templates/planner_agent/graph.py: Complete Plan-and-Execute StateGraph implementation with bounded replanning (python)
snippets/templates/planner_agent/schemas.py: Pydantic models for Plan, Response, and Act union (python)
snippets/templates/planner_agent/test_planner.py: Automated pytest suite covering normal execution and replan limit containment (python)

6. Verify it works¶

Execute the test suite to verify planner decomposition, execution loop, and replan limit enforcement:

python -m pytest snippets/templates/planner_agent/test_planner.py -v

Expected verification checks:

Happy Path Execution:
Input: "Compare the revenue of Contoso and Fabrikam for 2025."
Path: START -> planner -> executor -> replan_node -> executor -> replan_node -> END.
Expected Output: response is populated with comparative analysis, len(past_steps) >= 2, and replans == 0.
Dynamic Replanning Path:
Input requiring follow-up query:
Path: planner -> executor -> replan_node (updates plan) -> executor -> replan_node -> END.
Expected Output: replans == 1 and final response incorporates observations from both original and revised steps.
Replan Limit Soft Containment:
Input with forced repeating step mock:
Execution halts cleanly when replans == MAX_REPLANS.
Output contains "[Replan limit reached]" in response.
Verify no GraphRecursionError is raised.
LangSmith Trace Inspection:
Inspect trace in LangSmith. Confirm that "planner", "executor", and "replan_node" executions appear as sequential spans with prompt tokens and tool outputs captured.

7. Known issues and fixes¶

Symptom	Likely cause	Fix
Planner creates dozens of tiny trivial steps	Planner prompt lacks step count bounds and granularity instructions	Limit the plan to 3–5 high-level actionable steps in the prompt; provide few-shot examples of appropriate granularity.
Executor fails on steps it cannot execute	Planner is unaware of available tool capabilities	Inject tool names, descriptions, and parameter constraints into the planner's system prompt.
Replanner repeats completed tasks in an infinite loop	Past step results were not passed back to the replanner prompt	Pass the full `past_steps` list to the replanner prompt; add prompt rules strictly forbidding re-listing completed tasks.
Unhandled `GraphRecursionError` raised by LangGraph	Engine `recursion_limit` was reached before the soft replan limit	Size `recursion_limit` with the Step 7 formula and enforce the maximum plan length.
Pydantic `ValidationError` on replanner output	LLM structured output returned invalid JSON or neither `Response` nor `Plan`	Use `with_structured_output(Act)` with strict JSON mode; add a fallback error handler or default response step.

8. Monitoring and alerts¶

Signal	Alert when	Action
Average replans per task	Average above 1 for an hour	Review planner prompt quality, tool failure rates, and task ambiguity.
Rate of tasks terminating at replan limit	Above the agreed baseline (threshold TO VERIFY)	Investigate why plans fail to converge; adjust executor tools or increase step granularity.
Engine `GraphRecursionError` count	Greater than 0 occurrences	Critical incident: investigate infinite looping edges and verify `MAX_REPLANS` soft stop is executing.
Token consumption per planned task	Above the agent's token budget per run (value TO VERIFY with the agent owner)	Audit step outputs; summarize verbose tool results before appending to `past_steps`.

9. AWS and open-source equivalents¶

AWS Equivalents¶

AWS: The same graph runs on AgentCore Runtime.

Open-Source Equivalents¶

Microsoft Agent Framework: Magentic orchestration provides a manager that plans and replans with round, stall and reset limits (Microsoft Agent Framework: Magentic orchestration).
BabyAGI / Plan-and-Solve: Open-source architectures implementing upfront task generation and sequential execution with feedback loops.

10. Validation record¶

Pending. Planned dry-run scenario: A second engineer builds the plan-and-execute graph from the code snippets in a sandbox environment and executes the three verification test cases (happy path, dynamic replanning, and replan limit containment). The validation passes when all three tests complete with expected outputs, zero unhandled GraphRecursionError exceptions occur, and LangSmith traces confirm clean spans for planner, executor, and replanner nodes.