From f0c7b92613fa371da63a78848f99f4b064b89a18 Mon Sep 17 00:00:00 2001 From: Markov Andrey Date: Tue, 30 Jun 2026 22:27:55 +0000 Subject: [PATCH] =?UTF-8?q?=D0=A0=D0=B5=D0=B4=D0=B0=D0=BA=D1=82=D0=B8?= =?UTF-8?q?=D1=80=D0=BE=D0=B2=D0=B0=D1=82=D1=8C=20query=5Fprocessor.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- rag/query_processor.py | 181 ++++++++++++++++++++++++++++++++++++++--- 1 file changed, 170 insertions(+), 11 deletions(-) diff --git a/rag/query_processor.py b/rag/query_processor.py index 73368b6..e9c262d 100644 --- a/rag/query_processor.py +++ b/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):