Редактировать query_processor.py

rag/query_processor.py

@@ -3,10 +3,17 @@
 Обработчик обычных RAG-запросов (FACT, PROCEDURE, COMPARISON, CALCULATION, GENERAL).
 Выполняет расширение запроса, поиск в БЗ, переранжирование, синтез ответа,
 самокритику и перегенерацию при необходимости.
+
+ДОБАВЛЕНО:
+- Интеграция SOMA-анализа (LLM-as-judge)
+- Интеграция ReAct (вызов внешних инструментов)
+- Интеграция планирования действий (generate_plan)
+- Интеграция многоагентности (AgentCoordinator)
 """
 
 import logging
 import re
+import json
 from typing import Optional, Dict, List, Any
 
 from core.services.giga_client import GigaClient
@@ -18,12 +25,19 @@ from core.utils.text_utils import count_tokens
 from core.prompt_builder import PromptBuilder
 from core.config_models import AppConfig
 
+# Импорты новых модулей (интеграция)
+from .functions.soma_evaluate import soma_evaluate
+from .functions.react import react_loop
+from .functions.plan_generation import generate_plan
+from .agents.coordinator import AgentCoordinator
+
 logger = logging.getLogger(__name__)
 
 
 class QueryProcessor:
     """
     Обрабатывает обычные RAG-запросы (не специализированные намерения).
+    Интегрирует SOMA, ReAct, планирование и агентов.
     """
 
     def __init__(
@@ -31,13 +45,56 @@ class QueryProcessor:
         giga: GigaClient,
         kb: KBService,
         config: AppConfig,
-        default_prompts: Optional[Dict[str, str]] = None
+        default_prompts: Optional[Dict[str, str]] = None,
+        agent_coordinator: Optional[AgentCoordinator] = None,
     ) -> None:
+        """
+        Инициализация QueryProcessor.
+
+        Аргументы:
+            giga: клиент GigaChat.
+            kb: сервис базы знаний.
+            config: конфигурация приложения.
+            default_prompts: словарь промптов по умолчанию.
+            agent_coordinator: координатор агентов (если включена многоагентность).
+        """
         self.giga: GigaClient = giga
         self.kb: KBService = kb
         self.config: AppConfig = config
         self.default_prompts: Dict[str, str] = default_prompts or {}
         self.prompt_builder: PromptBuilder = PromptBuilder(config)
+        self.agent_coordinator: Optional[AgentCoordinator] = agent_coordinator
+
+    def _needs_external_data(self, query: str) -> bool:
+        """
+        Определяет, требует ли запрос вызова внешних инструментов.
+        Используется для активации ReAct.
+        """
+        keywords = ["найди", "поищи", "забронируй", "купи", "узнай курс", "какая погода",
+                    "погода", "билеты", "отель", "рейс", "тариф", "курс"]
+        return any(kw in query.lower() for kw in keywords)
+
+    def _get_available_tools(self) -> List[Dict[str, Any]]:
+        """
+        Возвращает список доступных инструментов для ReAct.
+        В реальности инструменты определяются в конфиге и регистрируются в приложении.
+        Здесь возвращаем заглушку, которая должна быть переопределена в рантайме.
+        """
+        # В будущем можно загружать из конфига: getattr(self.config, 'react_tools', [])
+        return [
+            {
+                "name": "search",
+                "description": "Поиск информации в интернете",
+                "parameters": {"query": "string"},
+                "handler": None  # будет передан извне (например, через регистрацию)
+            },
+            {
+                "name": "get_weather",
+                "description": "Получить текущую погоду в городе",
+                "parameters": {"city": "string"},
+                "handler": None
+            }
+        ]
 
     async def process(
         self,
@@ -51,7 +108,7 @@ class QueryProcessor:
         available_tokens_for_context: int,
     ) -> Dict[str, Any]:
         """
-        Выполняет полный RAG-пайплайн для обычного запроса.
+        Выполняет полный RAG-пайплайн для обычного запроса с интеграцией новых функций.
 
         Аргументы:
             query: текст запроса пользователя
@@ -66,7 +123,31 @@ class QueryProcessor:
         Возвращает:
             Словарь с ключами 'answer', 'context', 'sources', 'confidence'
         """
-        # 1. Расширение запроса
+
+        # ---- 1. Использование агентов (если включено) ----
+        if (getattr(self.config, 'enable_agents', False) and
+                self.agent_coordinator is not None):
+            logger.info("Запрос передан в AgentCoordinator")
+            try:
+                # Пытаемся обработать запрос через агентов
+                agent_result = await self.agent_coordinator.process(
+                    query=query,
+                    context="",  # агенты сами ищут контекст (или используют свой)
+                    history=history,
+                    tools=None  # можно передать общие инструменты
+                )
+                if agent_result:
+                    logger.info("AgentCoordinator вернул ответ")
+                    return {
+                        "answer": agent_result,
+                        "context": "",
+                        "sources": [],
+                        "confidence": None
+                    }
+            except Exception as e:
+                logger.error(f"Ошибка при работе AgentCoordinator: {e}, переходим к обычному RAG")
+
+        # ---- 2. Расширение запроса ----
         expanded: str = await expand_query(
             giga=self.giga,
             query=query,
@@ -75,14 +156,14 @@ class QueryProcessor:
         )
         search_query: str = expanded if expanded and expanded != query else query
 
-        # 2. Поиск релевантного контекста в базе знаний
+        # ---- 3. Поиск релевантного контекста в базе знаний ----
         context: str = 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})")
 
-        # 3. Обрезка контекста по токенам
+        # ---- 4. Обрезка контекста по токенам ----
         if context:
             context_tokens: int = count_tokens(context)
             if context_tokens > available_tokens_for_context:
@@ -96,7 +177,7 @@ class QueryProcessor:
                 else:
                     context = ""
 
-        # 4. Переранжирование контекста (если включено и контекст достаточно длинный)
+        # ---- 5. Переранжирование контекста ----
         rerank_min_length: int = getattr(self.config, 'rerank_min_length', 5000)
         if intent != "FACT" and len(context) > rerank_min_length:
             context = await rerank_context(
@@ -107,9 +188,23 @@ class QueryProcessor:
                 bot_config=self.config
             )
 
-        # 5. Формирование промта с помощью PromptBuilder
+        # ---- 6. Планирование (если включено) ----
+        plan = None
         extra_instructions: str = ""
-        if intent in ("CALCULATION", "PROCEDURE"):
+        if getattr(self.config, 'enable_planning', False) and intent in ("CALCULATION", "PROCEDURE"):
+            plan = await generate_plan(
+                query=query,
+                history=history,
+                config=self.config,
+                giga=self.giga,
+                prompts=self.default_prompts
+            )
+            if plan:
+                logger.info(f"Сгенерирован план из {len(plan)} шагов")
+                extra_instructions = f"Выполни следующие шаги по порядку: {json.dumps(plan, ensure_ascii=False)}"
+
+        # ---- 7. Формирование промта (с учётом возможного плана) ----
+        if intent in ("CALCULATION", "PROCEDURE") and not extra_instructions:
             extra_instructions = (
                 "Пожалуйста, покажи пошаговое решение перед итоговым ответом. "
                 "Опиши каждый шаг вычислений или действий в логической последовательности. "
@@ -130,7 +225,7 @@ class QueryProcessor:
         )
         logger.debug(f"Сформированный промт (первые 500 символов): {prompt[:500]}")
 
-        # 6. Генерация ответа
+        # ---- 8. Генерация ответа ----
         answer: str = await self.giga.chat(
             history=[],
             query=prompt,
@@ -139,7 +234,31 @@ class QueryProcessor:
             temperature=getattr(self.config, 'ai_temperature', 0.1)
         )
 
-        # 7. Самокритика (если включена)
+        # ---- 9. ReAct (если требуется внешний вызов) ----
+        if (getattr(self.config, 'enable_react', False) and
+                self._needs_external_data(query)):
+            logger.info("Запрос требует внешних данных, запускаем ReAct цикл")
+            try:
+                tools = self._get_available_tools()
+                # В реальности нужно передать реальные обработчики инструментов.
+                # Здесь они могут быть зарегистрированы в конфиге.
+                # Для демонстрации мы пытаемся вызвать react_loop.
+                react_answer = await react_loop(
+                    query=query,
+                    history=history,
+                    tools=tools,
+                    config=self.config,
+                    giga=self.giga,
+                    prompts=self.default_prompts,
+                    max_iterations=getattr(self.config, 'react_max_iterations', 5)
+                )
+                if react_answer:
+                    answer = react_answer
+                    logger.info("ReAct цикл завершён, ответ получен")
+            except Exception as e:
+                logger.error(f"ReAct цикл не удался: {e}, используем обычный ответ")
+
+        # ---- 10. Самокритика ----
         if getattr(self.config, 'enable_self_critique', False) and context:
             critique_prompt: str = prompts.get('critique', '')
             if critique_prompt:
@@ -171,7 +290,47 @@ class QueryProcessor:
                     ):
                         answer = "⚠️ Извините, я не уверен в точности ответа. Проверьте данные."
 
-        # 8. Извлечение источников
+        # ---- 11. SOMA-анализ (если включен) ----
+        if getattr(self.config, 'enable_soma', False):
+            logger.debug("Запуск SOMA-оценки")
+            try:
+                evaluation = await soma_evaluate(
+                    query=query,
+                    context=context,
+                    answer=answer,
+                    config=self.config,
+                    giga=self.giga,
+                    prompts=self.default_prompts
+                )
+                if evaluation.get('verdict') == 'fail':
+                    threshold = getattr(self.config, 'soma_threshold', 3.5)
+                    if evaluation.get('overall_score', 0) < threshold:
+                        logger.warning(f"SOMA оценка низкая ({evaluation['overall_score']}), перегенерация")
+                        # Перегенерация с повышенной температурой для разнообразия
+                        answer = await self.giga.chat(
+                            history=[],
+                            query=prompt,
+                            system_prompt=None,
+                            file_id=None,
+                            temperature=0.3
+                        )
+                        # Вторая оценка (опционально)
+                        evaluation2 = await soma_evaluate(
+                            query=query,
+                            context=context,
+                            answer=answer,
+                            config=self.config,
+                            giga=self.giga,
+                            prompts=self.default_prompts
+                        )
+                        if evaluation2.get('verdict') == 'fail':
+                            answer = "⚠️ Извините, я не уверен в точности ответа. Проверьте данные."
+                        else:
+                            logger.info("SOMA оценка после перегенерации успешна")
+            except Exception as e:
+                logger.error(f"Ошибка в SOMA-оценке: {e}, пропускаем")
+
+        # ---- 12. Извлечение источников ----
         sources: List[str] = []
         if context:
             for match in re.finditer(r'\[источник:\s*([^\]]+)\]', context):