177 lines
8.2 KiB
Python
177 lines
8.2 KiB
Python
import os
|
||
|
||
import aiohttp
|
||
import base64
|
||
import uuid
|
||
import time
|
||
from urllib.parse import urlencode
|
||
from typing import Optional, List, Dict, Any
|
||
|
||
from app.core.config import settings
|
||
from app.services.ai_agent_client import ai_agent_client
|
||
|
||
class GigaChatService:
|
||
"""
|
||
Сервис для работы с GigaChat через внешний AI-agent сервис.
|
||
Все запросы к GigaChat теперь проходят через внешний сервис.
|
||
"""
|
||
def __init__(self):
|
||
self.access_token: Optional[str] = None
|
||
self.token_expires_at: Optional[float] = None
|
||
|
||
async def _get_token(self) -> str:
|
||
"""
|
||
Получить OAuth токен.
|
||
|
||
ВНИМАНИЕ: Этот метод больше не используется, так как все запросы
|
||
к GigaChat теперь проходят через внешний AI-agent сервис.
|
||
Метод оставлен для возможной обратной совместимости.
|
||
"""
|
||
# Проверяем, не истек ли токен (оставляем запас 60 секунд)
|
||
if self.access_token and self.token_expires_at:
|
||
if time.time() < (self.token_expires_at - 60):
|
||
return self.access_token
|
||
|
||
# Проверяем наличие credentials
|
||
client_id = os.getenv("GIGACHAT_CLIENT_ID")
|
||
client_secret = os.getenv("GIGACHAT_CLIENT_SECRET")
|
||
|
||
if not client_id or not client_secret:
|
||
raise Exception(
|
||
"GigaChat credentials not configured. "
|
||
"Please set GIGACHAT_CLIENT_ID and GIGACHAT_CLIENT_SECRET in .env file"
|
||
)
|
||
|
||
# Формируем credentials и кодируем в Base64 с явным указанием UTF-8
|
||
credentials = f"{client_id}:{client_secret}".strip().encode('utf-8') # Обрезаем лишние символы
|
||
encoded_credentials = base64.b64encode(credentials).decode('utf-8')
|
||
|
||
headers = {
|
||
"Authorization": f"Basic {encoded_credentials}",
|
||
"Content-Type": "application/x-www-form-urlencoded",
|
||
"Accept": "application/json",
|
||
"RqUID": str(uuid.uuid4())
|
||
}
|
||
|
||
# Правильно кодируем данные формы (как в рабочем примере)
|
||
form_data = {
|
||
"grant_type": "client_credentials",
|
||
"scope": "GIGACHAT_API_PERS"
|
||
}
|
||
|
||
# Отключаем проверку SSL (только для разработки!)
|
||
# Используем ssl=False для полного отключения проверки сертификата
|
||
connector = aiohttp.TCPConnector(ssl=False)
|
||
async with aiohttp.ClientSession(connector=connector) as session:
|
||
async with session.post(
|
||
os.getenv("GIGACHAT_BASE_URL"),
|
||
headers=headers,
|
||
data=form_data
|
||
) as response:
|
||
if response.status != 200:
|
||
# Получаем детали ошибки из ответа
|
||
try:
|
||
error_body = await response.text()
|
||
# Пытаемся распарсить как JSON, если не получается - возвращаем текст
|
||
try:
|
||
error_json = await response.json()
|
||
error_detail = error_json.get("error_description") or error_json.get("error") or str(error_json)
|
||
except:
|
||
error_detail = error_body
|
||
except:
|
||
error_detail = "No error details available"
|
||
|
||
raise Exception(
|
||
f"Failed to get token: HTTP {response.status}. "
|
||
f"Error details: {error_detail}. "
|
||
f"Check your GIGACHAT_CLIENT_ID and GIGACHAT_CLIENT_SECRET_2 in .env file"
|
||
)
|
||
|
||
result = await response.json()
|
||
self.access_token = result.get("access_token")
|
||
if not self.access_token:
|
||
raise Exception(f"Token not found in response: {result}")
|
||
|
||
# Обрабатываем время истечения токена (может быть expires_at или expires_in)
|
||
expires_at = result.get("expires_at")
|
||
expires_in = result.get("expires_in")
|
||
|
||
if expires_at:
|
||
# expires_at может быть timestamp или количество секунд
|
||
if expires_at > 1000000000: # Это timestamp
|
||
self.token_expires_at = expires_at
|
||
else: # Это количество секунд
|
||
self.token_expires_at = time.time() + expires_at
|
||
elif expires_in:
|
||
# expires_in - это всегда количество секунд до истечения
|
||
self.token_expires_at = time.time() + expires_in
|
||
else:
|
||
# По умолчанию 30 минут (1800 секунд)
|
||
self.token_expires_at = time.time() + 1800
|
||
|
||
return self.access_token
|
||
|
||
async def chat(
|
||
self,
|
||
message: str,
|
||
context: Optional[List[Dict[str, Any]]] = None,
|
||
model: str = None
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
Отправить сообщение в GigaChat через внешний AI-agent сервис.
|
||
Сохраняет обратную совместимость с форматом ответа GigaChat API.
|
||
"""
|
||
try:
|
||
# Используем внешний AI-agent сервис
|
||
result = await ai_agent_client.chat(
|
||
message=message,
|
||
conversation_id=None, # Если нужен conversation_id, его нужно передавать отдельно
|
||
context=context
|
||
)
|
||
|
||
# Преобразуем ответ AI-agent сервиса в формат, совместимый с GigaChat API
|
||
# Предполагаем, что ai_agent_client возвращает структуру ChatResponse или аналогичную
|
||
if "response" in result:
|
||
# Если ответ в формате ChatResponse, преобразуем в формат GigaChat
|
||
return {
|
||
"model": result.get("model", model or settings.GIGACHAT_MODEL_CHAT or "GigaChat"),
|
||
"choices": [{
|
||
"message": {
|
||
"role": "assistant",
|
||
"content": result["response"]
|
||
},
|
||
"finish_reason": "stop"
|
||
}],
|
||
"usage": {
|
||
"total_tokens": result.get("tokens_used", 0),
|
||
"prompt_tokens": 0,
|
||
"completion_tokens": result.get("tokens_used", 0)
|
||
}
|
||
}
|
||
else:
|
||
# Если ответ уже в формате GigaChat, возвращаем как есть
|
||
return result
|
||
except Exception as e:
|
||
# Если внешний сервис недоступен, пробрасываем ошибку
|
||
raise Exception(f"AI-agent service error: {str(e)}")
|
||
|
||
async def generate_text(
|
||
self,
|
||
prompt: str,
|
||
model: str = None
|
||
) -> str:
|
||
"""
|
||
Генерация текста по промпту через внешний AI-agent сервис.
|
||
"""
|
||
try:
|
||
# Используем метод generate_text из ai_agent_client
|
||
response_text = await ai_agent_client.generate_text(prompt=prompt, model=model)
|
||
return response_text
|
||
except Exception as e:
|
||
# Если произошла ошибка, пробрасываем её
|
||
raise Exception(f"AI-agent service error: {str(e)}")
|
||
|
||
|
||
gigachat_service = GigaChatService()
|
||
|