Neste tutorial, construímos um sistema avançado de comunicação multi-agente usando uma arquitetura de barramento de mensagens estruturada alimentada por LangGraph e Pydantic. Definimos um esquema de mensagem estilo ACP rigoroso que permite que os agentes se comuniquem através de um estado compartilhado, em vez de chamarem uns aos outros diretamente, permitindo modularidade, rastreabilidade e orquestração de nível de produção. Implementamos três agentes especializados, um Planner, um Executor e um Validator, que se coordenam através de mensagens estruturadas, estado persistente e lógica de roteamento. Também integramos persistência baseada em SQLite para fornecer memória durável entre as execuções e visualizamos o fluxo de comunicação do agente para entender como as mensagens se propagam através do sistema. Copy Code Copiado Use um navegador diferente !pip -q install -U "pydantic==2.12.3" !pip -q install -U langgraph langchain-core networkx matplotlib !pip -q install -U langgraph-checkpoint-sqlite import os import json import uuid import sqlite3 from datetime import datetime, timezone from typing import Any, Dict, List, Literal, Optional, Tuple from pydantic import BaseModel, Field import networkx as nx import matplotlib.pyplot as plt from langgraph.graph import StateGraph, END from langgraph.checkpoint.sqlite import SqliteSaver Role = Literal["planner", "executor", "validator", "user", "system"] MsgType = Literal["task", "plan", "result", "validation", "error", "control"] class ACPMessage(BaseModel): msg_id: str = Field(default_factory=lambda: str(uuid.uuid4())) ts: str = Field(default_factory=lambda: datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")) sender: Role receiver: Role msg_type: MsgType content: str meta: Dict[str, Any] = Field(default_factory=dict) trace: Dict[str, Any] = Field(default_factory=dict) def acp_log_path() -> str: os.makedirs("acp_logs", exist_ok=True) return os.path.join("acp_logs", "acp_messages.jsonl") def append_acp_log(m: ACPMessage) -> None: with open(acp_log_path(), "a", encoding="utf-8") as f: f.write(m.model_dump_json() + "
") Instalamos e importamos todas as bibliotecas necessárias para construir um sistema de comunicação multi-agente estruturado. Definimos o esquema de mensagem estilo ACP usando Pydantic, o que nos permite impor um formato rigoroso e estruturado para a comunicação do agente. Também implementamos o registro estruturado para persistir cada mensagem trocada entre os agentes, permitindo a rastreabilidade e a observabilidade do sistema. Copy Code Copiado Use um navegador diferente class BusState(BaseModel): goal: str
Fonte: MarkTechPost
Publicado em 2026-03-01