intermediate⏱40 minutesLesson 8 of 10

Agent Communication and Supervision

Implement structured agent-to-agent messages, supervisory nodes, shared state coordination, and multi-agent orchestration patterns.

Agent Communication and Supervision

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.

Structured Agent Messages

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

ℹ️Note

Structured messages with sender, recipient, and type fields enable sophisticated routing. Agents can target messages to specific recipients or broadcast to all.

The Supervisor Pattern

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"}

Agent Nodes with Message Handling

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())}
    }]}

✅Success

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

Supervisor Routing

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 ────┘

Broadcast and Targeted Messages

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": {}
    }]}

Targeted (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": {}
    }]}

Dynamic Agent Selection

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()}

Sequential Multi-Agent with Handoff

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
    }

Complete Supervisor Example

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())

Shared State vs Agent-Scoped State

Pattern	Description	Use Case
Shared state	All agents read/write the same state dict	Simple coordination, few agents
Structured messages	Agents write to a shared message list	Complex multi-agent, history tracking
Subgraph state	Each agent has its own internal state, communicates via messages (next lesson)	Hierarchical agents, encapsulation

Practice Questions

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?

✅Success

Key Takeaways

Supervisor nodes coordinate multi-agent work by deciding the next step
Structured messages (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

Progress80%

Multi-Agent Basics

Streaming Tokens

Course Content

1. Advanced State Management 2. Persistence and Checkpointing 3. Conversation Memory 4. Summary Memory 5. Human-in-the-Loop with Interrupts 6. Approval Workflows 7. Multi-Agent Basics 8. Agent Communication and Supervision 9. Streaming Tokens 10. Intermediate Project: Multi-Agent Research System