Fundamentos de Multi-Agente

Aprende arquitectura multi-agente en LangGraph, múltiples nodos de agente comunicándose a través de estado compartido, roles especializados y patrones de coordinación.

Los sistemas multiagente usan múltiples agentes especializados que colaboran para resolver tareas complejas. La estructura de grafo de LangGraph es ideal para orquestar flujos de trabajo multiagente.

¿Por qué Multi-Agente?

Los agentes individuales tienen limitaciones:

Sobrecarga de contexto: Un solo agente que se encarga de todo supera las ventanas de contexto
Confusión de roles: Un agente individual no puede ser experto en todo
Modularidad: Es difícil sustituir o actualizar partes de un agente monolítico

La arquitectura multiagente lo resuelve dividiendo y conquistando: cada agente tiene un rol centrado y se comunica con los demás mediante el estado compartido.

Patrón 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 es un nodo separado con una función especializada. Se comunican leyendo y escribiendo en el estado compartido. Esto mantiene a cada agente simple y centrado.

Ejecución Multi-Agente Paralela

Cuando los agentes pueden trabajar de forma independiente, ejecútalos en 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)

✅Éxito

La ejecución paralela es una ventaja clave de los sistemas multiagente. Especialistas independientes trabajan simultáneamente y un nodo de fusión combina sus resultados.

Agentes Especializados con LLM

Cada agente puede tener su propio LLM con un 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"]})}

💡Consejo

Cada agente puede usar un modelo distinto. Usa modelos más baratos (gpt-4o-mini) para tareas más simples y reserva los modelos potentes (gpt-4o) para el razonamiento complejo.

Comunicación entre Agentes vía Estado Estructurado

Define campos estructurados para la comunicación 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 con Acceso a Herramientas

Cada agente puede tener su propio conjunto de herramientas:

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)

Manejo de Errores en Sistemas Multi-Agente

El fallo de un agente no debería tumbar al equipo:

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)

Ejemplo 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"])