Multi-Agent Systems: orchestrator-worker, swarms e handoffs

# Orchestrator-worker — versão minimal em Python
import asyncio
from anthropic import AsyncAnthropic
client = AsyncAnthropic()

async def worker(role: str, task: str) -> str:
    r = await client.messages.create(
        model="claude-haiku-4-5-20251001",        # workers no modelo barato
        max_tokens=800,
        system=f"Você é um {role}. Responda objetivo, máx 400 palavras.",
        messages=[{"role": "user", "content": task}],
    )
    return r.content[0].text

async def orchestrate(user_query: str) -> str:
    # 1) Planner decide sub-tarefas (modelo forte para planejar)
    plan = await client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=500,
        system=(
            "Decomponha a pergunta em 2-4 sub-tarefas independentes em JSON: "
            "[{role, task}]. Cada sub-tarefa deve ser executável em paralelo."
        ),
        messages=[{"role": "user", "content": user_query}],
    )
    import json
    sub_tasks = json.loads(plan.content[0].text)

    # 2) Workers em paralelo
    results = await asyncio.gather(*[
        worker(st["role"], st["task"]) for st in sub_tasks
    ])

    # 3) Orchestrator agrega
    synthesis = await client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1200,
        messages=[{
            "role": "user",
            "content": (
                f"Pergunta original: {user_query}\n\n"
                f"Resultados dos workers:\n\n" +
                "\n\n---\n\n".join(results) +
                "\n\nSintetize uma resposta coerente, citando resultados quando útil."
            ),
        }],
    )
    return synthesis.content[0].text

# OpenAI Agents SDK — handoffs nativos
from agents import Agent, Runner, handoff

triage_agent = Agent(
    name="Triage",
    instructions=(
        "Identifique a intenção do usuário e transfira para o agent certo. "
        "Se não tem certeza, peça esclarecimento."
    ),
)

billing_agent = Agent(
    name="Billing",
    instructions=(
        "Você cuida de cobrança, fatura e plano. Tem acesso a tools "
        "get_invoice, update_payment. Se o usuário precisar de refund "
        "acima de US$50, transfira para Human."
    ),
    tools=[get_invoice, update_payment],
)

human_agent = Agent(
    name="Human",
    instructions="Abra ticket crítico para um humano resolver.",
    tools=[open_ticket],
)

# Wire handoffs
triage_agent.handoffs = [billing_agent]
billing_agent.handoffs = [human_agent]

# Run
result = await Runner.run(triage_agent, "Meu cartão não passou na renovação")
# Triage → Billing → (resolve ou) → Human

# Panel of 3 experts + LLM judge escolhendo
import asyncio

PERSONAS = [
    "Você é um engenheiro sênior focado em segurança.",
    "Você é um engenheiro sênior focado em performance.",
    "Você é um engenheiro sênior focado em legibilidade.",
]

async def expert_review(persona: str, code: str) -> str:
    r = await client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=600,
        system=persona,
        messages=[{"role": "user", "content": f"Revise:\n\n{code}"}],
    )
    return r.content[0].text

async def panel_review(code: str) -> str:
    reviews = await asyncio.gather(*[expert_review(p, code) for p in PERSONAS])
    judge = await client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1000,
        messages=[{
            "role": "user",
            "content": (
                "Três revisões de código. Consolide em uma revisão única, "
                "resolvendo conflitos e priorizando severidade:\n\n" +
                "\n\n---\n\n".join(reviews)
            ),
        }],
    )
    return judge.content[0].text