En este tutorial, construimos un sistema avanzado de comunicación multiagente utilizando una arquitectura de bus de mensajes estructurados impulsada por LangGraph y Pydantic. Definimos un estricto esquema de mensajes estilo ACP que permite a los agentes comunicarse a través de un estado compartido en lugar de llamarse entre sí directamente, lo que permite modularidad, trazabilidad y orquestación de nivel de producción. Implementamos tres agentes especializados, un Planificador, un Ejecutor y un Validador, que se coordinan a través de mensajes estructurados, estado persistente y lógica de enrutamiento. También integramos la persistencia basada en SQLite para proporcionar memoria duradera entre ejecuciones y visualizar el flujo de comunicación del agente para comprender cómo se propagan los mensajes a través del sistema.
Tipo de mensaje = Literal[“task”, “plan”, “result”, “validation”, “error”, “control”]
clase 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”)) remitente: Rol receptor: Rol msg_type: MsgType contenido: str meta: Dict[str, Any] = Seguimiento de campo (default_factory = dict): Dict[str, Any] = Campo(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) -> Ninguno: con open(acp_log_path(), “a”, codificación=”utf-8″) como f: f.write(m.model_dump_json() + “\n”)
Instalamos e importamos todas las bibliotecas necesarias para construir un sistema estructurado de comunicación multiagente. Definimos el esquema de mensajes estilo ACP utilizando Pydantic, lo que nos permite aplicar un formato estricto y estructurado para la comunicación de los agentes. También implementamos registros estructurados para conservar cada mensaje intercambiado entre agentes, lo que permite la trazabilidad y observabilidad del sistema.
Definimos la estructura de estado compartido que actúa como bus de mensajes centralizado para todos los agentes. Implementamos la clase BusState para almacenar el objetivo, el buzón, la información de enrutamiento y el progreso de la ejecución. También creamos la función bus_update, que nos permite generar mensajes estructurados, actualizar el estado compartido y conservar registros de mensajes de manera consistente.
“Interpret the goal and extract requirements.”,
“Decide an execution strategy with clear outputs.”,
“Ask Executor to produce the result.”,
“Ask Validator to check correctness + completeness.”,
]
plan_text = “\n”.join([f”{i+1}. {p}” for i, p in enumerate(plan)]) return bus_update( estado, “planificador”, “ejecutor”, “plan”, plan_text, meta={“meta”: meta, “plan_steps”: len(plan)}, trace={“policy”: “deterministic_planner_v1”}, ) def executor_agent(state_dict: Dict[str, Any]) -> Dictar[str, Any]: estado = BusState.model_validate(state_dict) objetivo = estado.goal.strip() último_plan = Ninguno para m en reverso(estado.mailbox): if m.receiver == “ejecutor” y m.msg_type == “plan”: último_plan = m.content break resultado = { “objetivo”: objetivo, “supuestos”: [
“We can produce a concise, actionable output.”,
“We can validate via rule-based checks.”,
]”salida”: f”Tarea ejecutada para el objetivo: {objetivo}”, “entregables”: [
“A clear summary”,
“A step-by-step action list”,
“Any constraints and edge cases”,
]”plan_seen”: bool(latest_plan), } result_text = json.dumps(resultado, sangría=2) return bus_update( estado, “ejecutor”, “validador”, “resultado”, result_text, meta={“artifact_type”: “json”, “bytes”: len(result_text.encode(“utf-8”))}, trace={“policy”: “deterministic_executor_v1”}, )
Implementamos los agentes Planificador y Ejecutor, que se encargan de la planificación y ejecución de tareas. Diseñamos el agente Planner para interpretar el objetivo y generar un plan de ejecución estructurado, que luego pasa a través del bus de mensajes. Implementamos el agente Ejecutor para leer el plan, ejecutarlo y producir un artefacto de resultado estructurado que los agentes posteriores puedan validar.
return upd if last_result.msg_type == “error”: upd = bus_update( state, “validator”, “planner”, “validation”, f”La validación falló porque ocurrió un error ascendente: {latest_result.content}”, meta={“status”: “fail”}, ) upd[“done”] = Actualización verdadera[“errors”] = estado.errores + [latest_result.content]
devolver intento de actualización: parsed = json.loads(latest_result.content) excepto Excepción como e: upd = bus_update( state, “validator”, “planner”, “validation”, f”El resultado no es válido JSON: {e}”, meta={“status”: “fail”}, ) upd[“done”] = Actualización verdadera[“errors”] = estado.errores + [f”invalid_json: {e}”]
devolver problemas actualizados = []
if parsed.get(“meta”)! = objetivo: issues.append(“Result.goal no coincide con el objetivo de entrada.”) si los “entregables” no están analizados o no son instancia (analizado[“deliverables”]lista) o len(analizado[“deliverables”]) == 0: issues.append(“Lista de entregables faltantes o vacías.”) if issues: upd = bus_update( state, “validator”, “planner”, “validation”, “Validación fallida:\n- ” + “\n- “.join(issues), meta={“status”: “fail”, “issues”: issues}, ) upd[“done”] = Actualización verdadera[“errors”] = state.errors + problemas devuelven upd upd = bus_update( estado, “validador”, “usuario”, “validación”, “Validación aprobada ✅ El resultado parece consistente y completo.”, meta={“status”: “pass”}, ) upd[“done”] = Actualización verdadera[“errors”] = state.errors return upd def route_next(state_dict: Dict[str, Any]) -> str: if state_dict.get(“hecho”, False): devuelve END rol = state_dict.get(“active_role”, “usuario”) si rol == “planificador”: devuelve “planificador” si rol == “ejecutor”: devuelve “ejecutor” si rol == “validador”: devuelve “validador” devuelve END
Implementamos el agente Validador y la lógica de enrutamiento que controla el flujo de ejecución del agente. Diseñamos el Validador para inspeccionar los resultados de la ejecución, verificar la corrección y generar resultados de validación a través de controles estructurados. También implementamos la función de enrutamiento que determina dinámicamente qué agente debe ejecutarse a continuación, lo que permite una orquestación coordinada de múltiples agentes.
con open(ruta, “r”, codificación=”utf-8″) como f: líneas = f.readlines()[-n:]
devolver [json.loads(x) for x in lines]
print(“\nÚltimas entradas del registro ACP:”) para la fila en tail_jsonl(acp_log_path(), 6): print(f”{fila[‘msg_type’]:>10} | {fila[‘sender’]} -> {fila[‘receiver’]} | {fila[‘ts’]}”)
Construimos el gráfico de estado de LangGraph, habilitamos la persistencia basada en SQLite y ejecutamos el flujo de trabajo de múltiples agentes. Usamos un identificador de subproceso para garantizar que el estado del agente se pueda guardar y recuperar de manera confiable en todas las ejecuciones. También visualizamos los gráficos de orquestación y comunicación e inspeccionamos los registros persistentes, lo que nos permite comprender cómo interactúan los agentes a través del bus de mensajes estructurados.
En este tutorial, diseñamos e implementamos con éxito un marco de comunicación estructurado de múltiples agentes utilizando la arquitectura de estado compartido de LangGraph y los principios del bus de mensajes estilo ACP. Permitimos que los agentes operaran de forma independiente mientras se comunicaban a través de mensajes estructurados y persistentes, lo que mejora la confiabilidad, la observabilidad y la escalabilidad. Registramos cada interacción, mantuvimos el estado del agente en las ejecuciones y visualizamos patrones de comunicación para obtener una visión profunda de la coordinación de los agentes. Esta arquitectura nos permite construir sistemas multiagente robustos, modulares y listos para producción que se pueden ampliar con agentes adicionales, razonamiento LLM, sistemas de memoria y estrategias de enrutamiento complejas.
Consulte los códigos completos aquí. Además, no dude en seguirnos en Twitter y no olvide unirse a nuestro SubReddit de más de 120.000 ML y suscribirse a nuestro boletín. ¡Esperar! estas en telegrama? Ahora también puedes unirte a nosotros en Telegram.