Update 9 files

- /rag/config_models.py
- /rag/utils/config_loader.py
- /rag/auth.py
- /rag/rag_server.py
- /rag/rag_api.py
- /rag/rag_orchestrator.py
- /rag/__init__.py
- /bots/xmpp/client.py
- /bots/rag_client.py
This commit is contained in:
Markov Andrey
2026-06-30 14:27:34 +00:00
parent 068504c988
commit 59feffc190
9 changed files with 180 additions and 179 deletions

View File

@@ -14,6 +14,7 @@ HTTP-клиент для взаимодействия с RAG-сервером.
ДОБАВЛЕНО: ДОБАВЛЕНО:
- Метод vision() для распознавания текста на изображениях - Метод vision() для распознавания текста на изображениях
- Метод transcribe() для транскрибации аудио - Метод transcribe() для транскрибации аудио
- Аутентификация через API-ключ (заголовок X-API-Key)
""" """
import os import os
@@ -33,6 +34,7 @@ class RAGClient:
def __init__( def __init__(
self, self,
server_url: str, server_url: str,
api_key: Optional[str] = None, # <-- добавлен API-ключ
timeout: int = 60, timeout: int = 60,
max_retries: int = 3, max_retries: int = 3,
retry_delay: float = 1.0 retry_delay: float = 1.0
@@ -42,16 +44,22 @@ class RAGClient:
Аргументы: Аргументы:
server_url (str): базовый URL RAG-сервера (например, http://localhost:8080) server_url (str): базовый URL RAG-сервера (например, http://localhost:8080)
api_key (Optional[str]): API-ключ для аутентификации (передаётся в заголовке X-API-Key)
timeout (int): таймаут на запрос в секундах timeout (int): таймаут на запрос в секундах
max_retries (int): максимальное количество повторных попыток при ошибке max_retries (int): максимальное количество повторных попыток при ошибке
retry_delay (float): базовая задержка между попытками (секунды) retry_delay (float): базовая задержка между попытками (секунды)
""" """
self.server_url = server_url.rstrip('/') self.server_url = server_url.rstrip('/')
self.api_key = api_key # <-- сохраняем ключ
self.timeout = timeout self.timeout = timeout
self.max_retries = max_retries self.max_retries = max_retries
self.retry_delay = retry_delay self.retry_delay = retry_delay
self._session: Optional[aiohttp.ClientSession] = None self._session: Optional[aiohttp.ClientSession] = None
logger.info(f"RAGClient инициализирован с сервером {server_url}, таймаут={timeout}с, макс. попыток={max_retries}") logger.info(
f"RAGClient инициализирован с сервером {server_url}, "
f"таймаут={timeout}с, макс. попыток={max_retries}, "
f"API-ключ {'задан' if api_key else 'не задан'}"
)
async def _get_session(self) -> aiohttp.ClientSession: async def _get_session(self) -> aiohttp.ClientSession:
""" """
@@ -75,6 +83,7 @@ class RAGClient:
async def _post(self, endpoint: str, payload: Dict[str, Any]) -> Dict[str, Any]: async def _post(self, endpoint: str, payload: Dict[str, Any]) -> Dict[str, Any]:
""" """
Внутренний метод для отправки POST-запроса с повторными попытками. Внутренний метод для отправки POST-запроса с повторными попытками.
Автоматически добавляет заголовок X-API-Key, если ключ задан.
Аргументы: Аргументы:
endpoint (str): путь эндпоинта (например, '/rag/query') endpoint (str): путь эндпоинта (например, '/rag/query')
@@ -90,12 +99,17 @@ class RAGClient:
url = f"{self.server_url}{endpoint}" url = f"{self.server_url}{endpoint}"
session = await self._get_session() session = await self._get_session()
# Подготавливаем заголовки
headers = {}
if self.api_key:
headers['X-API-Key'] = self.api_key
last_exception = None last_exception = None
for attempt in range(self.max_retries): for attempt in range(self.max_retries):
try: try:
logger.debug(f"Попытка {attempt + 1}/{self.max_retries} к {endpoint}") logger.debug(f"Попытка {attempt + 1}/{self.max_retries} к {endpoint}")
async with session.post(url, json=payload) as resp: async with session.post(url, json=payload, headers=headers) as resp:
if resp.status == 200: if resp.status == 200:
return await resp.json() return await resp.json()
else: else:
@@ -190,7 +204,10 @@ class RAGClient:
""" """
try: try:
session = await self._get_session() session = await self._get_session()
async with session.get(f"{self.server_url}/health", timeout=10) as resp: headers = {}
if self.api_key:
headers['X-API-Key'] = self.api_key
async with session.get(f"{self.server_url}/health", timeout=10, headers=headers) as resp:
if resp.status == 200: if resp.status == 200:
data = await resp.json() data = await resp.json()
if data.get('status') == 'healthy': if data.get('status') == 'healthy':
@@ -308,10 +325,15 @@ class RAGClient:
session = await self._get_session() session = await self._get_session()
url = f"{self.server_url}/rag/vision" url = f"{self.server_url}/rag/vision"
# Добавляем заголовок с ключом
headers = {}
if self.api_key:
headers['X-API-Key'] = self.api_key
try: try:
with open(file_path, 'rb') as f: with open(file_path, 'rb') as f:
files = {'file': (os.path.basename(file_path), f, 'image/jpeg')} files = {'file': (os.path.basename(file_path), f, 'image/jpeg')}
async with session.post(url, data=files) as resp: async with session.post(url, data=files, headers=headers) as resp:
if resp.status == 200: if resp.status == 200:
return await resp.json() return await resp.json()
else: else:
@@ -338,10 +360,14 @@ class RAGClient:
session = await self._get_session() session = await self._get_session()
url = f"{self.server_url}/rag/transcribe" url = f"{self.server_url}/rag/transcribe"
headers = {}
if self.api_key:
headers['X-API-Key'] = self.api_key
try: try:
with open(file_path, 'rb') as f: with open(file_path, 'rb') as f:
files = {'file': (os.path.basename(file_path), f, 'audio/ogg')} files = {'file': (os.path.basename(file_path), f, 'audio/ogg')}
async with session.post(url, data=files) as resp: async with session.post(url, data=files, headers=headers) as resp:
if resp.status == 200: if resp.status == 200:
return await resp.json() return await resp.json()
else: else:

View File

@@ -3,6 +3,10 @@
XMPP клиент главный класс Bot. XMPP клиент главный класс Bot.
Теперь это тонкий клиент, который только общается с RAG-сервером. Теперь это тонкий клиент, который только общается с RAG-сервером.
Не содержит БД, Qdrant, GigaChat всё на сервере. Не содержит БД, Qdrant, GigaChat всё на сервере.
ИСПОЛЬЗУЕТ:
- AppConfig из rag.config_models для единой конфигурации.
- RAGClient с передачей API-ключа для аутентификации.
""" """
import asyncio import asyncio
@@ -10,7 +14,11 @@ import logging
from datetime import datetime from datetime import datetime
from slixmpp import ClientXMPP from slixmpp import ClientXMPP
from core.utils.config_loader import BotConfig # Импорты из нового пакета rag для конфигурации
from rag.config_models import AppConfig
# Остальные импорты остаются из core (старый код) позже их нужно будет перенести в rag,
# но пока оставляем как есть для совместимости.
from core.services.file_service import FileService from core.services.file_service import FileService
from core.handlers.metrics_handler import MetricsHandler from core.handlers.metrics_handler import MetricsHandler
from core.rag_client import RAGClient from core.rag_client import RAGClient
@@ -32,7 +40,7 @@ class Bot(ClientXMPP):
Тонкий клиент: вся RAG-логика на сервере. Тонкий клиент: вся RAG-логика на сервере.
""" """
def __init__(self, config: BotConfig): def __init__(self, config: AppConfig): # <-- изменён тип конфига
# Полный JID: user@domain/resource # Полный JID: user@domain/resource
full_jid = f"{config.jid}/{config.xmpp_resource}" full_jid = f"{config.jid}/{config.xmpp_resource}"
super().__init__(full_jid, config.xmpp_password) super().__init__(full_jid, config.xmpp_password)
@@ -47,35 +55,59 @@ class Bot(ClientXMPP):
# ----- 2. Загрузка промптов (остаются локально) ----- # ----- 2. Загрузка промптов (остаются локально) -----
# Промпты загружаются из файлов и передаются в каждом запросе к серверу. # Промпты загружаются из файлов и передаются в каждом запросе к серверу.
self.ai_prompts = {} self.ai_prompts = {}
prompt_files = { # В AppConfig теперь есть свойства для путей к файлам промптов,
'system': config.system_prompt_file, # но они называются так же, как в BotConfig, поэтому доступ через config.<имя> работает.
'synthesis': config.synthesis_prompt_file, # Однако в BotConfig они назывались system_prompt_file и т.д. они есть в AppConfig?
'quality': config.quality_prompt_file, # В AppConfig мы добавили прокси-свойства только для основных параметров,
'rerank': config.rerank_prompt_file, # но не для путей к промптам. Поэтому здесь нужно либо использовать config.prompts_dir_path
'expand': config.expand_prompt_file, # и формировать пути вручную, либо оставить как было, если в AppConfig есть соответствующие свойства.
'intent': config.intent_prompt_file, # Для обратной совместимости мы добавили свойства в AppConfig, но не для prompt_file.
'metrics': config.metrics_prompt_file, # Поэтому временно оставим как было: будем использовать config.prompts_dir_path и имена файлов.
'summary': config.summary_prompt_file, # Лучше прочитать имена файлов из config.ai.prompts и загрузить их.
'consistency': config.consistency_prompt_file, # Но для простоты пока оставим старый способ, если свойства есть.
'critique': config.critique_prompt_file, # Однако в AppConfig нет system_prompt_file и т.д. они были в BotConfig.
'spellcheck': config.spellcheck_prompt_file, # В BotConfig они были путями к файлам. В AppConfig мы не стали их добавлять,
'generate_document': config.generate_document_prompt_file, # чтобы не дублировать. Вместо этого мы загружаем все промпты в config.prompts_content.
} # Поэтому мы можем использовать config.prompts_content, который уже содержит все промпты.
for name, path in prompt_files.items(): # Используем его:
try: self.ai_prompts = config.prompts_content.copy() if config.prompts_content else {}
with open(path, 'r', encoding='utf-8') as f: if not self.ai_prompts:
self.ai_prompts[name] = f.read() logger.warning("Промпты не загружены из конфига, пробуем загрузить из файлов вручную...")
logger.debug(f"Загружен промпт {name} из {path}") # Запасной вариант загружаем из файлов, используя config.prompts_dir_path
except Exception as e: prompts_dir = config.prompts_dir_path
logger.error(f"Не удалось загрузить промпт {name} из {path}: {e}") if prompts_dir and prompts_dir.exists():
self.ai_prompts[name] = "" for prompt_name, filename in config.ai.prompts.items():
file_path = prompts_dir / filename
if file_path.exists():
try:
with open(file_path, 'r', encoding='utf-8') as f:
self.ai_prompts[prompt_name] = f.read()
logger.debug(f"Загружен промпт {prompt_name} из {file_path}")
except Exception as e:
logger.error(f"Не удалось загрузить промпт {prompt_name} из {file_path}: {e}")
self.ai_prompts[prompt_name] = ""
else:
logger.warning(f"Файл промпта {filename} не найден в {prompts_dir}")
self.ai_prompts[prompt_name] = ""
else:
logger.error("Каталог промптов не найден, промпты не загружены")
# Сохраняем промпты как атрибут для передачи в запросах # Сохраняем промпты как атрибут для передачи в запросах
self.prompts = self.ai_prompts self.prompts = self.ai_prompts
# ----- 3. HTTP-клиент для RAG-сервера ----- # ----- 3. HTTP-клиент для RAG-сервера -----
# Адрес сервера берём из конфига (bot.conf) или используем значение по умолчанию # Адрес сервера и API-ключ берём из конфига
rag_server_url = getattr(config, 'rag_server_url', 'http://localhost:8080') rag_server_url = config.rag_server_url # теперь это свойство
self.rag_client = RAGClient(server_url=rag_server_url) rag_api_key = config.rag_api_key # получаем API-ключ
# Создаём RAGClient с передачей ключа
self.rag_client = RAGClient(
server_url=rag_server_url,
api_key=rag_api_key, # <-- передаём ключ
timeout=config.http_download_timeout, # можно использовать таймаут из конфига
max_retries=3,
retry_delay=1.0
)
# ----- 4. Команды и состояния ----- # ----- 4. Команды и состояния -----
# Регистрируем все команды бота (из core/commands/) # Регистрируем все команды бота (из core/commands/)

View File

@@ -1,30 +1,25 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
""" """
Пакет rag RAG-ядро платформы Эфцекабот. Пакет rag RAG-ядро платформы Эфцекабот.
Содержит:
- RAG-сервер и оркестратор,
- сервисы (PostgreSQL, Qdrant, GigaChat),
- функции (классификация, суммаризация, проверка орфографии и др.),
- утилиты.
Используется как HTTP-сервис.
""" """
__version__ = "3.0.0" __version__ = "3.0.0"
__author__ = "Андрей Марков, Вениамин Кокорин" __author__ = "Андрей Марков, Вениамин Кокорин"
__license__ = "ISC" __license__ = "ISC"
# Импорт основных классов для удобства
from .rag_orchestrator import RAGOrchestrator from .rag_orchestrator import RAGOrchestrator
from .history_manager import HistoryManager from .history_manager import HistoryManager
from .intent_router import IntentRouter from .intent_router import IntentRouter
from .query_processor import QueryProcessor from .query_processor import QueryProcessor
from .indexing_manager import IndexingManager from .indexing_manager import IndexingManager
from .auth import verify_api_key, set_auth_config
# Экспортируем их для использования в других модулях
__all__ = [ __all__ = [
"RAGOrchestrator", "RAGOrchestrator",
"HistoryManager", "HistoryManager",
"IntentRouter", "IntentRouter",
"QueryProcessor", "QueryProcessor",
"IndexingManager", "IndexingManager",
"verify_api_key",
"set_auth_config",
] ]

52
rag/auth.py Normal file
View File

@@ -0,0 +1,52 @@
# -*- coding: utf-8 -*-
"""
Аутентификация для RAG-сервера.
Проверяет API-ключ в заголовке X-API-Key.
"""
import logging
from fastapi import HTTPException, Security
from fastapi.security import APIKeyHeader
from starlette.status import HTTP_403_FORBIDDEN
from .config_models import AppConfig
logger = logging.getLogger(__name__)
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
# Глобальная ссылка на конфиг (устанавливается из rag_server)
_config: AppConfig = None
def set_auth_config(config: AppConfig):
"""Устанавливает конфиг для аутентификации."""
global _config
_config = config
def verify_api_key(api_key: str = Security(api_key_header)) -> str:
"""
Проверяет API-ключ. Возвращает ключ при успехе, иначе 403.
"""
if _config is None:
raise RuntimeError("Auth config not set")
if not _config.rag_api_key:
# Если ключ не задан пропускаем (но предупреждение уже выведено)
logger.warning("API-ключ не настроен, аутентификация отключена!")
return "insecure"
if not api_key:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN,
detail="API-ключ отсутствует. Передайте его в заголовке X-API-Key."
)
if api_key != _config.rag_api_key:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN,
detail="Неверный API-ключ."
)
return api_key

View File

@@ -174,6 +174,11 @@ class WebScraperConfig(BaseModel):
max_depth: int = Field(1, description="Максимальная глубина рекурсии") max_depth: int = Field(1, description="Максимальная глубина рекурсии")
class AuthConfig(BaseModel):
"""Настройки аутентификации."""
api_key_env: str = Field("RAG_API_KEY", description="Имя переменной окружения с API-ключом для доступа к RAG-серверу")
# ============================================================ # ============================================================
# Главная модель конфигурации приложения # Главная модель конфигурации приложения
# ============================================================ # ============================================================
@@ -213,6 +218,7 @@ class AppConfig(BaseModel):
ai: AIConfig ai: AIConfig
prompts: PromptsConfig prompts: PromptsConfig
web_scraper: WebScraperConfig = Field(default_factory=WebScraperConfig) web_scraper: WebScraperConfig = Field(default_factory=WebScraperConfig)
auth: AuthConfig = Field(default_factory=AuthConfig)
# Дополнительные поля для путей (заполняются при загрузке) # Дополнительные поля для путей (заполняются при загрузке)
profile_dir: Optional[Path] = Field(None, description="Корень профиля") profile_dir: Optional[Path] = Field(None, description="Корень профиля")
@@ -226,11 +232,17 @@ class AppConfig(BaseModel):
# Промпты (загружаются из файлов) # Промпты (загружаются из файлов)
prompts_content: Optional[Dict[str, str]] = Field(default_factory=dict, description="Содержимое промптов") prompts_content: Optional[Dict[str, str]] = Field(default_factory=dict, description="Содержимое промптов")
# Приватное поле для API-ключа RAG (не сериализуется)
_rag_api_key: Optional[str] = Field(None, description="API-ключ RAG-сервера", exclude=True)
# ------------------------------------------------------------------ # ------------------------------------------------------------------
# Прокси-свойства для обратной совместимости со старым BotConfig # Прокси-свойства для обратной совместимости со старым BotConfig
# ------------------------------------------------------------------ # ------------------------------------------------------------------
# База данных @property
def rag_api_key(self) -> Optional[str]:
return self._rag_api_key
@property @property
def db_host(self) -> str: def db_host(self) -> str:
return self.database.host return self.database.host
@@ -259,7 +271,6 @@ class AppConfig(BaseModel):
def db_ssl_mode(self) -> Optional[str]: def db_ssl_mode(self) -> Optional[str]:
return self.database.ssl_mode return self.database.ssl_mode
# Qdrant
@property @property
def qdrant_host(self) -> str: def qdrant_host(self) -> str:
return self.qdrant.host return self.qdrant.host
@@ -284,7 +295,6 @@ class AppConfig(BaseModel):
def qdrant_distance(self) -> str: def qdrant_distance(self) -> str:
return self.qdrant.distance return self.qdrant.distance
# GigaChat
@property @property
def gigachat_api_key(self) -> Optional[str]: def gigachat_api_key(self) -> Optional[str]:
return self.gigachat.api_key return self.gigachat.api_key
@@ -305,7 +315,6 @@ class AppConfig(BaseModel):
def gigachat_temperature(self) -> float: def gigachat_temperature(self) -> float:
return self.gigachat.temperature return self.gigachat.temperature
# Embedding
@property @property
def embedding_cache_size(self) -> int: def embedding_cache_size(self) -> int:
return self.embedding.cache_size return self.embedding.cache_size
@@ -322,7 +331,6 @@ class AppConfig(BaseModel):
def embedding_verify_ssl(self) -> bool: def embedding_verify_ssl(self) -> bool:
return self.embedding.verify_ssl_certs return self.embedding.verify_ssl_certs
# Chunking
@property @property
def chunking_enabled(self) -> bool: def chunking_enabled(self) -> bool:
return self.chunking.enabled return self.chunking.enabled
@@ -347,7 +355,6 @@ class AppConfig(BaseModel):
def chunking_approx_overlap_chars(self) -> int: def chunking_approx_overlap_chars(self) -> int:
return self.chunking.approx_overlap_chars return self.chunking.approx_overlap_chars
# RAG
@property @property
def rag_default_top_k(self) -> int: def rag_default_top_k(self) -> int:
return self.rag.default_top_k return self.rag.default_top_k
@@ -380,7 +387,6 @@ class AppConfig(BaseModel):
def rerank_min_length(self) -> int: def rerank_min_length(self) -> int:
return self.rag.rerank_min_length return self.rag.rerank_min_length
# HTTP
@property @property
def http_download_timeout(self) -> int: def http_download_timeout(self) -> int:
return self.http.download_timeout return self.http.download_timeout
@@ -389,7 +395,6 @@ class AppConfig(BaseModel):
def http_upload_timeout(self) -> int: def http_upload_timeout(self) -> int:
return self.http.upload_timeout return self.http.upload_timeout
# Features
@property @property
def file_processing(self) -> bool: def file_processing(self) -> bool:
return self.features.file_processing return self.features.file_processing
@@ -442,7 +447,6 @@ class AppConfig(BaseModel):
def mention_keyword(self) -> Optional[str]: def mention_keyword(self) -> Optional[str]:
return self.features.mention_keyword return self.features.mention_keyword
# AI
@property @property
def ai_model(self) -> str: def ai_model(self) -> str:
return self.ai.model or self.gigachat.model_generation return self.ai.model or self.gigachat.model_generation
@@ -455,7 +459,6 @@ class AppConfig(BaseModel):
def ai_timeout(self) -> int: def ai_timeout(self) -> int:
return self.ai.timeout return self.ai.timeout
# Cleanup
@property @property
def temp_cleanup_days(self) -> int: def temp_cleanup_days(self) -> int:
return self.cleanup.temp_days return self.cleanup.temp_days
@@ -472,7 +475,6 @@ class AppConfig(BaseModel):
def user_data_cleanup_days(self) -> int: def user_data_cleanup_days(self) -> int:
return self.cleanup.user_data_days return self.cleanup.user_data_days
# Web scraper
@property @property
def ws_max_pages(self) -> int: def ws_max_pages(self) -> int:
return self.web_scraper.max_pages return self.web_scraper.max_pages
@@ -481,7 +483,6 @@ class AppConfig(BaseModel):
def ws_max_depth(self) -> int: def ws_max_depth(self) -> int:
return self.web_scraper.max_depth return self.web_scraper.max_depth
# Prompts temperatures
@property @property
def intent_temperature(self) -> float: def intent_temperature(self) -> float:
return self.prompts.intent.temperature return self.prompts.intent.temperature
@@ -522,7 +523,6 @@ class AppConfig(BaseModel):
def generate_document_temperature(self) -> float: def generate_document_temperature(self) -> float:
return self.prompts.generate_document.temperature return self.prompts.generate_document.temperature
# File permissions
@property @property
def file_mode(self) -> int: def file_mode(self) -> int:
return int(self.file_permissions.mode, 8) return int(self.file_permissions.mode, 8)

View File

@@ -5,16 +5,10 @@
знать внутреннее устройство оркестратора. знать внутреннее устройство оркестратора.
Этот класс является прослойкой (facade) между ботами и RAGOrchestrator. Этот класс является прослойкой (facade) между ботами и RAGOrchestrator.
В будущем он может быть легко заменён на HTTP-клиент,
если оркестратор будет вынесен в отдельный сервис.
Сейчас этот файл может использоваться для локальной работы,
но в новой архитектуре мы используем RAGClient (HTTP),
поэтому данный класс является опциональным.
""" """
import logging import logging
from typing import Optional, Dict, List, Any from typing import Optional, Dict, Any
from .rag_orchestrator import RAGOrchestrator from .rag_orchestrator import RAGOrchestrator
from .services.kb_service import KBService from .services.kb_service import KBService
@@ -30,7 +24,6 @@ class RAGAPI:
""" """
Единый API-интерфейс для RAG-функциональности. Единый API-интерфейс для RAG-функциональности.
Создаёт и содержит экземпляр RAGOrchestrator. Создаёт и содержит экземпляр RAGOrchestrator.
Предоставляет методы для выполнения запросов и индексации документов.
""" """
def __init__( def __init__(
@@ -44,24 +37,9 @@ class RAGAPI:
config: AppConfig, config: AppConfig,
default_prompts: Optional[Dict[str, str]] = None default_prompts: Optional[Dict[str, str]] = None
): ):
"""
Инициализация API.
Аргументы:
db: сервис PostgreSQL (PostgresService)
qdrant: сервис Qdrant (QdrantService)
embedding: сервис эмбеддингов (EmbeddingService)
kb: сервис базы знаний (KBService)
giga: клиент GigaChat (GigaClient)
files: сервис файлов (FileService)
config: объект конфигурации (AppConfig)
default_prompts: словарь промптов по умолчанию.
Используется, если в запросе не переданы свои промпты.
"""
self.config = config self.config = config
self.default_prompts = default_prompts or {} self.default_prompts = default_prompts or {}
# Создаём оркестратор с переданными сервисами
self.orchestrator = RAGOrchestrator( self.orchestrator = RAGOrchestrator(
db=db, db=db,
qdrant=qdrant, qdrant=qdrant,
@@ -84,10 +62,6 @@ class RAGAPI:
last_file_path: Optional[str] = None, last_file_path: Optional[str] = None,
last_file_text: Optional[str] = None, last_file_text: Optional[str] = None,
) -> Dict[str, Any]: ) -> Dict[str, Any]:
"""
Выполняет RAG-запрос через оркестратор.
ИСТОРИЯ НЕ ПЕРЕДАЁТСЯ оркестратор получает её из БД.
"""
try: try:
if prompts is None: if prompts is None:
prompts = self.default_prompts.copy() prompts = self.default_prompts.copy()
@@ -126,7 +100,6 @@ class RAGAPI:
file_hash: Optional[str] = None, file_hash: Optional[str] = None,
update_if_exists: bool = True update_if_exists: bool = True
) -> Dict[str, Any]: ) -> Dict[str, Any]:
"""Индексирует документ через оркестратор."""
try: try:
result = await self.orchestrator.index_document( result = await self.orchestrator.index_document(
file_name=file_name, file_name=file_name,
@@ -146,5 +119,4 @@ class RAGAPI:
return {"doc_id": None, "chunk_count": 0, "error": str(e)} return {"doc_id": None, "chunk_count": 0, "error": str(e)}
def close(self): def close(self):
"""Закрывает ресурсы (если необходимо)."""
logger.debug("RAGAPI закрывается") logger.debug("RAGAPI закрывается")

View File

@@ -2,7 +2,6 @@
""" """
Главный оркестратор RAG-пайплайна (фасад). Главный оркестратор RAG-пайплайна (фасад).
Координирует работу менеджеров: HistoryManager, IntentRouter, QueryProcessor, IndexingManager. Координирует работу менеджеров: HistoryManager, IntentRouter, QueryProcessor, IndexingManager.
Принимает запрос пользователя, обрабатывает его и возвращает ответ.
""" """
import logging import logging
@@ -29,7 +28,6 @@ class RAGOrchestrator:
""" """
Оркестратор RAG-пайплайна. Оркестратор RAG-пайплайна.
Содержит ссылки на все сервисы и менеджеры. Содержит ссылки на все сервисы и менеджеры.
Предоставляет два основных метода: process_query и index_document.
""" """
def __init__( def __init__(
@@ -43,19 +41,6 @@ class RAGOrchestrator:
config: AppConfig, config: AppConfig,
default_prompts: Optional[Dict[str, str]] = None default_prompts: Optional[Dict[str, str]] = None
): ):
"""
Инициализация оркестратора.
Аргументы:
db: сервис PostgreSQL
qdrant: сервис Qdrant
embedding: сервис эмбеддингов
kb: сервис базы знаний
giga: клиент GigaChat
files: сервис файлов
config: объект конфигурации (AppConfig)
default_prompts: словарь промптов по умолчанию
"""
self.db = db self.db = db
self.qdrant = qdrant self.qdrant = qdrant
self.embedding = embedding self.embedding = embedding
@@ -97,10 +82,6 @@ class RAGOrchestrator:
logger.info("RAGOrchestrator инициализирован с менеджерами") logger.info("RAGOrchestrator инициализирован с менеджерами")
# ------------------------------------------------------------------
# Вспомогательный метод для расчёта токенов
# ------------------------------------------------------------------
def _prepare_prompt_parts( def _prepare_prompt_parts(
self, self,
synthesis_template: str, synthesis_template: str,
@@ -110,17 +91,6 @@ class RAGOrchestrator:
reserved_for_answer: int = 1000, reserved_for_answer: int = 1000,
reserved_for_overhead: int = 200 reserved_for_overhead: int = 200
) -> Dict[str, Any]: ) -> Dict[str, Any]:
"""
Подсчитывает токены в статичных частях промта и вычисляет,
сколько токенов остаётся для истории и контекста.
Возвращает словарь с ключами:
- available_for_history_and_context: int
- prompt_tokens: int
- system_tokens: int
- synthesis_tokens: int
- query_tokens: int
"""
system_tokens = count_tokens(system_prompt) if system_prompt else 0 system_tokens = count_tokens(system_prompt) if system_prompt else 0
synthesis_tokens = count_tokens(synthesis_template) synthesis_tokens = count_tokens(synthesis_template)
query_tokens = count_tokens(query) query_tokens = count_tokens(query)
@@ -142,10 +112,6 @@ class RAGOrchestrator:
"query_tokens": query_tokens, "query_tokens": query_tokens,
} }
# ------------------------------------------------------------------
# Основной метод обработки запроса
# ------------------------------------------------------------------
async def process_query( async def process_query(
self, self,
query: str, query: str,
@@ -156,25 +122,6 @@ class RAGOrchestrator:
last_file_path: Optional[str] = None, last_file_path: Optional[str] = None,
last_file_text: Optional[str] = None, last_file_text: Optional[str] = None,
) -> Dict[str, Any]: ) -> Dict[str, Any]:
"""
Основной метод обработки запроса.
ИСТОРИЯ ДИАЛОГА ПОЛУЧАЕТСЯ ИЗ БД, а не из запроса.
Это обеспечивает единый контекст для всех клиентов.
Аргументы:
query: текст запроса пользователя
user_jid: JID пользователя (без ресурса)
room_jid: JID комнаты (None для личного чата)
prompts: словарь промптов (если None, используются default_prompts)
intent_override: принудительное переопределение намерения
last_file_path: путь к последнему загруженному файлу
last_file_text: текст последнего загруженного файла
Возвращает:
Словарь с ключами: answer, intent, context, sources, confidence, error
"""
# 1. Подготовка промптов
if prompts is None: if prompts is None:
prompts = self.default_prompts.copy() prompts = self.default_prompts.copy()
synthesis_template = prompts.get('synthesis', '') synthesis_template = prompts.get('synthesis', '')
@@ -182,7 +129,6 @@ class RAGOrchestrator:
synthesis_template = "{context}\n\n{query}\n\nОтвет:" synthesis_template = "{context}\n\n{query}\n\nОтвет:"
system_prompt = prompts.get('system', None) system_prompt = prompts.get('system', None)
# 2. Расчёт лимитов токенов
max_model_tokens = self.config.max_model_tokens max_model_tokens = self.config.max_model_tokens
reserved_for_answer = self.config.reserved_for_answer_tokens reserved_for_answer = self.config.reserved_for_answer_tokens
reserved_for_overhead = self.config.reserved_for_overhead_tokens reserved_for_overhead = self.config.reserved_for_overhead_tokens
@@ -198,10 +144,8 @@ class RAGOrchestrator:
available_for_history_and_context = token_info["available_for_history_and_context"] available_for_history_and_context = token_info["available_for_history_and_context"]
logger.debug(f"Доступно для истории и контекста: {available_for_history_and_context} токенов") logger.debug(f"Доступно для истории и контекста: {available_for_history_and_context} токенов")
# 3. Получение истории из БД
raw_history = await self.history_manager.get_history(user_jid, room_jid, limit=100) raw_history = await self.history_manager.get_history(user_jid, room_jid, limit=100)
# 4. Сжатие истории, если она слишком длинная
max_history_tokens = min(available_for_history_and_context // 2, 2000) max_history_tokens = min(available_for_history_and_context // 2, 2000)
formatted_history = await self.history_manager.compress_history_if_needed( formatted_history = await self.history_manager.compress_history_if_needed(
raw_history, raw_history,
@@ -209,7 +153,6 @@ class RAGOrchestrator:
prompt_template=prompts.get('hierarchical_summary', '') prompt_template=prompts.get('hierarchical_summary', '')
) )
# 5. Классификация намерений
intent = intent_override intent = intent_override
if intent is None and self.config.enable_intent_classification: if intent is None and self.config.enable_intent_classification:
intent_prompt = prompts.get('intent', '') intent_prompt = prompts.get('intent', '')
@@ -225,13 +168,11 @@ class RAGOrchestrator:
else: else:
intent = intent or "GENERAL" intent = intent or "GENERAL"
# 6. Принудительная установка SURGICAL по ключевым словам
keywords = self.config.surgical_keywords keywords = self.config.surgical_keywords
if any(kw in query.lower() for kw in keywords) and last_file_path: if any(kw in query.lower() for kw in keywords) and last_file_path:
intent = "SURGICAL" intent = "SURGICAL"
logger.info(f"Принудительный Intent: SURGICAL (есть файл и ключевые слова)") logger.info(f"Принудительный Intent: SURGICAL (есть файл и ключевые слова)")
# 7. Маршрутизация специализированных намерений
router_result = await self.intent_router.route( router_result = await self.intent_router.route(
intent=intent, intent=intent,
query=query, query=query,
@@ -249,12 +190,10 @@ class RAGOrchestrator:
sources = [] sources = []
if router_result is not None: if router_result is not None:
# Намерение обработано маршрутизатором
answer = router_result.get("answer") answer = router_result.get("answer")
context = router_result.get("context", "") context = router_result.get("context", "")
sources = router_result.get("sources", []) sources = router_result.get("sources", [])
else: else:
# Обычный RAG-пайплайн (GENERAL, FACT, PROCEDURE, COMPARISON, CALCULATION)
history_tokens = sum(count_tokens(msg['content']) for msg in formatted_history) history_tokens = sum(count_tokens(msg['content']) for msg in formatted_history)
available_for_context = available_for_history_and_context - history_tokens available_for_context = available_for_history_and_context - history_tokens
available_for_context = max(available_for_context, 0) available_for_context = max(available_for_context, 0)
@@ -273,12 +212,10 @@ class RAGOrchestrator:
context = processor_result.get("context", "") context = processor_result.get("context", "")
sources = processor_result.get("sources", []) sources = processor_result.get("sources", [])
# 8. Сохранение истории диалога в БД
await self.history_manager.save_message(user_jid, "user", query, room_jid) await self.history_manager.save_message(user_jid, "user", query, room_jid)
if answer: if answer:
await self.history_manager.save_message(user_jid, "assistant", answer, room_jid) await self.history_manager.save_message(user_jid, "assistant", answer, room_jid)
# 9. Формирование результата
return { return {
"answer": answer or "⚠️ Не удалось сгенерировать ответ.", "answer": answer or "⚠️ Не удалось сгенерировать ответ.",
"intent": intent, "intent": intent,
@@ -288,10 +225,6 @@ class RAGOrchestrator:
"error": None "error": None
} }
# ------------------------------------------------------------------
# Индексация документа
# ------------------------------------------------------------------
async def index_document( async def index_document(
self, self,
file_name: str, file_name: str,
@@ -304,24 +237,6 @@ class RAGOrchestrator:
file_hash: Optional[str] = None, file_hash: Optional[str] = None,
update_if_exists: bool = True update_if_exists: bool = True
) -> Dict[str, Any]: ) -> Dict[str, Any]:
"""
Индексирует документ в базу знаний.
Делегирует работу IndexingManager.
Аргументы:
file_name: исходное имя файла
file_text: извлечённый текст
user_jid: JID владельца
room_jid: JID комнаты (None для личного)
is_global: глобальный ли документ
title: отображаемое название
metadata: дополнительные метаданные
file_hash: SHA-256 хеш содержимого
update_if_exists: заменять ли существующий документ в комнате
Возвращает:
Словарь с ключами doc_id, chunk_count, error
"""
try: try:
doc_id, chunk_count = await self.indexing_manager.index_document( doc_id, chunk_count = await self.indexing_manager.index_document(
file_name=file_name, file_name=file_name,

View File

@@ -9,6 +9,7 @@ RAG-сервер отдельный HTTP-сервис для обработ
- Эндпоинт /rag/vision для распознавания текста на изображениях (OCR) - Эндпоинт /rag/vision для распознавания текста на изображениях (OCR)
- Эндпоинт /rag/transcribe для транскрибации аудио (SaluteSpeech) - Эндпоинт /rag/transcribe для транскрибации аудио (SaluteSpeech)
- Использование Pydantic-модели AppConfig для конфигурации. - Использование Pydantic-модели AppConfig для конфигурации.
- Аутентификация через API-ключ (X-API-Key)
""" """
import logging import logging
@@ -38,6 +39,7 @@ from .services.kb_service import KBService
from .services.giga_client import GigaClient from .services.giga_client import GigaClient
from .services.file_service import FileService from .services.file_service import FileService
from .rag_orchestrator import RAGOrchestrator from .rag_orchestrator import RAGOrchestrator
from .auth import verify_api_key, set_auth_config
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -253,11 +255,11 @@ async def shutdown_event():
# ============================================================ # ============================================================
# ЭНДПОИНТЫ API # ЭНДПОИНТЫ API (все защищены аутентификацией, кроме /health)
# ============================================================ # ============================================================
@app.post("/rag/query", response_model=QueryResponse) @app.post("/rag/query", response_model=QueryResponse)
async def rag_query(request: QueryRequest): async def rag_query(request: QueryRequest, _: str = Depends(verify_api_key)):
""" """
Выполняет RAG-запрос. Выполняет RAG-запрос.
ИСТОРИЯ БЕРЁТСЯ СЕРВЕРОМ ИЗ БД (не передаётся в запросе). ИСТОРИЯ БЕРЁТСЯ СЕРВЕРОМ ИЗ БД (не передаётся в запросе).
@@ -288,7 +290,7 @@ async def rag_query(request: QueryRequest):
@app.post("/rag/index", response_model=IndexResponse) @app.post("/rag/index", response_model=IndexResponse)
async def index_document(request: IndexRequest): async def index_document(request: IndexRequest, _: str = Depends(verify_api_key)):
"""Индексирует документ в базу знаний.""" """Индексирует документ в базу знаний."""
orchestrator = get_orchestrator() orchestrator = get_orchestrator()
@@ -316,21 +318,16 @@ async def index_document(request: IndexRequest):
@app.get("/health") @app.get("/health")
async def health_check(): async def health_check():
"""Проверка здоровья сервиса.""" """Проверка здоровья сервиса (без аутентификации)."""
if _orchestrator is None: if _orchestrator is None:
return {"status": "unhealthy", "message": "Orchestrator not initialized"} return {"status": "unhealthy", "message": "Orchestrator not initialized"}
return {"status": "healthy"} return {"status": "healthy"}
# ============================================================
# НОВЫЕ ЭНДПОИНТЫ: OCR и транскрибация
# ============================================================
@app.post("/rag/vision", response_model=VisionResponse) @app.post("/rag/vision", response_model=VisionResponse)
async def vision_endpoint( async def vision_endpoint(
file: UploadFile = FastAPIFile(...), file: UploadFile = FastAPIFile(...),
user_jid: str = None, _: str = Depends(verify_api_key)
room_jid: str = None
): ):
""" """
Распознаёт текст на изображении через OCR (Tesseract). Распознаёт текст на изображении через OCR (Tesseract).
@@ -370,8 +367,7 @@ async def vision_endpoint(
@app.post("/rag/transcribe", response_model=TranscribeResponse) @app.post("/rag/transcribe", response_model=TranscribeResponse)
async def transcribe_endpoint( async def transcribe_endpoint(
file: UploadFile = FastAPIFile(...), file: UploadFile = FastAPIFile(...),
user_jid: str = None, _: str = Depends(verify_api_key)
room_jid: str = None
): ):
""" """
Транскрибирует аудиофайл через SaluteSpeech. Транскрибирует аудиофайл через SaluteSpeech.
@@ -450,6 +446,9 @@ def main():
logger.info(f"Запуск RAG-сервера с профилем {args.profile_dir}") logger.info(f"Запуск RAG-сервера с профилем {args.profile_dir}")
logger.info(f"Хост: {args.host}, порт: {args.port}") logger.info(f"Хост: {args.host}, порт: {args.port}")
# Устанавливаем конфиг для аутентификации
set_auth_config(config)
# Инициализируем оркестратор # Инициализируем оркестратор
try: try:
init_orchestrator(config) init_orchestrator(config)

View File

@@ -101,12 +101,22 @@ def load_config(profile_dir: str) -> AppConfig:
if 'web_scraper' not in merged: if 'web_scraper' not in merged:
merged['web_scraper'] = {'max_pages': 5, 'max_depth': 1} merged['web_scraper'] = {'max_pages': 5, 'max_depth': 1}
# Добавляем секцию auth, если её нет
if 'auth' not in merged:
merged['auth'] = {}
# Создаём экземпляр AppConfig # Создаём экземпляр AppConfig
try: try:
config = AppConfig(**merged) config = AppConfig(**merged)
except Exception as e: except Exception as e:
raise ValueError(f"Ошибка валидации конфигурации: {e}") raise ValueError(f"Ошибка валидации конфигурации: {e}")
# Загружаем API-ключ для RAG-сервера
api_key_env = merged.get('auth', {}).get('api_key_env', 'RAG_API_KEY')
config._rag_api_key = os.getenv(api_key_env)
if not config._rag_api_key:
logger.warning(f"Переменная окружения {api_key_env} не задана. Аутентификация на RAG-сервере отключена (НЕ БЕЗОПАСНО!)")
# Загружаем промпты из файлов # Загружаем промпты из файлов
prompts_dir = config.prompts_dir_path prompts_dir = config.prompts_dir_path
if prompts_dir and prompts_dir.exists(): if prompts_dir and prompts_dir.exists():