diff --git a/rag/rag_orchestrator.py b/rag/rag_orchestrator.py index 9e49576..60e8fb1 100644 --- a/rag/rag_orchestrator.py +++ b/rag/rag_orchestrator.py @@ -1,43 +1,24 @@ # -*- coding: utf-8 -*- """ -Главный оркестратор RAG-пайплайна. -Принимает запрос пользователя, получает историю из БД, выполняет все этапы: -классификацию, расширение, поиск, переранжирование, синтез, самокритику. -Сохраняет историю в БД. - -Этот модуль не зависит от XMPP и может использоваться как в ботах, -так и в отдельном RAG-ядре (API-сервисе). - -Улучшено управление токенами: теперь история и контекст обрезаются с учётом -лимитов модели, резервирования для ответа и промптов. - -ДОБАВЛЕНО: -- Иерархическое резюмирование истории диалога. -- Иерархическое резюмирование больших документов при индексации. +Главный оркестратор RAG-пайплайна (фасад). +Координирует работу менеджеров: HistoryManager, IntentRouter, QueryProcessor, IndexingManager. +Принимает запрос пользователя, обрабатывает его и возвращает ответ. """ -import asyncio import logging -import os -import re from typing import Optional, Dict, List, Any -# Импорт сервисов и функций from core.services.postgres_service import PostgresService from core.services.qdrant_service import QdrantService from core.services.embedding_service import EmbeddingService from core.services.kb_service import KBService from core.services.giga_client import GigaClient from core.services.file_service import FileService +from core.history_manager import HistoryManager +from core.intent_router import IntentRouter +from core.query_processor import QueryProcessor +from core.indexing_manager import IndexingManager from core.functions.intent_classify import classify_intent -from core.functions.expand_query import expand_query -from core.functions.extract_metrics import extract_metrics -from core.functions.summarize_document import summarize_document -from core.functions.check_consistency import check_consistency -from core.functions.critique_answer import critique_answer -from core.functions.rerank_context import rerank_context -from core.functions.check_spelling import check_spelling -from core.functions.hierarchical_summarize import hierarchical_summarize from core.utils.text_utils import count_tokens logger = logging.getLogger(__name__) @@ -46,7 +27,8 @@ logger = logging.getLogger(__name__) class RAGOrchestrator: """ Оркестратор RAG-пайплайна. - Содержит ссылки на все сервисы и выполняет полный цикл обработки запроса. + Содержит ссылки на все сервисы и менеджеры. + Предоставляет два основных метода: process_query и index_document. """ def __init__( @@ -64,12 +46,12 @@ class RAGOrchestrator: Инициализация оркестратора. Аргументы: - db: сервис PostgreSQL (для истории и метаданных) - qdrant: сервис Qdrant (векторный поиск) - embedding: сервис эмбеддингов (GigaChat) - kb: сервис базы знаний (индексация, поиск) - giga: клиент GigaChat (генерация) - files: сервис файлов (извлечение текста) + db: сервис PostgreSQL + qdrant: сервис Qdrant + embedding: сервис эмбеддингов + kb: сервис базы знаний + giga: клиент GigaChat + files: сервис файлов config: объект конфигурации (BotConfig) default_prompts: словарь промптов по умолчанию """ @@ -81,10 +63,41 @@ class RAGOrchestrator: self.files = files self.config = config self.default_prompts = default_prompts or {} - logger.info("RAGOrchestrator инициализирован") + + # Инициализация менеджеров + 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 + ) + + self.indexing_manager = IndexingManager( + kb=kb, + giga=giga, + config=config, + default_prompts=self.default_prompts + ) + + logger.info("RAGOrchestrator инициализирован с менеджерами") # ------------------------------------------------------------------ - # Вспомогательный метод для расчёта токенов с резервированием + # Вспомогательный метод для расчёта токенов # ------------------------------------------------------------------ def _prepare_prompt_parts( @@ -92,111 +105,42 @@ class RAGOrchestrator: synthesis_template: str, system_prompt: Optional[str], query: str, - context: str, max_total_tokens: int = 8192, reserved_for_answer: int = 1000, reserved_for_overhead: int = 200 ) -> Dict[str, Any]: """ - Подготавливает части промта и обрезает историю/контекст по токенам. + Подсчитывает токены в статичных частях промта и вычисляет, + сколько токенов остаётся для истории и контекста. Возвращает словарь с ключами: - - history: отфильтрованная история (список сообщений) - - context: отфильтрованный контекст (строка) - - prompt_tokens: количество токенов в промте (без истории и контекста) - - total_used: общее использованное количество токенов - - available_for_history_and_context: сколько токенов выделено для истории и контекста + - available_for_history_and_context: int + - prompt_tokens: int + - system_tokens: int + - synthesis_tokens: int + - query_tokens: int """ - # 1. Подсчёт токенов в статичных частях промта system_tokens = count_tokens(system_prompt) if system_prompt else 0 synthesis_tokens = count_tokens(synthesis_template) query_tokens = count_tokens(query) - - # 2. Резервирование prompt_tokens = system_tokens + synthesis_tokens + query_tokens - available_for_history_and_context = ( - max_total_tokens - - prompt_tokens - - reserved_for_answer - - reserved_for_overhead - ) - if available_for_history_and_context <= 0: + available = max_total_tokens - prompt_tokens - reserved_for_answer - reserved_for_overhead + if available < 0: logger.warning( - f"Недостаточно токенов для истории и контекста: {available_for_history_and_context}. " + f"Недостаточно токенов для истории и контекста: {available}. " f"Увеличьте max_total_tokens или уменьшите размер промптов." ) - available_for_history_and_context = max(available_for_history_and_context, 100) + available = max(available, 100) return { - "available_for_history_and_context": available_for_history_and_context, + "available_for_history_and_context": available, "prompt_tokens": prompt_tokens, "system_tokens": system_tokens, "synthesis_tokens": synthesis_tokens, "query_tokens": query_tokens, } - # ------------------------------------------------------------------ - # Метод для сжатия больших документов при индексации - # ------------------------------------------------------------------ - - async def _compress_document_if_needed(self, text: str, file_name: str) -> str: - """ - Если текст превышает порог (max_tokens_for_document), сжимает его - с помощью иерархического резюмирования. - Возвращает либо исходный текст, либо сжатый. - """ - # Читаем настройки из конфига - summarization_config = getattr(self.config, 'summarization', {}) - enabled = summarization_config.get('enable_hierarchical_summarization', True) - if not enabled: - return text - - max_tokens = summarization_config.get('max_tokens_for_document', 8000) - target_tokens = summarization_config.get('target_tokens_after_summary', 3000) - chunk_size = summarization_config.get('chunk_size_tokens', 500) - max_depth = summarization_config.get('max_depth', 2) - - # Подсчитываем токены - text_tokens = count_tokens(text) - if text_tokens <= max_tokens: - logger.debug(f"Документ '{file_name}' ({text_tokens} токенов) не требует сжатия") - return text - - logger.info(f"Документ '{file_name}' ({text_tokens} токенов) превышает лимит {max_tokens}, применяем иерархическое резюмирование") - - # Загружаем промпт для резюмирования - summary_prompt = self.default_prompts.get('hierarchical_summary', '') - if not summary_prompt: - try: - prompt_path = self.config.prompts_dir / 'hierarchical_summary.txt' - with open(prompt_path, 'r', encoding='utf-8') as f: - summary_prompt = f.read() - except Exception as e: - logger.warning(f"Не удалось загрузить промпт hierarchical_summary: {e}, используем стандартный") - summary_prompt = ( - "Ты — профессиональный реферант. Кратко изложи суть текста, сохранив ключевые факты, цифры и логические связи.\n" - "Объём: не более 30% от исходного, но не менее 2 предложений.\n" - "Стиль: деловой, нейтральный, без оценок.\n\n" - "ТЕКСТ:\n{text}\n\nРЕЗЮМЕ:" - ) - - try: - compressed = await hierarchical_summarize( - text=text, - giga=self.giga, - prompt_template=summary_prompt, - target_tokens=target_tokens, - chunk_size_tokens=chunk_size, - max_depth=max_depth, - temperature=getattr(self.config, 'ai_temperature', 0.1) - ) - logger.info(f"Документ '{file_name}' сжат с {text_tokens} до {count_tokens(compressed)} токенов") - return compressed - except Exception as e: - logger.error(f"Ошибка при сжатии документа '{file_name}': {e}, используем исходный текст") - return text - # ------------------------------------------------------------------ # Основной метод обработки запроса # ------------------------------------------------------------------ @@ -215,27 +159,21 @@ class RAGOrchestrator: Основной метод обработки запроса. ИСТОРИЯ ДИАЛОГА ПОЛУЧАЕТСЯ ИЗ БД, а не из запроса. - Это обеспечивает единый контекст для всех клиентов (XMPP, Telegram и т.д.). + Это обеспечивает единый контекст для всех клиентов. Аргументы: - query (str): текст запроса пользователя - user_jid (str): JID пользователя (без ресурса) - room_jid (Optional[str]): JID комнаты (None для личного чата) - prompts (Optional[Dict[str, str]]): словарь промптов для текущего запроса. - Если не передан, используются default_prompts. - intent_override (Optional[str]): принудительное переопределение намерения - last_file_path (Optional[str]): путь к последнему загруженному файлу - last_file_text (Optional[str]): текст последнего загруженного файла + query: текст запроса пользователя + user_jid: JID пользователя (без ресурса) + room_jid: JID комнаты (None для личного чата) + prompts: словарь промптов (если None, используются default_prompts) + intent_override: принудительное переопределение намерения + last_file_path: путь к последнему загруженному файлу + last_file_text: текст последнего загруженного файла Возвращает: - Dict[str, Any]: словарь с ключами: - - answer (str): итоговый ответ - - intent (str): распознанное намерение - - context (str): использованный контекст (для отладки) - - sources (List[str]): список источников - - confidence (float): оценка уверенности (если есть) + Словарь с ключами: answer, intent, context, sources, confidence, error """ - # ----- 1. Подготовка промптов ----- + # 1. Подготовка промптов if prompts is None: prompts = self.default_prompts.copy() synthesis_template = prompts.get('synthesis', '') @@ -243,10 +181,7 @@ class RAGOrchestrator: synthesis_template = "{context}\n\n{query}\n\nОтвет:" system_prompt = prompts.get('system', None) - # ----- 2. Получение истории диалога из БД (необрезанной) ----- - history = await self.db.get_history(user_jid, room_jid, limit=100) - - # ----- 3. Расчёт лимитов токенов ----- + # 2. Расчёт лимитов токенов max_model_tokens = getattr(self.config, 'max_model_tokens', 8192) reserved_for_answer = getattr(self.config, 'reserved_for_answer_tokens', 1000) reserved_for_overhead = getattr(self.config, 'reserved_for_overhead_tokens', 200) @@ -255,81 +190,25 @@ class RAGOrchestrator: synthesis_template=synthesis_template, system_prompt=system_prompt, query=query, - context="", 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} токенов") - logger.debug( - f"Доступно для истории и контекста: {available_for_history_and_context} токенов" + # 3. Получение истории из БД + 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) + formatted_history = await self.history_manager.compress_history_if_needed( + raw_history, + max_tokens=max_history_tokens, + prompt_template=prompts.get('hierarchical_summary', '') ) - # ----- 4. Сжатие истории (иерархическое резюмирование) ----- - max_history_tokens = min(available_for_history_and_context // 2, 2000) - - # Если история слишком длинная, сжимаем её иерархически - history_text = "\n".join([f"{rec['role']}: {rec['content']}" for rec in history]) - if count_tokens(history_text) > max_history_tokens: - logger.info( - f"История слишком длинная ({count_tokens(history_text)} токенов), " - f"применяем иерархическое резюмирование" - ) - summary_prompt = prompts.get('hierarchical_summary', '') - if not summary_prompt: - try: - prompt_path = self.config.prompts_dir / 'hierarchical_summary.txt' - with open(prompt_path, 'r', encoding='utf-8') as f: - summary_prompt = f.read() - except Exception: - summary_prompt = ( - "Кратко изложи суть диалога, сохранив ключевые факты и вопросы.\n\n" - "ТЕКСТ:\n{text}" - ) - try: - compressed_history = await hierarchical_summarize( - text=history_text, - giga=self.giga, - prompt_template=summary_prompt, - target_tokens=max_history_tokens, - chunk_size_tokens=500, - max_depth=2, - temperature=getattr(self.config, 'ai_temperature', 0.1) - ) - formatted_history = [ - {"role": "system", "content": "Сжатая история диалога:\n" + compressed_history} - ] - logger.info( - f"История сжата с {count_tokens(history_text)} до {count_tokens(compressed_history)} токенов" - ) - except Exception as e: - logger.error(f"Ошибка при иерархическом резюмировании истории: {e}") - # Fallback: простая обрезка - truncated_history = [] - total_tokens = 0 - for record in reversed(history): - tokens = count_tokens(record['content']) - if total_tokens + tokens <= max_history_tokens: - truncated_history.append(record) - total_tokens += tokens - else: - break - truncated_history.reverse() - formatted_history = [ - {"role": rec['role'], "content": rec['content']} - for rec in truncated_history - ] - else: - # Если история укладывается, форматируем последние сообщения - # Берём максимум 20 сообщений, чтобы не перегружать - truncated_history = history[-20:] - formatted_history = [ - {"role": rec['role'], "content": rec['content']} - for rec in truncated_history - ] - - # ----- 5. Классификация намерений ----- + # 5. Классификация намерений intent = intent_override if intent is None and getattr(self.config, 'enable_intent_classification', True): intent_prompt = prompts.get('intent', '') @@ -345,320 +224,72 @@ class RAGOrchestrator: else: intent = intent or "GENERAL" - # ----- 6. Принудительная установка SURGICAL по ключевым словам ----- + # 6. Принудительная установка SURGICAL по ключевым словам keywords = getattr(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 (есть файл и ключевые слова)") - # ----- 7. Обработка специализированных намерений ----- + # 7. Маршрутизация специализированных намерений + 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 = None - ctx_for_critique = None - synthesis_template_for_critique = None - - # --- 7.1. METRICS --- - if intent == "METRICS": - context = await self.kb.find_relevant_info( - query, user_jid, room_jid, - top_k=getattr(self.config, 'rag_metrics_top_k', 30) - ) - if not context: - answer = "Не найдено данных для извлечения метрик." - else: - metrics_prompt = prompts.get('metrics', '') - metrics = await extract_metrics( - giga=self.giga, - context=context, - prompt_text=metrics_prompt, - bot_config=self.config - ) - if metrics: - lines = [f"- {m.get('metric_name')}: {m.get('value')} {m.get('unit', '')}" for m in metrics[:10]] - answer = "📊 **Извлечённые метрики:**\n" + "\n".join(lines) - else: - answer = "Не удалось извлечь метрики." - - # --- 7.2. SUMMARY --- - elif intent == "SUMMARY": - if not last_file_text: - answer = "Нет документа для суммаризации." - else: - summary_prompt = prompts.get('summary', '') - answer = await summarize_document( - giga=self.giga, - text=last_file_text, - title="Ваш документ", - prompt_text=summary_prompt, - bot_config=self.config - ) - - # --- 7.3. CONTRADICTION --- - elif intent == "CONTRADICTION": - context = await self.kb.find_relevant_info( - query, user_jid, room_jid, - top_k=getattr(self.config, 'rag_contradiction_top_k', 10) - ) - if not context: - answer = "Недостаточно данных для проверки противоречий." - else: - chunks = [c.strip() for c in context.split("\n\n") if c.strip()] - if len(chunks) < 2: - answer = "Недостаточно фрагментов." - else: - consistency_prompt = prompts.get('consistency', '') - consistency = await check_consistency( - giga=self.giga, - chunks=chunks, - query=query, - prompt_text=consistency_prompt, - bot_config=self.config - ) - if "[CONFLICT]" in consistency: - answer = f"⚠️ **Обнаружены противоречия:**\n{consistency}" - else: - answer = "✅ Противоречий не обнаружено." - - # --- 7.4. TEMPLATE_FILL --- - elif intent == "TEMPLATE_FILL": - template_text = last_file_text or "" - if not template_text and last_file_path and os.path.exists(last_file_path): - result = await asyncio.to_thread(self.files.process_any_file, last_file_path) - if isinstance(result, tuple) and len(result) == 2: - template_text = result[0] - else: - template_text = str(result) - if not template_text: - answer = "❌ Нет шаблона документа для заполнения. Загрузите файл .docx." - else: - truncated_template = template_text[:5000] - search_query = f"{query}\n{truncated_template}" - context = await self.kb.find_relevant_info(search_query, user_jid, room_jid, top_k=15) - fill_prompt = prompts.get('generate_document', '') - if not fill_prompt: - fill_prompt = "Заполни плейсхолдеры, используя базу знаний." - prompt = ( - f"Перед тобой шаблон документа. Заполни плейсхолдеры, используя ТОЛЬКО базу знаний.\n\n" - f"[ТЕКСТ ШАБЛОНА]:\n{truncated_template}\n\n" - f"[ДАННЫЕ ИЗ БАЗЫ ЗНАНИЙ]:\n{context}\n\n" - f"Инструкция по заполнению:\n{fill_prompt}\n\n" - f"Формат ответа: [SURGICAL_REPLACE]\nинструкция_из_шаблона ||| текст_из_БЗ\n[/SURGICAL_REPLACE]" - ) - answer = await self.giga.chat( - history=formatted_history, - query=prompt, - system_prompt=system_prompt, - file_id=None, - temperature=getattr(self.config, 'ai_temperature', 0.1) - ) - - # --- 7.5. SPELLCHECK --- - elif intent == "SPELLCHECK": - if room_jid is not None: - answer = "⚠️ Проверка орфографии доступна только в личном чате." - else: - if not last_file_text or not last_file_path: - answer = "❌ Нет документа для проверки. Сначала отправьте файл .docx." - elif not last_file_path.lower().endswith('.docx'): - answer = "❌ Проверка орфографии поддерживается только для файлов .docx." - else: - answer = "⚠️ Проверка орфографии требует доработки (передача промптов)." - - # --- 7.6. GREETING --- - elif intent == "GREETING": - answer = await self.giga.chat( - history=formatted_history, - query=query, - system_prompt=system_prompt, - file_id=None, - temperature=getattr(self.config, 'ai_temperature', 0.1) - ) - - # --- 7.7. SURGICAL --- - elif intent == "SURGICAL": - if not last_file_path: - answer = "❌ Нет загруженного документа для замены." - else: - import re - q = re.sub(r'\s+', ' ', query.lower()).strip() - m = re.search(r'замени\s+(.+?)\s+на\s+(.+)', q) - if not m: - m = re.search(r'заменить\s+(.+?)\s+на\s+(.+)', q) - if not m: - answer = "❌ Не удалось распознать, что на что заменять. Используйте формат: замени слово на слово" - else: - old_word = m.group(1).strip() - new_word = m.group(2).strip() - try: - from mawo_pymorphy3 import create_analyzer - morph = create_analyzer() - old_forms = set() - parsed_old = morph.parse(old_word)[0] - old_forms.add(old_word) - old_forms.add(parsed_old.normal_form) - cases = ['nomn', 'gent', 'datv', 'accs', 'ablt', 'loct'] - numbers = ['sing', 'plur'] - for number in numbers: - for case in cases: - inflected = parsed_old.inflect({case, number}) - if inflected: - old_forms.add(inflected.word) - replacements = {} - for old_form in old_forms: - parsed_old_form = morph.parse(old_form)[0] - tags = set() - if parsed_old_form.tag.case: - tags.add(parsed_old_form.tag.case) - if parsed_old_form.tag.number: - tags.add(parsed_old_form.tag.number) - if parsed_old_form.tag.gender: - tags.add(parsed_old_form.tag.gender) - parsed_new = morph.parse(new_word)[0] - new_form = parsed_new.inflect(tags) - replacements[old_form] = new_form.word if new_form else new_word - replacements[old_word] = new_word - replacements[parsed_old.normal_form] = new_word - - new_path = await asyncio.to_thread(self.files.surgical_replace, last_file_path, replacements) - if new_path: - answer = f"Замена '{old_word}' → '{new_word}' выполнена. Файл сохранён: {new_path}" - else: - answer = f"❌ Ошибка при замене '{old_word}' → '{new_word}'." - except ImportError: - answer = "❌ Библиотека mawo-pymorphy3 не установлена. Установите её для морфологической замены." - - # ----- 8. Обычный RAG (для GENERAL, FACT, PROCEDURE, COMPARISON, CALCULATION) ----- - if answer is None: - # 8.1. Расширение запроса - expanded = await expand_query( - giga=self.giga, - query=query, - prompt_text=prompts.get('expand', ''), - bot_config=self.config - ) - search_query = expanded if expanded and expanded != query else query - - # 8.2. Поиск релевантного контекста в базе знаний - context = await self.kb.find_relevant_info( - search_query, user_jid, room_jid, - top_k=getattr(self.config, 'rag_default_top_k', 30) - ) - logger.info(f"Найден контекст длиной {len(context)} символов (room={room_jid})") - - # 8.3. Обрезка контекста по токенам - # Оставшиеся токены после истории и промптов – используем для контекста - max_context_tokens = available_for_history_and_context - sum(count_tokens(msg['content']) for msg in formatted_history) - max_context_tokens = max(max_context_tokens, 0) - - if max_context_tokens > 0 and context: - context_tokens = count_tokens(context) - if context_tokens > max_context_tokens: - logger.warning( - f"Контекст слишком длинный ({context_tokens} токенов), обрезаем до {max_context_tokens}" - ) - max_context_chars = int(max_context_tokens * 3.5) - if max_context_chars > 0: - context = context[:max_context_chars] - else: - context = "" - - # 8.4. Переранжирование контекста - rerank_min_length = getattr(self.config, 'rerank_min_length', 5000) - if intent != "FACT" and len(context) > rerank_min_length: - context = await rerank_context( - bot=None, - query=query, - context=context, - prompt_text=None, - bot_config=self.config - ) - - # 8.5. Синтез ответа с добавлением цепочки рассуждений (CoT) - synthesis_template = synthesis_template - if intent in ("CALCULATION", "PROCEDURE"): - cot_instruction = ( - "\n\nПожалуйста, покажи пошаговое решение перед итоговым ответом. " - "Опиши каждый шаг вычислений или действий в логической последовательности. " - "После всех шагов дай итоговый ответ." - ) - synthesis_template += cot_instruction - - full_query = synthesis_template.format(context=context, query=query) - logger.debug(f"Полный запрос к GigaChat (первые 500 символов): {full_query[:500]}") - - # 8.6. Генерация ответа - answer = await self.giga.chat( - history=formatted_history, - query=full_query, - system_prompt=system_prompt, - file_id=None, - temperature=getattr(self.config, 'ai_temperature', 0.1) - ) - - ctx_for_critique = context - synthesis_template_for_critique = synthesis_template - - # ----- 9. Самокритика (если включена) ----- - if (getattr(self.config, 'enable_self_critique', False) and - intent not in ("METRICS", "SUMMARY", "CONTRADICTION") and - ctx_for_critique is not None and - answer is not None): - critique_prompt = prompts.get('critique', '') - if critique_prompt: - logger.debug("Запуск самокритики") - is_ok = await critique_answer( - giga=self.giga, - query=query, - context=ctx_for_critique, - answer=answer, - prompt_text=critique_prompt, - bot_config=self.config - ) - if not is_ok: - logger.warning("Ответ не прошёл самокритику, перегенерация") - full_query_retry = synthesis_template_for_critique.format( - context=ctx_for_critique, - query=query - ) - answer = await self.giga.chat( - history=formatted_history, - query=full_query_retry, - system_prompt=system_prompt, - file_id=None, - temperature=getattr(self.config, 'ai_temperature', 0.1) - ) - if not await critique_answer( - giga=self.giga, - query=query, - context=ctx_for_critique, - answer=answer, - prompt_text=critique_prompt, - bot_config=self.config - ): - answer = "⚠️ Извините, я не уверен в точности ответа. Проверьте данные." - - # ----- 10. Сохранение истории диалога в БД ----- - await self.db.add_history(user_jid, "user", query, room_jid) - if answer: - await self.db.add_history(user_jid, "assistant", answer, room_jid) - - # ----- 11. Извлечение источников из контекста ----- + context = "" sources = [] - if context: - for match in re.finditer(r'\[источник:\s*([^\]]+)\]', context): - sources.append(match.group(1)) - # ----- 12. Формирование результата ----- + if router_result is not None: + # Намерение обработано маршрутизатором + answer = router_result.get("answer") + context = router_result.get("context", "") + sources = router_result.get("sources", []) + else: + # Обычный RAG-пайплайн (GENERAL, FACT, PROCEDURE, COMPARISON, CALCULATION) + # Вычисляем, сколько токенов осталось для контекста после истории + 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", []) + + # 8. Сохранение истории диалога в БД + 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) + + # 9. Формирование результата return { - "answer": answer, + "answer": answer or "⚠️ Не удалось сгенерировать ответ.", "intent": intent, "context": context, - "sources": list(set(sources)), - "confidence": None + "sources": sources, + "confidence": None, + "error": None } # ------------------------------------------------------------------ - # Индексация документа (с возможным сжатием) + # Индексация документа # ------------------------------------------------------------------ async def index_document( @@ -674,23 +305,36 @@ class RAGOrchestrator: update_if_exists: bool = True ) -> Dict[str, Any]: """ - Индексация документа в базу знаний. - Перед индексацией применяет иерархическое резюмирование, - если документ слишком большой. - """ - # Сжимаем документ, если он слишком большой - compressed_text = await self._compress_document_if_needed(file_text, file_name) + Индексирует документ в базу знаний. + Делегирует работу IndexingManager. - # Передаём сжатый текст в сервис базы знаний - doc_id, chunk_count = await self.kb.add_document( - file_name=file_name, - file_text=compressed_text, - user_jid=user_jid, - is_global=is_global, - title=title, - metadata=metadata, - room_jid=room_jid, - file_hash=file_hash, - update_if_exists=update_if_exists - ) - return {"doc_id": doc_id, "chunk_count": chunk_count} \ No newline at end of file + Аргументы: + 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: + 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)} \ No newline at end of file