293 lines
11 KiB
Python
293 lines
11 KiB
Python
#!/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
|
||
)
|