Action: web_search("Python release year")
Action: web_search("Java release year")
Action: web_search("Go release year")
Action: calculator("...")
(occasionally repeats a search or takes extra steps)

Task
 │
 ▼
[Plan Node]     ← LLM generates 3-7 step plan (no execution, just planning)
 │
 ▼
[Execute Node]  ← Execute current step (embedded ReAct, can call tools)
 │
 ├─ Step failed? ─→ [Replan Node] ← Re-plan remaining steps based on progress so far
 │                      │
 │                      └──────────────┐
 │                                     ▼
 ├─ More steps? ─→ back to Execute    Execute (continue)
 │
 └─ All done? ─→ [Finalize Node] ← Output final answer
                       │
                       ▼
                      END

from typing import TypedDict

class PlanSolveState(TypedDict):
    task: str                    # original user task
    plan: list[str]              # current plan (list of steps)
    completed_steps: list[str]   # completed steps with result summaries
    current_step_index: int      # which step we're on (0-based)
    step_result: str             # result of the current step
    replan_count: int            # how many times we've replanned
    final_answer: str            # the final answer

def plan_node(state: PlanSolveState) -> dict:
    messages = [
        SystemMessage(content=PLANNER_SYSTEM),  # planner expert prompt
        HumanMessage(content=f"Task: {state['task']}"),
    ]
    response = llm.invoke(messages)
    plan = parse_plan(response.content)  # parse "1. xxx\n2. xxx" format

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

PLANNER_SYSTEM = """You are a task planning expert.
Rules:
1. Break the task into 3-7 independent steps
2. Each step must be concrete and actionable
3. Steps must have clear dependencies (later steps can use earlier results)
4. The final step should be "synthesize all information and deliver the answer"

Output format (only the step list, nothing else):
1. [step description]
2. [step description]
...
"""

def execute_node(state: PlanSolveState) -> dict:
    idx = state["current_step_index"]
    current_step = state["plan"][idx]

    # Build execution context (includes results from completed steps)
    system_prompt = EXECUTOR_SYSTEM.format(
        completed_steps=format_completed_steps(state["completed_steps"]),
        current_step=current_step,
    )

    # Use a ReAct sub-agent to execute a single step (may need tools)
    sub_agent = create_react_agent(model=llm, tools=[calculator, web_search])
    result = sub_agent.invoke(
        {"messages": [
            SystemMessage(content=system_prompt),
            HumanMessage(content=f"Execute this step: {current_step}"),
        ]},
        config={"recursion_limit": 8},
    )

    step_result = result["messages"][-1].content
    new_completed = state["completed_steps"] + [
        f"{current_step} → {step_result[:100]}"
    ]

    return {
        "step_result": step_result,
        "completed_steps": new_completed,
        "current_step_index": idx + 1,
    }

MAX_REPLAN = 2

def should_continue(state) -> Literal["execute", "replan", "finalize"]:
    idx = state["current_step_index"]
    total = len(state["plan"])

    if idx >= total:
        return "finalize"  # all steps complete

    # detect step failure
    result = state.get("step_result", "")
    failed = any(kw in result for kw in ["Calculation error", "Search failed", "Error"])

    if failed and state["replan_count"] < MAX_REPLAN:
        return "replan"  # failed, still have retry budget

    return "execute"  # keep going

from langgraph.graph import END, START, StateGraph

graph = StateGraph(PlanSolveState)

graph.add_node("plan", plan_node)
graph.add_node("execute", execute_node)
graph.add_node("replan", replan_node)
graph.add_node("finalize", finalize_node)

graph.add_edge(START, "plan")
graph.add_edge("plan", "execute")
graph.add_conditional_edges(
    "execute",
    should_continue,
    {"execute": "execute", "replan": "replan", "finalize": "finalize"},
)
graph.add_conditional_edges(
    "replan", after_replan,
    {"execute": "execute", "finalize": "finalize"},
)
graph.add_edge("finalize", END)

agent = graph.compile()

1. Search "China population", "US population", "India population"
   to get the latest figures.
2. Record China, US, and India's population numbers.
3. Add China, US, and India's populations to get the three-country total.
4. Calculate China's population as a percentage of the three-country total.
5. Synthesize all information and deliver the final answer.

[Step 1] web_search("China population") → 1.40489 billion
         web_search("US population")    → 341 million
         web_search("India population") → 1.451 billion

[Step 2] Record results (no tool call, model consolidates)
         → China 1.40489B, US 341M, India: no data available ← ⚠️

[Step 3] calculator("14048900000.0 + 3400000000.0")
         → 17448900000 ← ⚠️ India missing!

[Step 4] calculator("14.0489 / 17.4489 * 100")
         → 80.5145%

[Final answer] Three-country total: 1.74489B, China's share: 80.5145%

[Step 6] Need to round 8836.45
  → calculator("round(8836.45)")
  → Error: unsupported AST node: Call
  → calculator("round(8836.45, 0)")
  → Error: unsupported AST node: Call
  → Result: Sorry, need more steps to process this request.

1299 USD × 6.8025 = 8836.45 CNY
Rounded to approximately 8836 CNY

1. Calculate 2 to the power of 10
2. Calculate 3 to the power of 5
3. Add the results of steps 1 and 2
4. Synthesize all information and give the final answer

Task analysis
│
├─ Fewer than 3 steps?
│   └─ Use ReAct (lightweight, fast)
│
├─ Strong dependencies between steps?
│   (later steps need precise results from earlier steps)
│   └─ Plan-and-Solve (explicit plan enforces dependency order)
│
├─ Clear task boundary, enumerable steps?
│   └─ Plan-and-Solve or even Workflow-Driven
│
├─ Open-ended task, fuzzy boundaries?
│   └─ ReAct (adapts to unknowns)
│
└─ Long-horizon planning (10+ steps)?
    └─ Consider multi-Agent architecture (later article)

# Instead of natural language summaries:
completed_steps.append(f"Search China population → {step_result[:100]}")

# Use structured data:
step_data = {
    "step_index": idx,
    "description": current_step,
    "result": step_result,          # full result, no truncation
    "extracted_values": {},         # have the model extract key numbers
}

class PlanSolveState(TypedDict):
    # ... other fields ...
    collected_data: dict[str, Any]  # dedicated storage for gathered data