From 41f32964fe04a51537a9934d410a4b98f30eb7e5 Mon Sep 17 00:00:00 2001 From: Markov Andrey Date: Tue, 30 Jun 2026 16:03:15 +0000 Subject: [PATCH] Update 2 files - /requirements.txt - /rag/rag_server.py --- rag/rag_server.py | 142 ++++++++++++++++++++++++++-------------------- requirements.txt | 5 +- 2 files changed, 85 insertions(+), 62 deletions(-) diff --git a/rag/rag_server.py b/rag/rag_server.py index 476430d..7a498ca 100644 --- a/rag/rag_server.py +++ b/rag/rag_server.py @@ -6,9 +6,8 @@ RAG-сервер – отдельный HTTP-сервис для обработ ИСТОРИЯ ХРАНИТСЯ НА СЕРВЕРЕ (в PostgreSQL). ДОБАВЛЕНО: -- Эндпоинт /rag/vision для распознавания текста на изображениях (OCR) -- Эндпоинт /rag/transcribe для транскрибации аудио (SaluteSpeech) -- Использование Pydantic-модели AppConfig для конфигурации. +- Rate limiting через slowapi (защита от перегрузки) +- Централизованное логирование всех запросов (middleware) - Аутентификация через API-ключ (X-API-Key) - Полная OpenAPI-документация с примерами и тегами """ @@ -19,18 +18,24 @@ import sys import tempfile import shutil import asyncio +import time from pathlib import Path 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, status +from fastapi import FastAPI, HTTPException, Depends, UploadFile, File as FastAPIFile, Request, status from fastapi.middleware.cors import CORSMiddleware from fastapi.openapi.utils import get_openapi -from pydantic import BaseModel, Field, HttpUrl, constr +from pydantic import BaseModel, Field import uvicorn +# Rate limiting +from slowapi import Limiter, _rate_limit_exceeded_handler +from slowapi.util import get_remote_address +from slowapi.errors import RateLimitExceeded + # Импорты наших модулей from .utils.config_loader import load_config from .utils.logger import setup_logging @@ -336,10 +341,13 @@ def init_orchestrator(config): # ============================================================ -# FASTAPI ПРИЛОЖЕНИЕ С КАСТОМНОЙ ОPENAPI +# FASTAPI ПРИЛОЖЕНИЕ С КАСТОМНОЙ OPENAPI, RATE LIMITING И ЛОГИРОВАНИЕМ # ============================================================ -# Создаём приложение с базовыми метаданными +# 1. Настройка rate limiter +limiter = Limiter(key_func=get_remote_address) + +# 2. Создание приложения app = FastAPI( title="Эфцекабот RAG-сервер", description=""" @@ -357,6 +365,12 @@ app = FastAPI( ### Аутентификация Все эндпоинты (кроме `/health`) требуют передачи API-ключа в заголовке `X-API-Key`. + + ### Rate Limiting + - `/rag/query`: 10 запросов в минуту + - `/rag/index`: 5 запросов в минуту + - `/rag/vision`: 15 запросов в минуту + - `/rag/transcribe`: 10 запросов в минуту """, version="3.0.0", contact={ @@ -372,7 +386,11 @@ app = FastAPI( openapi_url="/openapi.json", ) -# Разрешаем CORS для возможности запросов с других доменов (для тестов) +# 3. Подключаем обработчик ошибок rate limiting +app.state.limiter = limiter +app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) + +# 4. CORS app.add_middleware( CORSMiddleware, allow_origins=["*"], @@ -381,8 +399,22 @@ app.add_middleware( allow_headers=["*"], ) +# 5. Централизованное логирование запросов (middleware) +@app.middleware("http") +async def log_requests(request: Request, call_next): + """Логирует все входящие запросы с временем выполнения и IP-адресом.""" + start_time = time.time() + response = await call_next(request) + duration = time.time() - start_time + client_host = request.client.host if request.client else "unknown" + logger.info( + f"{request.method} {request.url.path} → {response.status_code} " + f"({duration:.3f}s) from {client_host}" + ) + return response -# Кастомизируем OpenAPI-схему для добавления примеров и кодов ошибок + +# 6. Кастомизация OpenAPI def custom_openapi(): if app.openapi_schema: return app.openapi_schema @@ -405,17 +437,10 @@ def custom_openapi(): "description": "API-ключ для доступа к сервису. Получите у администратора.", } } - - # Добавляем глобальную security requirement openapi_schema["security"] = [{"APIKeyHeader": []}] - - # Исключаем /health из требований безопасности - # (это делается через параметр `include_in_schema=False` в эндпоинте) - app.openapi_schema = openapi_schema return app.openapi_schema - app.openapi = custom_openapi @@ -448,30 +473,9 @@ async def shutdown_event(): # ============================================================ -# ЭНДПОИНТЫ API С ТЕГАМИ И ПОДРОБНЫМИ ОПИСАНИЯМИ +# ЭНДПОИНТЫ API С ТЕГАМИ, ПОДРОБНЫМИ ОПИСАНИЯМИ И RATE LIMITING # ============================================================ -# Теги для группировки в документации -tags_metadata = [ - { - "name": "RAG", - "description": "Основные RAG-операции: запросы и индексация документов", - }, - { - "name": "Vision", - "description": "Распознавание текста на изображениях (OCR)", - }, - { - "name": "Audio", - "description": "Транскрибация аудиофайлов (SaluteSpeech)", - }, - { - "name": "Health", - "description": "Проверка состояния сервиса", - }, -] - - @app.post( "/rag/query", response_model=QueryResponse, @@ -490,24 +494,27 @@ tags_metadata = [ **Требования**: заголовок `X-API-Key` с валидным ключом. **История**: сервер автоматически получает историю диалога из БД. + **Rate limit**: 10 запросов в минуту. """, response_description="Ответ содержит сгенерированный текст, намерение, источники и уверенность" ) -async def rag_query(request: QueryRequest, _: str = Depends(verify_api_key)): - """ - Выполняет RAG-запрос. - """ +@limiter.limit("10/minute") +async def rag_query( + request: Request, # обязательный параметр для slowapi + query_req: QueryRequest, + _: str = Depends(verify_api_key) +): 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 + query=query_req.query, + user_jid=query_req.user_jid, + room_jid=query_req.room_jid, + prompts=query_req.prompts, + intent_override=query_req.intent_override, + last_file_path=query_req.last_file_path, + last_file_text=query_req.last_file_text ) return QueryResponse( answer=result.get('answer', ''), @@ -546,24 +553,30 @@ async def rag_query(request: QueryRequest, _: str = Depends(verify_api_key)): **Обновление:** если документ с таким именем уже существует в комнате, он будет заменён новой версией (по хешу). + + **Rate limit**: 5 запросов в минуту. """, response_description="Возвращает идентификатор документа и количество чанков" ) -async def index_document(request: IndexRequest, _: str = Depends(verify_api_key)): - """Индексирует документ в базу знаний.""" +@limiter.limit("5/minute") +async def index_document( + request: Request, + index_req: IndexRequest, + _: str = Depends(verify_api_key) +): 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 + file_name=index_req.file_name, + file_text=index_req.file_text, + user_jid=index_req.user_jid, + room_jid=index_req.room_jid, + is_global=index_req.is_global, + title=index_req.title, + metadata=index_req.metadata, + file_hash=index_req.file_hash, + update_if_exists=index_req.update_if_exists ) return IndexResponse( doc_id=result.get('doc_id'), @@ -594,10 +607,13 @@ async def index_document(request: IndexRequest, _: str = Depends(verify_api_key) - Статус: `ok`, `no_text_found`, `error` **Требования:** заголовок `X-API-Key`. + **Rate limit**: 15 запросов в минуту. """, response_description="Текст, распознанный на изображении" ) +@limiter.limit("15/minute") async def vision_endpoint( + request: Request, file: UploadFile = FastAPIFile( ..., description="Файл изображения (JPEG, PNG, BMP, TIFF)" @@ -656,10 +672,14 @@ async def vision_endpoint( **Возвращает:** - Распознанный текст - Статус: `ok`, `no_speech_detected`, `error` + + **Rate limit**: 10 запросов в минуту. """, response_description="Текст, распознанный из аудио" ) +@limiter.limit("10/minute") async def transcribe_endpoint( + request: Request, file: UploadFile = FastAPIFile( ..., description="Аудиофайл (OGG, WAV, MP3, AMR, M4A)" @@ -705,7 +725,7 @@ async def transcribe_endpoint( - `healthy` – сервер работает - `unhealthy` – оркестратор не инициализирован - **Примечание:** этот эндпоинт **не требует** аутентификации. + **Примечание:** этот эндпоинт **не требует** аутентификации и **не имеет** rate limiting. """, response_description="Статус сервиса" ) diff --git a/requirements.txt b/requirements.txt index ddc681d..74b2131 100644 --- a/requirements.txt +++ b/requirements.txt @@ -64,4 +64,7 @@ fastapi >= 0.115.0 uvicorn >= 0.30.0 # Для CORS поддержки -python-multipart >= 0.0.9 \ No newline at end of file +python-multipart >= 0.0.9 + +# Защита от перегрузки эндпоинтов +slowapi>=0.1.0 \ No newline at end of file