Una guía de codificación para sistemas agentes semicentralizados estilo Anemoi que utilizan bucles críticos de igual a igual en LangGraph

En este tutorial, demostramos cómo funciona un sistema multiagente semicentralizado estilo Anemoi al permitir que dos agentes pares negocien directamente sin un gerente o supervisor. Mostramos cómo un redactor y un crítico refinan iterativamente un resultado a través de comentarios entre pares, reduciendo los gastos generales de coordinación y preservando la calidad. Implementamos este patrón de un extremo a otro en Colab utilizando LangGraph, centrándonos en la claridad, el flujo de control y la ejecución práctica en lugar de la teoría de orquestación abstracta. Consulta los CÓDIGOS COMPLETOS aquí.

!pip -q install -U langgraph langchain-openai langchain-core importar os importar json de getpass importar getpass de escribir importar TypedDict de langchain_openai importar ChatOpenAI de langgraph.graph importar StateGraph, FINALIZAR si no os.environ.get(“OPENAI_API_KEY”): os.environ[“OPENAI_API_KEY”] = getpass(“Ingrese OPENAI_API_KEY (oculto): “) MODELO = os.environ.get(“OPENAI_MODEL”, “gpt-4o-mini”) llm = ChatOpenAI(modelo=MODELO, temperatura=0.2)

Configuramos el entorno Colab instalando los paquetes LangGraph y LangChain necesarios y recopilando de forma segura la clave API de OpenAI como entrada oculta. Inicializamos el modelo de lenguaje que será compartido por todos los agentes, manteniendo la configuración mínima y reproducible. Consulta los CÓDIGOS COMPLETOS aquí.

clase AnemoiState (TypedDict): tarea: str max_rounds: int round: int draft: str crítica: str acordado: bool final: str trace: bool

Definimos un estado escrito que actúa como superficie de comunicación compartida entre agentes durante la negociación. Realizamos un seguimiento explícito de la tarea, el borrador, la crítica, el indicador de acuerdo y el recuento de iteraciones para mantener el flujo transparente y depurable. Este estado elimina la necesidad de un administrador central o de memoria implícita. Consulta los CÓDIGOS COMPLETOS aquí.

DRAFTER_SYSTEM = “””Usted es el Agente A (Redactor) en un bucle de igual a igual. Escribe una solución de alta calidad para la tarea del usuario. Si recibe críticas, las revisa decisivamente y las incorpora. Devuelve solo el texto borrador mejorado.””” def drafter_node(estado: AnemoiState) -> AnemoiState: tarea = estado[“task”]
critique = state.get(“critique”, “”).strip() r = state.get(“round”, 0) + 1 if critique: user_msg = f”””TASK: {task} CRÍTICA: {critique} Revisar el borrador.””” else: user_msg = f”””TASK: {task} Escribir el primer borrador.””” draft = llm.invoke(
[
{“role”: “system”, “content”: DRAFTER_SYSTEM},
{“role”: “user”, “content”: user_msg},
]
).content.strip() if state.get(“trace”, False): print(f”\n— Ronda de redacción {r} —\n{borrador}\n”) return {**estado, “ronda”: r, “borrador”: borrador, “acordado”: False}

Implementamos el agente Drafter, que produce la respuesta inicial y la revisa cada vez que hay comentarios de pares disponibles. Mantenemos al Drafter enfocado exclusivamente en mejorar el borrador de cara al usuario, sin tener en cuenta la lógica de control o las condiciones de terminación. Refleja la idea de Anemoi de que los agentes optimicen localmente mientras observan las señales de sus pares. Consulta los CÓDIGOS COMPLETOS aquí.

CRITIC_SYSTEM = “””Usted es el Agente B (Crítico). Devuelve JSON estricto: {“agree”: true/false, “critique”: “…”}””” def critic_node(state: AnemoiState) -> AnemoiState: task = state[“task”]
borrador = state.get(“borrador”, “”) raw = llm.invoke(
[
{“role”: “system”, “content”: CRITIC_SYSTEM},
{
“role”: “user”,
“content”: f”TASK:\n{task}\n\nDRAFT:\n{draft}”,
},
]
).content.strip() cleaned = raw.strip(““`”).replace(“json”, “”).strip() intente: data = json.loads(cleaned) de acuerdo = bool(data.get(“agree”, False)) critica = str(data.get(“critique”, “”)).strip() excepto Excepción: de acuerdo = False crítica = raw if state.get(“trace”, False): print(f”— Decisión crítica —\nAGREE: {agree}\n{critique}\n”) final = borrador si está de acuerdo else state.get(“final”, “”) return {**state, “agreed”: de acuerdo, “critique”: critica, “final”: final}

Implementamos el agente Crítico, que evalúa el borrador y decide si está listo para enviarse o necesita revisión. Aplicamos una decisión estricta de acordar o revisar para evitar comentarios vagos y garantizar una convergencia rápida. Este paso de evaluación por pares permite el control de calidad sin introducir un agente supervisor. Consulta los CÓDIGOS COMPLETOS aquí.

def continuar_or_end(estado: AnemoiState) -> str: if state.get(“acordado”, False): devuelve “fin” if state.get(“round”, 0) >= state.get(“max_rounds”, 3): devuelve “force_ship” return “loop” def force_ship_node(estado: AnemoiState) -> AnemoiState: devuelve {**state, “final”: state.get(“final”) o state.get(“draft”, “”)} gráfico = StateGraph(AnemoiState) Graph.add_node(“drafter”, drafter_node) Graph.add_node(“critic”, critic_node) Graph.add_node(“force_ship”, force_ship_node) Graph.set_entry_point(“drafter”) Graph.add_edge(“drafter”, “critic”) graph.add_conditional_edges( “critic”, continue_or_end, {“loop”: “drafter”, “force_ship”: “force_ship”, “end”: END}, ) graph.add_edge(“force_ship”, END) anemoi_critic_loop = graph.compile() demo_task = “””Explique el patrón de agente semicentralizado de Anemoi y por qué se reducen los bucles críticos entre pares cuellos de botella.””” resultado = anemoi_critic_loop.invoke( { “task”: demo_task, “max_rounds”: 3, “round”: 0, “draft”: “”, “critique”: “”, “agreed”: False, “final”: “”, “trace”: False, } ) print(“\n===========================”) print(“✅ FINAL SALIDA”) print(“====================\n”) print(resultado[“final”])

Montamos el flujo de trabajo LangGraph que dirige el control entre Drafter y Critic hasta que se llega a un acuerdo o se alcanza el límite máximo de rondas. Nos basamos en un enrutamiento condicional simple en lugar de una planificación centralizada, preservando así la naturaleza semicentralizada del sistema. Finalmente, ejecutamos el gráfico y devolvemos el mejor resultado disponible al usuario.

En conclusión, demostramos que la negociación entre pares al estilo Anemoi es una alternativa práctica a las arquitecturas gerente-trabajador, ya que ofrece una latencia más baja, una menor sobrecarga de contexto y una coordinación de agentes más simple. Al permitir que los agentes se monitorearan y corrigieran entre sí directamente, logramos la convergencia con menos tokens y menos complejidad de orquestación. En este tutorial, proporcionamos un modelo reutilizable para crear sistemas de agentes semicentralizados y escalables. Sienta las bases para extender el patrón a mallas de múltiples pares, bucles de equipo rojo o interoperabilidad de agentes basada en protocolos.

Consulta los CÓDIGOS COMPLETOS aquí. Además, no dude en seguirnos en Twitter y no olvide unirse a nuestro SubReddit de más de 100.000 ML y suscribirse a nuestro boletín. ¡Esperar! estas en telegrama? Ahora también puedes unirte a nosotros en Telegram.