Update file rag_server.py

This commit is contained in:
Markov Andrey
2026-06-30 14:33:41 +00:00
parent f3fb6b1f21
commit 3ff96463ae

View File

@@ -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"}
# ============================================================
# ТОЧКА ВХОДА
# ============================================================