Decision-Making
& Planning

cd agent-lab && source .venv/bin/activate
touch plan_execute_agent.py

import os, json, random
import anthropic
from dotenv import load_dotenv

load_dotenv()
client = anthropic.Anthropic()

TOOLS = [
    {
        "name": "web_search",
        "description": "Search the web. Returns a short text result.",
        "input_schema": {
            "type": "object",
            "properties": {"query": {"type": "string"}},
            "required": ["query"]
        }
    },
    {
        "name": "calculator",
        "description": "Evaluate a Python arithmetic expression. Returns the result as a string.",
        "input_schema": {
            "type": "object",
            "properties": {
                "expression": {"type": "string", "description": "A safe arithmetic expression, e.g. '365 * 24 * 60 * 60'"}
            },
            "required": ["expression"]
        }
    },
    {
        "name": "flaky_api",
        "description": "Fetch data from an external API. Sometimes fails — use web_search as a fallback.",
        "input_schema": {
            "type": "object",
            "properties": {"endpoint": {"type": "string"}},
            "required": ["endpoint"]
        }
    },
]

SEARCH_DB = {
    "python popularity": "Python is ranked #1 in the TIOBE index as of 2024–2026.",
    "tiobe index": "TIOBE index ranks Python #1, C #2, C++ #3, Java #4 (2024–2026).",
    "javascript popularity": "JavaScript dominates web development, ranked top 5 in all major surveys.",
}

def execute_tool(name: str, tool_input: dict) -> str:
    try:
        if name == "web_search":
            q = tool_input["query"].lower()
            for key, val in SEARCH_DB.items():
                if key in q:
                    return val
            return f'No results for "{tool_input["query"]}".'
        if name == "calculator":
            result = eval(tool_input["expression"], {"__builtins__": {}})
            return str(result)
        if name == "flaky_api":
            if random.random() < 0.5:
                raise ConnectionError("API timeout after 30s")
            return f"API response for {tool_input['endpoint']}: status=200, data=42"
        return f"Unknown tool: {name}"
    except Exception as e:
        # Always return errors as strings — never let them propagate
        return f"Tool error: {type(e).__name__}: {e}"

PLANNER_SYSTEM = """You are a planning assistant.
Given a user goal, produce a step-by-step plan as a JSON array.
Each step: {"step": integer, "description": string, "tool_hint": string}
tool_hint must be one of: web_search, calculator, flaky_api, none
Return ONLY the JSON array, no other text."""


def generate_plan(goal: str, context: str = "") -> list[dict]:
    prompt = f"Goal: {goal}"
    if context:
        prompt += f"\n\nContext (work done and failures so far):\n{context}"
        prompt += "\n\nRevise the remaining steps given this context."

    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=512,
        system=PLANNER_SYSTEM,
        messages=[{"role": "user", "content": prompt}]
    )
    raw = response.content[0].text.strip()
    try:
        return json.loads(raw)
    except json.JSONDecodeError:
        start, end = raw.find("["), raw.rfind("]") + 1
        return json.loads(raw[start:end])

EXECUTOR_SYSTEM = """You are an execution assistant.
Complete the given step using available tools.
When done, summarise what you did and what you found in 1-2 sentences."""

MAX_REPLAN = 2


def execute_step(step: dict) -> tuple[str, bool]:
    """Execute one plan step. Returns (summary, success)."""
    messages = [{"role": "user", "content": f"Complete this step: {step['description']}"}]

    for _ in range(5):
        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=512,
            system=EXECUTOR_SYSTEM,
            tools=TOOLS,
            messages=messages
        )
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason == "end_turn":
            summary = "".join(b.text for b in response.content if hasattr(b, "text"))
            failed = "error" in summary.lower() or "failed" in summary.lower()
            return summary, not failed

        if response.stop_reason == "tool_use":
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    result = execute_tool(block.name, block.input)
                    print(f"    tool: {block.name}({block.input}) → {result[:80]}")
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result
                    })
            messages.append({"role": "user", "content": tool_results})

    return "Step did not complete within iteration budget.", False


def run_plan_execute(goal: str) -> None:
    print(f"\n{'='*60}\nGOAL: {goal}\n{'='*60}")

    plan = generate_plan(goal)
    replan_count = 0
    completed: list[str] = []

    while plan and replan_count <= MAX_REPLAN:
        step = plan.pop(0)
        print(f"\n[step {step['step']}] {step['description']}")

        summary, success = execute_step(step)
        print(f"  → {summary}")

        if success:
            completed.append(f"Step {step['step']}: {summary}")
        else:
            print(f"\n  ⚠ Step failed. Replanning... ({replan_count + 1}/{MAX_REPLAN})")
            context = "\n".join(completed) + f"\nFailed: {step['description']}\nReason: {summary}"
            plan = generate_plan(goal, context=context)
            replan_count += 1
            print(f"  Revised plan has {len(plan)} remaining steps.")

    print(f"\n{'='*60}\nCOMPLETED STEPS:")
    for c in completed:
        print(f"  ✓ {c}")
    print(f"{'='*60}")


if __name__ == "__main__":
    run_plan_execute(
        "Research the most popular programming language in 2024–2026, "
        "then calculate how many seconds are in a year."
    )

python plan_execute_agent.py

============================================================
GOAL: Research the most popular programming language ...
============================================================

[step 1] Search for the most popular programming language 2024-2026
    tool: web_search({'query': 'python popularity'}) → Python is ranked #1...
  → Found that Python is ranked #1 by the TIOBE index for 2024–2026.

[step 2] Fetch supporting data from the TIOBE API
    tool: flaky_api({'endpoint': 'tiobe/rankings'}) → Tool error: ConnectionError: API timeout
  → Step failed due to API timeout error.

  ⚠ Step failed. Replanning... (1/2)
  Revised plan has 2 remaining steps.

[step 2] Search for TIOBE index rankings as fallback
    tool: web_search({'query': 'tiobe index'}) → TIOBE index ranks Python #1, C #2...
  → Confirmed via web search: Python #1, C #2, C++ #3.

[step 3] Calculate seconds in a year
    tool: calculator({'expression': '365 * 24 * 60 * 60'}) → 31536000
  → There are 31,536,000 seconds in a year.

============================================================
COMPLETED STEPS:
  ✓ Step 1: Python is ranked #1 by TIOBE for 2024–2026.
  ✓ Step 2: Confirmed Python #1 via web search fallback.
  ✓ Step 3: 31,536,000 seconds in a year.
============================================================

# Print plan and request approval before execution
print("\nProposed plan:")
for s in plan:
    print(f"  {s['step']}. {s['description']} [tool: {s['tool_hint']}]")

approval = input("\nApprove this plan? (yes/no): ").strip().lower()
if approval != "yes":
    print("Plan rejected. Exiting.")
    return