From 3ff96463ae13d7b399806dac970e565dbf0dd0c3 Mon Sep 17 00:00:00 2001 From: Markov Andrey Date: Tue, 30 Jun 2026 14:33:41 +0000 Subject: [PATCH] Update file rag_server.py --- rag/rag_server.py | 471 +++++++++++++++++++++++++++++++--------------- 1 file changed, 322 insertions(+), 149 deletions(-) diff --git a/rag/rag_server.py b/rag/rag_server.py index 0bad34f..476430d 100644 --- a/rag/rag_server.py +++ b/rag/rag_server.py @@ -10,7 +10,7 @@ RAG-сервер – отдельный HTTP-сервис для обработ - Эндпоинт /rag/transcribe для транскрибации аудио (SaluteSpeech) - Использование Pydantic-модели AppConfig для конфигурации. - Аутентификация через API-ключ (X-API-Key) -- Документирование API через OpenAPI (Swagger UI и ReDoc) +- Полная OpenAPI-документация с примерами и тегами """ import logging @@ -25,10 +25,10 @@ from typing import Optional, Dict, List, Any # Добавляем путь к rag, если запускаем из корня проекта sys.path.insert(0, str(Path(__file__).parent.parent)) -from fastapi import FastAPI, HTTPException, Depends, UploadFile, File as FastAPIFile +from fastapi import FastAPI, HTTPException, Depends, UploadFile, File as FastAPIFile, status from fastapi.middleware.cors import CORSMiddleware from fastapi.openapi.utils import get_openapi -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, HttpUrl, constr import uvicorn # Импорты наших модулей @@ -55,58 +55,181 @@ class QueryRequest(BaseModel): Модель запроса к RAG-серверу на выполнение RAG-запроса. ИСТОРИЯ НЕ ПЕРЕДАЁТСЯ – сервер получает её из БД. """ - query: str = Field(..., description="Текст запроса пользователя", example="Какой OEE на линии?") - user_jid: str = Field(..., description="JID пользователя (без ресурса)", example="user@domain.ru") - room_jid: Optional[str] = Field(None, description="JID комнаты (None для личного чата)", example="room@conference.domain.ru") + query: str = Field( + ..., + description="Текст запроса пользователя", + example="Какое тактовое время установлено на линии сборки?" + ) + user_jid: str = Field( + ..., + description="JID пользователя (без ресурса)", + example="user@domain.ru" + ) + room_jid: Optional[str] = Field( + None, + description="JID комнаты (None для личного чата)", + example="room@conference.domain.ru" + ) prompts: Dict[str, str] = Field( default_factory=dict, description="Словарь промптов (system, synthesis, intent, expand, critique и т.д.)" ) - intent_override: Optional[str] = Field(None, description="Принудительное переопределение намерения", example="METRICS") - last_file_path: Optional[str] = Field(None, description="Путь к последнему загруженному файлу") - last_file_text: Optional[str] = Field(None, description="Текст последнего загруженного файла") + intent_override: Optional[str] = Field( + None, + description="Принудительное переопределение намерения", + example="FACT" + ) + last_file_path: Optional[str] = Field( + None, + description="Путь к последнему загруженному файлу на сервере" + ) + last_file_text: Optional[str] = Field( + None, + description="Текст последнего загруженного файла" + ) + + class Config: + schema_extra = { + "example": { + "query": "Какое тактовое время установлено на линии сборки?", + "user_jid": "user@domain.ru", + "room_jid": None, + "prompts": {}, + "intent_override": None, + "last_file_path": None, + "last_file_text": None + } + } 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="Сообщение об ошибке, если есть") + answer: str = Field( + ..., + description="Итоговый ответ", + example="Тактовое время на линии сборки составляет 120 секунд. [Источник: Регламент_линии_2025.docx]" + ) + intent: str = Field( + default="GENERAL", + description="Распознанное намерение", + example="FACT" + ) + context: Optional[str] = Field( + None, + description="Использованный контекст (для отладки)" + ) + sources: List[str] = Field( + default_factory=list, + description="Список источников", + example=["Регламент_линии_2025.docx"] + ) + confidence: Optional[float] = Field( + None, + description="Оценка уверенности (0-1)", + example=0.95 + ) + error: Optional[str] = Field( + None, + description="Сообщение об ошибке, если есть" + ) class IndexRequest(BaseModel): """Модель запроса на индексацию документа.""" - file_name: str = Field(..., description="Исходное имя файла", example="report.docx") - file_text: str = Field(..., description="Извлечённый текст документа") - user_jid: str = Field(..., description="JID владельца документа", example="user@domain.ru") - 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="Заменять ли существующий документ") + file_name: str = Field( + ..., + description="Исходное имя файла", + example="Регламент_отдела.docx" + ) + file_text: str = Field( + ..., + description="Извлечённый текст документа", + example="Полный текст документа..." + ) + user_jid: str = Field( + ..., + description="JID владельца документа", + example="user@domain.ru" + ) + room_jid: Optional[str] = Field( + None, + description="JID комнаты (если документ комнатный)", + example="room@conference.domain.ru" + ) + 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="Сообщение об ошибке, если есть") + doc_id: Optional[int] = Field( + None, + description="Идентификатор документа в БД", + example=123 + ) + chunk_count: int = Field( + 0, + description="Количество проиндексированных чанков", + example=15 + ) + error: Optional[str] = Field( + None, + description="Сообщение об ошибке, если есть" + ) class VisionResponse(BaseModel): """Модель ответа на распознавание изображения.""" - text: str = Field(..., description="Распознанный текст") - status: str = Field(..., description="Статус: ok, no_text_found, error") + text: str = Field( + ..., + description="Распознанный текст", + example="Текст с изображения..." + ) + status: str = Field( + ..., + description="Статус: ok, no_text_found, error", + example="ok" + ) + error: Optional[str] = Field( + None, + description="Сообщение об ошибке" + ) class TranscribeResponse(BaseModel): """Модель ответа на транскрибацию аудио.""" - text: str = Field(..., description="Распознанный текст") - status: str = Field(..., description="Статус: ok, no_speech_detected, error") + text: str = Field( + ..., + description="Распознанный текст", + example="Распознанная речь..." + ) + status: str = Field( + ..., + description="Статус: ok, no_speech_detected, error", + example="ok" + ) + error: Optional[str] = Field( + None, + description="Сообщение об ошибке" + ) # ============================================================ @@ -213,57 +336,43 @@ def init_orchestrator(config): # ============================================================ -# FASTAPI ПРИЛОЖЕНИЕ с документацией +# FASTAPI ПРИЛОЖЕНИЕ С КАСТОМНОЙ ОPENAPI # ============================================================ -# Описание тегов для группировки эндпоинтов в документации -tags_metadata = [ - { - "name": "RAG", - "description": "Основные RAG-операции: выполнение запроса и индексация документов", - }, - { - "name": "Media", - "description": "Обработка медиафайлов: OCR для изображений и транскрибация аудио", - }, - { - "name": "Health", - "description": "Проверка состояния сервиса", - } -] - +# Создаём приложение с базовыми метаданными app = FastAPI( - title="RAG-сервер Эфцекабот", + title="Эфцекабот RAG-сервер", description=""" - Единый сервис для RAG-обработки запросов и индексации документов. + ## Единый RAG-сервис для интеллектуальной обработки запросов - ## Аутентификация - Для всех защищённых эндпоинтов (кроме `/health`) требуется передача API-ключа - в заголовке `X-API-Key`. + **Эфцекабот** – корпоративная платформа для гибридного RAG на базе GigaChat. - ## История диалога - Сервер самостоятельно хранит историю диалогов в PostgreSQL и использует её - для контекста при ответах. Клиентам не нужно передавать историю в запросах. + ### Основные возможности + - **Гибридный поиск** (dense + sparse) с RRF-ранжированием + - **Самокритика и перегенерация** ответов + - **Иерархическое резюмирование** длинных документов и диалогов + - **Поддержка личных, комнатных и глобальных документов** + - **Распознавание текста на изображениях (OCR)** + - **Транскрибация аудио** через SaluteSpeech - ## Поддерживаемые форматы файлов - - Документы: DOCX, PDF, XLSX, PPTX, TXT, CSV, JSON - - Изображения: JPEG, PNG, BMP, TIFF (OCR через Tesseract) - - Аудио: OGG, WAV, MP3, AMR, M4A (транскрибация через SaluteSpeech) - - Архивы: ZIP, 7z (рекурсивная распаковка) + ### Аутентификация + Все эндпоинты (кроме `/health`) требуют передачи API-ключа в заголовке `X-API-Key`. """, version="3.0.0", - openapi_tags=tags_metadata, contact={ - "name": "Андрей Марков, Вениамин Кокорин", + "name": "Команда Эфцекабот", "email": "support@fckbot.ru", }, license_info={ - "name": "ISC", + "name": "ISC License", "url": "https://opensource.org/licenses/ISC", }, + docs_url="/docs", + redoc_url="/redoc", + openapi_url="/openapi.json", ) -# Добавляем CORS +# Разрешаем CORS для возможности запросов с других доменов (для тестов) app.add_middleware( CORSMiddleware, allow_origins=["*"], @@ -272,7 +381,8 @@ app.add_middleware( allow_headers=["*"], ) -# Настраиваем OpenAPI для отображения схемы авторизации + +# Кастомизируем OpenAPI-схему для добавления примеров и кодов ошибок def custom_openapi(): if app.openapi_schema: return app.openapi_schema @@ -282,35 +392,35 @@ def custom_openapi(): version=app.version, description=app.description, routes=app.routes, - tags=app.openapi_tags, contact=app.contact, license_info=app.license_info, ) - # Добавляем схему безопасности для API-ключа + # Добавляем глобальные компоненты безопасности openapi_schema["components"]["securitySchemes"] = { "APIKeyHeader": { "type": "apiKey", "in": "header", "name": "X-API-Key", - "description": "API-ключ для доступа к защищённым эндпоинтам", + "description": "API-ключ для доступа к сервису. Получите у администратора.", } } - # Применяем security ко всем эндпоинтам кроме /health - for path in openapi_schema["paths"]: - if path != "/health": - for method in openapi_schema["paths"][path]: - openapi_schema["paths"][path][method]["security"] = [{"APIKeyHeader": []}] + # Добавляем глобальную security requirement + openapi_schema["security"] = [{"APIKeyHeader": []}] + + # Исключаем /health из требований безопасности + # (это делается через параметр `include_in_schema=False` в эндпоинте) app.openapi_schema = openapi_schema return app.openapi_schema + app.openapi = custom_openapi # ============================================================ -# ОБРАБОТЧИКИ СОБЫТИЙ +# LIFECYCLE EVENTS # ============================================================ @app.on_event("startup") @@ -338,26 +448,54 @@ async def shutdown_event(): # ============================================================ -# ЭНДПОИНТЫ API (все защищены аутентификацией, кроме /health) +# ЭНДПОИНТЫ API С ТЕГАМИ И ПОДРОБНЫМИ ОПИСАНИЯМИ # ============================================================ +# Теги для группировки в документации +tags_metadata = [ + { + "name": "RAG", + "description": "Основные RAG-операции: запросы и индексация документов", + }, + { + "name": "Vision", + "description": "Распознавание текста на изображениях (OCR)", + }, + { + "name": "Audio", + "description": "Транскрибация аудиофайлов (SaluteSpeech)", + }, + { + "name": "Health", + "description": "Проверка состояния сервиса", + }, +] + + @app.post( "/rag/query", response_model=QueryResponse, tags=["RAG"], summary="Выполнить RAG-запрос", - description="Отправляет запрос пользователя в RAG-пайплайн. " - "Сервер автоматически подгружает историю диалога из БД.", - responses={ - 200: {"description": "Успешный ответ", "model": QueryResponse}, - 403: {"description": "Неверный или отсутствующий API-ключ"}, - 500: {"description": "Внутренняя ошибка сервера"}, - } + description=""" + Отправляет текстовый запрос пользователя, выполняет полный RAG-пайплайн: + + 1. **Классификация намерений** – определяет тип запроса (FACT, PROCEDURE, METRICS и т.д.) + 2. **Расширение запроса** – добавляет синонимы и связанные термины + 3. **Гибридный поиск** (dense + sparse) в Qdrant + 4. **Переранжирование** через кросс-энкодер + 5. **Генерация ответа** через GigaChat + 6. **Самокритика** и перегенерация при необходимости + 7. **Сохранение истории** диалога в PostgreSQL + + **Требования**: заголовок `X-API-Key` с валидным ключом. + **История**: сервер автоматически получает историю диалога из БД. + """, + response_description="Ответ содержит сгенерированный текст, намерение, источники и уверенность" ) async def rag_query(request: QueryRequest, _: str = Depends(verify_api_key)): """ Выполняет RAG-запрос. - ИСТОРИЯ БЕРЁТСЯ СЕРВЕРОМ ИЗ БД (не передаётся в запросе). """ orchestrator = get_orchestrator() @@ -381,22 +519,35 @@ async def rag_query(request: QueryRequest, _: str = Depends(verify_api_key)): ) except Exception as e: logger.error(f"Ошибка в /rag/query: {e}", exc_info=True) - raise HTTPException(status_code=500, detail=str(e)) + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=str(e) + ) @app.post( "/rag/index", response_model=IndexResponse, tags=["RAG"], - summary="Проиндексировать документ", - description="Загружает текст документа в базу знаний. " - "Документ разбивается на чанки, для каждого генерируются эмбеддинги " - "и сохраняются в Qdrant. Метаданные хранятся в PostgreSQL.", - responses={ - 200: {"description": "Индексация выполнена", "model": IndexResponse}, - 403: {"description": "Неверный или отсутствующий API-ключ"}, - 500: {"description": "Внутренняя ошибка сервера"}, - } + summary="Индексировать документ", + description=""" + Индексирует документ в базу знаний. + + **Процесс:** + 1. Документ разбивается на чанки (с учётом токенов) + 2. Для каждого чанка вычисляются dense- и sparse-эмбеддинги + 3. Чанки сохраняются в Qdrant с метаданными + 4. Метаданные сохраняются в PostgreSQL с флагом `indexed` + + **Поддерживаемые типы:** + - Личные (`is_global=False`, `room_jid=None`) + - Комнатные (`room_jid` указан) + - Глобальные (`is_global=True`, только для админов) + + **Обновление:** если документ с таким именем уже существует в комнате, + он будет заменён новой версией (по хешу). + """, + response_description="Возвращает идентификатор документа и количество чанков" ) async def index_document(request: IndexRequest, _: str = Depends(verify_api_key)): """Индексирует документ в базу знаний.""" @@ -421,55 +572,47 @@ async def index_document(request: IndexRequest, _: str = Depends(verify_api_key) ) except Exception as e: logger.error(f"Ошибка в /rag/index: {e}", exc_info=True) - raise HTTPException(status_code=500, detail=str(e)) - - -@app.get( - "/health", - tags=["Health"], - summary="Проверка состояния сервиса", - description="Возвращает статус работоспособности RAG-сервера. " - "Не требует аутентификации.", - responses={ - 200: {"description": "Сервис работает", "content": {"application/json": {"example": {"status": "healthy"}}}}, - 503: {"description": "Сервис недоступен"}, - } -) -async def health_check(): - """Проверка здоровья сервиса (без аутентификации).""" - if _orchestrator is None: - return {"status": "unhealthy", "message": "Orchestrator not initialized"} - return {"status": "healthy"} + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=str(e) + ) @app.post( "/rag/vision", response_model=VisionResponse, - tags=["Media"], - summary="Распознать текст на изображении (OCR)", - description="Отправляет изображение на распознавание текста через Tesseract OCR. " - "Поддерживаются форматы: JPEG, PNG, BMP, TIFF.", - responses={ - 200: {"description": "Текст распознан", "model": VisionResponse}, - 400: {"description": "Неподдерживаемый формат файла"}, - 403: {"description": "Неверный или отсутствующий API-ключ"}, - 500: {"description": "Внутренняя ошибка сервера"}, - } + tags=["Vision"], + summary="Распознать текст на изображении", + description=""" + Отправляет изображение на OCR-распознавание через Tesseract. + + **Поддерживаемые форматы:** + - JPEG, PNG, BMP, TIFF + + **Возвращает:** + - Распознанный текст + - Статус: `ok`, `no_text_found`, `error` + + **Требования:** заголовок `X-API-Key`. + """, + response_description="Текст, распознанный на изображении" ) async def vision_endpoint( - file: UploadFile = FastAPIFile(...), + file: UploadFile = FastAPIFile( + ..., + description="Файл изображения (JPEG, PNG, BMP, TIFF)" + ), _: str = Depends(verify_api_key) ): - """ - Распознаёт текст на изображении через OCR (Tesseract). - Возвращает JSON: {"text": str, "status": "ok" | "no_text_found" | "error"} - """ orchestrator = get_orchestrator() # Проверяем расширение ext = os.path.splitext(file.filename)[1].lower() if ext not in ('.jpg', '.jpeg', '.png', '.bmp', '.tiff'): - raise HTTPException(status_code=400, detail="Поддерживаются только изображения (jpg, png, bmp, tiff)") + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Поддерживаются только изображения (jpg, png, bmp, tiff)" + ) # Сохраняем во временный файл with tempfile.NamedTemporaryFile(delete=False, suffix=ext) as tmp: @@ -477,7 +620,6 @@ async def vision_endpoint( shutil.copyfileobj(file.file, tmp) try: - # Извлекаем текст через существующий FileService (синхронный метод) loop = asyncio.get_running_loop() text = await loop.run_in_executor( None, @@ -489,7 +631,7 @@ async def vision_endpoint( return VisionResponse(text=text, status="ok") except Exception as e: logger.error(f"Ошибка OCR: {e}", exc_info=True) - raise HTTPException(status_code=500, detail=str(e)) + return VisionResponse(text="", status="error", error=str(e)) finally: if os.path.exists(tmp_path): os.unlink(tmp_path) @@ -498,31 +640,41 @@ async def vision_endpoint( @app.post( "/rag/transcribe", response_model=TranscribeResponse, - tags=["Media"], - summary="Транскрибировать аудиофайл", - description="Отправляет аудиофайл на распознавание речи через SaluteSpeech. " - "Поддерживаются форматы: OGG, WAV, MP3, AMR, M4A.", - responses={ - 200: {"description": "Речь распознана", "model": TranscribeResponse}, - 400: {"description": "Неподдерживаемый формат файла"}, - 403: {"description": "Неверный или отсутствующий API-ключ"}, - 500: {"description": "Внутренняя ошибка сервера"}, - } + tags=["Audio"], + summary="Транскрибировать аудио", + description=""" + Отправляет аудиофайл на транскрибацию через SaluteSpeech. + + **Поддерживаемые форматы:** + - OGG, WAV, MP3, AMR, M4A + + **Требования:** + - Заголовок `X-API-Key` + - Переменная окружения `SALUTE_SPEECH_AUTH` + - Установленный `ffmpeg` на сервере + + **Возвращает:** + - Распознанный текст + - Статус: `ok`, `no_speech_detected`, `error` + """, + response_description="Текст, распознанный из аудио" ) async def transcribe_endpoint( - file: UploadFile = FastAPIFile(...), + file: UploadFile = FastAPIFile( + ..., + description="Аудиофайл (OGG, WAV, MP3, AMR, M4A)" + ), _: str = Depends(verify_api_key) ): - """ - Транскрибирует аудиофайл через SaluteSpeech. - Возвращает JSON: {"text": str, "status": "ok" | "no_speech_detected" | "error"} - """ orchestrator = get_orchestrator() # Проверяем расширение ext = os.path.splitext(file.filename)[1].lower() if ext not in ('.ogg', '.wav', '.mp3', '.amr', '.m4a'): - raise HTTPException(status_code=400, detail="Поддерживаются только аудио (ogg, wav, mp3, amr, m4a)") + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Поддерживаются только аудио (ogg, wav, mp3, amr, m4a)" + ) # Сохраняем во временный файл with tempfile.NamedTemporaryFile(delete=False, suffix=ext) as tmp: @@ -530,19 +682,40 @@ async def transcribe_endpoint( shutil.copyfileobj(file.file, tmp) try: - # Транскрибируем через существующий FileService (асинхронный метод) text = await orchestrator.files.transcribe_audio(tmp_path) if not text: return TranscribeResponse(text="", status="no_speech_detected") return TranscribeResponse(text=text, status="ok") except Exception as e: logger.error(f"Ошибка транскрибации: {e}", exc_info=True) - raise HTTPException(status_code=500, detail=str(e)) + return TranscribeResponse(text="", status="error", error=str(e)) finally: if os.path.exists(tmp_path): os.unlink(tmp_path) +@app.get( + "/health", + tags=["Health"], + summary="Проверка состояния сервиса", + description=""" + Проверяет, запущен ли RAG-сервер и инициализирован ли оркестратор. + + **Ответ:** + - `healthy` – сервер работает + - `unhealthy` – оркестратор не инициализирован + + **Примечание:** этот эндпоинт **не требует** аутентификации. + """, + response_description="Статус сервиса" +) +async def health_check(): + """Проверка здоровья сервиса (без аутентификации).""" + if _orchestrator is None: + return {"status": "unhealthy", "message": "Orchestrator not initialized"} + return {"status": "healthy"} + + # ============================================================ # ТОЧКА ВХОДА # ============================================================