intermediate⏱40 minutesLección 8 de 10

Comunicación y Supervisión entre Agentes

Implementa mensajes estructurados agente-a-agente, nodos supervisores, coordinación de estado compartido y patrones de orquestación multi-agente.

Comunicación y Supervisión entre Agentes

In production multi-agent systems, agents need structured communication and a supervisor that coordinates their work. This lesson covers message protocols, supervisor patterns, and shared state coordination.

Mensajes Estructurados de Agente

Use typed dictionaries for agent-to-agent messages:

python

from typing_extensions import TypedDict, Annotated
from typing import List, Optional
from datetime import datetime
from operator import add

class AgentMessage(TypedDict):
    sender: str
    recipient: str
    content: str
    message_type: str  # task, result, question, error
    timestamp: str
    metadata: dict

class SupervisorState(TypedDict):
    task: str
    conversation: Annotated[List[AgentMessage], add]
    current_agent: str
    completed_agents: List[str]
    final_output: str

[!NOTA] Mensajes estructurados with sender, recipient, and type fields enable sophisticated routing. Agents can target messages to specific recipients or broadcast to all.

El Patrón Supervisor

A supervisor is a coordinator agent that decides which agent should work next:

python

from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI

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

class SupervisorState(TypedDict):
    task: str
    conversation: Annotated[List[AgentMessage], add]
    next_agent: str
    completed: bool

def supervisor_node(state: SupervisorState) -> dict:
    """Decides which agent should work next based on progress."""

    recent = state["conversation"][-3:] if state["conversation"] else []
    context = "\n".join(f"{m['sender']} → {m['recipient']}: {m['content']}"
                        for m in recent)

    prompt = f"""Task: {state['task']}
Recent activity:
{context}

Available agents: researcher, writer, reviewer, finalizer
Which agent should work next? Or respond with 'complete' if done.
Reply with one word: researcher, writer, reviewer, finalizer, or complete."""

    response = llm.invoke(prompt)
    next_agent = response.content.strip().lower()

    return {"next_agent": next_agent,
            "completed": next_agent == "complete"}

Nodos de Agente con Manejo de Mensajes

python

def researcher_agent(state: SupervisorState) -> dict:
    result = llm.invoke(f"Research this topic: {state['task']}")

    return {"conversation": [{
        "sender": "researcher",
        "recipient": "supervisor",
        "content": result.content,
        "message_type": "result",
        "timestamp": datetime.now().isoformat(),
        "metadata": {"sources": 3}
    }]}

def writer_agent(state: SupervisorState) -> dict:
    # Find the latest research result
    research = [m for m in reversed(state["conversation"])
                if m["sender"] == "researcher"][0]

    content = llm.invoke(f"Write based on: {research['content']}")
    return {"conversation": [{
        "sender": "writer",
        "recipient": "supervisor",
        "content": content.content,
        "message_type": "result",
        "timestamp": datetime.now().isoformat(),
        "metadata": {"word_count": len(content.content.split())}
    }]}

[!ÉXITO] The supervisor acts as a router — it reads the conversation, decides the next step, and the graph routes execution accordingly.

Enrutamiento del Supervisor

python

def route_from_supervisor(state: SupervisorState) -> str:
    if state["completed"]:
        return "finalize"
    return state["next_agent"]

builder = StateGraph(SupervisorState)
builder.add_node("supervisor", supervisor_node)
builder.add_node("researcher", researcher_agent)
builder.add_node("writer", writer_agent)
builder.add_node("finalizer", finalizer_agent)

builder.add_edge(START, "supervisor")
builder.add_conditional_edges(
    "supervisor",
    route_from_supervisor,
    {
        "researcher": "researcher",
        "writer": "writer",
        "finalize": "finalizer"
    }
)
# After any agent works, go back to supervisor
builder.add_edge("researcher", "supervisor")
builder.add_edge("writer", "supervisor")
builder.add_edge("finalizer", END)

START → Supervisor → (researcher → Supervisor → writer → Supervisor → finalizer → END)
         ↑              ↑
         └──── loop ────┘

Mensajes de Difusión y Dirigidos

Broadcast (to all agents)

python

def supervisor_broadcast(state: SupervisorState) -> dict:
    return {"conversation": [{
        "sender": "supervisor",
        "recipient": "*",  # Broadcast
        "content": f"New task: {state['task']}",
        "message_type": "announcement",
        "timestamp": datetime.now().isoformat(),
        "metadata": {}
    }]}

Objetivoed (to specific agent)

python

def researcher_requests_clarification(state: SupervisorState) -> dict:
    return {"conversation": [{
        "sender": "researcher",
        "recipient": "supervisor",
        "content": "Need clarification on the scope",
        "message_type": "question",
        "timestamp": datetime.now().isoformat(),
        "metadata": {}
    }]}

Selección Dinámica de Agentes

Let the supervisor pick the right agent based on the task:

python

def dynamic_supervisor(state: SupervisorState) -> dict:
    available_agents = [{
        "name": "researcher",
        "skills": "web research, fact-finding, data gathering",
        "status": "ready"
    }, {
        "name": "analyst",
        "skills": "data analysis, pattern recognition, statistics",
        "status": "ready"
    }, {
        "name": "writer",
        "skills": "content creation, summarization, formatting",
        "status": "ready"
    }]

    prompt = f"""Task: {state['task']}
Available agents: {available_agents}
Recent messages: {state['conversation'][-2:]}

Which agent should work next, or has the task been completed?
Respond with: agent_name or 'complete'."""

    response = llm.invoke(prompt)
    return {"next_agent": response.content.strip().lower()}

Multi-Agente Secuencial con Transferencia

Agents can hand off to each other explicitly:

python

def analyst_agent(state: SupervisorState) -> dict:
    analysis = llm.invoke(f"Analyze: {state['task']}")

    return {
        "conversation": [{
            "sender": "analyst",
            "recipient": "writer",
            "content": analysis.content,
            "message_type": "handoff",
            "timestamp": datetime.now().isoformat(),
            "metadata": {}
        }],
        "current_agent": "writer"  # Explicit handoff
    }

Ejemplo Completo de Supervisor

python

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

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

class Message(TypedDict):
    sender: str
    content: str
    type: str

class AgentState(TypedDict):
    task: str
    messages: Annotated[List[Message], add]
    next_agent: str
    done: bool

def supervisor(state: AgentState) -> dict:
    context = "\n".join(f"{m['sender']}: {m['content']}" for m in state["messages"][-4:])
    prompt = f"Task: {state['task']}\nRecent:\n{context}\n\nNext agent (researcher/writer/complete):"
    resp = llm.invoke(prompt).content.strip().lower()
    return {"next_agent": resp, "done": resp == "complete"}

def researcher(state: AgentState) -> dict:
    resp = llm.invoke(f"Research: {state['task']}")
    return {"messages": [{"sender": "researcher", "content": resp.content, "type": "research"}]}

def writer(state: AgentState) -> dict:
    research = [m for m in reversed(state["messages"]) if m["sender"] == "researcher"]
    content = research[0]["content"] if research else state["task"]
    resp = llm.invoke(f"Write based on: {content}")
    return {"messages": [{"sender": "writer", "content": resp.content, "type": "draft"}]}

def router(state: AgentState) -> str:
    if state["done"]:
        return "done"
    return state["next_agent"]

builder = StateGraph(AgentState)
builder.add_node("supervisor", supervisor)
builder.add_node("researcher", researcher)
builder.add_node("writer", writer)
builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor", router, {
    "researcher": "researcher",
    "writer": "writer",
    "done": END
})
builder.add_edge("researcher", "supervisor")
builder.add_edge("writer", "supervisor")

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

Estado Compartido vs Estado con Ámbito de Agente

Pattern	Descripción	Use Case
Estado compartido	All agents read/write the same state dict	Coordinación simple, pocos agentes
Mensajes estructurados	Agents write to a shared message list	Multi-agente complejo, seguimiento histórico
Estado de subgrafo	Each agent has its own internal state, communicates via messages (next lesson)	Agentes jerárquicos, encapsulamiento

Preguntas Prácticas

Practice Question

What is the role of a supervisor node in a multi-agent system?

Practice Question

How does the supervisor communicate its decision to the graph?

Practice Question

What is the loop pattern in a supervisor-based graph?

Practice Question

What field structure enables targeted agent-to-agent messaging?

Practice Question

How does the graph terminate in a supervisor pattern?

Practice Question

What is a broadcast message in a multi-agent system?

Practice Question

What happens after an agent completes its work in the supervisor loop?

Practice Question

What is the benefit of structured messages over plain text in agent communication?

Practice Question

What routing mechanism executes the supervisor's decision?

Practice Question

What happens if an agent writes a message but the supervisor decides to route to a different agent?

[!ÉXITO]

Conclusiones Clave

Supervisor nodes coordinate multi-agent work by deciding the next step
Mensajes estructurados (sender, recipient, content, type) enable sophisticated communication
The supervisor loop pattern: Supervisor → Agent → Supervisor → Agent → ... → END
Conditional edges route based on the supervisor's decision
Broadcast messages go to all agents; targeted messages go to specific recipients
Messages accumulate in state for full conversation history
Dynamic agent selection lets the supervisor choose the right agent for each step

Progreso80%

Fundamentos de Multi-Agente

Streaming de Tokens

Contenido del Curso

1. Gestión Avanzada de Estado 2. Persistencia y Puntos de Control 3. Memoria de Conversación 4. Memoria de Resumen 5. Humano-en-el-bucle con Interrupciones 6. Flujos de Aprobación 7. Fundamentos de Multi-Agente 8. Comunicación y Supervisión entre Agentes 9. Streaming de Tokens 10. Proyecto Intermedio: Sistema de Investigación Multi-Agente