Agent Planning
Planning is how agents decide what to do and in what order. Without planning, agents react impulsively. With planning, they break down complex tasks, execute step by step, and self-correct. This covers the major planning strategies.
Table of Contents
Installation / Setup
pip install langchain-openai langgraph
Core Syntax
Planning strategies comparison
| Strategy |
How it works |
Best for |
| ReAct |
Think → Act → Observe loop |
Most tasks (default) |
| Plan-and-execute |
Plan all steps first, then execute |
Complex multi-step tasks |
| Tree-of-thought |
Explore multiple reasoning paths |
Ambiguous problems |
| Reflection |
Generate → critique → improve |
High-quality output |
| Reflexion |
Try → fail → reflect → retry |
Learning from failures |
ReAct (default strategy)
graph TD
A[Thought: I need to X] --> B[Action: tool_call]
B --> C[Observation: result]
C --> D{Goal achieved?}
D -->|No| A
D -->|Yes| E[Final Answer]
# ReAct is the default in LangGraph's create_react_agent
from langgraph.prebuilt import create_react_agent
agent = create_react_agent(llm, tools)
result = agent.invoke({"messages": [("human", "Research and summarize X")]})
Plan-and-execute
graph LR
A[Goal] --> B[Plan: Step 1, 2, 3]
B --> C[Execute Step 1]
C --> D[Execute Step 2]
D --> E[Execute Step 3]
E --> F[Final Answer]
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o")
def plan_and_execute(goal, tools):
# step 1: create a plan
plan_prompt = f"""Create a step-by-step plan for: {goal}
Return as numbered list."""
plan = llm.invoke(plan_prompt).content
steps = [s.strip() for s in plan.split("\n") if s.strip()]
# step 2: execute each step
results = []
for step in steps:
result = llm.invoke(f"Execute this step: {step}\n\nPrevious results: {results}").content
results.append({"step": step, "result": result})
# step 3: synthesize
summary = llm.invoke(
f"Goal: {goal}\n\nResults:\n{results}\n\nSynthesize a final answer."
).content
return summary
Tree-of-thought
graph TD
A[Problem] --> B[Thought 1]
A --> C[Thought 2]
A --> D[Thought 3]
B --> E[Eval: 0.8]
C --> F[Eval: 0.3]
D --> G[Eval: 0.9]
G --> H[Continue from best path]
def tree_of_thought(problem, branches=3, depth=2):
def evaluate_thought(thought):
return llm.invoke(f"Rate this reasoning 0-1:\n{thought}").content
# generate multiple thoughts at each level
current = [problem]
for level in range(depth):
candidates = []
for path in current:
thoughts = llm.invoke(
f"Generate {branches} different approaches to:\n{path}"
).content.split("\n\n")
for t in thoughts:
score = float(evaluate_thought(t).split()[0])
candidates.append((score, t))
# keep top paths
current = [t for _, t in sorted(candidates, reverse=True)[:branches]]
return current[0]
Reflection
def reflect_and_improve(task, iterations=3):
draft = llm.invoke(f"Complete this task: {task}").content
for i in range(iterations):
critique = llm.invoke(
f"Critically review this output. What's wrong?\n\n{draft}"
).content
draft = llm.invoke(
f"Improve based on feedback:\n\nFeedback: {critique}\n\nDraft: {draft}"
).content
return draft
Common Operations
Task decomposition
def decompose(task, llm):
"""Break a complex task into subtasks."""
response = llm.invoke(
f"Break this into 3-5 concrete subtasks:\n{task}\n\nReturn as numbered list."
)
return [s.strip().lstrip("0123456789. ") for s in response.content.split("\n") if s.strip()]
Adaptive planning (switch strategies based on complexity)
def adaptive_plan(task, llm):
complexity = llm.invoke(
f"Rate this task's complexity 1-5 (1=simple, 5=very complex):\n{task}"
).content
if int(complexity[0]) <= 2:
return "react" # simple: just do it
elif int(complexity[0]) <= 4:
return "plan-execute" # medium: plan first
else:
return "tree-of-thought" # complex: explore alternatives
Self-correction loop
def self_correct(task, llm, tools, max_attempts=3):
for attempt in range(max_attempts):
result = agent.invoke(task)
# check if result is satisfactory
check = llm.invoke(
f"Is this result correct and complete?\nTask: {task}\nResult: {result}\nAnswer yes/no."
).content
if "yes" in check.lower():
return result
# if not, add feedback and retry
task = f"{task}\n\nPrevious attempt failed: {result}\nTry again."
return result # return last attempt
Practical Examples
Example 1 — Plan-and-execute research agent
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o")
def research_agent(topic):
# plan
plan = llm.invoke(f"""Create a research plan for: {topic}
Return 4-5 numbered steps.""").content
steps = [s.strip() for s in plan.split("\n") if s.strip()]
findings = []
# execute each step
for step in steps:
finding = llm.invoke(
f"Execute this research step: {step}\n\nTopic: {topic}"
).content
findings.append(finding)
# synthesize
report = llm.invoke(
f"Write a report on {topic} based on these findings:\n\n" +
"\n\n".join(f"{i+1}. {f}" for i, f in enumerate(findings))
).content
return report
Example 2 — Reflection for code quality
def write_quality_code(spec, llm, iterations=2):
code = llm.invoke(f"Write Python code for: {spec}").content
for i in range(iterations):
review = llm.invoke(
f"""Review this code for bugs, style, and efficiency:
{code}
Provide specific improvements."""
).content
code = llm.invoke(
f"""Improve this code based on review:
Review: {review}
Current code:
{code}
Return the improved code."""
).content
return code
Common Patterns
Pattern: planning with rollback
def plan_with_rollback(goal, llm, max_steps=5):
plan = decompose(goal, llm)
results = []
for i, step in enumerate(plan[:max_steps]):
try:
result = execute_step(step, results)
results.append(result)
except Exception as e:
# rollback: remove last result, revise plan
results.pop()
plan = llm.invoke(
f"Step '{step}' failed: {e}\nRevise the remaining plan."
).content
continue
return synthesize(goal, results, llm)
Pattern: parallel planning
from concurrent.futures import ThreadPoolExecutor
def parallel_plan(task, llm, num_approaches=3):
"""Generate and evaluate multiple approaches in parallel."""
approaches = llm.invoke(
f"Generate {num_approaches} different approaches to: {task}"
).content.split("\n\n")
with ThreadPoolExecutor(max_workers=num_approaches) as pool:
scores = list(pool.map(
lambda a: float(llm.invoke(f"Rate approach 0-1:\n{a}").content.split()[0]),
approaches
))
best = max(zip(scores, approaches), key=lambda x: x[0])
return best[1]
Best Practices
- ReAct is the default — use it for most tasks; it's simple and effective.
- Plan-and-execute for complex tasks — planning upfront saves wasted steps.
- Reflection for quality-critical output — 2–3 iterations usually suffice.
- Decompose first — break complex goals into concrete subtasks.
- Limit plan depth — 5–10 steps max; deeper plans lose coherence.
- Adapt strategy to complexity — don't use tree-of-thought for simple queries.
- Log plans — they help debug when agents go wrong.
Common Mistakes
| Mistake |
Fix |
| Over-planning simple tasks |
Use ReAct for straightforward queries |
| Plans too long (>10 steps) |
Decompose into phases; plan within phases |
| No reflection step |
Add critique-improve cycles for quality |
| Ignoring intermediate results |
Pass previous results to each step |
| Rigid plans |
Allow plan revision when steps fail |
Troubleshooting
| Symptom |
Cause / Fix |
| Agent doesn't follow plan |
Plan too vague — make steps concrete |
| Steps produce wrong results |
Add validation after each step |
| Plan is too complex |
Break into smaller phases |
| Reflection doesn't improve output |
Critique is too generic — be specific |
Official Documentation
Last Updated
2026-09-06 — Initial version.