270 lines
10 KiB
Python
270 lines
10 KiB
Python
# ============================================================
|
||
# rag/rag_orchestrator.py
|
||
# Полная версия с поддержкой ToolHandler
|
||
# ============================================================
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
Главный оркестратор RAG-пайплайна (фасад).
|
||
Координирует работу менеджеров: HistoryManager, IntentRouter, QueryProcessor, IndexingManager.
|
||
Добавлена поддержка ToolHandler для ReAct.
|
||
"""
|
||
|
||
import logging
|
||
from typing import Optional, Dict, List, Any
|
||
|
||
from .services.postgres_service import PostgresService
|
||
from .services.qdrant_service import QdrantService
|
||
from .services.embedding_service import EmbeddingService
|
||
from .services.kb_service import KBService
|
||
from .services.giga_client import GigaClient
|
||
from .services.file_service import FileService
|
||
from .history_manager import HistoryManager
|
||
from .intent_router import IntentRouter
|
||
from .query_processor import QueryProcessor
|
||
from .indexing_manager import IndexingManager
|
||
from .functions.intent_classify import classify_intent
|
||
from .utils.text_utils import count_tokens
|
||
from .config_models import AppConfig
|
||
from .functions.tools import ToolHandler
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class RAGOrchestrator:
|
||
"""
|
||
Оркестратор RAG-пайплайна.
|
||
Содержит ссылки на все сервисы и менеджеры.
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
db: PostgresService,
|
||
qdrant: QdrantService,
|
||
embedding: EmbeddingService,
|
||
kb: KBService,
|
||
giga: GigaClient,
|
||
files: FileService,
|
||
config: AppConfig,
|
||
default_prompts: Optional[Dict[str, str]] = None,
|
||
tool_handler: Optional[ToolHandler] = None,
|
||
):
|
||
self.db = db
|
||
self.qdrant = qdrant
|
||
self.embedding = embedding
|
||
self.kb = kb
|
||
self.giga = giga
|
||
self.files = files
|
||
self.config = config
|
||
self.default_prompts = default_prompts or {}
|
||
self.tool_handler = tool_handler
|
||
|
||
# Инициализация менеджеров
|
||
self.history_manager = HistoryManager(
|
||
db=db,
|
||
giga=giga,
|
||
config=config,
|
||
default_prompts=self.default_prompts
|
||
)
|
||
|
||
self.intent_router = IntentRouter(
|
||
giga=giga,
|
||
kb=kb,
|
||
files=files,
|
||
config=config,
|
||
default_prompts=self.default_prompts
|
||
)
|
||
|
||
self.query_processor = QueryProcessor(
|
||
giga=giga,
|
||
kb=kb,
|
||
config=config,
|
||
default_prompts=self.default_prompts,
|
||
agent_coordinator=None, # будет установлен позже, если включены агенты
|
||
tool_handler=tool_handler
|
||
)
|
||
|
||
self.indexing_manager = IndexingManager(
|
||
kb=kb,
|
||
giga=giga,
|
||
config=config,
|
||
default_prompts=self.default_prompts
|
||
)
|
||
|
||
logger.info("RAGOrchestrator инициализирован с менеджерами")
|
||
|
||
def set_agent_coordinator(self, agent_coordinator):
|
||
"""Устанавливает координатор агентов в QueryProcessor."""
|
||
self.query_processor.agent_coordinator = agent_coordinator
|
||
logger.info("AgentCoordinator передан в QueryProcessor")
|
||
|
||
def _prepare_prompt_parts(
|
||
self,
|
||
synthesis_template: str,
|
||
system_prompt: Optional[str],
|
||
query: str,
|
||
max_total_tokens: int = 8192,
|
||
reserved_for_answer: int = 1000,
|
||
reserved_for_overhead: int = 200
|
||
) -> Dict[str, Any]:
|
||
system_tokens = count_tokens(system_prompt) if system_prompt else 0
|
||
synthesis_tokens = count_tokens(synthesis_template)
|
||
query_tokens = count_tokens(query)
|
||
prompt_tokens = system_tokens + synthesis_tokens + query_tokens
|
||
|
||
available = max_total_tokens - prompt_tokens - reserved_for_answer - reserved_for_overhead
|
||
if available < 0:
|
||
logger.warning(
|
||
f"Недостаточно токенов для истории и контекста: {available}. "
|
||
f"Увеличьте max_total_tokens или уменьшите размер промптов."
|
||
)
|
||
available = max(available, 100)
|
||
|
||
return {
|
||
"available_for_history_and_context": available,
|
||
"prompt_tokens": prompt_tokens,
|
||
"system_tokens": system_tokens,
|
||
"synthesis_tokens": synthesis_tokens,
|
||
"query_tokens": query_tokens,
|
||
}
|
||
|
||
async def process_query(
|
||
self,
|
||
query: str,
|
||
user_jid: str,
|
||
room_jid: Optional[str],
|
||
prompts: Optional[Dict[str, str]] = None,
|
||
intent_override: Optional[str] = None,
|
||
last_file_path: Optional[str] = None,
|
||
last_file_text: Optional[str] = None,
|
||
) -> Dict[str, Any]:
|
||
if prompts is None:
|
||
prompts = self.default_prompts.copy()
|
||
synthesis_template = prompts.get('synthesis', '')
|
||
if not synthesis_template:
|
||
synthesis_template = "{context}\n\n{query}\n\nОтвет:"
|
||
system_prompt = prompts.get('system', None)
|
||
|
||
max_model_tokens = self.config.max_model_tokens
|
||
reserved_for_answer = self.config.reserved_for_answer_tokens
|
||
reserved_for_overhead = self.config.reserved_for_overhead_tokens
|
||
|
||
token_info = self._prepare_prompt_parts(
|
||
synthesis_template=synthesis_template,
|
||
system_prompt=system_prompt,
|
||
query=query,
|
||
max_total_tokens=max_model_tokens,
|
||
reserved_for_answer=reserved_for_answer,
|
||
reserved_for_overhead=reserved_for_overhead
|
||
)
|
||
available_for_history_and_context = token_info["available_for_history_and_context"]
|
||
logger.debug(f"Доступно для истории и контекста: {available_for_history_and_context} токенов")
|
||
|
||
raw_history = await self.history_manager.get_history(user_jid, room_jid, limit=100)
|
||
|
||
max_history_tokens = min(available_for_history_and_context // 2, 2000)
|
||
formatted_history = await self.history_manager.compress_history_if_needed(
|
||
raw_history,
|
||
max_tokens=max_history_tokens,
|
||
prompt_template=prompts.get('hierarchical_summary', '')
|
||
)
|
||
|
||
intent = intent_override
|
||
if intent is None and self.config.enable_intent_classification:
|
||
intent_prompt = prompts.get('intent', '')
|
||
if intent_prompt:
|
||
intent = await classify_intent(
|
||
giga=self.giga,
|
||
query=query,
|
||
prompt_text=intent_prompt,
|
||
bot_config=self.config
|
||
)
|
||
else:
|
||
intent = "GENERAL"
|
||
else:
|
||
intent = intent or "GENERAL"
|
||
|
||
keywords = self.config.surgical_keywords
|
||
if any(kw in query.lower() for kw in keywords) and last_file_path:
|
||
intent = "SURGICAL"
|
||
logger.info(f"Принудительный Intent: SURGICAL (есть файл и ключевые слова)")
|
||
|
||
router_result = await self.intent_router.route(
|
||
intent=intent,
|
||
query=query,
|
||
user_jid=user_jid,
|
||
room_jid=room_jid,
|
||
prompts=prompts,
|
||
last_file_path=last_file_path,
|
||
last_file_text=last_file_text,
|
||
history=formatted_history,
|
||
system_prompt=system_prompt
|
||
)
|
||
|
||
answer = None
|
||
context = ""
|
||
sources = []
|
||
|
||
if router_result is not None:
|
||
answer = router_result.get("answer")
|
||
context = router_result.get("context", "")
|
||
sources = router_result.get("sources", [])
|
||
else:
|
||
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 = max(available_for_context, 0)
|
||
|
||
processor_result = await self.query_processor.process(
|
||
query=query,
|
||
user_jid=user_jid,
|
||
room_jid=room_jid,
|
||
prompts=prompts,
|
||
intent=intent,
|
||
history=formatted_history,
|
||
system_prompt=system_prompt,
|
||
available_tokens_for_context=available_for_context
|
||
)
|
||
answer = processor_result.get("answer")
|
||
context = processor_result.get("context", "")
|
||
sources = processor_result.get("sources", [])
|
||
|
||
await self.history_manager.save_message(user_jid, "user", query, room_jid)
|
||
if answer:
|
||
await self.history_manager.save_message(user_jid, "assistant", answer, room_jid)
|
||
|
||
return {
|
||
"answer": answer or "⚠️ Не удалось сгенерировать ответ.",
|
||
"intent": intent,
|
||
"context": context,
|
||
"sources": sources,
|
||
"confidence": None,
|
||
"error": None
|
||
}
|
||
|
||
async def index_document(
|
||
self,
|
||
file_name: str,
|
||
file_text: str,
|
||
user_jid: str,
|
||
room_jid: Optional[str],
|
||
is_global: bool = False,
|
||
title: Optional[str] = None,
|
||
metadata: Optional[Dict] = None,
|
||
file_hash: Optional[str] = None,
|
||
update_if_exists: bool = True
|
||
) -> Dict[str, Any]:
|
||
try:
|
||
doc_id, chunk_count = await self.indexing_manager.index_document(
|
||
file_name=file_name,
|
||
file_text=file_text,
|
||
user_jid=user_jid,
|
||
room_jid=room_jid,
|
||
is_global=is_global,
|
||
title=title,
|
||
metadata=metadata,
|
||
file_hash=file_hash,
|
||
update_if_exists=update_if_exists
|
||
)
|
||
return {"doc_id": doc_id, "chunk_count": chunk_count, "error": None}
|
||
except Exception as e:
|
||
logger.error(f"Ошибка индексации документа {file_name}: {e}", exc_info=True)
|
||
return {"doc_id": None, "chunk_count": 0, "error": str(e)} |