Мультиагентные системы: Оркестрация взаимодействия

Введение: От солиста к оркестру

Приветствую вас в шестом модуле! До этого момента мы рассматривали Gemini 3 API как мощный, но одиночный инструмент. Мы учились писать идеальные промпты, настраивать параметры и использовать инструменты. Это похоже на обучение виртуозного пианиста.

Однако в реальном Enterprise-мире задачи редко решаются в одиночку. Сложные бизнес-процессы требуют не просто «умного чат-бота», а целой команды специалистов. Представьте разработку ПО: здесь нужен продукт-менеджер, архитектор, разработчик и тестировщик. Если вы попытаетесь запихнуть все эти роли в один системный промпт, модель начнет «галлюцинировать» от перегрузки контекста или путаться в обязанностях.

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

Топология взаимодействия агентов

Прежде чем писать код, нам нужно выбрать архитектурный паттерн. В мире LLM существует несколько проверенных схем оркестрации:

• Последовательная цепь (The Chain):
  Самый простой вариант. Выход Агента А становится входом Агента Б.
  Пример: Агент-Переводчик -> Агент-Редактор -> Агент-Форматировщик. Идеально для линейных задач.
• Маршрутизатор (The Router):
  Центральный узел анализирует запрос и направляет его нужному специалисту.
  Пример: Служба поддержки. Запрос «не работает вход» идет к Техническому агенту, а «верните деньги» — к Финансовому агенту.
• Иерархия (Manager-Worker):
  Наиболее мощный паттерн для сложных задач. Есть «Супервизор» (Manager), который разбивает большую задачу на подзадачи и делегирует их «Рабочим» (Workers). Рабочие возвращают результат Менеджеру, который собирает итоговый ответ.
• Совместная работа (Joint Collaboration/Debate):
  Агенты выступают как равноправные участники, обсуждая проблему. Один генерирует идею, другой критикует, третий предлагает альтернативу.

В этом уроке мы сосредоточимся на паттерне Иерархия (Supervisor), так как он наиболее востребован в бизнесе и позволяет максимально раскрыть потенциал Gemini 3.

Ключевой компонент: Общее состояние (Shared State)

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

Вместо этого мы используем концепцию общего состояния. Это структурированный объект (чаще всего JSON или словарь Python), который содержит:

1. Текущую цель (Goal).
2. Собранные факты/данные (Knowledge).
3. Историю действий (Plan execution status).
4. Последнее сообщение (Last message).

Оркестратор (код на Python) управляет этим состоянием, передавая агентам только то, что им нужно знать в данный момент.

import google.generativeai as genai
from typing import List, Dict, Any
import json

# Настройка API (предполагаем, что ключ уже в окружении)
# genai.configure(api_key="YOUR_API_KEY")

class Agent:
    """
    Базовый класс агента. 
    Каждый агент имеет свою специализацию (System Instruction).
    """
    def __init__(self, name: str, role: str, model_name: str = "gemini-3-pro"):
        self.name = name
        self.role = role
        self.model = genai.GenerativeModel(
            model_name=model_name,
            system_instruction=f"Ты - {name}. Твоя роль: {role}. Отвечай кратко и по делу."
        )
        self.chat = self.model.start_chat(history=[])

    def send_message(self, message: str) -> str:
        response = self.chat.send_message(message)
        return response.text

# Пример создания специалистов
researcher = Agent(
    name="Исследователь", 
    role="Ты ищешь точную информацию по запросу. Используй факты."
)

writer = Agent(
    name="Копирайтер", 
    role="Ты пишешь увлекательные тексты на основе предоставленных фактов."
)

reviewer = Agent(
    name="Редактор", 
    role="Ты проверяешь текст на ошибки и стиль. Если текст плохой, укажи, что исправить."
)

Оркестрация: Ручное управление потоком

Теперь создадим «Мозг» нашей системы. Это простой скрипт, который будет координировать передачу данных между агентами. Обратите внимание: мы не используем LLM для самой маршрутизации в этом простом примере (хотя могли бы). Мы жестко задаем логику процесса (workflow), что часто надежнее для бизнес-задач.

Сценарий: Написание поста для блога.

1. Исследователь находит информацию по теме.
2. Копирайтер пишет черновик.
3. Редактор оценивает результат.

def orchestrate_blog_post(topic: str):
    print(f"--- НАЧАЛО РАБОТЫ: {topic} ---\n")

    # ШАГ 1: Исследование
    print(f"🤖 {researcher.name} приступает к работе...")
    facts = researcher.send_message(f"Найди 3 ключевых факта о теме: {topic}")
    print(f"> Факты собраны:\n{facts}\n")

    # ШАГ 2: Написание
    print(f"🤖 {writer.name} пишет черновик...")
    # Мы передаем писателю не просто тему, а уже найденные факты
    draft = writer.send_message(
        f"Напиши короткий пост для блога на тему '{topic}', используя эти факты:\n{facts}"
    )
    print(f"> Черновик готов:\n{draft}\n")

    # ШАГ 3: Редактура
    print(f"🤖 {reviewer.name} проверяет...")
    review = reviewer.send_message(
        f"Проверь этот текст. Если он хороший, ответь 'OK'. Если нет, напиши замечания.\nТекст:\n{draft}"
    )
    print(f"> Вердикт редактора: {review}\n")

    # Простейшая логика ветвления
    if "OK" in review:
        return draft
    else:
        # В реальной системе здесь был бы цикл доработки (loop)
        print("⚠️ Требуется доработка! Отправляем обратно писателю...")
        fix_request = writer.send_message(f"Редактор недоволен. Вот замечания: {review}. Перепиши текст.")
        return fix_request

# Запуск
final_post = orchestrate_blog_post("Влияние ИИ на образование")
print(f"=== ИТОГОВЫЙ РЕЗУЛЬТАТ ===\n{final_post}")

Продвинутая техника: Агент-Оркестратор (LLM as Router)

В коде выше логика была линейной (`A -> B -> C`). Но что, если мы не знаем заранее, какой агент нужен? Здесь на сцену выходит Агент-Оркестратор.

Мы создаем агента, цель которого — не выполнять задачу, а делегировать её. Для этого идеально подходит механизм Function Calling (вызов функций) в Gemini 3.

Мы описываем агентов как инструменты. Оркестратор «вызывает функцию» (например, `call_researcher`), а наш код перехватывает этот вызов и передает сообщение реальному агенту.

Структура промпта для Оркестратора:

«Ты — главный менеджер проекта. У тебя есть команда: [Исследователь, Математик, Кодер]. Твоя задача — понять запрос пользователя и выбрать, к кому обратиться. Не отвечай на вопрос сам, используй инструменты.»

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

# Концептуальный пример использования Function Calling для маршрутизации

tools_schema = [
    {
        "function_declarations": [
            {
                "name": "ask_researcher",
                "description": "Используй это для поиска фактов и общей информации.",
                "parameters": {
                    "type": "OBJECT",
                    "properties": {
                        "query": {"type": "STRING", "description": "Вопрос к исследователю"}
                    },
                    "required": ["query"]
                }
            },
            {
                "name": "ask_coder",
                "description": "Используй это для написания или отладки кода.",
                "parameters": {
                    "type": "OBJECT",
                    "properties": {
                        "task": {"type": "STRING", "description": "Задание для программиста"}
                    },
                    "required": ["task"]
                }
            }
        ]
    }
]

model = genai.GenerativeModel('gemini-3-pro', tools=tools_schema)
chat = model.start_chat()

# Запрос пользователя
response = chat.send_message("Мне нужно написать скрипт на Python для парсинга сайта")

# Gemini 3 вернет вызов функции:
# function_call { name: "ask_coder", args: { task: "написать скрипт на Python для парсинга сайта" } }

# Ваш код должен распарсить этот ответ и передать его Агенту-Программисту

Проблемы и лучшие практики

Построение мультиагентных систем сопряжено с рядом рисков. Давайте разберем, как их минимизировать.

• Бесконечные циклы: Агент А просит Агента Б уточнить, Агент Б просит Агента А переформулировать.
  Решение: Всегда устанавливайте жесткий лимит итераций (например, `max_turns = 10`). Если лимит исчерпан, Оркестратор должен прервать цикл и вернуть ошибку или лучший имеющийся результат.
• Потеря контекста: При передаче сообщения от агента к агенту детали могут теряться.
  Решение: Используйте «структурированный вывод» (JSON). Пусть каждый агент возвращает ответ в формате: `{"thought": "мои мысли", "action": "что я сделал", "output": "результат"}`. Это помогает следующему агенту понять логику предыдущего.
• Стоимость и задержка: 5 агентов = 5 вызовов API. Это дороже и дольше.
  Решение: Используйте более легкие модели (Gemini Flash) для простых задач (рабочие агенты) и мощные модели (Gemini Pro/Ultra) для Оркестратора и финальной проверки.

python
import google.generativeai as genai

# Настройка агентов
city_agent = genai.GenerativeModel(
    'gemini-3-flash',
    system_instruction="Ты помощник туриста. При получении страны верни ТОЛЬКО 3 популярных города через запятую. Без лишних слов."
)

guide_agent = genai.GenerativeModel(
    'gemini-3-pro',
    system_instruction="Ты опытный гид. Для каждого полученного города напиши ОДНО главное занятие."
)

def plan_trip(country):
    # Шаг 1
    cities_response = city_agent.generate_content(country)
    cities = cities_response.text.strip()
    print(f"Выбраны города: {cities}")
    
    # Шаг 2
    itinerary = guide_agent.generate_content(f"Составь план для этих городов: {cities}")
    return itinerary.text

# Тест
print(plan_trip("Япония"))