Implementing Anemoi-Style Peer-to-Peer Systems in LangGraph

Overview

In this tutorial, we demonstrate how a semi-centralized Anemoi-style multi-agent system works by letting two peer agents negotiate directly without a manager or supervisor. We show how a Drafter and a Critic iteratively refine an output through peer-to-peer feedback, reducing coordination overhead while preserving quality. We implement this pattern end-to-end in Colab using LangGraph, focusing on clarity, control flow, and practical execution rather than abstract orchestration theory.

Setting Up the Environment

We set up the Colab environment by installing the required LangGraph and LangChain packages and securely collecting the OpenAI API key as a hidden input. We initialize the language model that will be shared by all agents, keeping the configuration minimal and reproducible.

!pip -q install -U langgraph langchain-openai langchain-core
 
import os
import json
from getpass import getpass
from typing import TypedDict
 
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
 
if not os.environ.get("OPENAI_API_KEY"):
    os.environ["OPENAI_API_KEY"] = getpass("Enter OPENAI_API_KEY (hidden): ")
 
MODEL = os.environ.get("OPENAI_MODEL", "gpt-4o-mini")
llm = ChatOpenAI(model=MODEL, temperature=0.2)

Defining the Anemoi State

We define a typed state that acts as the shared communication surface between agents during negotiation. We explicitly track the task, draft, critique, agreement flag, and iteration count to keep the flow transparent and debuggable.

class AnemoiState(TypedDict):
    task: str
    max_rounds: int
    round: int
    draft: str
    critique: str
    agreed: bool
    final: str
    trace: bool

Creating the Drafter Agent

We implement the Drafter agent, which produces the initial response and revises it whenever peer feedback is available. The Drafter focuses purely on improving the user-facing draft.

DRAFTER_SYSTEM = """You are Agent A (Drafter) in a peer-to-peer loop.
You write a high-quality solution to the user's task.
If you receive critique, you revise decisively and incorporate it.
Return only the improved draft text."""
 
def drafter_node(state: AnemoiState) -> AnemoiState:
    task = state["task"]
    critique = state.get("critique", "").strip()
    r = state.get("round", 0) + 1
 
    if critique:
        user_msg = f"""TASK:
{task}
 
CRITIQUE:
{critique}
 
Revise the draft."""
    else:
        user_msg = f"""TASK:
{task}
 
Write the first draft."""
 
    draft = llm.invoke([
        {"role": "system", "content": DRAFTER_SYSTEM},
        {"role": "user", "content": user_msg},
    ]).content.strip()
 
    if state.get("trace", False):
        print(f"\n--- Drafter Round {r} ---\n{draft}\n")
 
    return {**state, "round": r, "draft": draft, "agreed": False}

Implementing the Critic Agent

We implement the Critic agent, which evaluates the draft and decides whether it is ready or needs revision. This peer evaluation step allows quality control without introducing a supervisory agent.

CRITIC_SYSTEM = """You are Agent B (Critic).
Return strict JSON:
{"agree": true/false, "critique": "..."}"""
 
def critic_node(state: AnemoiState) -> AnemoiState:
    task = state["task"]
    draft = state.get("draft", "")
 
    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()
 
    try:
        data = json.loads(cleaned)
        agree = bool(data.get("agree", False))
        critique = str(data.get("critique", "")).strip()
    except Exception:
        agree = False
        critique = raw
 
    if state.get("trace", False):
        print(f"--- Critic Decision ---\nAGREE: {agree}\n{critique}\n")
 
    final = draft if agree else state.get("final", "")
    return {**state, "agreed": agree, "critique": critique, "final": final}

Managing the Workflow

We assemble the LangGraph workflow that routes control between Drafter and Critic until agreement is reached or the maximum round limit is reached.

def continue_or_end(state: AnemoiState) -> str:
    if state.get("agreed", False):
        return "end"
    if state.get("round", 0) >= state.get("max_rounds", 3):
        return "force_ship"
    return "loop"
 
# Additional methods and graph setup...

Results

We execute the graph and return the best available output to the user. This approach lays the foundation for extending the architecture to complex agent systems.

Why It Matters

We demonstrated that Anemoi-style peer negotiation is a practical alternative to manager-worker architectures. It offers reduced latency and simpler coordination. This tutorial provides a reusable blueprint for building scalable, semi-centralized agent systems.