Fundamentos de Multi-Agentes

Aprenda arquitetura multi-agente em LangGraph, múltiplos nós de agente comunicando através de estado compartilhado, papéis especializados e padrões de coordenação.

Sistemas multi-agente usam múltiplos agentes especializados que colaboram para resolver tarefas complexas. A estrutura de grafo do LangGraph é ideal para orquestrar fluxos de trabalho multi-agente.

Por que Multi-Agente?

Agentes únicos têm limitaciones:

Sobrecarga de contexto: Um único agente cuidando de tudo excede as janelas de contexto
Confusão de papéis: Um agente único não pode ser especialista em tudo
Modularidade: É difícil trocar ou atualizar partes de um agente monolítico

A arquitetura multi-agente resolve isso dividindo para conquistar: cada agente tem um papel focado e se comunica com os outros por meio do estado compartilhado.

Padrão Básico Multi-Agente

python

from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
from typing import List, Annotated
from operator import add

class TeamState(TypedDict):
    input: str
    research_results: str
    code_output: str
    review_comments: str
    final_output: str
    logs: Annotated[List[str], add]

def researcher(state: TeamState) -> dict:
    """Specialist: researches the topic."""
    research = f"Research findings for: {state['input']}"
    return {"research_results": research,
            "logs": ["Researcher completed"]}

def coder(state: TeamState) -> dict:
    """Specialist: writes code based on research."""
    code = f"# Code implementing: {state['research_results']}"
    return {"code_output": code,
            "logs": ["Coder completed"]}

def reviewer(state: TeamState) -> dict:
    """Specialist: reviews code."""
    review = f"Review of code: looks good"
    return {"review_comments": review,
            "logs": ["Reviewer completed"]}

def assembler(state: TeamState) -> dict:
    """Combines all outputs into final result."""
    final = f"## Result\n\nResearch: {state['research_results']}\n\nCode: {state['code_output']}\n\nReview: {state['review_comments']}"
    return {"final_output": final}

# Build sequential multi-agent graph
builder = StateGraph(TeamState)
builder.add_node("researcher", researcher)
builder.add_node("coder", coder)
builder.add_node("reviewer", reviewer)
builder.add_node("assembler", assembler)
builder.add_edge(START, "researcher")
builder.add_edge("researcher", "coder")
builder.add_edge("coder", "reviewer")
builder.add_edge("reviewer", "assembler")
builder.add_edge("assembler", END)

ℹ️Nota

Cada agente é um nó separado com uma função especializada. Eles se comunicam lendo e escrevendo no estado compartilhado. Isso mantém cada agente simples e focado.

Execução Multi-Agente Paralela

Quando os agentes podem trabalhar de forma independente, execute-os em paralelo:

python

def agent_a(state: State) -> dict:
    return {"result_a": "Agent A output"}

def agent_b(state: State) -> dict:
    return {"result_b": "Agent B output"}

def agent_c(state: State) -> dict:
    return {"result_c": "Agent C output"}

def merger(state: State) -> dict:
    combined = f"{state['result_a']}\n{state['result_b']}\n{state['result_c']}"
    return {"merged": combined}

builder = StateGraph(State)
builder.add_node("agent_a", agent_a)
builder.add_node("agent_b", agent_b)
builder.add_node("agent_c", agent_c)
builder.add_node("merger", merger)

# Fan-out: all three agents run in parallel
builder.add_edge(START, "agent_a")
builder.add_edge(START, "agent_b")
builder.add_edge(START, "agent_c")

# Fan-in: merger waits for all three
builder.add_edge("agent_a", "merger")
builder.add_edge("agent_b", "merger")
builder.add_edge("agent_c", "merger")
builder.add_edge("merger", END)

✅Sucesso

A execução paralela é uma vantagem-chave dos sistemas multi-agente. Especialistas independentes trabalham simultaneamente e um nó de mesclagem combina os resultados deles.

Agentes Especializados com LLM

Cada agente pode ter seu próprio LLM com um prompt de sistema especializado:

python

from langchain_openai import ChatOpenAI

# Each agent has its own model
research_llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
code_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.1)
review_llm = ChatOpenAI(model="gpt-4o", temperature=0.0)

def research_agent(state: TeamState) -> dict:
    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are a research specialist. Find and summarize "
                   "key information about the given topic. Be thorough and cite sources."),
        ("human", "{input}")
    ])
    chain = prompt | research_llm | StrOutputParser()
    return {"research_results": chain.invoke({"input": state["input"]})}

def code_agent(state: TeamState) -> dict:
    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are a code specialist. Write clean, well-documented "
                   "Python code based on the research provided."),
        ("human", "Research: {research}\n\nWrite code to implement this.")
    ])
    chain = prompt | code_llm | StrOutputParser()
    return {"code_output": chain.invoke({"research": state["research_results"]})}

def review_agent(state: TeamState) -> dict:
    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are a code reviewer. Check for bugs, security issues, "
                   "and best practices. Provide actionable feedback."),
        ("human", "Code to review:\n{code}")
    ])
    chain = prompt | review_llm | StrOutputParser()
    return {"review_comments": chain.invoke({"code": state["code_output"]})}

💡Dica

Cada agente pode usar um modelo diferente. Use modelos mais baratos (gpt-4o-mini) para tarefas mais simples e reserve modelos poderosos (gpt-4o) para raciocínio complexo.

Comunicação entre Agentes via Estado Estruturado

Defina campos estruturados para a comunicação entre agentes:

python

class AgentMessage(TypedDict):
    from_agent: str
    to_agent: str
    content: str
    message_type: str  # "request", "response", "feedback"

class MultiAgentState(TypedDict):
    task: str
    agent_messages: Annotated[List[AgentMessage], add]
    status: str

def researcher_v2(state: MultiAgentState) -> dict:
    research = llm.invoke(f"Research: {state['task']}")
    return {
        "agent_messages": [{
            "from_agent": "researcher",
            "to_agent": "writer",
            "content": research.content,
            "message_type": "research_results"
        }],
        "status": "research_done"
    }

def writer_v2(state: MultiAgentState) -> dict:
    # Find the latest research message addressed to writer
    research_msg = [m for m in reversed(state["agent_messages"])
                    if m["to_agent"] == "writer"][0]

    content = llm.invoke(f"Write based on: {research_msg['content']}")
    return {
        "agent_messages": [{
            "from_agent": "writer",
            "to_agent": "reviewer",
            "content": content.content,
            "message_type": "draft"
        }],
        "status": "writing_done"
    }

Agente com Acesso a Ferramentas

Cada agente pode ter seu próprio conjunto de ferramentas:

python

from langchain_core.tools import tool
from langgraph.prebuilt import ToolExecutor

@tool
def search_web(q: str) -> str:
    """Search the web."""
    return f"Search results for: {q}"

@tool
def query_database(sql: str) -> str:
    """Execute SQL queries."""
    return f"DB results for: {sql}"

@tool
def send_email(to: str, body: str) -> str:
    """Send an email."""
    return f"Email sent to {to}"

# Researcher agent only gets search tools
researcher_tools = [search_web]
researcher_llm = ChatOpenAI(model="gpt-4o").bind_tools(researcher_tools)

# Data agent only gets database tools
data_tools = [query_database]
data_llm = ChatOpenAI(model="gpt-4o").bind_tools(data_tools)

# Email agent only gets communication tools
email_tools = [send_email]
email_llm = ChatOpenAI(model="gpt-4o").bind_tools(email_tools)

Tratamento de Erros em Sistemas Multi-Agente

A falha de um agente não deve derrubar a equipe:

python

def safe_agent(state: MultiAgentState, agent_name: str, agent_func) -> dict:
    try:
        return agent_func(state)
    except Exception as e:
        return {
            "agent_messages": [{
                "from_agent": agent_name,
                "to_agent": "supervisor",
                "content": f"Failed: {str(e)}",
                "message_type": "error"
            }],
            "status": f"{agent_name}_failed"
        }

# Wrap each agent
def researcher_safe(state: MultiAgentState) -> dict:
    return safe_agent(state, "researcher", research_agent)

def writer_safe(state: MultiAgentState) -> dict:
    return safe_agent(state, "writer", write_agent)

Exemplo Completo Multi-Agente

python

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from typing_extensions import TypedDict, Annotated
from typing import List
from operator import add

llm = ChatOpenAI(model="gpt-4o-mini")

class ProjectState(TypedDict):
    topic: str
    research: str
    outline: str
    draft: str
    reviewed: str
    final: str
    logs: Annotated[List[str], add]

def researcher_node(state: ProjectState) -> dict:
    r = llm.invoke(f"Research: {state['topic']}")
    return {"research": r.content, "logs": ["Research done"]}

def planner_node(state: ProjectState) -> dict:
    p = llm.invoke(f"Outline based on: {state['research']}")
    return {"outline": p.content, "logs": ["Planning done"]}

def writer_node(state: ProjectState) -> dict:
    w = llm.invoke(f"Write based on outline: {state['outline']}")
    return {"draft": w.content, "logs": ["Writing done"]}

def reviewer_node(state: ProjectState) -> dict:
    r = llm.invoke(f"Review this draft: {state['draft']}")
    return {"reviewed": r.content, "logs": ["Review done"]}

def finalizer_node(state: ProjectState) -> dict:
    f = llm.invoke(f"Finalize: Draft={state['draft']}, Review={state['reviewed']}")
    return {"final": f.content, "logs": ["Finalized"]}

builder = StateGraph(ProjectState)
builder.add_node("researcher", researcher_node)
builder.add_node("planner", planner_node)
builder.add_node("writer", writer_node)
builder.add_node("reviewer", reviewer_node)
builder.add_node("finalizer", finalizer_node)

builder.add_edge(START, "researcher")
builder.add_edge("researcher", "planner")
builder.add_edge("planner", "writer")
builder.add_edge("writer", "reviewer")
builder.add_edge("reviewer", "finalizer")
builder.add_edge("finalizer", END)

app = builder.compile(checkpointer=MemorySaver())

result = app.invoke({
    "topic": "Benefits of vector databases for AI",
    "research": "", "outline": "", "draft": "",
    "reviewed": "", "final": "", "logs": []
})
print(result["final"])