Создание масштабируемых кастомных AI-агентов для автоматизации в бизнесе с мониторингом и оркестрацией

Построение кастомного фреймворка AI-агентов

В данном руководстве описывается разработка и реализация кастомного фреймворка AI-агентов с использованием PyTorch и ключевых библиотек Python. Рассматривается оборачивание основных функций в классы CustomTool с мониторингом, оркестрация нескольких агентов с системными подсказками и определение end-to-end рабочих процессов для автоматизации таких задач, как конкурентный анализ сайтов и обработка данных. Включены логика повторных попыток, логирование и метрики производительности для надежного развертывания и масштабирования в корпоративной среде.

Основные библиотеки и настройка

Первоначальная настройка включает установку и импорт основных библиотек: PyTorch, Transformers, pandas, NumPy, BeautifulSoup для веб-скрапинга и scikit-learn для машинного обучения. Настроено стандартное логирование и определены глобальные константы для таймаутов API и ограничений на повторные попытки, что обеспечивает предсказуемую работу инструментов в продакшене.

Классы CustomTool и ToolResult

Dataclass ToolResult стандартизирует результаты выполнения инструментов, фиксируя успешность, время выполнения, возвращённые данные и ошибки. Базовый класс CustomTool оборачивает функции методом execute, который отслеживает количество вызовов, вычисляет среднее время выполнения и логирует ошибки, обеспечивая единообразный мониторинг и наблюдаемость.

@dataclass
class ToolResult:
   """Структура результата инструмента"""
   success: bool
   data: Any
   error: Optional[str] = None
   execution_time: float = 0.0
   metadata: Dict[str, Any] = None
 
class CustomTool:
   def __init__(self, name: str, description: str, func: Callable):
       self.name = name
       self.description = description
       self.func = func
       self.calls = 0
       self.avg_execution_time = 0.0
       self.error_rate = 0.0
  
   def execute(self, *args, **kwargs) -> ToolResult:
       start_time = time.time()
       self.calls += 1
       try:
           result = self.func(*args, **kwargs)
           execution_time = time.time() - start_time
           self.avg_execution_time = ((self.avg_execution_time * (self.calls - 1)) + execution_time) / self.calls
           return ToolResult(success=True, data=result, execution_time=execution_time, metadata={'tool_name': self.name, 'call_count': self.calls})
       except Exception as e:
           execution_time = time.time() - start_time
           self.error_rate = (self.error_rate * (self.calls - 1) + 1) / self.calls
           logger.error(f"Tool {self.name} failed: {str(e)}")
           return ToolResult(success=False, data=None, error=str(e), execution_time=execution_time, metadata={'tool_name': self.name, 'call_count': self.calls})

Класс CustomAgent для выполнения задач

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

class CustomAgent:
   def __init__(self, name: str, system_prompt: str = "", max_iterations: int = 5):
       self.name = name
       self.system_prompt = system_prompt
       self.max_iterations = max_iterations
       self.tools = {}
       self.conversation_history = []
       self.performance_metrics = {}
  
   def add_tool(self, tool: CustomTool):
       self.tools[tool.name] = tool
  
   def run(self, task: str) -> Dict[str, Any]:
       logger.info(f"Agent {self.name} executing task: {task}")
       task_lower = task.lower()
       results = []
       if any(keyword in task_lower for keyword in ['analyze', 'website', 'url', 'web']):
           if 'advanced_web_intelligence' in self.tools:
               import re
               url_pattern = r'https?://[^\s]+'
               urls = re.findall(url_pattern, task)
               if urls:
                   result = self.tools['advanced_web_intelligence'].execute(urls[0])
                   results.append(result)
       elif any(keyword in task_lower for keyword in ['data', 'analyze', 'stats', 'csv']):
           if 'advanced_data_science_toolkit' in self.tools:
               if 'name,age,salary' in task:
                   data_start = task.find('name,age,salary')
                   data_part = task[data_start:]
                   result = self.tools['advanced_data_science_toolkit'].execute(data_part, 'stats')
                   results.append(result)
       elif any(keyword in task_lower for keyword in ['generate', 'code', 'api', 'client']):
           if 'advanced_code_generator' in self.tools:
               result = self.tools['advanced_code_generator'].execute(task)
               results.append(result)
       return {
           'agent': self.name,
           'task': task,
           'results': [r.data if r.success else {'error': r.error} for r in results],
           'execution_summary': {
               'tools_used': len(results),
               'success_rate': sum(1 for r in results if r.success) / len(results) if results else 0,
               'total_time': sum(r.execution_time for r in results)
           }
       }

Специализированные инструменты с мониторингом производительности

Три ключевых инструмента, обернутых декоратором @performance_monitor, обеспечивают:

advanced_web_intelligence: комплексный или сентиментальный анализ сайта.
advanced_data_science_toolkit: статистический анализ и кластеризацию CSV или JSON данных.
advanced_code_generator: генерацию готового к производству кода для API клиентов и обработки данных.

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

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

Примеры и метрики системы

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

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