A Coding Guide to Build Intelligent Multi-Agent Systems with the PEER Pattern

Создание многоагентной системы на основе PEER-паттерна (План, Выполнение, Выражение, Обзор) предоставляет уникальную возможность оптимизации бизнес-процессов. В этой статье мы рассмотрим, как реализовать такую систему с помощью Google Colab, интегрируя специализированных агентов и используя модель Gemini 1.5 Flash через бесплатный API. Мы будем исследовать, как каждый агент взаимодействует для решения сложных задач в различных секторах, включая финансы, технологии и креативные стратегии. Этот практический подход поможет вам понять архитектуру и рабочие процессы, необходимые для получения качественных результатов от ИИ.

Установка и настройка

Для начала установим необходимые библиотеки, такие как agentUniverse и google-generativeai, для настройки многоагентной системы:

!pip install agentUniverse google-generativeai python-dotenv pydantic

Далее, необходимо настроить API Gemini, используя ваш бесплатный API-ключ, чтобы включить генерацию контента с помощью ИИ:

import os
import asyncio
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
from enum import Enum
import json
import time
import google.generativeai as genai

GEMINI_API_KEY = 'Ваш API Ключ' 
genai.configure(api_key=GEMINI_API_KEY)

Роли агентов и управление задачами

В нашей системе мы определяем четыре основные роли агентов: Планировщик, Исполнитель, Выразитель и Рецензент. Создаем класс Task для управления метаданными задач, включая статус, результаты и отзывы. Класс BaseAgent служит основой для всех агентов, позволяя им обрабатывать задачи и взаимодействовать с API Gemini:

class AgentRole(Enum):
   PLANNER = "planner"
   EXECUTOR = "executor"
   EXPRESSER = "expresser"
   REVIEWER = "reviewer"

@dataclass
class Task:
   id: str
   description: str
   context: Dict[str, Any]
   status: str = "pending"
   result: Optional[str] = None
   feedback: Optional[str] = None

class BaseAgent:
   def __init__(self, name: str, role: AgentRole, system_prompt: str):
       self.name = name
       self.role = role
       self.system_prompt = system_prompt
       self.memory: List[Dict] = []

   async def process(self, task: Task) -> str:
       prompt = f"{self.system_prompt}\n\nTask: {task.description}\nContext: {json.dumps(task.context)}"
       result = await self._simulate_llm_call(prompt, task)
       self.memory.append({
           "task_id": task.id,
           "input": task.description,
           "output": result,
           "timestamp": time.time()
       })
       return result

Реализация PEER-паттерна

PEER-паттерн реализуется через класс PEERAgent, который координирует действия четырех специализированных агентов для совместного решения задач. Каждая итерация проходит через этапы — Планирование, Выполнение, Выражение и Обзор, что позволяет улучшить результаты на основе структурированных процессов:

class PEERAgent:
   def __init__(self):
       self.planner = BaseAgent("Планировщик", AgentRole.PLANNER, "Вы агент стратегического планирования. Разбейте сложные задачи на выполнимые шаги.")
       self.executor = BaseAgent("Исполнитель", AgentRole.EXECUTOR, "Вы агент исполнения. Завершайте задачи эффективно, используя доступные инструменты и знания.")
       self.expresser = BaseAgent("Выразитель результатов", AgentRole.EXPRESSER, "Вы агент коммуникации. Ясно и профессионально представляйте результаты.")
       self.reviewer = BaseAgent("Рецензент качества", AgentRole.REVIEWER, "Вы агент контроля качества. Проверяйте результаты и давайте обратную связь на улучшение.")
       self.iteration_count = 0
       self.max_iterations = 3

   async def collaborate(self, task: Task) -> Dict[str, Any]:
       results = {"iterations": [], "final_result": None}
       while self.iteration_count < self.max_iterations:
           iteration_result = {}
           plan = await self.planner.process(task)
           iteration_result["plan"] = plan
           task.context["current_plan"] = plan
           execution = await self.executor.process(task)
           iteration_result["execution"] = execution
           task.context["execution_result"] = execution
           expression = await self.expresser.process(task)
           iteration_result["expression"] = expression
           task.result = expression
           review = await self.reviewer.process(task)
           iteration_result["review"] = review
           task.feedback = review
           results["iterations"].append(iteration_result)
           if "high" in review.lower() and self.iteration_count >= 1:
               results["final_result"] = expression
               break
           self.iteration_count += 1
           task.context["previous_feedback"] = review
       return results

Оркестрация многоагентного взаимодействия

Класс MultiAgentOrchestrator управляет всей многоагентной системой. Он обрабатывает сложные задачи, используя PEER-паттерн, улучшая результаты с помощью доменных агентов при необходимости:

class MultiAgentOrchestrator:
   def __init__(self):
       self.agents = {}
       self.peer_system = PEERAgent()
       self.task_queue = []

   def register_agent(self, agent: BaseAgent):
       self.agents[agent.name] = agent

   async def process_complex_task(self, description: str, domain: str = "general") -> Dict[str, Any]:
       task = Task(
           id=f"task_{int(time.time())}",
           description=description,
           context={"domain": domain, "complexity": "high"}
       )
       peer_results = await self.peer_system.collaborate(task)
       if domain in ["financial", "technical", "creative"]:
           domain_agent = self._get_domain_agent(domain)
           if domain_agent:
               domain_result = await domain_agent.process(task)
               peer_results["domain_enhancement"] = domain_result
       return {
           "task_id": task.id,
           "original_request": description,
           "peer_results": peer_results,
           "status": "completed",
           "processing_time": f"{len(peer_results['iterations'])} iterations"
       }

   def _get_domain_agent(self, domain: str) -> Optional[BaseAgent]:
       domain_agents = {
           "financial": BaseAgent("Финансовый аналитик", AgentRole.EXECUTOR, "Вы старший финансовый аналитик с опытом в рыночном анализе и инвестиционных стратегиях."),
           "technical": BaseAgent("Технический эксперт", AgentRole.EXECUTOR, "Вы ведущий архитектор программного обеспечения с опытом в системном дизайне и масштабируемости."),
           "creative": BaseAgent("Креативный директор", AgentRole.EXPRESSER, "Вы награжденный креативный директор с опытом в бренд-стратегии и контенте.")
       }
       return domain_agents.get(domain)

Запуск демонстрации

Мы объединяем компоненты в функции run_advanced_demo, которая тестирует конвейер с финансовыми, техническими и креативными задачами:

async def run_advanced_demo():
   orchestrator = MultiAgentOrchestrator()
   financial_task = "Проанализировать влияние роста процентных ставок на портфель акций технологий"
   result1 = await orchestrator.process_complex_task(financial_task, "financial")
   technical_task = "Разработать масштабируемую архитектуру микросервисов для высоконагруженной электронной коммерции"
   result2 = await orchestrator.process_complex_task(technical_task, "technical")
   creative_task = "Создать комплексную стратегию бренда для стартапа в области устойчивой моды"
   result3 = await orchestrator.process_complex_task(creative_task, "creative")
   return {
       "demo_results": [result1, result2, result3],
       "agent_stats": {
           "total_tasks": 3,
           "success_rate": "100%",
           "avg_iterations": sum(len(r['peer_results']['iterations']) for r in [result1, result2, result3]) / 3
       }
   }

Заключение

В заключение, этот учебник демонстрирует, как многоагентная система может систематически решать сложные проблемы, комбинируя специфическое для домена рассуждение, структурированную коммуникацию и итерационные проверки качества. Мы видим, как PEER-рамка раскрывает потенциал совместной работы агентов, а модель Gemini улучшает результаты каждого агента. Это подчеркивает полезность модульных ИИ-систем для создания масштабируемых и надежных приложений.

Для получения дополнительных ресурсов посетите нашу страницу на GitHub для учебных материалов, кодов и блокнотов. Следите за нами в Twitter и присоединяйтесь к нашему сообществу для получения обновлений.