Agent Design Patterns

# react_loop.py - a bounded ReAct loop on the raw Anthropic SDK

import anthropic

client = anthropic.Anthropic()

# --- Tool definition (what the model is allowed to call) ---
tools = [{
    "name": "search_docs",
    "description": "Search internal documentation for a query.",
    "input_schema": {
        "type": "object",
        "properties": {"query": {"type": "string"}},
        "required": ["query"],
    },
}]

def search_docs(query: str) -> str:
    # Real implementation would hit a vector store or search API
    return f"[results for '{query}': the deploy step runs 'make release']"

# --- The ReAct loop: reason -> act -> observe -> repeat ---
messages = [{"role": "user", "content": "What command cuts a release?"}]
MAX_STEPS = 6  # a loop without a budget is an anti-pattern (section 8)

for step in range(1, MAX_STEPS + 1):
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=1024,
        tools=tools,
        messages=messages,
    )
    messages.append({"role": "assistant", "content": response.content})

    # No tool requested -> the model has its final answer
    if response.stop_reason != "tool_use":
        answer = "".join(b.text for b in response.content if b.type == "text")
        print(f"step {step}: answering")
        print(answer)
        break

    # Execute every tool call the model asked for, feed the results back
    # in a single user message: the model may request several at once.
    tool_results = []
    for block in response.content:
        if block.type == "tool_use":
            print(f"step {step}: {block.name}({block.input['query']!r})")
            result = search_docs(**block.input)
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": result,
            })
    messages.append({"role": "user", "content": tool_results})
else:
    raise RuntimeError(f"agent did not finish within {MAX_STEPS} steps")

# reflection.py - draft, critique, revise, with the rounds capped

import anthropic


def text_of(message) -> str:
    """Pull the text out of a response.

    content is a list of typed blocks, and on a thinking model the first one
    is a ThinkingBlock, so select by type rather than by index.
    """
    return next(b.text for b in message.content if b.type == "text")


def reflect_and_improve(client, task: str, max_rounds: int = 2) -> str:
    """Generate a draft, then critique and revise it until approved."""

    draft = text_of(client.messages.create(
        model="claude-sonnet-5", max_tokens=1024,
        messages=[{"role": "user", "content": task}],
    ))

    for round_no in range(1, max_rounds + 1):
        critique = text_of(client.messages.create(
            model="claude-sonnet-5", max_tokens=1024,
            messages=[{"role": "user", "content": (
                "Critique the draft against the task. List concrete, "
                "actionable flaws only. If it is genuinely good, reply "
                f"exactly 'APPROVED'.\n\nTask: {task}\n\nDraft:\n{draft}"
            )}],
        ))

        if "APPROVED" in critique:
            print(f"round {round_no}: approved")
            break
        print(f"round {round_no}: revising")

        draft = text_of(client.messages.create(
            model="claude-sonnet-5", max_tokens=1024,
            messages=[{"role": "user", "content": (
                "Revise the draft to fix every point in the critique. "
                f"Return only the improved draft.\n\nCritique:\n{critique}"
                f"\n\nDraft:\n{draft}"
            )}],
        ))

    return draft


if __name__ == "__main__":
    client = anthropic.Anthropic()
    print(reflect_and_improve(client, "Write a one-sentence release note for a "
                              "--dry-run flag that prints planned changes "
                              "without applying them."))

# planner_executor.py - Sonnet decomposes the goal, Haiku executes each step

import anthropic
from pydantic import BaseModel

client = anthropic.Anthropic()


class Plan(BaseModel):
    steps: list[str]


def make_plan(goal: str) -> list[str]:
    """Planner: decompose the goal into ordered steps, typed rather than prose."""

    response = client.messages.parse(
        model="claude-sonnet-5",
        max_tokens=600,
        system=("Decompose the goal into 3 to 5 ordered steps. Each step is one "
                "short sentence naming a single concrete action."),
        messages=[{"role": "user", "content": goal}],
        output_format=Plan,
    )
    return response.parsed_output.steps


def execute(goal: str, step: str, done: list[str]) -> str:
    """Executor: carry out one step, with the goal and prior results in view."""

    so_far = "\n".join(f"- {d}" for d in done) or "(nothing yet)"
    reply = client.messages.create(
        model="claude-haiku-4-5",  # execution is the cheap half of the pattern
        max_tokens=200,
        system=("Carry out exactly the step you are given. Reply with one sentence "
                "of at most 15 words, no headings and no lists."),
        messages=[{"role": "user", "content": (
            f"Goal: {goal}\n\nDone so far:\n{so_far}\n\nYour step: {step}"
        )}],
    )
    return next(b.text for b in reply.content if b.type == "text")


if __name__ == "__main__":
    goal = "Prepare a release checklist for a small Python CLI tool."
    steps = make_plan(goal)

    # The plan is data, not prose. This is where you would print it for review,
    # edit the list, or require a sign-off before anything actually runs.
    for i, step in enumerate(steps, 1):
        print(f"plan {i}. {step}")

    results: list[str] = []
    for i, step in enumerate(steps, 1):
        results.append(execute(goal, step, results))
        print(f"done {i}. {results[-1]}")

# router.py - a Haiku classifier in front of three specialized handlers

import anthropic

# Each handler is a mini-agent of its own: different model, prompt, and
# tools. Stubbed here so the example runs end to end.
def handle_billing(query: str) -> str:
    return "-> billing agent"


def handle_technical(query: str) -> str:
    return "-> technical agent"


def handle_sales(query: str) -> str:
    return "-> sales agent"


def handle_general(query: str) -> str:
    return "-> general agent"


def route(client, query: str) -> str:
    """Classify the query once with a cheap model, then dispatch."""

    reply = client.messages.create(
        model="claude-haiku-4-5",  # fast + cheap for classification
        max_tokens=10,
        system=(
            "Classify the user query into exactly one label: "
            "BILLING, TECHNICAL, or SALES. Reply with only the label."
        ),
        messages=[{"role": "user", "content": query}],
    )
    label = next(b.text for b in reply.content if b.type == "text").strip().upper()

    handlers = {
        "BILLING": handle_billing,      # the fallback catches anything the
        "TECHNICAL": handle_technical,  # classifier returns that is not one
        "SALES": handle_sales,          # of the three expected labels
    }
    handler = handlers.get(label, handle_general)
    return f"{label:10} {handler(query)}"


if __name__ == "__main__":
    client = anthropic.Anthropic()
    for q in ["My card was charged twice this month.",
              "The webhook returns 502 after I rotate the key.",
              "Do you offer volume discounts for 50 seats?"]:
        print(route(client, q))