Multi-agent orchestration patterns avançados

from typing import TypedDict, Annotated, Literal
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool

# Tools dos workers
@tool
def search_arxiv(query: str) -> str:
    """Pesquisa papers no arXiv."""
    return f"3 papers encontrados sobre '{query}'"

@tool
def run_code(code: str) -> str:
    """Executa Python em sandbox."""
    return f"Output: {eval(code)}"  # NÃO faça isso em prod (use sandbox real)

llm = ChatAnthropic(model="claude-sonnet-4-7")

# Workers especialistas
researcher = create_react_agent(llm, tools=[search_arxiv])
coder = create_react_agent(llm, tools=[run_code])

# Supervisor: LLM com tool "delegate"
class State(TypedDict):
    messages: Annotated[list, operator.add]
    next: str

def supervisor(state: State) -> dict:
    sys = """Você coordena dois workers: 'researcher' (papers) e 'coder' (executa código).
    Decida qual chamar ou se a task está completa. Retorne JSON: {"next": "researcher"|"coder"|"FINISH"}"""
    response = llm.invoke([{"role": "system", "content": sys}] + state["messages"])
    import json
    decision = json.loads(response.content)
    return {"next": decision["next"]}

def call_researcher(state):
    result = researcher.invoke({"messages": state["messages"]})
    return {"messages": [result["messages"][-1]]}

def call_coder(state):
    result = coder.invoke({"messages": state["messages"]})
    return {"messages": [result["messages"][-1]]}

graph = StateGraph(State)
graph.add_node("supervisor", supervisor)
graph.add_node("researcher", call_researcher)
graph.add_node("coder", call_coder)
graph.add_edge(START, "supervisor")
graph.add_conditional_edges(
    "supervisor",
    lambda s: END if s["next"] == "FINISH" else s["next"],
)
graph.add_edge("researcher", "supervisor")  # supervisor decide next
graph.add_edge("coder", "supervisor")
app = graph.compile()

# Adapted from Liang et al. 2023 — arxiv.org/abs/2305.19118
import asyncio
from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-sonnet-4-7")

async def debate(question: str, n_agents: int = 3, n_rounds: int = 2) -> str:
    history = [[] for _ in range(n_agents)]

    # Round 0: cada agente responde independentemente
    for i in range(n_agents):
        response = await llm.ainvoke([{"role": "user", "content": question}])
        history[i].append(response.content)

    # Rounds 1..N: cada agente vê argumentos dos outros e refina
    for r in range(1, n_rounds + 1):
        new_responses = []
        for i in range(n_agents):
            others = [history[j][r - 1] for j in range(n_agents) if j != i]
            others_text = "\n\n".join(f"Argumento {k+1}: {a}" for k, a in enumerate(others))
            prompt = (
                f"Pergunta original: {question}\n\n"
                f"Outros agentes responderam:\n{others_text}\n\n"
                f"Sua resposta anterior: {history[i][r-1]}\n\n"
                "Considere os argumentos. Refine sua resposta. Se mantiver, justifique."
            )
            response = await llm.ainvoke([{"role": "user", "content": prompt}])
            new_responses.append(response.content)
        for i in range(n_agents):
            history[i].append(new_responses[i])

    # Round final: judge agrega
    final_args = [h[-1] for h in history]
    judge_prompt = (
        f"Pergunta: {question}\n\n"
        + "\n\n".join(f"Agente {i+1}: {a}" for i, a in enumerate(final_args))
        + "\n\nComo juiz, dê a resposta consolidada e correta."
    )
    final = await llm.ainvoke([{"role": "user", "content": judge_prompt}])
    return final.content

# answer = asyncio.run(debate("Prove que sqrt(2) é irracional.", n_agents=3, n_rounds=2))

# Self-consistency (Wang et al. 2023, ICLR — arxiv.org/abs/2203.11171)
import asyncio
from collections import Counter
from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-sonnet-4-7", temperature=0.7)

async def self_consistency(question: str, n: int = 10) -> str:
    """N samples paralelos com temperatura > 0, vote majoritário."""
    # Paralelo via gather
    tasks = [llm.ainvoke([{"role": "user", "content": question}]) for _ in range(n)]
    responses = await asyncio.gather(*tasks)
    answers = [r.content for r in responses]

    # Normaliza (extrair "answer: X" — depende do domínio)
    # Para math: extrair último número
    import re
    extracted = []
    for ans in answers:
        m = re.search(r"\\boxed\{(.+?)\}|answer:?\s*(\S+)", ans, re.IGNORECASE)
        if m:
            extracted.append(m.group(1) or m.group(2))
    counts = Counter(extracted)
    most_common, _ = counts.most_common(1)[0]
    return most_common

# Custo: N× tokens. Latência: 1× (paralelo). Accuracy boost ~5–15% em math.