Руководство по Anemoi-Style полусамоуправляемым системам

Обзор

В этом руководстве мы демонстрируем, как работает полусамоуправляемая многоагентная система в стиле Anemoi, позволяя двум агентам напрямую договариваться без менеджера. Мы показываем, как Драфтер и Критик итеративно уточняют вывод через обратную связь друг с другом, уменьшая накладные расходы на координацию при сохранении качества. Мы реализуем этот подход в Colab, используя LangGraph, сосредотачиваясь на ясности и практическом применении.

Настройка окружения

Мы настраиваем окружение Colab, устанавливая необходимые пакеты LangGraph и LangChain и собирая API-ключ OpenAI в скрытом режиме. Мы инициализируем языковую модель, которая будет использоваться всеми агентами, сохраняя минимальную и воспроизводимую конфигурацию.

!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("Введите OPENAI_API_KEY (скрыто): ")
 
MODEL = os.environ.get("OPENAI_MODEL", "gpt-4o-mini")
llm = ChatOpenAI(model=MODEL, temperature=0.2)

Определение состояния Anemoi

Мы определяем тип состояния, который действует как общая поверхность общения между агентами в процессе переговоров. Мы отслеживаем задачу, черновик, критику, флаг согласия и счетчик итераций, чтобы сохранять прозрачность.

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

Создание агента Драфтера

Мы реализуем агента Драфтера, который создает первоначальный ответ и пересматривает его, когда доступна обратная связь от коллег. Драфтер сосредоточен исключительно на улучшении пользовательского черновика.

DRAFTER_SYSTEM = """Вы агент A (Драфтер) в петле.
Ваша задача - написать качественное решение.
Если вы получаете критику, пересмотрите и учтите.
Верните только улучшенный текст."""
 
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}
 
КРИТИКА:
{critique}
 
Перепишите черновик."""
    else:
        user_msg = f"""ЗАДАЧА:
{task}
 
Напишите первый черновик."""
 
    draft = llm.invoke([
        {"role": "system", "content": DRAFTER_SYSTEM},
        {"role": "user", "content": user_msg},
    ]).content.strip()
 
    if state.get("trace", False):
        print(f"\n--- Драфтер Раунд {r} ---\n{draft}\n")
 
    return {**state, "round": r, "draft": draft, "agreed": False}

Реализация агента Критика

Мы реализуем агента Критика, который оценивает черновик и решает, готов ли он или требуются доработки. Этот этап оценки позволяет контролировать качество без введения управляющего агента.

CRITIC_SYSTEM = """Вы агент B (Критик).
Верните строгий 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"ЗАДАЧА:\n{task}\n\nЧЕРНОВИК:\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"--- Решение Критика ---\nСОГЛАСИЕ: {agree}\n{critique}\n")
 
    final = draft if agree else state.get("final", "")
    return {**state, "agreed": agree, "critique": critique, "final": final}

Управление потоком работы

Мы собираем рабочий процесс LangGraph, который управляет контролем между Драфтером и Критиком до тех пор, пока не будет достигнуто согласие или лимит раундов.

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"
 
# Дополнительные методы и установка графа...

Результаты

Мы выполняем граф и возвращаем лучший доступный вывод пользователю. Этот подход закладывает основу для расширения архитектуры для сложных агентов.

Почему это важно

Мы продемонстрировали, что переговоры в стиле Anemoi являются практичной альтернативой архитектуре управляющего работника. Они предлагают меньшую задержку и простую координацию. Это руководство предоставляет многократно используемый шаблон для построения масштабируемых полусамоуправляемых систем агентов.