320 lines
14 KiB
Python
320 lines
14 KiB
Python
#!/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_executor',
|
||
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_manager',
|
||
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_search',
|
||
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_reader',
|
||
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()
|