6 Agent Design Patterns

import anthropic

client = anthropic.Anthropic()

# Define tools the agent can call
tools = [
    {
        "name": "search_web",
        "description": "Search the web for current information",
        "input_schema": {
            "type": "object",
            "properties": {"query": {"type": "string"}},
            "required": ["query"]
        }
    }
]

messages = [{"role": "user", "content": "What's the current price of Bitcoin?"}]

# Agent loop — runs until model returns end_turn
while True:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        tools=tools,
        messages=messages
    )

    if response.stop_reason == "end_turn":
        # Model is done — extract and return final text
        for block in response.content:
            if hasattr(block, 'text'):
                print(block.text)
        break

    if response.stop_reason == "tool_use":
        messages.append({"role": "assistant", "content": response.content})
        tool_results = []

        for block in response.content:
            if block.type == "tool_use":
                # Execute the tool and capture result
                result = execute_tool(block.name, block.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result
                })

        # Return results to model and loop again
        messages.append({"role": "user", "content": tool_results})

import anthropic
import json

client = anthropic.Anthropic()

ROUTER_PROMPT = """Classify this customer message into exactly one category.
Return a JSON object with a single key "category" and one of these values:
- "billing" — payment, invoice, refund, subscription, pricing
- "technical" — bugs, errors, API, performance, features
- "retention" — cancel, too expensive, switching, leaving
- "general" — anything else

Respond with JSON only. No explanation."""

def route_request(message: str) -> str:
    """Use a fast model to classify intent"""
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",  # Cheap router
        max_tokens=64,
        system=ROUTER_PROMPT,
        messages=[{"role": "user", "content": message}]
    )
    result = json.loads(response.content[0].text)
    return result["category"]

def run_specialist(category: str, message: str) -> str:
    """Delegate to the right specialist with its system prompt"""
    specialists = {
        "billing": {
            "model": "claude-haiku-4-5-20251001",
            "system": "You are a billing specialist. You have access to Stripe...",
        },
        "technical": {
            "model": "claude-opus-4-6",  # Complex — use big model
            "system": "You are a technical support engineer...",
        },
        "retention": {
            "model": "claude-opus-4-6",
            "system": "You are a customer success specialist. Your goal...",
        },
    }

    spec = specialists.get(category, specialists["technical"])
    response = client.messages.create(
        model=spec["model"],
        max_tokens=1024,
        system=spec["system"],
        messages=[{"role": "user", "content": message}]
    )
    return response.content[0].text

# Usage
user_message = "I want to cancel my subscription, it's too expensive"
category = route_request(user_message)        # → "retention"
reply = run_specialist(category, user_message) # → retention specialist runs

import anthropic

client = anthropic.Anthropic()

def run_stage(stage_name: str, system: str, content: str, model: str = "claude-opus-4-6") -> str:
    """Run a single pipeline stage and return its output"""
    print(f"[Pipeline] Running stage: {stage_name}")
    response = client.messages.create(
        model=model,
        max_tokens=2048,
        system=system,
        messages=[{"role": "user", "content": content}]
    )
    output = response.content[0].text
    print(f"  → {len(output)} chars output")
    return output

def run_content_pipeline(raw_articles: str) -> str:
    """Run the full content pipeline sequentially"""

    # Stage 1: Summarize raw content
    summaries = run_stage(
        "summarizer",
        system="Extract the key points from each article as bullet points. Be concise.",
        content=raw_articles,
        model="claude-haiku-4-5-20251001"  # Cheap model for summarization
    )

    # Stage 2: Fact-check (only run on summaries, not full articles)
    verified = run_stage(
        "fact-checker",
        system="Review these summaries. Flag any claims that seem unverified or exaggerated.",
        content=summaries,
        model="claude-opus-4-6"  # Use powerful model for reasoning
    )

    # Stage 3: Format into final output
    formatted = run_stage(
        "formatter",
        system="Format the verified summaries into a professional markdown newsletter.",
        content=verified,
        model="claude-haiku-4-5-20251001"  # Cheap model for formatting
    )

    return formatted

# Run the pipeline
raw = fetch_articles()  # Your scraper returns raw text
newsletter = run_content_pipeline(raw)
publish(newsletter)

import anthropic

client = anthropic.Anthropic()

GENERATOR_PROMPT = """You write high-quality Python code. When given a task:
- Write clean, well-commented code
- Handle edge cases and errors
- Include type hints and docstrings"""

CRITIC_PROMPT = """You are a senior code reviewer. Evaluate this code strictly.
Return a JSON object:
{
  "passed": true/false,
  "score": 0-10,
  "issues": ["list of specific problems found"],
  "verdict": "brief explanation"
}
Be strict. Score below 8 = fail. No issues = pass."""

def generate(task: str, feedback: str = "") -> str:
    prompt = task
    if feedback:
        prompt += f"\n\nPrevious attempt failed critique:\n{feedback}\n\nFix all issues."
    response = client.messages.create(
        model="claude-opus-4-6", max_tokens=2048,
        system=GENERATOR_PROMPT,
        messages=[{"role": "user", "content": prompt}]
    )
    return response.content[0].text

def critique(code: str, task: str) -> dict:
    import json
    response = client.messages.create(
        model="claude-opus-4-6", max_tokens=512,
        system=CRITIC_PROMPT,
        messages=[{"role": "user", "content": f"Task: {task}\n\nCode:\n{code}"}]
    )
    return json.loads(response.content[0].text)

def reflection_agent(task: str, max_iterations: int = 4) -> str:
    feedback = ""
    for i in range(max_iterations):
        draft = generate(task, feedback)
        review = critique(draft, task)
        print(f"Iteration {i+1}: score={review['score']}, passed={review['passed']}")
        if review["passed"]:
            print("  ✓ Passed critique!")
            return draft
        feedback = "\n".join(review["issues"])
    return draft  # Return best attempt

result = reflection_agent("Write a Python function to parse CSV with mixed date formats")

import anthropic
import json

client = anthropic.Anthropic()

# Actions that ALWAYS require human approval
ALWAYS_APPROVE = {"send_email", "delete_event", "make_purchase"}
CONFIDENCE_THRESHOLD = 0.85

def assess_action(action: str, inputs: dict) -> dict:
    """Ask the model to assess risk and confidence"""
    response = client.messages.create(
        model="claude-opus-4-6", max_tokens=256,
        system="Assess this action. Return JSON: {confidence: 0-1, risk: low/medium/high, reason: str}",
        messages=[{"role": "user", "content": f"Action: {action}\nInputs: {json.dumps(inputs)}"}]
    )
    return json.loads(response.content[0].text)

def needs_approval(action: str, assessment: dict) -> bool:
    if action in ALWAYS_APPROVE:
        return True
    if assessment["risk"] == "high":
        return True
    if assessment["confidence"] < CONFIDENCE_THRESHOLD:
        return True
    return False

def request_human_approval(action: str, inputs: dict, assessment: dict) -> bool:
    """Present action to human and wait for input"""
    print(f"\n⚠️  APPROVAL REQUIRED")
    print(f"Action: {action}")
    print(f"Inputs: {json.dumps(inputs, indent=2)}")
    print(f"Risk: {assessment['risk']} | Confidence: {assessment['confidence']:.0%}")
    print(f"Reason: {assessment['reason']}")
    answer = input("Approve? (y/n): ").strip().lower()
    return answer == "y"

def execute_with_approval(action: str, inputs: dict) -> str:
    assessment = assess_action(action, inputs)
    if needs_approval(action, assessment):
        approved = request_human_approval(action, inputs, assessment)
        if not approved:
            return "Action rejected by human. Taking alternative approach."
    return run_action(action, inputs)  # Execute approved action

import asyncio
import anthropic

client = anthropic.Anthropic()

async def run_agent(name: str, task: str, system: str) -> dict:
    """Run a single specialist agent asynchronously"""
    loop = asyncio.get_event_loop()
    response = await loop.run_in_executor(None, lambda: client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        system=system,
        messages=[{"role": "user", "content": task}]
    ))
    print(f"  ✓ {name} completed")
    return {"agent": name, "output": response.content[0].text}

async def run_parallel_agents(company: str) -> str:
    print(f"Launching parallel agents for: {company}")

    # Define specialist agents
    agents = [
        ("competitor-researcher",
         f"Research competitors of {company}. List top 5 with features.",
         "You are a competitive intelligence analyst."),
        ("pricing-analyst",
         f"Analyze pricing strategies in the {company} market.",
         "You are a pricing strategy consultant."),
        ("sentiment-analyst",
         f"Summarize customer sentiment about {company}'s competitors.",
         "You are a customer experience researcher."),
    ]

    # Run all agents in parallel
    tasks = [run_agent(name, task, system) for name, task, system in agents]
    results = await asyncio.gather(*tasks)

    # Aggregate: combine all outputs and synthesize
    combined = "\n\n".join([f"## {r['agent']}\n{r['output']}" for r in results])
    synthesis = client.messages.create(
        model="claude-opus-4-6", max_tokens=2048,
        system="Synthesize these research findings into a strategic report with actionable recommendations.",
        messages=[{"role": "user", "content": combined}]
    )
    return synthesis.content[0].text

# Run the multi-agent system
report = asyncio.run(run_parallel_agents("Notion"))
print(report)