ReAct & Planning

# Colab cell 1 — run once. Set your key in the 🔑 panel (name it
# ANTHROPIC_API_KEY) or just paste it when prompted.
!pip install -q anthropic

import os
try:
    from google.colab import userdata
    os.environ["ANTHROPIC_API_KEY"] = userdata.get("ANTHROPIC_API_KEY")
except Exception:
    from getpass import getpass
    os.environ.setdefault("ANTHROPIC_API_KEY", getpass("Anthropic API key: "))

import re
import anthropic

client = anthropic.Anthropic()
MODEL = "claude-sonnet-5"

# A tiny in-memory knowledge base so the actions do real work — no setup,
# no files. search[] finds titles; lookup[] reads one article in full.
KB = {
    "ReAct": "ReAct interleaves Thought, Action, and Observation steps. "
             "Reason-only models hallucinate facts; act-only models make "
             "impulsive, unrecoverable moves. Interleaving the two beats both.",
    "Agent loop": "An agent is an LLM calling tools in a loop where the model — "
                  "not your code — chooses the next action each iteration.",
    "Exponential backoff": "On 429/5xx, retry with wait 2**attempt seconds plus "
                           "jitter, capped at 60s; give up after 5 tries.",
}

def search(query: str) -> str:
    words = [w for w in query.lower().split() if len(w) > 2]
    hits = [t for t in KB if any(w in t.lower() for w in words)]
    # on a miss, name what DOES exist — an empty dead-end invites the model
    # to stop searching and answer from memory (ungrounded). See the loop below.
    return "Titles: " + ", ".join(hits) if hits else (
        "No match. The KB contains: " + ", ".join(KB))

def lookup(title: str) -> str:
    return KB.get(title, f"No article titled {title!r}.")

REACT_PROMPT = """Answer the question by interleaving Thought, Action, and
Observation steps.

Available actions:
  search[query]   - search the knowledge base
  lookup[title]   - read a full article
  finish[answer]  - give the final answer

Respond with exactly one Thought and one Action, then STOP:
Thought: <reasoning about what to do next>
Action: <one action>

I will run the action and reply with:
Observation: <result>

Question: {question}"""

def react_step(messages) -> tuple[str, str]:
    resp = client.messages.create(
        model=MODEL, max_tokens=512,
        stop_sequences=["Observation:"],   # forbid hallucinating results
        messages=messages,
    )
    text = resp.content[0].text
    match = re.search(r"Action: *(\w+)\[(.*)\]", text)
    if match is None:
        raise ValueError("model broke the ReAct format:\n" + text)
    return match.group(1), match.group(2)   # e.g. ("search", "ReAct")

def run_react(question: str, max_steps: int = 6) -> str:
    messages = [{"role": "user", "content": REACT_PROMPT.format(question=question)}]
    for _ in range(max_steps):
        action, arg = react_step(messages)
        arg = arg.strip().strip("'").strip('"')   # models love quoting args
        print(f"Action: {action}[{arg}]")
        if action == "finish":
            return arg
        obs = search(arg) if action == "search" else lookup(arg)
        print(f"Observation: {obs}")
        # feed the action + real observation back as the next turn of context
        messages.append({"role": "assistant", "content": f"Action: {action}[{arg}]"})
        messages.append({"role": "user", "content": f"Observation: {obs}"})
    raise RuntimeError("max steps exceeded")

print(run_react("What problem does the ReAct pattern solve?"))

# Colab cell 2 — run the ReAct cell above first (it installs the SDK and
# creates client and MODEL). This cell swaps the fake KB for a fake repo.

# A tiny in-memory "repo" so list_dir/grep/read_file do real work.
REPO = {
    "README.md": "Sample service. Retry policy is configured in config/app.yaml.",
    "config/app.yaml": "retries: 5\nbackoff: exponential\ntimeout_s: 30\n",
    "src/retry.py": "def backoff(attempt):\n    return min(2 ** attempt, 60)\n",
}

def list_dir(path: str = "") -> str:
    hits = [p for p in REPO if p.startswith(path)]
    return "\n".join(sorted(hits)) if hits else f"Nothing under {path!r}."

def grep(pattern: str) -> str:
    hits = [f"{p}: {line}" for p, body in REPO.items()
            for line in body.splitlines() if pattern.lower() in line.lower()]
    return "\n".join(hits) if hits else f"No lines match {pattern!r}."

def read_file(path: str) -> str:
    return REPO.get(path, f"No file at {path!r}.")

WORK_IMPL = {"list_dir": list_dir, "grep": grep, "read_file": read_file}
WORK_TOOLS = [
    {"name": "list_dir", "description": "List repo paths under a prefix.",
     "input_schema": {"type": "object",
                      "properties": {"path": {"type": "string"}}, "required": []}},
    {"name": "grep", "description": "Find lines matching a substring.",
     "input_schema": {"type": "object",
                      "properties": {"pattern": {"type": "string"}},
                      "required": ["pattern"]}},
    {"name": "read_file", "description": "Read one file in full, by path.",
     "input_schema": {"type": "object",
                      "properties": {"path": {"type": "string"}},
                      "required": ["path"]}},
]

PLAN_TOOL = {
    "name": "submit_plan",
    "description": "Record a step-by-step plan before doing any work.",
    "input_schema": {
        "type": "object",
        "properties": {
            "steps": {"type": "array", "items": {"type": "string"},
                      "minItems": 1, "maxItems": 6},
        },
        "required": ["steps"],
    },
}

def make_plan(question: str) -> list[str]:
    resp = client.messages.create(
        model=MODEL, max_tokens=1024,
        tools=[PLAN_TOOL],
        tool_choice={"type": "tool", "name": "submit_plan"},  # forced
        messages=[{"role": "user", "content":
            "Plan how to answer this question using list_dir, grep and "
            "read_file tools. At most 6 concrete steps.\n\n"
            "Question: " + question}],
    )
    block = next(b for b in resp.content if b.type == "tool_use")
    return block.input["steps"]

def run_with_plan(question: str, max_iterations: int = 12) -> str:
    plan = make_plan(question)
    plan_text = "\n".join(f"{i + 1}. {s}" for i, s in enumerate(plan))
    print("Initial plan:\n" + plan_text + "\n")
    task = (
        f"Question: {question}\n\nYour plan:\n{plan_text}\n\n"
        "Follow the plan, but treat it as a hypothesis. If an observation "
        "contradicts a step, do NOT push on: call submit_plan again with a "
        "revised plan, then continue."
    )
    messages = [{"role": "user", "content": task}]
    revisions = 0
    for _ in range(max_iterations):
        resp = client.messages.create(
            model=MODEL, max_tokens=1024,
            tools=[PLAN_TOOL] + WORK_TOOLS, messages=messages,
        )
        if resp.stop_reason != "tool_use":
            return next(b.text for b in resp.content if b.type == "text")
        messages.append({"role": "assistant", "content": resp.content})
        results = []
        for block in resp.content:
            if block.type != "tool_use":
                continue
            if block.name == "submit_plan":          # re-plan escape hatch
                revisions += 1
                revised = "\n".join(f"{i + 1}. {s}"
                                     for i, s in enumerate(block.input["steps"]))
                print(f"Re-plan #{revisions}:\n{revised}\n")
                output = "Plan updated."
            else:
                output = WORK_IMPL[block.name](**block.input)
                print(f"{block.name}({block.input}) -> {output[:60]!r}")
            results.append({"type": "tool_result",
                            "tool_use_id": block.id, "content": output})
        messages.append({"role": "user", "content": results})
    raise RuntimeError("max iterations exceeded")

print(run_with_plan("How does this service configure retries?"))