379 lines
15 KiB
Python
379 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
RAG-сервер – отдельный HTTP-сервис для обработки RAG-запросов.
|
||
Предоставляет API для выполнения RAG-запросов и индексации документов.
|
||
|
||
Этот сервер запускается как отдельный процесс и может обслуживать
|
||
несколько ботов и других систем одновременно.
|
||
|
||
ИСПОЛЬЗУЕТ: FastAPI для создания REST API и Uvicorn для запуска.
|
||
ИСТОРИЯ ХРАНИТСЯ НА СЕРВЕРЕ (в PostgreSQL).
|
||
"""
|
||
|
||
import logging
|
||
import os
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Optional, Dict, List, Any
|
||
|
||
# Добавляем путь к core, если запускаем из корня проекта
|
||
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||
|
||
from fastapi import FastAPI, HTTPException, Depends
|
||
from fastapi.middleware.cors import CORSMiddleware
|
||
from pydantic import BaseModel, Field
|
||
import uvicorn
|
||
|
||
# Импортируем наши модули
|
||
from core.utils.config_loader import BotConfig
|
||
from core.utils.logger import setup_logging
|
||
from core.services.postgres_service import PostgresService
|
||
from core.services.qdrant_service import QdrantService
|
||
from core.services.embedding_service import EmbeddingService
|
||
from core.services.kb_service import KBService
|
||
from core.services.giga_client import GigaClient
|
||
from core.services.file_service import FileService
|
||
from core.rag_orchestrator import RAGOrchestrator
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# ============================================================
|
||
# МОДЕЛИ ДАННЫХ (Pydantic) ДЛЯ API
|
||
# ============================================================
|
||
|
||
class QueryRequest(BaseModel):
|
||
"""
|
||
Модель запроса к RAG-серверу на выполнение RAG-запроса.
|
||
ИСТОРИЯ НЕ ПЕРЕДАЁТСЯ – сервер получает её из БД.
|
||
"""
|
||
query: str = Field(..., description="Текст запроса пользователя")
|
||
user_jid: str = Field(..., description="JID пользователя (без ресурса)")
|
||
room_jid: Optional[str] = Field(None, description="JID комнаты (None для личного чата)")
|
||
|
||
# history: List[Dict[str, str]] = Field(...) # УБРАНО!
|
||
|
||
prompts: Dict[str, str] = Field(
|
||
default_factory=dict,
|
||
description="Словарь промптов (system, synthesis, intent, expand, critique и т.д.)"
|
||
)
|
||
|
||
intent_override: Optional[str] = Field(None, description="Принудительное переопределение намерения")
|
||
last_file_path: Optional[str] = Field(None, description="Путь к последнему загруженному файлу")
|
||
last_file_text: Optional[str] = Field(None, description="Текст последнего загруженного файла")
|
||
|
||
|
||
class QueryResponse(BaseModel):
|
||
"""Модель ответа RAG-сервера на RAG-запрос."""
|
||
answer: str = Field(..., description="Итоговый ответ")
|
||
intent: str = Field(default="GENERAL", description="Распознанное намерение")
|
||
context: Optional[str] = Field(None, description="Использованный контекст (для отладки)")
|
||
sources: List[str] = Field(default_factory=list, description="Список источников")
|
||
confidence: Optional[float] = Field(None, description="Оценка уверенности (0-1)")
|
||
error: Optional[str] = Field(None, description="Сообщение об ошибке, если есть")
|
||
|
||
|
||
class IndexRequest(BaseModel):
|
||
"""Модель запроса на индексацию документа."""
|
||
file_name: str = Field(..., description="Исходное имя файла")
|
||
file_text: str = Field(..., description="Извлечённый текст документа")
|
||
user_jid: str = Field(..., description="JID владельца документа")
|
||
room_jid: Optional[str] = Field(None, description="JID комнаты (если документ комнатный)")
|
||
is_global: bool = Field(False, description="Глобальный ли документ")
|
||
title: Optional[str] = Field(None, description="Отображаемое название")
|
||
metadata: Optional[Dict[str, Any]] = Field(None, description="Дополнительные метаданные")
|
||
file_hash: Optional[str] = Field(None, description="SHA-256 хеш содержимого")
|
||
update_if_exists: bool = Field(True, description="Заменять ли существующий документ")
|
||
|
||
|
||
class IndexResponse(BaseModel):
|
||
"""Модель ответа на индексацию."""
|
||
doc_id: Optional[int] = Field(None, description="Идентификатор документа в БД")
|
||
chunk_count: int = Field(0, description="Количество проиндексированных чанков")
|
||
error: Optional[str] = Field(None, description="Сообщение об ошибке, если есть")
|
||
|
||
|
||
# ============================================================
|
||
# ГЛОБАЛЬНЫЙ ОРКЕСТРАТОР (инициализируется один раз при старте)
|
||
# ============================================================
|
||
|
||
_orchestrator: Optional[RAGOrchestrator] = None
|
||
_config: Optional[BotConfig] = None
|
||
|
||
|
||
def get_orchestrator() -> RAGOrchestrator:
|
||
"""Возвращает экземпляр RAGOrchestrator (инициализирует при первом вызове)."""
|
||
global _orchestrator
|
||
if _orchestrator is None:
|
||
raise RuntimeError("RAGOrchestrator не инициализирован. Сервер не запущен корректно.")
|
||
return _orchestrator
|
||
|
||
|
||
def init_orchestrator(config: BotConfig) -> RAGOrchestrator:
|
||
"""Инициализирует все сервисы и создаёт RAGOrchestrator."""
|
||
global _orchestrator, _config
|
||
_config = config
|
||
logger.info("Инициализация RAG-сервера...")
|
||
|
||
# ---- 1. Инициализация сервисов ----
|
||
# PostgreSQL (история, документы)
|
||
db = PostgresService(
|
||
host=config.db_host,
|
||
port=config.db_port,
|
||
user=config.db_user,
|
||
password=config.db_password,
|
||
db_name=config.db_name
|
||
)
|
||
|
||
# Qdrant (векторный поиск)
|
||
qdrant = QdrantService(
|
||
host=config.qdrant_host,
|
||
port=config.qdrant_port,
|
||
grpc_port=config.qdrant_grpc_port,
|
||
collection_name=config.qdrant_collection,
|
||
vector_size=config.qdrant_vector_size,
|
||
distance=config.qdrant_distance,
|
||
prefer_grpc=False
|
||
)
|
||
|
||
# Эмбеддинги (GigaChat)
|
||
embedding = EmbeddingService(
|
||
api_key=config.gigachat_api_key,
|
||
model=config.embedding_model,
|
||
timeout=config.embedding_timeout,
|
||
cache_size=config.embedding_cache_size,
|
||
verify_ssl=config.embedding_verify_ssl
|
||
)
|
||
|
||
# База знаний (индексация, поиск)
|
||
kb = KBService(
|
||
db=db,
|
||
qdrant=qdrant,
|
||
embedding=embedding,
|
||
collection_name=config.qdrant_collection,
|
||
chunk_size_tokens=config.chunk_size_tokens,
|
||
overlap_tokens=config.overlap_tokens,
|
||
approx_chunk_chars=config.chunking_approx_chunk_chars,
|
||
approx_overlap_chars=config.chunking_approx_overlap_chars
|
||
)
|
||
|
||
# GigaChat клиент (генерация)
|
||
giga = GigaClient(
|
||
api_key=config.gigachat_api_key,
|
||
model=config.ai_model,
|
||
temperature=config.ai_temperature,
|
||
timeout=config.ai_timeout,
|
||
verify_ssl=False
|
||
)
|
||
|
||
# Файловый сервис (извлечение текста)
|
||
files = FileService(config)
|
||
|
||
# ---- 2. Создание оркестратора ----
|
||
# Загружаем дефолтные промпты (опционально)
|
||
default_prompts = {}
|
||
system_prompt_path = getattr(config, 'system_prompt_file', None)
|
||
if system_prompt_path and system_prompt_path.exists():
|
||
try:
|
||
with open(system_prompt_path, 'r', encoding='utf-8') as f:
|
||
default_prompts['system'] = f.read()
|
||
except Exception as e:
|
||
logger.warning(f"Не удалось загрузить системный промпт: {e}")
|
||
|
||
orchestrator = RAGOrchestrator(
|
||
db=db,
|
||
qdrant=qdrant,
|
||
embedding=embedding,
|
||
kb=kb,
|
||
giga=giga,
|
||
files=files,
|
||
config=config,
|
||
default_prompts=default_prompts
|
||
)
|
||
|
||
_orchestrator = orchestrator
|
||
logger.info("RAG-сервер успешно инициализирован")
|
||
return orchestrator
|
||
|
||
|
||
# ============================================================
|
||
# FASTAPI ПРИЛОЖЕНИЕ
|
||
# ============================================================
|
||
|
||
app = FastAPI(
|
||
title="RAG-сервер",
|
||
description="Единый сервис для RAG-обработки запросов и индексации документов",
|
||
version="1.0.0"
|
||
)
|
||
|
||
# Разрешаем CORS для возможности запросов с других доменов (для тестов)
|
||
app.add_middleware(
|
||
CORSMiddleware,
|
||
allow_origins=["*"],
|
||
allow_credentials=True,
|
||
allow_methods=["*"],
|
||
allow_headers=["*"],
|
||
)
|
||
|
||
|
||
@app.on_event("startup")
|
||
async def startup_event():
|
||
"""Подключаемся к БД при старте сервера."""
|
||
if _orchestrator is None:
|
||
logger.error("Оркестратор не инициализирован.")
|
||
else:
|
||
try:
|
||
await _orchestrator.db.connect()
|
||
logger.info("Подключение к PostgreSQL установлено")
|
||
except Exception as e:
|
||
logger.error(f"Ошибка подключения к PostgreSQL: {e}")
|
||
|
||
|
||
@app.on_event("shutdown")
|
||
async def shutdown_event():
|
||
"""Закрываем соединение с БД при завершении сервера."""
|
||
if _orchestrator:
|
||
try:
|
||
await _orchestrator.db.close()
|
||
logger.info("Подключение к PostgreSQL закрыто")
|
||
except Exception as e:
|
||
logger.error(f"Ошибка закрытия PostgreSQL: {e}")
|
||
|
||
|
||
# ============================================================
|
||
# ЭНДПОИНТЫ API
|
||
# ============================================================
|
||
|
||
@app.post("/rag/query", response_model=QueryResponse)
|
||
async def rag_query(request: QueryRequest):
|
||
"""
|
||
Выполняет RAG-запрос.
|
||
ИСТОРИЯ БЕРЁТСЯ СЕРВЕРОМ ИЗ БД (не передаётся в запросе).
|
||
"""
|
||
orchestrator = get_orchestrator()
|
||
|
||
try:
|
||
result = await orchestrator.process_query(
|
||
query=request.query,
|
||
user_jid=request.user_jid,
|
||
room_jid=request.room_jid,
|
||
prompts=request.prompts,
|
||
intent_override=request.intent_override,
|
||
last_file_path=request.last_file_path,
|
||
last_file_text=request.last_file_text
|
||
)
|
||
return QueryResponse(
|
||
answer=result.get('answer', ''),
|
||
intent=result.get('intent', 'GENERAL'),
|
||
context=result.get('context'),
|
||
sources=result.get('sources', []),
|
||
confidence=result.get('confidence'),
|
||
error=result.get('error')
|
||
)
|
||
except Exception as e:
|
||
logger.error(f"Ошибка в /rag/query: {e}", exc_info=True)
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
|
||
@app.post("/rag/index", response_model=IndexResponse)
|
||
async def index_document(request: IndexRequest):
|
||
"""Индексирует документ в базу знаний."""
|
||
orchestrator = get_orchestrator()
|
||
|
||
try:
|
||
result = await orchestrator.index_document(
|
||
file_name=request.file_name,
|
||
file_text=request.file_text,
|
||
user_jid=request.user_jid,
|
||
room_jid=request.room_jid,
|
||
is_global=request.is_global,
|
||
title=request.title,
|
||
metadata=request.metadata,
|
||
file_hash=request.file_hash,
|
||
update_if_exists=request.update_if_exists
|
||
)
|
||
return IndexResponse(
|
||
doc_id=result.get('doc_id'),
|
||
chunk_count=result.get('chunk_count', 0),
|
||
error=result.get('error')
|
||
)
|
||
except Exception as e:
|
||
logger.error(f"Ошибка в /rag/index: {e}", exc_info=True)
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
|
||
@app.get("/health")
|
||
async def health_check():
|
||
"""Проверка здоровья сервиса."""
|
||
if _orchestrator is None:
|
||
return {"status": "unhealthy", "message": "Orchestrator not initialized"}
|
||
return {"status": "healthy"}
|
||
|
||
|
||
# ============================================================
|
||
# ТОЧКА ВХОДА
|
||
# ============================================================
|
||
|
||
def main():
|
||
"""Главная функция запуска RAG-сервера."""
|
||
import argparse
|
||
|
||
parser = argparse.ArgumentParser(description="RAG-сервер")
|
||
parser.add_argument(
|
||
"--profile-dir",
|
||
required=True,
|
||
help="Путь к директории профиля бота (например, /usr/local/etc/bots/metabot)"
|
||
)
|
||
parser.add_argument(
|
||
"--host",
|
||
default="0.0.0.0",
|
||
help="Хост для привязки сервера"
|
||
)
|
||
parser.add_argument(
|
||
"--port",
|
||
type=int,
|
||
default=8080,
|
||
help="Порт для привязки сервера"
|
||
)
|
||
parser.add_argument(
|
||
"--log-level",
|
||
default="info",
|
||
choices=["debug", "info", "warning", "error", "critical"],
|
||
help="Уровень логирования"
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
# Загружаем конфигурацию
|
||
try:
|
||
config = BotConfig(args.profile_dir)
|
||
except Exception as e:
|
||
print(f"Ошибка загрузки конфигурации: {e}")
|
||
sys.exit(1)
|
||
|
||
# Настраиваем логирование
|
||
log_level = getattr(logging, args.log_level.upper(), logging.INFO)
|
||
setup_logging(config.name, config.log_file, log_level=log_level)
|
||
|
||
logger.info(f"Запуск RAG-сервера с профилем {args.profile_dir}")
|
||
logger.info(f"Хост: {args.host}, порт: {args.port}")
|
||
|
||
# Инициализируем оркестратор
|
||
try:
|
||
init_orchestrator(config)
|
||
except Exception as e:
|
||
logger.error(f"Ошибка инициализации оркестратора: {e}", exc_info=True)
|
||
sys.exit(1)
|
||
|
||
# Запускаем Uvicorn
|
||
uvicorn.run(
|
||
"core.rag_server:app",
|
||
host=args.host,
|
||
port=args.port,
|
||
log_level=args.log_level,
|
||
reload=False
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main() |