FDKost
e04933b9c1
- пофикшен баг с авторизацией; - поменен README.md, более подробно описан запуск проекта; - починен .env для проекта. Checked: - docker-compose работает; - auth работает; - чат с нейросетью работает, но кидает 400 из за NEWPLANET-AI-AGENTS,нужно настроить подключение.
116 lines
4.3 KiB
Python
116 lines
4.3 KiB
Python
import aiohttp
|
|
import ssl
|
|
import base64
|
|
import uuid
|
|
import time
|
|
from typing import Optional, List, Dict, Any
|
|
from app.core.config import settings
|
|
|
|
|
|
class GigaChatService:
|
|
def __init__(self):
|
|
self.access_token: Optional[str] = None
|
|
self.token_expires_at: Optional[float] = None
|
|
|
|
async def _get_token(self) -> str:
|
|
"""Получить OAuth токен"""
|
|
# Проверяем, не истек ли токен (оставляем запас 60 секунд)
|
|
if self.access_token and self.token_expires_at:
|
|
if time.time() < (self.token_expires_at - 60):
|
|
return self.access_token
|
|
|
|
credentials = f"{settings.GIGACHAT_CLIENT_ID}:{settings.GIGACHAT_CLIENT_SECRET}"
|
|
encoded_credentials = base64.b64encode(credentials.encode()).decode()
|
|
|
|
headers = {
|
|
"Authorization": f"Basic {encoded_credentials}",
|
|
"Content-Type": "application/x-www-form-urlencoded",
|
|
"Accept": "application/json",
|
|
"RqUID": str(uuid.uuid4())
|
|
}
|
|
|
|
data = {"scope": "GIGACHAT_API_PERS"}
|
|
|
|
# Создаем SSL контекст без проверки сертификата (только для разработки!)
|
|
ssl_context = ssl.create_default_context()
|
|
ssl_context.check_hostname = False
|
|
ssl_context.verify_mode = ssl.CERT_NONE
|
|
|
|
connector = aiohttp.TCPConnector(ssl=ssl_context)
|
|
async with aiohttp.ClientSession(connector=connector) as session:
|
|
async with session.post(
|
|
settings.GIGACHAT_AUTH_URL,
|
|
headers=headers,
|
|
data=data
|
|
) as response:
|
|
if response.status != 200:
|
|
raise Exception(f"Failed to get token: {response.status}")
|
|
|
|
result = await response.json()
|
|
self.access_token = result.get("access_token")
|
|
expires_in = result.get("expires_at", 1800)
|
|
# expires_at может быть timestamp или количество секунд
|
|
if expires_in > 1000000000: # Это timestamp
|
|
self.token_expires_at = expires_in
|
|
else: # Это количество секунд
|
|
self.token_expires_at = time.time() + expires_in
|
|
|
|
return self.access_token
|
|
|
|
async def chat(
|
|
self,
|
|
message: str,
|
|
context: Optional[List[Dict[str, Any]]] = None,
|
|
model: str = None
|
|
) -> Dict[str, Any]:
|
|
"""Отправить сообщение в GigaChat"""
|
|
token = await self._get_token()
|
|
model = model or settings.GIGACHAT_MODEL_CHAT
|
|
|
|
messages = context or []
|
|
messages.append({"role": "user", "content": message})
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer {token}",
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
payload = {
|
|
"model": model,
|
|
"messages": messages,
|
|
"temperature": 0.7,
|
|
"max_tokens": 2000
|
|
}
|
|
|
|
# Создаем SSL контекст без проверки сертификата (только для разработки!)
|
|
ssl_context = ssl.create_default_context()
|
|
ssl_context.check_hostname = False
|
|
ssl_context.verify_mode = ssl.CERT_NONE
|
|
|
|
connector = aiohttp.TCPConnector(ssl=ssl_context)
|
|
async with aiohttp.ClientSession(connector=connector) as session:
|
|
async with session.post(
|
|
f"{settings.GIGACHAT_BASE_URL}/chat/completions",
|
|
headers=headers,
|
|
json=payload
|
|
) as response:
|
|
if response.status != 200:
|
|
error_text = await response.text()
|
|
raise Exception(f"GigaChat API error: {response.status} - {error_text}")
|
|
|
|
result = await response.json()
|
|
return result
|
|
|
|
async def generate_text(
|
|
self,
|
|
prompt: str,
|
|
model: str = None
|
|
) -> str:
|
|
"""Генерация текста по промпту"""
|
|
result = await self.chat(prompt, model=model)
|
|
return result.get("choices", [{}])[0].get("message", {}).get("content", "")
|
|
|
|
|
|
gigachat_service = GigaChatService()
|
|
|