#!/usr/bin/env python3 """ AI Agent Module - автономный агент с инструментами. Агент может самостоятельно принимать решения об использовании инструментов на основе контекста запроса пользователя. """ import logging from typing import Optional, List, Dict, Any from dataclasses import dataclass from datetime import datetime from bot.tools import tools_registry, ToolResult logger = logging.getLogger(__name__) @dataclass class AgentDecision: """Решение агента об использовании инструмента.""" should_use_tool: bool tool_name: Optional[str] = None tool_args: Optional[Dict[str, Any]] = None confidence: float = 0.0 reasoning: str = "" class AIAgent: """ AI-агент с доступом к инструментам. Агент анализирует запрос и решает, нужно ли использовать какой-либо инструмент для выполнения задачи. """ # Триггеры для поиска в интернете SEARCH_TRIGGERS = [ 'найди', 'поиск', 'погугли', 'узнай', 'проверь в интернете', 'что нового', 'последние новости', 'свежая информация', 'как сделать', 'руководство', 'документация', 'tutorial', 'weather', 'news', 'search', 'find', 'look up', 'что это', 'кто такой', 'где находится', 'когда выйдет', 'скачай', 'загрузи', 'найди информацию', 'посмотри в сети' ] # Триггеры для RSS RSS_TRIGGERS = [ 'новости', 'rss', 'лента', 'feed', 'дайджест', 'что нового в linux', 'новости it', 'tech news', 'почитай новости', 'последние статьи', 'свежие новости', 'новости технологий', 'opensource новости', 'linux новости', 'покажи новости', 'что в лентах', 'есть новые статьи' ] # Триггеры для SSH-команд SSH_TRIGGERS = [ 'выполни команду', 'ssh', 'запусти на сервере', 'проверь сервер', 'посмотри логи', 'покажи процесс', 'сколько места', 'df', 'top', 'перезапусти', 'останови', 'запусти сервис', 'systemctl', 'проверь нагрузку', 'uptime', 'кто залогинен', 'who', 'last', 'посмотри в /var/log', 'проверь диск', 'мониторинг', 'выполни на 192.168.1', 'запусти скрипт', 'cron' ] # Триггеры для Cron-задач CRON_TRIGGERS = [ 'напомни', 'запланируй', 'каждый день', 'каждый час', 'периодически', 'по расписанию', 'автоматически', 'создай задачу', 'добавь в cron', 'регулярно', 'повторяй', 'каждую неделю', 'ежедневно', 'ежечасно' ] def __init__(self): self.registry = tools_registry self._tool_use_history: List[Dict] = [] self._user_preferences: Dict[int, Dict] = {} # preferences per user def _should_search(self, message: str) -> tuple[bool, float]: """Проверить, нужен ли поиск в интернете.""" message_lower = message.lower() score = 0.0 # Прямые триггеры — высокий приоритет for trigger in self.SEARCH_TRIGGERS: if trigger in message_lower: return True, 0.9 # Вопросы с "что", "как", "где", "когда" о внешних фактах question_words = ['что такое', 'как сделать', 'где найти', 'когда будет'] for qword in question_words: if qword in message_lower: score = max(score, 0.7) # Упоминания текущих событий current_events = ['сегодня', 'сейчас', 'в этом году', 'recent', 'latest', '2024', '2025', '2026'] for event in current_events: if event in message_lower: score = max(score, 0.6) # Если есть вопросительные слова + внешние факты if any(word in message_lower for word in ['почему', 'зачем', 'как работает']): score = max(score, 0.65) return score >= 0.65, score def _should_read_rss(self, message: str) -> tuple[bool, float]: """Проверить, нужно ли читать RSS ленты.""" message_lower = message.lower() score = 0.0 for trigger in self.RSS_TRIGGERS: if trigger in message_lower: return True, 0.9 # Если пользователь спрашивает про новости технологий tech_news = ['новости linux', 'it новости', 'tech news', 'opensource'] for topic in tech_news: if topic in message_lower: score = max(score, 0.8) # Контекстные подсказки if any(word in message_lower for word in ['почитай', 'посмотри', 'покажи']) and \ any(word in message_lower for word in ['статьи', 'материалы', 'публикации']): score = max(score, 0.75) return score >= 0.75, score def _should_use_ssh(self, message: str) -> tuple[bool, float]: """Проверить, нужна ли SSH-команда.""" message_lower = message.lower() score = 0.0 # Прямые триггеры for trigger in self.SSH_TRIGGERS: if trigger in message_lower: return True, 0.9 # Команды системного администрирования sysadmin_tasks = ['проверь', 'посмотри', 'покажи', 'выполни', 'запусти'] sysadmin_objects = ['сервер', 'лог', 'процесс', 'диск', 'память', 'сервис', 'демон'] has_task = any(task in message_lower for task in sysadmin_tasks) has_object = any(obj in message_lower for obj in sysadmin_objects) if has_task and has_object: score = max(score, 0.75) # Упоминания конкретных утилит utils = ['systemctl', 'journalctl', 'top', 'htop', 'df', 'du', 'free', 'ps', 'netstat'] for util in utils: if util in message_lower: score = max(score, 0.8) return score >= 0.75, score def _should_use_cron(self, message: str) -> tuple[bool, float]: """Проверить, нужна ли cron-задача.""" message_lower = message.lower() score = 0.0 # Прямые триггеры for trigger in self.CRON_TRIGGERS: if trigger in message_lower: return True, 0.85 # Расписания schedules = ['каждый', 'каждую', 'ежедневно', 'ежечасно', 'еженедельно', 'раз в'] for sched in schedules: if sched in message_lower: score = max(score, 0.8) # Напоминания и периодические задачи if any(word in message_lower for word in ['напомни', 'запланируй', 'повторяй']): score = max(score, 0.85) return score >= 0.8, score async def decide(self, message: str, context: Optional[Dict] = None) -> AgentDecision: """ Принять решение об использовании инструмента. Args: message: Сообщение пользователя context: Дополнительный контекст (история, состояние) Returns: AgentDecision с решением агента """ user_id = context.get('user_id') if context else None # Приоритет: SSH > Cron > Поиск > RSS # Проверяем в порядке приоритета # 1. Проверка на SSH-команды (системные задачи) should_ssh, ssh_conf = self._should_use_ssh(message) if should_ssh and ssh_conf > 0.75: return AgentDecision( should_use_tool=True, tool_name='ssh_tool', tool_args={'command': self._extract_ssh_command(message)}, confidence=ssh_conf, reasoning='Пользователю нужно выполнить команду на сервере' ) # 2. Проверка на Cron-задачи (планирование) should_cron, cron_conf = self._should_use_cron(message) if should_cron and cron_conf > 0.75: return AgentDecision( should_use_tool=True, tool_name='cron_tool', tool_args={'action': 'list'}, # Показываем список задач confidence=cron_conf, reasoning='Пользователь хочет создать или управлять задачей' ) # 3. Проверка на поиск should_search, search_conf = self._should_search(message) if should_search and search_conf > 0.7: query = self._extract_search_query(message) return AgentDecision( should_use_tool=True, tool_name='ddgs_tool', tool_args={'query': query, 'max_results': 5}, confidence=search_conf, reasoning='Пользователю нужна информация из интернета' ) # 4. Проверка на RSS should_rss, rss_conf = self._should_read_rss(message) if should_rss and rss_conf > 0.7: return AgentDecision( should_use_tool=True, tool_name='rss_tool', tool_args={'action': 'list', 'limit': 10, 'undigested_only': True}, confidence=rss_conf, reasoning='Пользователь хочет прочитать новости из лент' ) # Инструменты не нужны return AgentDecision( should_use_tool=False, confidence=0.0, reasoning='Инструменты не требуются' ) def _extract_search_query(self, message: str) -> str: """Извлечь поисковый запрос из сообщения.""" triggers_to_remove = self.SEARCH_TRIGGERS + ['покажи', 'напиши', 'дай', 'расскажи', 'хочу', 'надо', 'нужно'] query = message.lower() for trigger in triggers_to_remove: query = query.replace(trigger, '') query = query.strip(' ?:.,!') if not query: query = message return query.strip() def _extract_ssh_command(self, message: str) -> str: """Извлечь SSH-команду из сообщения.""" message_lower = message.lower() # Если есть явная команда в кавычках import re quoted = re.search(r'["\']([^"\']+)["\']', message) if quoted: return quoted.group(1).strip() # Если команда после триггера for trigger in ['выполни команду', 'запусти', 'ssh']: if trigger in message_lower: idx = message_lower.find(trigger) return message[idx + len(trigger):].strip() # Возвращаем оригинальное сообщение как команду return message async def execute_tool(self, tool_name: str, **kwargs) -> ToolResult: """Выполнить инструмент и сохранить историю.""" logger.info(f"🤖 AI-агент выполняет инструмент: {tool_name} с аргументами: {kwargs}") result = await self.registry.execute_tool(tool_name, **kwargs) # Сохраняем историю использования self._tool_use_history.append({ 'tool_name': tool_name, 'args': kwargs, 'result': result.to_dict(), 'timestamp': datetime.now().isoformat() }) # Ограничиваем историю if len(self._tool_use_history) > 100: self._tool_use_history = self._tool_use_history[-50:] logger.info(f"✅ Инструмент {tool_name} выполнен: success={result.success}") return result def get_tool_history(self, limit: int = 10) -> List[Dict]: """Получить историю использования инструментов.""" return self._tool_use_history[-limit:] def set_user_preference(self, user_id: int, preference: str, value: Any): """Установить предпочтение пользователя для инструментов.""" if user_id not in self._user_preferences: self._user_preferences[user_id] = {} self._user_preferences[user_id][preference] = value logger.info(f"Установлено предпочтение для пользователя {user_id}: {preference} = {value}") def get_user_preference(self, user_id: int, preference: str, default: Any = None) -> Any: """Получить предпочтение пользователя.""" return self._user_preferences.get(user_id, {}).get(preference, default) # Глобальный агент ai_agent = AIAgent()