v0.5.4: Добавлена поддержка GigaChat API (Сбер)

- Новый инструмент gigachat_tool для работы с GigaChat API
- GigaChatProvider в qwen_integration как альтернатива Qwen Code
- Настройки GigaChat и YandexGPT в .env.example
- Документация по настройке в README.md
- GigaChatCapability зарегистрирован в реестре инструментов
- Поддержка системных промптов, температуры, лимита токенов
- Автоматическое получение и кэширование OAuth токена

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

.env.example

@@ -36,6 +36,40 @@ SERVERS=
 # SSH ключ для подключения (альтернатива паролю)
 # SSH_KEY_PATH=/home/user/.ssh/id_ed25519
 
+# ===========================================
+# GigaChat API (Сбер)
+# ===========================================
+# Получите credentials в SberDevices Developer Portal:
+# https://developers.sber.ru/docs/ru/gigachat
+#
+# GIGACHAT_CLIENT_ID - ID клиента (UUID)
+# GIGACHAT_CLIENT_SECRET - Секрет клиента
+# GIGACHAT_SCOPE - Область доступа (обычно GIGACHAT_API_PERS)
+# GIGACHAT_AUTH_URL - URL авторизации (https://ngw.devices.sberbank.ru:9443/api/v2/oauth)
+# GIGACHAT_MODEL - Модель по умолчанию (GigaChat-Pro или GigaChat-Max)
+#
+# Пример:
+GIGACHAT_CLIENT_ID=your-client-id-here
+GIGACHAT_CLIENT_SECRET=your-client-secret-here
+GIGACHAT_SCOPE=GIGACHAT_API_PERS
+GIGACHAT_AUTH_URL=https://ngw.devices.sberbank.ru:9443/api/v2/oauth
+GIGACHAT_MODEL=GigaChat-Pro
+
+# ===========================================
+# YandexGPT API (Яндекс)
+# ===========================================
+# Получите credentials в Yandex Cloud Console:
+# https://cloud.yandex.ru/docs/fundamentals/concepts/infrastructure
+#
+# YANDEX_FOLDER_ID - ID каталога в Yandex Cloud
+# YANDEX_API_KEY - API ключ (или используйте IAM-токен)
+# YANDEX_MODEL - Модель по умолчанию (yandexgpt/latest или yandexgpt-lite/latest)
+#
+# Пример:
+# YANDEX_FOLDER_ID=b1gxxxxxxxxxxxxxxxx
+# YANDEX_API_KEY=your-api-key-here
+# YANDEX_MODEL=yandexgpt/latest
+
 # ===========================================
 # SOCKS5 Proxy (опционально)
 # ===========================================

README.md

@@ -288,6 +288,52 @@ WORKING_DIRECTORY=/home/user
 | `ALLOWED_USERS` | Список разрешённых user ID через запятую (пусто = все) |
 | `WORKING_DIRECTORY` | Рабочая директория для выполнения команд |
 
+### Настройка GigaChat API (Сбер)
+
+Бот поддерживает альтернативный AI-провайдер — **GigaChat** от Сбера. Для использования:
+
+1. Получите credentials в [SberDevices Developer Portal](https://developers.sber.ru/docs/ru/gigachat)
+2. Добавьте в `.env`:
+
+```bash
+# GigaChat API (Сбер)
+GIGACHAT_CLIENT_ID=ваш-client-id-uuid
+GIGACHAT_CLIENT_SECRET=ваш-client-secret
+GIGACHAT_SCOPE=GIGACHAT_API_PERS
+GIGACHAT_AUTH_URL=https://ngw.devices.sberbank.ru:9443/api/v2/oauth
+GIGACHAT_MODEL=GigaChat-Pro
+```
+
+3. Перезапустите бота
+
+**Параметры:**
+
+| Параметр | Описание |
+|----------|----------|
+| `GIGACHAT_CLIENT_ID` | ID клиента (UUID из SberDevices Portal) |
+| `GIGACHAT_CLIENT_SECRET` | Секрет клиента |
+| `GIGACHAT_SCOPE` | Область доступа (обычно `GIGACHAT_API_PERS`) |
+| `GIGACHAT_AUTH_URL` | URL авторизации OAuth |
+| `GIGACHAT_MODEL` | Модель: `GigaChat-Pro` или `GigaChat-Max` |
+
+**Инструмент GigaChat:**
+- `gigachat` — генерация ответов через GigaChat API
+- Используется как альтернатива Qwen Code
+- Поддерживает системные промпты, температуру, лимит токенов
+
+### Настройка YandexGPT API (Яндекс)
+
+Для использования YandexGPT добавьте в `.env`:
+
+```bash
+# YandexGPT API (Яндекс)
+YANDEX_FOLDER_ID=ваш-folder-id
+YANDEX_API_KEY=ваш-api-key
+YANDEX_MODEL=yandexgpt/latest
+```
+
+Получите credentials в [Yandex Cloud Console](https://cloud.yandex.ru/docs/fundamentals/concepts/infrastructure).
+
 ⚠️ **Важно:** После изменения `.env` требуется перезапуск бота.
 
 ## Безопасность

bot/tools/__init__.py

@@ -124,4 +124,4 @@ def register_tool(tool_class: type) -> type:
 
 # Авто-импорт инструментов для регистрации
 # Импортируем после определения register_tool чтобы декоратор сработал
-from bot.tools import ddgs_tool, rss_tool, ssh_tool, cron_tool
+from bot.tools import ddgs_tool, rss_tool, ssh_tool, cron_tool, gigachat_tool

bot/tools/gigachat_tool.py

@@ -0,0 +1,385 @@
+"""
+GigaChat API Tool для Telegram CLI Bot
+
+Инструмент для работы с GigaChat API (Сбер).
+Поддерживает генерацию текста, чат-сессии и различные модели.
+
+Документация: https://developers.sber.ru/docs/ru/gigachat
+"""
+
+import os
+import base64
+import httpx
+import asyncio
+from typing import Optional, List, Dict, Any
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta
+
+
+@dataclass
+class GigaChatMessage:
+    """Сообщение для чата с GigaChat"""
+    role: str  # 'user', 'assistant', 'system'
+    content: str
+
+
+@dataclass
+class GigaChatConfig:
+    """Конфигурация подключения к GigaChat API"""
+    client_id: str
+    client_secret: str
+    scope: str = "GIGACHAT_API_PERS"
+    auth_url: str = "https://ngw.devices.sberbank.ru:9443/api/v2/oauth"
+    model: str = "GigaChat-Pro"
+    api_url: str = "https://gigachat.devices.sberbank.ru/api/v2"
+    timeout: int = 60
+
+
+class GigaChatTool:
+    """
+    Инструмент для работы с GigaChat API
+    
+    Пример использования:
+        config = GigaChatConfig(
+            client_id=os.getenv("GIGACHAT_CLIENT_ID"),
+            client_secret=os.getenv("GIGACHAT_CLIENT_SECRET"),
+        )
+        tool = GigaChatTool(config)
+        
+        # Простой запрос
+        response = await tool.chat("Привет, как дела?")
+        
+        # Чат с историей
+        messages = [
+            GigaChatMessage(role="system", content="Ты полезный ассистент."),
+            GigaChatMessage(role="user", content="Расскажи про Python"),
+        ]
+        response = await tool.chat(messages=messages)
+    """
+    
+    def __init__(self, config: Optional[GigaChatConfig] = None):
+        self.config = config or self._load_config_from_env()
+        self._access_token: Optional[str] = None
+        self._token_expires: Optional[datetime] = None
+        self._chat_history: List[GigaChatMessage] = []
+    
+    def _load_config_from_env(self) -> GigaChatConfig:
+        """Загрузка конфигурации из переменных окружения"""
+        return GigaChatConfig(
+            client_id=os.getenv("GIGACHAT_CLIENT_ID", ""),
+            client_secret=os.getenv("GIGACHAT_CLIENT_SECRET", ""),
+            scope=os.getenv("GIGACHAT_SCOPE", "GIGACHAT_API_PERS"),
+            auth_url=os.getenv("GIGACHAT_AUTH_URL", "https://ngw.devices.sberbank.ru:9443/api/v2/oauth"),
+            model=os.getenv("GIGACHAT_MODEL", "GigaChat-Pro"),
+        )
+    
+    def _get_auth_headers(self) -> Dict[str, str]:
+        """Получение заголовков для авторизации"""
+        credentials = f"{self.config.client_id}:{self.config.client_secret}"
+        encoded_credentials = base64.b64encode(credentials.encode()).decode()
+        return {
+            "Authorization": f"Basic {encoded_credentials}",
+            "Content-Type": "application/x-www-form-urlencoded",
+            "RqUID": "00000000-0000-0000-0000-000000000000",
+        }
+    
+    async def _get_access_token(self) -> str:
+        """Получение access токена для API"""
+        # Проверяем кэш токена
+        if self._access_token and self._token_expires:
+            if datetime.now() < self._token_expires - timedelta(minutes=5):
+                return self._access_token
+        
+        # Запрашиваем новый токен
+        async with httpx.AsyncClient(verify=False) as client:
+            response = await client.post(
+                self.config.auth_url,
+                headers=self._get_auth_headers(),
+                data={"scope": self.config.scope},
+                timeout=30,
+            )
+            response.raise_for_status()
+            data = response.json()
+            
+            self._access_token = data["access_token"]
+            # Токен действителен 30 минут, кэшируем на 25 минут
+            self._token_expires = datetime.now() + timedelta(minutes=25)
+            
+            return self._access_token
+    
+    async def chat(
+        self,
+        messages: Optional[List[GigaChatMessage]] = None,
+        model: Optional[str] = None,
+        temperature: float = 0.7,
+        max_tokens: int = 2000,
+        top_p: float = 0.1,
+        repetition_penalty: float = 1.0,
+        use_history: bool = True,
+    ) -> Dict[str, Any]:
+        """
+        Отправка запроса к GigaChat API
+        
+        Args:
+            messages: Список сообщений (если None, используется история чата)
+            model: Модель для генерации (если None, используется модель из конфига)
+            temperature: Температура генерации (0.0 - 2.0)
+            max_tokens: Максимальное количество токенов в ответе
+            top_p: Параметр top-p sampling
+            repetition_penalty: Штраф за повторения
+            use_history: Использовать ли историю чата
+        
+        Returns:
+            Dict с ответом API:
+                - content: Текст ответа
+                - model: Использованная модель
+                - usage: Статистика использования токенов
+                - finish_reason: Причина завершения
+        """
+        token = await self._get_access_token()
+        
+        # Формируем сообщения
+        if messages is None:
+            if use_history:
+                messages = self._chat_history.copy()
+            else:
+                messages = []
+        elif use_history:
+            # Добавляем новые сообщения к истории
+            self._chat_history.extend(messages)
+            messages = self._chat_history.copy()
+        
+        # Преобразуем сообщения в формат API
+        api_messages = [
+            {"role": msg.role, "content": msg.content}
+            for msg in messages
+        ]
+        
+        payload = {
+            "model": model or self.config.model,
+            "messages": api_messages,
+            "temperature": temperature,
+            "max_tokens": max_tokens,
+            "top_p": top_p,
+            "repetition_penalty": repetition_penalty,
+        }
+        
+        headers = {
+            "Authorization": f"Bearer {token}",
+            "Content-Type": "application/json",
+        }
+        
+        async with httpx.AsyncClient(verify=False) as client:
+            response = await client.post(
+                f"{self.config.api_url}/chat/completions",
+                headers=headers,
+                json=payload,
+                timeout=self.config.timeout,
+            )
+            response.raise_for_status()
+            data = response.json()
+        
+        # Добавляем ответ ассистента в историю
+        if use_history and data.get("choices"):
+            assistant_message = data["choices"][0]["message"]
+            self._chat_history.append(GigaChatMessage(
+                role=assistant_message["role"],
+                content=assistant_message["content"],
+            ))
+        
+        return {
+            "content": data["choices"][0]["message"]["content"] if data.get("choices") else "",
+            "model": data.get("model", self.config.model),
+            "usage": data.get("usage", {}),
+            "finish_reason": data["choices"][0]["finish_reason"] if data.get("choices") else "",
+        }
+    
+    def clear_history(self):
+        """Очистка истории чата"""
+        self._chat_history = []
+    
+    def get_history(self) -> List[GigaChatMessage]:
+        """Получение истории чата"""
+        return self._chat_history.copy()
+    
+    def set_system_prompt(self, prompt: str):
+        """Установка системного промпта (добавляется в начало истории)"""
+        # Удаляем старый системный промпт если есть
+        self._chat_history = [
+            msg for msg in self._chat_history if msg.role != "system"
+        ]
+        # Добавляем новый в начало
+        self._chat_history.insert(0, GigaChatMessage(role="system", content=prompt))
+    
+    async def generate_image(
+        self,
+        prompt: str,
+        model: str = " Kandinsky-2",
+        size: str = "1024x1024",
+    ) -> Dict[str, Any]:
+        """
+        Генерация изображений через GigaChat (Kandinsky)
+        
+        Args:
+            prompt: Текстовое описание изображения
+            model: Модель для генерации
+            size: Размер изображения
+        
+        Returns:
+            Dict с результатом генерации
+        """
+        token = await self._get_access_token()
+        
+        payload = {
+            "model": model,
+            "prompt": prompt,
+            "size": size,
+        }
+        
+        headers = {
+            "Authorization": f"Bearer {token}",
+            "Content-Type": "application/json",
+        }
+        
+        async with httpx.AsyncClient(verify=False) as client:
+            # Запуск генерации
+            response = await client.post(
+                f"{self.config.api_url}/images/generations",
+                headers=headers,
+                json=payload,
+                timeout=self.config.timeout,
+            )
+            response.raise_for_status()
+            data = response.json()
+        
+        return data
+    
+    async def get_models(self) -> List[str]:
+        """Получение списка доступных моделей"""
+        token = await self._get_access_token()
+        
+        headers = {
+            "Authorization": f"Bearer {token}",
+        }
+        
+        async with httpx.AsyncClient(verify=False) as client:
+            response = await client.get(
+                f"{self.config.api_url}/models",
+                headers=headers,
+                timeout=30,
+            )
+            response.raise_for_status()
+            data = response.json()
+        
+        return [model["id"] for model in data.get("data", [])]
+
+
+# Утилита для создания инструмента в формате бота
+def create_gigachat_tool():
+    """
+    Создает экземпляр GigaChatTool с конфигурацией из окружения
+    
+    Returns:
+        GigaChatTool или None если конфигурация не задана
+    """
+    if not os.getenv("GIGACHAT_CLIENT_ID") or not os.getenv("GIGACHAT_CLIENT_SECRET"):
+        return None
+    
+    return GigaChatTool()
+
+
+if __name__ == "__main__":
+    # Пример использования
+    async def main():
+        tool = create_gigachat_tool()
+        if not tool:
+            print("GigaChat не настроен. Проверьте переменные окружения.")
+            return
+        
+        # Простой запрос
+        response = await tool.chat("Привет! Расскажи кратко про себя.")
+        print(f"Ответ: {response['content']}")
+        print(f"Модель: {response['model']}")
+        print(f"Токены: {response['usage']}")
+    
+    asyncio.run(main())
+
+# ===========================================
+# Интеграция с реестром инструментов бота
+# ===========================================
+
+from bot.tools import BaseTool, ToolResult, register_tool
+
+
+@register_tool
+class GigaChatCapability(BaseTool):
+    """
+    Capability-обёртка для GigaChat API.
+    
+    Позволяет использовать GigaChat через реестр инструментов бота.
+    """
+    
+    name = "gigachat"
+    description = "Генерация ответов AI через GigaChat API (Сбер). Альтернатива Qwen Code."
+    category = "ai"
+    
+    def __init__(self):
+        self._provider = None
+    
+    def _get_provider(self):
+        """Ленивая инициализация провайдера"""
+        if self._provider is None:
+            from qwen_integration import GigaChatProvider
+            self._provider = GigaChatProvider()
+        return self._provider
+    
+    async def execute(
+        self,
+        prompt: str,
+        system_prompt: Optional[str] = None,
+        temperature: float = 0.7,
+        max_tokens: int = 2000,
+        **kwargs
+    ) -> ToolResult:
+        """
+        Выполнить запрос к GigaChat API.
+        
+        Args:
+            prompt: Запрос пользователя
+            system_prompt: Системный промпт (роль ассистента)
+            temperature: Температура генерации (0.0-2.0)
+            max_tokens: Максимум токенов в ответе
+        """
+        provider = self._get_provider()
+        
+        if not provider.is_available():
+            return ToolResult(
+                success=False,
+                error=provider.get_error() or "GigaChat не доступен",
+            )
+        
+        result = await provider.chat(
+            prompt=prompt,
+            system_prompt=system_prompt,
+            temperature=temperature,
+            max_tokens=max_tokens,
+        )
+        
+        if result.get("success"):
+            return ToolResult(
+                success=True,
+                data={
+                    "content": result.get("content", ""),
+                    "model": result.get("model", "GigaChat-Pro"),
+                    "usage": result.get("usage", {}),
+                },
+                metadata={
+                    "model": result.get("model"),
+                    "tokens": result.get("usage"),
+                }
+            )
+        else:
+            return ToolResult(
+                success=False,
+                error=result.get("error", "Неизвестная ошибка GigaChat"),
+            )

qwen_integration.py

@@ -484,5 +484,133 @@ class QwenCodeManager:
         return '\n'.join(cleaned) if cleaned else output
 
 
+class GigaChatProvider:
+    """
+    AI-провайдер для работы с GigaChat API.
+    
+    Альтернатива Qwen Code для генерации ответов.
+    Использует GigaChatTool для взаимодействия с API Сбера.
+    """
+    
+    def __init__(self):
+        self._tool = None
+        self._initialized = False
+        self._config_error: Optional[str] = None
+    
+    def _ensure_initialized(self):
+        """Ленивая инициализация инструмента"""
+        if self._initialized:
+            return
+        
+        try:
+            from bot.tools.gigachat_tool import create_gigachat_tool
+            self._tool = create_gigachat_tool()
+            
+            if not self._tool:
+                self._config_error = "GigaChat не настроен. Проверьте GIGACHAT_CLIENT_ID и GIGACHAT_CLIENT_SECRET в .env"
+                logger.warning(self._config_error)
+            else:
+                logger.info("GigaChatProvider инициализирован")
+        except ImportError as e:
+            self._config_error = f"Ошибка импорта GigaChat: {e}"
+            logger.error(self._config_error)
+        except Exception as e:
+            self._config_error = f"Ошибка инициализации GigaChat: {e}"
+            logger.error(self._config_error)
+        
+        self._initialized = True
+    
+    async def chat(
+        self,
+        prompt: str,
+        system_prompt: Optional[str] = None,
+        temperature: float = 0.7,
+        max_tokens: int = 2000,
+        on_chunk: Optional[Callable[[str], Any]] = None,
+    ) -> Dict[str, Any]:
+        """
+        Отправка запроса к GigaChat API
+        
+        Args:
+            prompt: Запрос пользователя
+            system_prompt: Системный промпт (роль ассистента)
+            temperature: Температура генерации
+            max_tokens: Максимум токенов в ответе
+            on_chunk: Callback для потоковой отправки (не используется, GigaChat отдаёт целиком)
+        
+        Returns:
+            Dict с полями:
+                - success: bool
+                - content: str - текст ответа
+                - error: str - ошибка если есть
+                - model: str - использованная модель
+                - usage: dict - статистика токенов
+        """
+        self._ensure_initialized()
+        
+        if not self._tool:
+            return {
+                "success": False,
+                "error": self._config_error or "GigaChat не инициализирован",
+                "content": "",
+            }
+        
+        try:
+            from bot.tools.gigachat_tool import GigaChatMessage
+            
+            # Формируем сообщения
+            messages = []
+            
+            if system_prompt:
+                messages.append(GigaChatMessage(role="system", content=system_prompt))
+            
+            messages.append(GigaChatMessage(role="user", content=prompt))
+            
+            # Вызываем GigaChat API
+            response = await self._tool.chat(
+                messages=messages,
+                temperature=temperature,
+                max_tokens=max_tokens,
+                use_history=False,  # Не используем встроенную историю — у нас своя
+            )
+            
+            # Потоковая отправка если есть callback
+            if on_chunk and response.get("content"):
+                await on_chunk(response["content"])
+            
+            return {
+                "success": True,
+                "content": response.get("content", ""),
+                "model": response.get("model", "GigaChat-Pro"),
+                "usage": response.get("usage", {}),
+            }
+            
+        except Exception as e:
+            logger.error(f"Ошибка GigaChat API: {e}")
+            return {
+                "success": False,
+                "error": str(e),
+                "content": "",
+            }
+    
+    def clear_session(self):
+        """Очистка сессии (истории чата)"""
+        if self._tool:
+            self._tool.clear_history()
+    
+    def is_available(self) -> bool:
+        """Проверка доступности провайдера"""
+        self._ensure_initialized()
+        return self._tool is not None
+    
+    def get_error(self) -> Optional[str]:
+        """Получение ошибки инициализации"""
+        self._ensure_initialized()
+        return self._config_error
+
+
 # Глобальный менеджер
 qwen_manager = QwenCodeManager()
+
+# Глобальный GigaChat провайдер
+gigachat_provider = GigaChatProvider()