05-03-02 — Comunicação entre agentes: shared state vs message passing

Shared State (estado compartilhado)

Todos os agentes acessam e modificam um objeto de estado centralizado. É como um quadro branco que todos podem ver e escrever.

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
import operator

# Define o estado compartilhado (typed)
class ResearchState(TypedDict):
    query: str
    search_results: list[str]
    analysis: str
    final_report: str
    messages: Annotated[list, operator.add]  # append-only

# Cada nó (agente) lê e escreve no estado
def researcher(state: ResearchState) -> ResearchState:
    results = search_web(state["query"])
    return {"search_results": results}  # Atualiza apenas seu campo

def analyst(state: ResearchState) -> ResearchState:
    # Lê os resultados do pesquisador
    analysis = analyze(state["search_results"])
    return {"analysis": analysis}

def writer(state: ResearchState) -> ResearchState:
    report = write_report(state["analysis"])
    return {"final_report": report}

# Grafo com estado compartilhado (LangGraph)
graph = StateGraph(ResearchState)
graph.add_node("researcher", researcher)
graph.add_node("analyst", analyst)
graph.add_node("writer", writer)
graph.add_edge("researcher", "analyst")
graph.add_edge("analyst", "writer")
graph.add_edge("writer", END)

app = graph.compile()
result = app.invoke({"query": "Market share ERP Brasil 2024"})

Message Passing (passagem de mensagens)

Agentes se comunicam enviando mensagens estruturadas. Não há estado global — cada agente tem seu próprio contexto e recebe apenas o que é enviado a ele.

import asyncio
from dataclasses import dataclass
from typing import Any

@dataclass
class AgentMessage:
    sender: str
    recipient: str
    task: str
    data: Any
    correlation_id: str

# Message bus simples
class MessageBus:
    def __init__(self):
        self.queues: dict[str, asyncio.Queue] = {}
    
    def register(self, agent_name: str):
        self.queues[agent_name] = asyncio.Queue()
    
    async def send(self, message: AgentMessage):
        await self.queues[message.recipient].put(message)
    
    async def receive(self, agent_name: str) -> AgentMessage:
        return await self.queues[agent_name].get()

# Agente que usa message passing
class ResearcherAgent:
    def __init__(self, bus: MessageBus):
        self.bus = bus
        bus.register("researcher")
    
    async def run(self):
        while True:
            msg = await self.bus.receive("researcher")
            results = await search_web(msg.data["query"])
            
            # Envia resultado para o próximo agente
            await self.bus.send(AgentMessage(
                sender="researcher",
                recipient="analyst",
                task="analyze",
                data={"results": results},
                correlation_id=msg.correlation_id
            ))

Trade-offs: quando usar cada um