691 lines
29 KiB
Python
691 lines
29 KiB
Python
"""
|
||
GigaChat API Tool для Telegram CLI Bot
|
||
|
||
Инструмент для работы с GigaChat API (Сбер).
|
||
Поддерживает генерацию текста, чат-сессии и различные модели.
|
||
|
||
Документация: https://developers.sber.ru/docs/ru/gigachat
|
||
"""
|
||
|
||
import os
|
||
import base64
|
||
import httpx
|
||
import asyncio
|
||
import uuid
|
||
import logging
|
||
from typing import Optional, List, Dict, Any
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime, timedelta
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
@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" # Модель по умолчанию
|
||
model_lite: str = "GigaChat" # Lite модель для простых запросов
|
||
model_pro: str = "GigaChat-Pro" # Pro модель для сложных запросов
|
||
api_url: str = "https://gigachat.devices.sberbank.ru/api/v1"
|
||
timeout: int = 60
|
||
# Пороги для переключения моделей
|
||
complexity_token_threshold: int = 50 # Если токенов в запросе > порога → Pro
|
||
complexity_keyword_threshold: int = 2 # Если ключевых слов сложности >= порога → Pro
|
||
|
||
|
||
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"),
|
||
model_lite=os.getenv("GIGACHAT_MODEL_LITE", "GigaChat"),
|
||
model_pro=os.getenv("GIGACHAT_MODEL_PRO", "GigaChat-Pro"),
|
||
complexity_token_threshold=int(os.getenv("GIGACHAT_TOKEN_THRESHOLD", "50")),
|
||
complexity_keyword_threshold=int(os.getenv("GIGACHAT_KEYWORD_THRESHOLD", "2")),
|
||
)
|
||
|
||
def _get_auth_headers(self) -> Dict[str, str]:
|
||
"""Получение заголовков для авторизации"""
|
||
# GigaChat требует RqUID (UUID) и Content-Type для OAuth
|
||
return {
|
||
"Content-Type": "application/x-www-form-urlencoded",
|
||
"RqUID": str(uuid.uuid4()),
|
||
}
|
||
|
||
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
|
||
|
||
# Запрашиваем новый токен с использованием Basic Auth
|
||
credentials = f"{self.config.client_id}:{self.config.client_secret}"
|
||
encoded_credentials = base64.b64encode(credentials.encode()).decode()
|
||
|
||
# GigaChat использует самоподписанные сертификаты - отключаем верификацию
|
||
async with httpx.AsyncClient(verify=False) as client:
|
||
response = await client.post(
|
||
self.config.auth_url,
|
||
headers={
|
||
"Authorization": f"Basic {encoded_credentials}",
|
||
"Content-Type": "application/x-www-form-urlencoded",
|
||
"RqUID": str(uuid.uuid4()),
|
||
},
|
||
data={"scope": self.config.scope},
|
||
timeout=30,
|
||
)
|
||
|
||
# Логируем для отладки
|
||
logger.debug(f"GigaChat auth status: {response.status_code}")
|
||
logger.debug(f"GigaChat auth response: {response.text[:200]}")
|
||
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
|
||
self._access_token = data["access_token"]
|
||
# Токен действителен 30 минут, кэшируем на 25 минут
|
||
self._token_expires = datetime.now() + timedelta(minutes=25)
|
||
|
||
# Логируем начало токена для проверки (первые 50 символов)
|
||
logger.info(f"GigaChat токен получен: {self._access_token[:50]}...")
|
||
|
||
return self._access_token
|
||
|
||
def _estimate_query_complexity(self, messages: List[GigaChatMessage]) -> dict:
|
||
"""
|
||
Оценить сложность запроса для выбора модели (Lite или Pro).
|
||
|
||
Критерии сложности:
|
||
1. Длина запроса (количество токенов/слов)
|
||
2. Наличие ключевых слов для сложных задач
|
||
3. Наличие инструментов (tool calls)
|
||
4. Технические термины
|
||
|
||
Returns:
|
||
Dict с оценкой сложности и рекомендуемой моделью
|
||
"""
|
||
# Собираем весь текст из сообщений пользователя
|
||
user_text = ""
|
||
for msg in messages:
|
||
if msg.role == "user":
|
||
user_text += " " + msg.content
|
||
|
||
user_text = user_text.lower()
|
||
|
||
# 1. Оценка по длине (считаем слова как грубая оценка токенов)
|
||
word_count = len(user_text.split())
|
||
token_estimate = word_count * 1.3 # Примерная конверсия слов в токены
|
||
|
||
# 2. Ключевые слова для сложных задач
|
||
complex_keywords = [
|
||
# Программирование и код
|
||
'код', 'функция', 'класс', 'метод', 'переменная', 'цикл', 'условие',
|
||
'алгоритм', 'структура данных', 'массив', 'словарь', 'список',
|
||
'импорт', 'экспорт', 'модуль', 'пакет', 'библиотека', 'фреймворк',
|
||
'дебаг', 'отладк', 'тест', 'юнит тест', 'интеграционн',
|
||
'рефактор', 'оптимиз', 'производительност',
|
||
# Анализ и работа с данными
|
||
'анализ', 'анализиров', 'сравни', 'сравнени', 'исследовани',
|
||
'закономерност', 'паттерн', 'тенденци', 'прогноз',
|
||
# Системные задачи
|
||
'конфигурац', 'настройк', 'деплой', 'развертывани', 'оркестрац',
|
||
'контейнер', 'docker', 'kubernetes', 'k8s', 'helm',
|
||
'мониторинг', 'логировани', 'трассировк', 'метрик',
|
||
# Сложные запросы
|
||
'объясни', 'расскажи подробно', 'детальн', 'подробн',
|
||
'почему', 'зачем', 'как работает', 'принцип работы',
|
||
'спроектируй', 'спроектировать', 'архитектур', 'архитектура',
|
||
'реализуй', 'реализовать', 'напиши код', 'создай функцию',
|
||
]
|
||
|
||
complexity_keywords_count = sum(
|
||
1 for keyword in complex_keywords
|
||
if keyword in user_text
|
||
)
|
||
|
||
# 3. Наличие технических терминов
|
||
tech_terms = [
|
||
'api', 'http', 'rest', 'graphql', 'grpc', 'websocket',
|
||
'sql', 'nosql', 'postgres', 'mysql', 'mongodb', 'redis',
|
||
'git', 'merge', 'commit', 'branch', 'pull request', 'merge request',
|
||
'ci/cd', 'pipeline', 'jenkins', 'gitlab', 'github',
|
||
'linux', 'bash', 'shell', 'terminal', 'ssh',
|
||
'python', 'javascript', 'typescript', 'java', 'go', 'rust', 'cpp',
|
||
'react', 'vue', 'angular', 'django', 'flask', 'fastapi', 'express',
|
||
]
|
||
|
||
tech_terms_count = sum(
|
||
1 for term in tech_terms
|
||
if term in user_text
|
||
)
|
||
|
||
# 4. Наличие инструментов в контексте
|
||
has_tools = any(
|
||
'tool' in msg.content.lower() or 'инструмент' in msg.content.lower()
|
||
for msg in messages
|
||
)
|
||
|
||
# Принятие решения
|
||
use_pro = False
|
||
reasons = []
|
||
|
||
# Если токенов больше порога → Pro
|
||
if token_estimate > self.config.complexity_token_threshold:
|
||
use_pro = True
|
||
reasons.append(f"длинный запрос ({word_count} слов, ~{int(token_estimate)} токенов)")
|
||
|
||
# Если много ключевых слов сложности → Pro
|
||
if complexity_keywords_count >= self.config.complexity_keyword_threshold:
|
||
use_pro = True
|
||
reasons.append(f"сложная задача ({complexity_keywords_count} ключевых слов)")
|
||
|
||
# Если есть технические термины + инструменты → Pro
|
||
if tech_terms_count >= 2 and has_tools:
|
||
use_pro = True
|
||
reasons.append(f"техническая задача с инструментами ({tech_terms_count} терминов)")
|
||
|
||
# Если есть явные запросы на работу с кодом/файлами → Pro
|
||
if any(phrase in user_text for phrase in [
|
||
'исходник', 'source code', 'посмотри код', 'проанализируй код',
|
||
'работай с файлом', 'прочитай файл', 'изучи код'
|
||
]):
|
||
use_pro = True
|
||
reasons.append("работа с кодом/файлами")
|
||
|
||
model = self.config.model_pro if use_pro else self.config.model_lite
|
||
|
||
return {
|
||
"use_pro": use_pro,
|
||
"model": model,
|
||
"word_count": word_count,
|
||
"token_estimate": int(token_estimate),
|
||
"complexity_keywords": complexity_keywords_count,
|
||
"tech_terms": tech_terms_count,
|
||
"has_tools": has_tools,
|
||
"reasons": reasons
|
||
}
|
||
|
||
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,
|
||
user_id: Optional[str] = None,
|
||
) -> 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: Использовать ли историю чата
|
||
user_id: ID пользователя для заголовка X-User-Id
|
||
|
||
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()
|
||
|
||
# Автоматически выбираем модель на основе сложности запроса
|
||
# Если модель явно не указана
|
||
selected_model = model
|
||
model_info = None
|
||
if selected_model is None:
|
||
model_info = self._estimate_query_complexity(messages)
|
||
selected_model = model_info["model"]
|
||
logger.info(f"📊 GigaChat выбор модели: {selected_model} (причины: {', '.join(model_info['reasons']) if model_info['reasons'] else 'простой запрос'})")
|
||
|
||
# Преобразуем сообщения в формат API
|
||
api_messages = [
|
||
{"role": msg.role, "content": msg.content}
|
||
for msg in messages
|
||
]
|
||
|
||
payload = {
|
||
"model": selected_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",
|
||
"X-User-Id": str(user_id) if user_id else "telegram-bot",
|
||
}
|
||
|
||
# Логируем запрос для отладки
|
||
logger.debug(f"GigaChat API URL: {self.config.api_url}/chat/completions")
|
||
logger.debug(f"GigaChat headers: {headers}")
|
||
logger.debug(f"GigaChat payload: model={selected_model}, messages={len(api_messages)}, max_tokens={max_tokens}")
|
||
|
||
# GigaChat использует самоподписанные сертификаты - отключаем верификацию
|
||
async with httpx.AsyncClient(verify=False) as client:
|
||
try:
|
||
response = await client.post(
|
||
f"{self.config.api_url}/chat/completions",
|
||
headers=headers,
|
||
json=payload,
|
||
timeout=self.config.timeout,
|
||
)
|
||
|
||
# Логируем для отладки
|
||
logger.debug(f"GigaChat chat status: {response.status_code}")
|
||
logger.debug(f"GigaChat response headers: {dict(response.headers)}")
|
||
if response.status_code != 200:
|
||
logger.error(f"GigaChat error response: {response.text[:1000]}")
|
||
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
except httpx.HTTPStatusError as e:
|
||
logger.error(f"GigaChat HTTP error: {e}")
|
||
logger.error(f"Response: {e.response.text[:500]}")
|
||
return {
|
||
"content": "",
|
||
"error": f"HTTP {e.response.status_code}: {e.response.text[:200]}",
|
||
}
|
||
except httpx.HTTPError as e:
|
||
logger.error(f"GigaChat request error: {e}")
|
||
return {
|
||
"content": "",
|
||
"error": f"Request error: {str(e)}",
|
||
}
|
||
|
||
# Добавляем ответ ассистента в историю
|
||
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", selected_model),
|
||
"usage": data.get("usage", {}),
|
||
"finish_reason": data["choices"][0]["finish_reason"] if data.get("choices") else "",
|
||
"complexity_info": model_info, # Информация о выборе модели для отладки
|
||
}
|
||
|
||
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 chat_with_functions(
|
||
self,
|
||
messages: List[Dict[str, Any]],
|
||
functions: Optional[List[Dict[str, Any]]] = None,
|
||
model: Optional[str] = None,
|
||
temperature: float = 0.7,
|
||
max_tokens: int = 2000,
|
||
top_p: float = 0.1,
|
||
repetition_penalty: float = 1.0,
|
||
user_id: Optional[str] = None,
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
Отправка запроса к GigaChat API с поддержкой function calling.
|
||
|
||
Args:
|
||
messages: Список сообщений в формате API
|
||
functions: Массив функций для вызова
|
||
model: Модель для генерации
|
||
temperature: Температура генерации
|
||
max_tokens: Максимум токенов
|
||
top_p: Параметр top-p sampling
|
||
repetition_penalty: Штраф за повторения
|
||
user_id: ID пользователя
|
||
|
||
Returns:
|
||
Dict с ответом API включая возможный function_call
|
||
"""
|
||
token = await self._get_access_token()
|
||
|
||
# Выбираем модель на основе сложности запроса
|
||
selected_model = model
|
||
model_info = None
|
||
if selected_model is None:
|
||
# Преобразуем messages в формат GigaChatMessage для оценки сложности
|
||
gc_messages = [GigaChatMessage(role=msg["role"], content=msg.get("content", "")) for msg in messages]
|
||
model_info = self._estimate_query_complexity(gc_messages)
|
||
selected_model = model_info["model"]
|
||
logger.info(f"📊 GigaChat выбор модели: {selected_model} (причины: {', '.join(model_info['reasons']) if model_info['reasons'] else 'простой запрос'})")
|
||
|
||
# Формируем payload
|
||
payload = {
|
||
"model": selected_model,
|
||
"messages": messages,
|
||
"temperature": temperature,
|
||
"max_tokens": max_tokens,
|
||
"top_p": top_p,
|
||
"repetition_penalty": repetition_penalty,
|
||
}
|
||
|
||
# Добавляем functions если есть
|
||
if functions:
|
||
payload["functions"] = functions
|
||
# function_call: "auto" позволяет модели самой решать когда вызывать функции
|
||
payload["function_call"] = "auto"
|
||
logger.info(f"🔧 GigaChat function calling: {len(functions)} функций доступно")
|
||
|
||
headers = {
|
||
"Authorization": f"Bearer {token}",
|
||
"Content-Type": "application/json",
|
||
"X-User-Id": str(user_id) if user_id else "telegram-bot",
|
||
}
|
||
|
||
logger.info(f"📤 GigaChat API: model={selected_model}, messages={len(messages)}, functions={len(functions) if functions else 0}")
|
||
|
||
# GigaChat использует самоподписанные сертификаты - отключаем верификацию
|
||
async with httpx.AsyncClient(verify=False) as client:
|
||
try:
|
||
response = await client.post(
|
||
f"{self.config.api_url}/chat/completions",
|
||
headers=headers,
|
||
json=payload,
|
||
timeout=self.config.timeout,
|
||
)
|
||
|
||
logger.debug(f"GigaChat chat_with_functions status: {response.status_code}")
|
||
logger.debug(f"GigaChat response: {response.text[:500]}")
|
||
|
||
if response.status_code != 200:
|
||
logger.error(f"GigaChat error response: {response.text[:1000]}")
|
||
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
except httpx.HTTPStatusError as e:
|
||
logger.error(f"GigaChat HTTP error: {e}")
|
||
logger.error(f"Response: {e.response.text[:500]}")
|
||
return {
|
||
"content": "",
|
||
"error": f"HTTP {e.response.status_code}: {e.response.text[:200]}",
|
||
"choices": [],
|
||
}
|
||
except httpx.HTTPError as e:
|
||
logger.error(f"GigaChat request error: {e}")
|
||
return {
|
||
"content": "",
|
||
"error": f"Request error: {str(e)}",
|
||
"choices": [],
|
||
}
|
||
|
||
# Извлекаем content и function_call
|
||
content = ""
|
||
function_call = None
|
||
functions_state_id = None
|
||
|
||
if data.get("choices"):
|
||
choice = data["choices"][0]
|
||
message = choice.get("message", {})
|
||
content = message.get("content", "")
|
||
function_call = message.get("function_call")
|
||
functions_state_id = message.get("functions_state_id")
|
||
|
||
logger.info(f"📬 GigaChat ответ: content_len={len(content)}, function_call={function_call is not None}, functions_state_id={functions_state_id}")
|
||
|
||
return {
|
||
"content": content,
|
||
"function_call": function_call,
|
||
"functions_state_id": functions_state_id,
|
||
"model": data.get("model", selected_model),
|
||
"usage": data.get("usage", {}),
|
||
"finish_reason": data["choices"][0]["finish_reason"] if data.get("choices") else "",
|
||
"choices": data.get("choices", []),
|
||
}
|
||
|
||
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",
|
||
}
|
||
|
||
# GigaChat использует самоподписанные сертификаты - отключаем верификацию
|
||
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}",
|
||
}
|
||
|
||
# GigaChat использует самоподписанные сертификаты - отключаем верификацию
|
||
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"),
|
||
)
|