AI Agent Planning Patterns

Real-World Scenario

A research agent is asked to “write a competitive analysis of the top 3 CRM vendors.” Without planning, it calls tools randomly and produces an uneven report. With a planning step, it first decomposes the task: identify the top 3 vendors, research each one’s pricing/features/reviews, compare them, write the report. The resulting document is complete and structured.

Plan-Then-Execute Pattern

import anthropic
import json

client = anthropic.Anthropic()

PLANNING_TOOLS = [
    {
        "name": "create_plan",
        "description": "Create a step-by-step execution plan to accomplish the user's goal. Call this before taking any action.",
        "input_schema": {
            "type": "object",
            "properties": {
                "goal":  {"type": "string", "description": "The overall goal to accomplish"},
                "steps": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {
                            "step_id":     {"type": "integer"},
                            "description": {"type": "string"},
                            "tool":        {"type": "string", "description": "Which tool to use"},
                            "depends_on":  {"type": "array", "items": {"type": "integer"}},
                        },
                        "required": ["step_id", "description"]
                    }
                }
            },
            "required": ["goal", "steps"]
        }
    },
    {
        "name": "web_search",
        "description": "Search the web for information.",
        "input_schema": {
            "type": "object",
            "properties": {
                "query":  {"type": "string"},
                "detail": {"type": "string", "enum": ["brief", "detailed"], "default": "brief"},
            },
            "required": ["query"]
        }
    },
    {
        "name": "write_section",
        "description": "Write a section of the final document.",
        "input_schema": {
            "type": "object",
            "properties": {
                "section_title": {"type": "string"},
                "content":       {"type": "string"},
            },
            "required": ["section_title", "content"]
        }
    },
]

def mock_tool(name: str, inputs: dict) -> str:
    """Simulate tool execution."""
    if name == "create_plan":
        steps = "\n".join(f"  Step {s['step_id']}: {s['description']}" for s in inputs["steps"])
        return f"Plan created with {len(inputs['steps'])} steps:\n{steps}"
    if name == "web_search":
        return f"[Search results for: {inputs['query']}] Found 5 relevant articles about {inputs['query']}."
    if name == "write_section":
        return f"Section '{inputs['section_title']}' written ({len(inputs['content'])} chars)."
    return "Unknown tool"


class PlanningAgent:
    def __init__(self, system: str, tools: list[dict], max_iterations: int = 20):
        self.system         = system
        self.tools          = tools
        self.max_iterations = max_iterations
        self.plan           = None
        self.completed_steps: set[int] = set()

    def run(self, task: str) -> str:
        messages = [{"role": "user", "content": task}]
        iterations = 0

        while iterations < self.max_iterations:
            iterations += 1
            response = client.messages.create(
                model="claude-sonnet-4-6",
                max_tokens=2048,
                system=self.system,
                tools=self.tools,
                messages=messages,
            )

            # Extract text output
            text_blocks = [b.text for b in response.content if hasattr(b, "text")]
            if text_blocks:
                print(f"[Agent] {text_blocks[0][:200]}")

            if response.stop_reason == "end_turn":
                return text_blocks[0] if text_blocks else "Task complete."

            # Process tool calls
            messages.append({"role": "assistant", "content": response.content})
            tool_results = []

            for block in response.content:
                if block.type != "tool_use":
                    continue

                print(f"[Tool] {block.name}({json.dumps(block.input)[:100]}...)")
                result = mock_tool(block.name, block.input)

                # Track the plan when created
                if block.name == "create_plan":
                    self.plan = block.input.get("steps", [])

                tool_results.append({
                    "type":        "tool_result",
                    "tool_use_id": block.id,
                    "content":     result,
                })

            messages.append({"role": "user", "content": tool_results})

        return "Max iterations reached."


SYSTEM = """You are a research agent. When given a complex task:
1. ALWAYS start by calling create_plan to break the task into clear steps
2. Execute each step in order using the available tools
3. Track your progress and mark steps as complete
4. After all steps are done, write a final summary

Be systematic and thorough. Do not skip the planning step."""

agent = PlanningAgent(SYSTEM, PLANNING_TOOLS)
result = agent.run(
    "Write a brief competitive analysis comparing the top 3 cloud database services."
)

Tree-of-Thought: Exploring Multiple Paths

import anthropic
import json
from dataclasses import dataclass, field

client = anthropic.Anthropic()

@dataclass
class ThoughtNode:
    thought:  str
    score:    float = 0.0
    children: list["ThoughtNode"] = field(default_factory=list)
    is_terminal: bool = False

def generate_thoughts(problem: str, context: str, n: int = 3) -> list[str]:
    """Generate N candidate next thoughts for a problem."""
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": f"""Problem: {problem}

Current reasoning: {context if context else "None yet"}

Generate {n} different next reasoning steps. Each should explore a different approach.
Return as JSON: {{"thoughts": ["thought1", "thought2", "thought3"]}}"""
        }]
    )
    text = response.content[0].text
    try:
        start = text.index("{")
        data  = json.loads(text[start:text.rindex("}") + 1])
        return data.get("thoughts", [])
    except (ValueError, json.JSONDecodeError):
        return [text]

def score_thought(problem: str, thought_chain: str) -> float:
    """Score a reasoning chain on a 0-1 scale."""
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=64,
        messages=[{
            "role": "user",
            "content": f"""Rate this reasoning chain for solving the problem.
Problem: {problem}
Reasoning: {thought_chain}

Score 0.0 (wrong/irrelevant) to 1.0 (correct/complete). Reply with only a number."""
        }]
    )
    try:
        return float(response.content[0].text.strip())
    except ValueError:
        return 0.5

def tree_of_thought(
    problem: str,
    depth: int = 3,
    breadth: int = 3,
) -> tuple[str, float]:
    """
    Explore a tree of reasoning paths and return the best chain.
    depth:   how many reasoning steps to explore
    breadth: how many alternative thoughts per step
    """
    # BFS through the thought tree
    current_paths = [("", 1.0)]   # (thought_chain, score)

    for step in range(depth):
        next_paths = []
        for chain, _ in current_paths[:breadth]:
            thoughts = generate_thoughts(problem, chain, n=breadth)
            for thought in thoughts:
                new_chain = f"{chain}\nStep {step+1}: {thought}" if chain else f"Step {step+1}: {thought}"
                score     = score_thought(problem, new_chain)
                next_paths.append((new_chain, score))

        # Keep only the top `breadth` paths
        next_paths.sort(key=lambda x: x[1], reverse=True)
        current_paths = next_paths[:breadth]
        print(f"Step {step+1}: best score = {current_paths[0][1]:.2f}")

    best_chain, best_score = current_paths[0]
    return best_chain, best_score


# Example usage
problem = "A train leaves Chicago at 9am going 60mph. Another leaves NYC at 10am going 80mph. The cities are 790 miles apart. Where do they meet?"
best_reasoning, score = tree_of_thought(problem, depth=3, breadth=3)
print(f"\nBest reasoning chain (score={score:.2f}):\n{best_reasoning}")

Self-Correcting Agent

import anthropic

client = anthropic.Anthropic()

def verify_output(task: str, output: str) -> dict:
    """Ask a critic model to verify and critique the agent's output."""
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": f"""Critically evaluate this output for the given task.

Task: {task}
Output: {output}

Check for:
- Completeness: does it fully address the task?
- Accuracy: are there factual errors or logical flaws?
- Quality: is it well-structured and clear?

Respond with JSON: {{"passes": true/false, "issues": ["issue1", ...], "suggestions": ["fix1", ...]}}"""
        }]
    )
    text = response.content[0].text
    try:
        start = text.index("{")
        return __import__("json").loads(text[start:text.rindex("}") + 1])
    except Exception:
        return {"passes": True, "issues": [], "suggestions": []}


def self_correcting_agent(task: str, max_revisions: int = 3) -> str:
    """Generate output, critique it, revise until it passes or hits max revisions."""
    messages = [{"role": "user", "content": task}]

    for revision in range(max_revisions + 1):
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            messages=messages,
        )
        output = response.content[0].text

        if revision == max_revisions:
            print(f"Max revisions reached. Using revision {revision}.")
            return output

        critique = verify_output(task, output)
        print(f"Revision {revision}: passes={critique['passes']}, issues={len(critique['issues'])}")

        if critique["passes"]:
            return output

        # Build correction prompt
        issues_text = "\n".join(f"- {issue}" for issue in critique["issues"])
        suggestions  = "\n".join(f"- {s}" for s in critique["suggestions"])

        messages.append({"role": "assistant", "content": output})
        messages.append({
            "role": "user",
            "content": f"""Your response has the following issues:
{issues_text}

Suggestions for improvement:
{suggestions}

Please revise your response to address these issues."""
        })

    return output


result = self_correcting_agent(
    "Write a 3-step guide for setting up a production PostgreSQL database with connection pooling."
)
print(result)