Мультиагентная система

В этой главе представлена мультиагентная система.

Мультиагентная система выполняет сложные задачи за счёт взаимодействия нескольких агентов.

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

Основы мультиагентной архитектуры

Ключевые проблемы мультиагентной системы включают три аспекта.

Разделение труда: как рационально распределять задачи между разными агентами.

Коммуникация: как информация передаётся между агентами.

Координация: как обеспечить согласованность и эффективность действий нескольких агентов.

Базовые архитектурные паттерны

Иерархическая архитектура

Иерархическая архитектура использует древовидную структуру, в которой главный агент отвечает за планирование.

Главный агент (оркестратор) отвечает за декомпозицию задач и интеграцию результатов.

Дочерний агент (субагент) отвечает за выполнение конкретных задач.

Orchestrator (调度者)
    |
    +-- Agent A (专家A)
    +-- Agent B (专家B)
    +-- Agent C (专家C)

Одноранговая архитектура

В одноранговой архитектуре все агенты равны и общаются и сотрудничают напрямую.

Подходит для сценариев, где агентам требуется частое одноранговое взаимодействие.

Agent A <--> Agent B
    |           |
    v           v
Agent C <--> Agent D

Разделение ролей и сотрудничество

Эффективная мультиагентная система требует чётких определений ролей.

Каждый агент должен иметь собственную специализацию и область ответственности.

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

Общие определения ролей

Планировщик: разбивает сложные задачи и формулирует планы выполнения.

Исполнитель: вызывает инструменты и выполняет конкретные операции.

Проверяющий: оценивает результаты и предоставляет предложения по улучшению.

Координатор: управляет коммуникацией и распределением задач между агентами.

Реализация кода

Базовая реализация мультиагентной системы

class Agent:
"""
Базовый класс агента
Каждый агент имеет свою роль, инструменты и логику выполнения.
"""

def __init__(self, role, llm, tools):
# Название роли агента
self.role = role
# Связанная языковая модель
self.llm = llm
# Список доступных инструментов
self.tools = tools

def execute(self, task, context=None):
"""
Выполнить задачу
:param task: Описание задачи
:param context: Информация о контексте
:return: Результат выполнения
"""
# Собрать промпт
prompt = self.build_prompt(task, context)

# Сгенерировать ответ
response = self.llm.generate(prompt)

# Если нужен вызов инструмента
if response.needs_tool_call:
return self.execute_tool(response)

return response.content

def build_prompt(self, task, context):
"""Построить промпт агента"""
return f"Роль: {self.role}\nЗадача: {task}\nКонтекст: {context}"

def execute_tool(self, response):
"""Выполнить вызов инструмента"""
tool_name = response.tool_name
tool_input = response.tool_input

# Найти соответствующий инструмент в списке инструментов
tool = self.get_tool(tool_name)

# Выполнить инструмент
return tool.execute(tool_input)

class MultiAgentSystem:
"""
Мультиагентная система совместной работы
Отвечает за управление совместной работой нескольких агентов
"""

def __init__(self, agents, coordinator):
# Словарь агентов, ключ — имя роли
self.agents = {a.role: a for a in agents}
# Агент-координатор
self.coordinator = coordinator

def solve(self, task):
"""
Решать сложные задачи
Реализовано путем координации нескольких агентов для совместной работы
"""
# Шаг 1: координатор анализирует задачу и определяет, какие агенты должны участвовать
plan = self.coordinator.decompose(task)

# Шаг 2: Распределите задачи соответствующим агентам
results = {}
for role, subtask in plan.subtasks.items():
# Получить агента с соответствующей ролью
agent = self.agents.get(role)
if agent:
# Выполнить подзадачу
results[role] = agent.execute(subtask)

# Шаг 3: Координатор интегрирует результаты
final_result = self.coordinator.integrate(results)

return final_result

class CoordinatorAgent:
"""Агент-координатор"""

def __init__(self, llm):
self.llm = llm

def decompose(self, task):
"""
Декомпозировать задачу
Проанализируйте задачу, определите, какие агенты нужны и как разделить работу
"""
prompt = f"""
Проанализируйте следующую задачу, разбейте её на подзадачи и назначьте их соответствующим агентам.

Задача: {task}

Пожалуйста, выведите:
1. Какие агенты нужны (роли)
2. За какую подзадачу отвечает каждый агент
3. Зависимости между подзадачами

Формат вывода: JSON
"""
response = self.llm.generate(prompt)
return Plan.parse(response)

def integrate(self, results):
"""
Объедините результаты нескольких агентов
"""
prompt = f"""
Объедините результаты выполнения следующих агентов, чтобы сгенерировать окончательный ответ.

Список результатов:
{results}

Пожалуйста, объедините и выведите окончательный результат.
"""
return self.llm.generate(prompt)

Фреймворк AutoGen

AutoGen — это мультиагентный фреймворк программирования, разработанный Microsoft.

Он предоставляет лаконичный API для создания мультиагентных диалоговых систем.

AutoGen значительно упрощает сложность разработки мультиагентных систем.

Основные концепции

AssistantAgent: Интеллектуальный агент-ассистент, который может вызывать инструменты.

UserProxyAgent: Представляет поведение пользователя, может выполнять код и вызовы инструментов.

GroupChat: Поддерживает совместную работу в групповом чате между несколькими агентами.

GroupChatManager: Управляет взаимодействиями агентов в групповом чате.

Примеры кода

Базовое использование AutoGen

# Импорт фреймворка autogen
import autogen
from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager

# ==================== Настройка LLM ====================
# Создание словаря конфигурации
llm_config = {
"model": "gpt-4",
"api-key": "your-api-key", # В реальном использовании считывается из переменных среды
"temperature": 0.7
}

# ==================== Создание одного агента ====================
# Создание агента-ассистента
assistant = AssistantAgent(
name="assistant",
system_message="""
Вы — полезный помощник по программированию на Python.
Вы можете помогать пользователям писать, отлаживать и оптимизировать код.
""",
llm_config=llm_config
)

# Создайте пользовательского прокси-агента
# human_input_mode="NEVER" означает, что ручной ввод не требуется
# max_consecutive_auto_reply=10 означает до 10 последовательных автоматических ответов
user_proxy = UserProxyAgent(
name="user_proxy",
human_input_mode="NEVER",
max_consecutive_auto_reply=10
)

# ==================== Разговор с одним агентом ====================
# Начать разговор
user_proxy.initiate_chat(
assistant,
message="Помогите мне написать алгоритм быстрой сортировки"
)

# ==================== Групповой чат мультиагентов ====================
# Создание нескольких агентов
coder = AssistantAgent(
name="coder",
system_message="Вы эксперт по программированию на Python, отвечаете за написание кода.",
llm_config=llm_config
)

reviewer = AssistantAgent(
name="reviewer",
system_message="Вы эксперт по проверке кода, отвечаете за оценку качества кода.",
llm_config=llm_config
)

# Создать групповой чат
group_chat = GroupChat(
agents=[coder, reviewer],
messages=[],
max_round=10 # До 10 раундов разговора
)

# Создать менеджер группового чата
manager = GroupChatManager(
groupchat=group_chat,
llm_config=llm_config
)

# Запустить групповой чат
user_proxy.initiate_chat(
manager,
message="Напишите алгоритм быстрой сортировки и проверьте код"
)

Более сложный пример AutoGen

Вызов инструментов AutoGen

import autogen
from autogen import AssistantAgent, UserProxyAgent

# Определить список инструментов
# Эти инструменты будут зарегистрированы в списке инструментов агента
tools = [
{
"name": "calculate",
"description": "Выполнить математические вычисления",
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "Математическое выражение для вычисления, например, '2+3*5'"
}
},
"required": ["expression"]
}
},
{
"name": "search_web",
"description": "Поиск информации в интернете",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Ключевое слово для поиска"
}
},
"required": ["query"]
}
}
]

# Создать агента с поддержкой вызова инструментов
assistant = AssistantAgent(
name="assistant",
system_message="""
Вы — полезный ассистент.
Когда вам нужно выполнить вычисления или найти информацию, вы можете вызвать соответствующие инструменты.
""",
llm_config={
"model": "gpt-4",
"api_key": "your-api-key"
},
tools=tools # Регистрация инструментов
)

# Создать прокси пользователя
user_proxy = UserProxyAgent(
name="user_proxy",
human_input_mode="NEVER"
)

# Начать разговор, Агент автоматически вызовет инструменты
user_proxy.initiate_chat(
assistant,
message="Чему равно (15 + 25) * 2?"
)

# Пример ответа:
# Агент может ответить:
# "Let me calculate it: (15 + 25) * 2 = 40 * 2 = 80"
# Или напрямую вызвать инструмент calculate, чтобы получить результат

Протоколы A2A и MCP

Мультиагентные системы требуют стандартизированных протоколов связи.

A2A и MCP — два важных стандарта протоколов.

Протокол A2A (Agent-to-Agent)

Протокол A2A определяет стандартный формат обмена данными между агентами.

Он поддерживает обнаружение агентов, совместное выполнение задач и синхронизацию состояния.

Основные возможности

Обнаружение агентов: агенты могут обнаруживать возможности и сервисы других агентов.

Совместная работа над задачами: несколько агентов могут совместно выполнять сложные задачи.

Синхронизация состояния: агенты могут синхронизировать информацию о состоянии друг с другом.

Протокол MCP (Model Context Protocol)

MCP — это открытый стандартный протокол.

Он позволяет ИИ-моделям безопасно подключаться к внешним инструментам и источникам данных.

MCP — стандартный протокол для интеграции инструментов, аналогичный по роли интерфейсу USB.

Основные компоненты

Хост: Хост-среда для запуска ИИ-приложений.

Клиент: Клиент, устанавливающий соединения с MCP-серверами.

Сервер: Сервер, предоставляющий инструменты и ресурсы.

Пример определения инструмента MCP

Определение инструмента MCP

{
"name": "filesystem",
"description": "Инструмент для операций с файловой системой",
"version": "1.0.0",
"tools": [
{
"name": "read_file",
"description": "Чтение содержимого файла",
"inputSchema": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Путь к файлу, обязательно"
},
"encoding": {
"type": "string",
"description": "Кодировка файла, необязательно, по умолчанию utf-8"
}
},
"required": ["path"]
}
},
{
"name": "write_file",
"description": "Записать содержимое файла",
"inputSchema": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Путь к файлу, обязательно"
},
"content": {
"type": "string",
"description": "Содержимое файла, обязательно"
}
},
"required": ["path", "content"]
}
},
{
"name": "list_directory",
"description": "Вывести содержимое каталога",
"inputSchema": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Путь к каталогу"
}
}
}
}
],
"resources": [
{
"uri": "file://config",
"description": "Файл конфигурации"
}
]
}

Примечание: A2A и MCP решают разные задачи. A2A отвечает за коммуникацию между агентами, а MCP — за подключение агентов к внешним инструментам. Оба можно использовать вместе для построения полноценной многоагентной системы.

Паттерн агента «ведущий-ведомый»

Режим «мастер-субагент» — классический паттерн мультиагентной архитектуры.

Главный агент (оркестратор) отвечает за декомпозицию задач и интеграцию результатов.

Субагент (Subagent) отвечает за выполнение конкретных задач.

Применимые сценарии

Сценарии, в которых задачи можно чётко декомпозировать на независимые подзадачи.

Сценарии, требующие сотрудничества агентов с разной специализацией.

Сложные рабочие процессы, требующие центральной координации и управления.

Реализация кода

Реализация паттерна агента «ведущий-ведомый»

class OrchestratorAgent:
"""
Главный агент оркестрации (Master Agent)
Отвечает за декомпозицию задач, назначение и интеграцию результатов
"""

def __init__(self, llm, subagents):
# Языковая модель
self.llm = llm
# Словарь субагентов
self.subagents = {a.role: a for a in subagents}

def handle_request(self, request):
"""
Обработка запросов пользователей
Полный поток оркестрации
"""
# ==================== Фаза анализа ====================
# Проанализировать запрос, чтобы определить, необходимо ли взаимодействие нескольких агентов
plan = self.llm.plan(request)

if len(plan.subtasks) == 1:
# Единственная задача — напрямую направить соответствующему агенту
target_role = plan.target_role
return self.subagents[target_role].execute(plan.subtasks[0])

# ==================== Multi-Agent Collaboration Phase ====================
# Create task queue
task_queue = self.create_task_queue(plan.subtasks)

# Store completed results
results = []

# Processing loop until all tasks are complete
while task_queue:
# Get tasks that can be executed in parallel
parallel_tasks = self.get_parallel_tasks(task_queue)

# Execute in parallel
parallel_results = self.execute_in_parallel(parallel_tasks)
results.extend(parallel_results)

# Обновить очередь задач
# Обработать зависимости, удалить завершённые задачи, добавить новые задачи
task_queue = self.update_queue(task_queue, parallel_results)

# ==================== Этап интеграции ====================
# Интегрировать все результаты
return self.integrate_results(results)

def create_task_queue(self, subtasks):
"""Создать очередь задач"""
return TaskQueue(subtasks)

def get_parallel_tasks(self, task_queue):
"""
Получить задачи, которые можно выполнять параллельно
Исключить задачи с зависимостями
"""
ready_tasks = []
for task in task_queue.tasks:
# Проверить, все ли зависимости задач выполнены
if all(dep in task_queue.completed for dep in task.dependencies):
ready_tasks.append(task)
return ready_tasks

def execute_in_parallel(self, tasks):
"""Выполнить несколько задач параллельно"""
import concurrent.futures

results = []
with concurrent.futures.ThreadPoolExecutor() as executor:
# Отправить все задачи
future_to_task = {
executor.submit(self.execute_task, task): task
for task in tasks
}

# Собрать результаты
for future in concurrent.futures.as_completed(future_to_task):
task = future_to_task[future]
try:
result = future.result()
results.append(result)
except Exception as e:
results.append(TaskResult(task=task, error=str(e)))

return results

def execute_task(self, task):
"""Выполнить одну задачу"""
agent = self.subagents.get(task.role)
if not agent:
return TaskResult(task=task, error=f"Неизвестная роль: {task.role}")

return agent.execute(task.description)

def update_queue(self, task_queue, completed_results):
"""Обновить очередь задач"""
# Отметить выполненные задачи как завершённые
for result in completed_results:
task_queue.completed.add(result.task.id)

# Удалить выполненные задачи
task_queue.tasks = [
t for t in task_queue.tasks
if t.id not in task_queue.completed
]

return task_queue

def integrate_results(self, results):
"""
Интегрировать результаты от нескольких агентов
Сгенерировать итоговый ответ
"""
prompt = f"""
Интегрируйте результаты выполнения следующих агентов и сгенерируйте итоговый ответ.

Список результатов:
{results}

Пожалуйста, объедините в полный, связный ответ.
"""
return self.llm.generate(prompt)

class Subagent:
"""Из базового класса Agent"""

def __init__(self, role, llm, tools=Отсутствует):
self.role = role
self.llm = llm
self.tools = tools or []

def execute(self, task):
"""Конкретная логика выполнения задачи"""
raise NotImplementedError

Плюсы и минусы шаблона «ведущий-ведомый»

Преимущества	Недостатки
Понятная структура, легко понять и реализовать	Мастер-агент становится единой точкой отказа
Централизованная координация, легко контролировать	Мастер-агент может стать узким местом производительности
Подходит для хорошо разделенных на уровни задач	Низкая гибкость, сложно динамически настраивать
Удобная отладка, каждая подзадача может быть отслежена	Не подходит для сценариев с частым одноранговым взаимодействием

Краткое содержание главы

В этой главе представлены основные концепции и методы реализации мультиагентных систем.

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

Разделение ролей и сотрудничествоПовышение эффективности системы за счет четкого определения ролей.

Фреймворк AutoGenПредоставляет программный фреймворк, упрощающий разработку мультиагентных систем.

Протоколы A2A и MCPЭто стандартные протоколы для взаимодействия агентов и интеграции инструментов.

Паттерн агентов «ведущий–ведомый»Подходит для сложных задач с чёткой иерархической структурой.

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

Для простых сценариев достаточно одного агента, чтобы удовлетворить требованиям.

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

Другие расширения

Инструменты агента

Python: реализация интеллектуальных агентов

Мультиагентная система

Основы мультиагентной архитектуры

Базовые архитектурные паттерны

Иерархическая архитектура

Одноранговая архитектура

Разделение ролей и сотрудничество

Общие определения ролей

Реализация кода

Базовая реализация мультиагентной системы

Фреймворк AutoGen

Основные концепции

Примеры кода

Базовое использование AutoGen

Более сложный пример AutoGen

Вызов инструментов AutoGen

Протоколы A2A и MCP

Протокол A2A (Agent-to-Agent)

Основные возможности

Протокол MCP (Model Context Protocol)

Основные компоненты

Пример определения инструмента MCP

Определение инструмента MCP

Паттерн агента «ведущий-ведомый»

Применимые сценарии

Реализация кода

Реализация паттерна агента «ведущий-ведомый»

Плюсы и минусы шаблона «ведущий-ведомый»

Краткое содержание главы