#!/usr/bin/env python3 """ GigaChat AI Provider - адаптер GigaChat для работы с инструментами. Реализует интерфейс BaseAIProvider для единой работы с инструментами независимо от AI-провайдера. """ import logging from typing import Optional, Dict, Any, Callable, List import json import re from bot.base_ai_provider import ( BaseAIProvider, ProviderResponse, AIMessage, ToolCall, ToolCallStatus, ) from bot.tools.gigachat_tool import GigaChatTool, GigaChatMessage, GigaChatConfig logger = logging.getLogger(__name__) class GigaChatProvider(BaseAIProvider): """ GigaChat AI Provider с поддержкой инструментов. Использует эвристики для извлечения вызовов инструментов из текста, так как GigaChat не поддерживает нативные tool calls. """ def __init__(self, config: Optional[GigaChatConfig] = None): self._tool = GigaChatTool(config) self._available: Optional[bool] = None @property def provider_name(self) -> str: return "GigaChat" @property def supports_tools(self) -> bool: # GigaChat не поддерживает нативные tool calls # Но мы эмулируем через парсинг текста return True @property def supports_streaming(self) -> bool: return False def is_available(self) -> bool: """Проверить доступность GigaChat.""" if self._available is not None: return self._available # Проверяем наличие токенов try: import os client_id = os.getenv("GIGACHAT_CLIENT_ID") client_secret = os.getenv("GIGACHAT_CLIENT_SECRET") self._available = bool(client_id and client_secret) if not self._available: logger.warning("GigaChat недоступен: не настроены GIGACHAT_CLIENT_ID или GIGACHAT_CLIENT_SECRET") else: logger.info("GigaChat доступен") except Exception as e: self._available = False logger.error(f"Ошибка проверки доступности GigaChat: {e}") return self._available def get_tools_schema(self, tools_registry: Dict[str, Any]) -> List[Dict[str, Any]]: """ Получить схему инструментов для промпта. Формирует описание инструментов в формате понятном GigaChat. """ schema = [] for name, tool in tools_registry.items(): if hasattr(tool, 'get_schema'): tool_schema = tool.get_schema() schema.append({ "name": name, "description": tool_schema.get("description", ""), "parameters": tool_schema.get("parameters", {}) }) elif hasattr(tool, 'description'): schema.append({ "name": name, "description": tool.description, "parameters": getattr(tool, 'parameters', {}) }) return schema def _build_tools_prompt(self, tools_schema: List[Dict[str, Any]]) -> str: """ Построить текстовое описание инструментов для промпта. GigaChat не поддерживает нативные tool calls, поэтому описываем инструменты в тексте и просим модель использовать специальный формат. """ if not tools_schema: return "" prompt_parts = [ "\n\n🛠️ ДОСТУПНЫЕ ИНСТРУМЕНТЫ:", "Ты можешь использовать следующие инструменты. Для вызова инструмента используй формат:", "```tool", '{"name": "имя_инструмента", "arguments": {аргументы}}', '```', "", "Список инструментов:" ] for tool in tools_schema: name = tool.get("name", "unknown") desc = tool.get("description", "Нет описания") params = tool.get("parameters", {}) prompt_parts.append(f"\n**{name}**") prompt_parts.append(f"Описание: {desc}") if params: prompt_parts.append(f"Параметры: {json.dumps(params, ensure_ascii=False)}") prompt_parts.extend([ "", "После вызова инструмента ты получишь результат и сможешь продолжить ответ." ]) return "\n".join(prompt_parts) def _parse_tool_calls(self, content: str) -> List[ToolCall]: """ Извлечь вызовы инструментов из текста ответа. Ищет блоки вида: ```tool {"name": "ssh_tool", "arguments": {"command": "df -h"}} ``` """ tool_calls = [] # Ищем блоки ```tool {...}``` pattern = r'```tool\s*\n({.*?})\s*\n```' matches = re.findall(pattern, content, re.DOTALL) for match in matches: try: tool_data = json.loads(match) tool_name = tool_data.get("name") tool_args = tool_data.get("arguments", {}) if tool_name: tool_calls.append(ToolCall( tool_name=tool_name, tool_args=tool_args, tool_call_id=f"gc_{len(tool_calls)}" )) except json.JSONDecodeError as e: logger.warning(f"Ошибка парсинга tool call: {e}") return tool_calls def _remove_tool_blocks(self, content: str) -> str: """Удалить блоки вызовов инструментов из текста.""" pattern = r'```tool\s*\n\{.*?\}\s*\n```' return re.sub(pattern, '', content, flags=re.DOTALL).strip() async def chat( self, prompt: str, system_prompt: Optional[str] = None, context: Optional[List[Dict[str, str]]] = None, tools: Optional[List[Dict[str, Any]]] = None, on_chunk: Optional[Callable[[str], Any]] = None, user_id: Optional[int] = None, **kwargs ) -> ProviderResponse: """ Отправить запрос GigaChat. Args: prompt: Запрос пользователя system_prompt: Системный промпт context: История диалога tools: Доступные инструменты (схема) on_chunk: Callback для потокового вывода (не используется) user_id: ID пользователя **kwargs: Дополнительные параметры Returns: ProviderResponse с ответом и возможными вызовами инструментов """ try: # Формируем системный промпт с инструментами full_system_prompt = system_prompt or "" if tools: tools_prompt = self._build_tools_prompt(tools) full_system_prompt += tools_prompt # Формируем сообщения messages = [] if full_system_prompt: messages.append(GigaChatMessage(role="system", content=full_system_prompt)) if context: for msg in context: role = msg.get("role", "user") content = msg.get("content", "") if role in ("user", "assistant", "system"): messages.append(GigaChatMessage(role=role, content=content)) if prompt: messages.append(GigaChatMessage(role="user", content=prompt)) # Выполняем запрос result = await self._tool.chat( messages=messages, user_id=str(user_id) if user_id else None, temperature=kwargs.get("temperature", 0.7), max_tokens=kwargs.get("max_tokens", 2000), ) if not result.get("content"): if result.get("error"): return ProviderResponse( success=False, error=result["error"], provider_name=self.provider_name ) else: return ProviderResponse( success=False, error="Пустой ответ от GigaChat", provider_name=self.provider_name ) content = result["content"] # Парсим вызовы инструментов tool_calls = self._parse_tool_calls(content) # Очищаем контент от блоков инструментов clean_content = self._remove_tool_blocks(content) return ProviderResponse( success=True, message=AIMessage( content=clean_content, tool_calls=tool_calls, metadata={ "model": result.get("model", "GigaChat"), "usage": result.get("usage", {}) } ), provider_name=self.provider_name, usage=result.get("usage") ) except Exception as e: logger.error(f"Ошибка GigaChat провайдера: {e}") return ProviderResponse( success=False, error=str(e), provider_name=self.provider_name ) async def execute_tool( self, tool_name: str, tool_args: Dict[str, Any], tool_call_id: Optional[str] = None, **kwargs ) -> ToolCall: """ Выполнить инструмент (заглушка). GigaChat не выполняет инструменты напрямую - это делает AIProviderManager через process_with_tools. """ return ToolCall( tool_name=tool_name, tool_args=tool_args, tool_call_id=tool_call_id, status=ToolCallStatus.PENDING )