From 67d86d6c1dea075e8048283b7eb17cfebe68fb5e Mon Sep 17 00:00:00 2001 From: Markov Andrey Date: Tue, 30 Jun 2026 14:57:03 +0000 Subject: [PATCH] Update 5 files - /rag/config_models.py - /rag/utils/config_loader.py - /rag/prompt_builder.py - /rag/query_processor.py - /template_bot_profile/data/fewshot_examples.json --- rag/config_models.py | 15 ++ rag/prompt_builder.py | 212 ++++++++++++++++++ rag/query_processor.py | 96 ++++---- rag/utils/config_loader.py | 4 + .../data/fewshot_examples.json | 26 +++ 5 files changed, 298 insertions(+), 55 deletions(-) create mode 100644 rag/prompt_builder.py create mode 100644 template_bot_profile/data/fewshot_examples.json diff --git a/rag/config_models.py b/rag/config_models.py index 9fc4115..bc22fc6 100644 --- a/rag/config_models.py +++ b/rag/config_models.py @@ -179,6 +179,12 @@ class AuthConfig(BaseModel): api_key_env: str = Field("RAG_API_KEY", description="Имя переменной окружения с API-ключом для доступа к RAG-серверу") +class FewshotConfig(BaseModel): + """Настройки few-shot примеров.""" + max_examples: int = Field(3, description="Максимальное количество примеров, включаемых в промт") + examples_file: str = Field("data/fewshot_examples.json", description="Путь к файлу с примерами (относительно профиля)") + + # ============================================================ # Главная модель конфигурации приложения # ============================================================ @@ -219,6 +225,7 @@ class AppConfig(BaseModel): prompts: PromptsConfig web_scraper: WebScraperConfig = Field(default_factory=WebScraperConfig) auth: AuthConfig = Field(default_factory=AuthConfig) + fewshot: FewshotConfig = Field(default_factory=FewshotConfig) # Дополнительные поля для путей (заполняются при загрузке) profile_dir: Optional[Path] = Field(None, description="Корень профиля") @@ -527,5 +534,13 @@ class AppConfig(BaseModel): def file_mode(self) -> int: return int(self.file_permissions.mode, 8) + @property + def max_fewshot_examples(self) -> int: + return self.fewshot.max_examples + + @property + def fewshot_examples_file(self) -> str: + return self.fewshot.examples_file + class Config: extra = "forbid" # Запрещаем неизвестные поля \ No newline at end of file diff --git a/rag/prompt_builder.py b/rag/prompt_builder.py new file mode 100644 index 0000000..956413f --- /dev/null +++ b/rag/prompt_builder.py @@ -0,0 +1,212 @@ +# -*- coding: utf-8 -*- +""" +Построитель промтов для RAG-системы. +Реализует динамический подбор few-shot примеров и структуру «сэндвич». +""" + +import json +import logging +import re +from pathlib import Path +from typing import List, Dict, Any, Optional + +from .config_models import AppConfig + +logger = logging.getLogger(__name__) + + +class PromptBuilder: + """ + Формирует промт для LLM, используя динамические few-shot примеры + и структуру «сэндвич» (важное в начале и в конце). + """ + + def __init__(self, config: AppConfig): + """ + Инициализация. + + Аргументы: + config: объект AppConfig (содержит пути и настройки few-shot). + """ + self.config = config + self.max_examples = config.fewshot.max_examples + examples_path = Path(config.fewshot.examples_file) + if not examples_path.is_absolute(): + examples_path = config.profile_dir / examples_path + self.examples_file = examples_path + self._examples = None # кэш загруженных примеров + + def load_examples(self) -> List[Dict[str, Any]]: + """ + Загружает few-shot примеры из JSON-файла или возвращает статические. + """ + if self._examples is not None: + return self._examples + + if self.examples_file and self.examples_file.exists(): + try: + with open(self.examples_file, 'r', encoding='utf-8') as f: + data = json.load(f) + if isinstance(data, list): + self._examples = data + logger.info(f"Загружено {len(self._examples)} few-shot примеров из {self.examples_file}") + return self._examples + else: + logger.warning(f"Файл {self.examples_file} должен содержать список JSON-объектов") + except Exception as e: + logger.error(f"Ошибка загрузки few-shot примеров: {e}") + + # Статические примеры (запасные) + static_examples = [ + { + "keywords": ["OEE", "эффективность", "оборудование"], + "intents": ["PROCEDURE", "CALCULATION"], + "question": "Как рассчитать OEE для станка?", + "answer": "OEE = Доступность × Производительность × Качество. Доступность = (запланированное время - простои) / запланированное время. Производительность = (фактический выпуск / теоретический выпуск) × 100%. Качество = (годные изделия / общий выпуск) × 100%. Итоговый OEE = Доступность × Производительность × Качество." + }, + { + "keywords": ["время", "переналадка", "SMED"], + "intents": ["FACT"], + "question": "Какое время переналадки установлено для линии А?", + "answer": "Время переналадки линии А составляет 35 минут согласно регламенту." + }, + { + "keywords": ["5S", "рабочее место", "сортировка"], + "intents": ["PROCEDURE"], + "question": "Как внедрить 5S на рабочем месте?", + "answer": "1. Сортировка (Seiri) – удалить всё лишнее. 2. Соблюдение порядка (Seiton) – расположить инструменты рационально. 3. Сияние (Seiso) – поддерживать чистоту. 4. Стандартизация (Seiketsu) – создать стандарты. 5. Совершенствование (Shitsuke) – соблюдать дисциплину." + }, + { + "keywords": ["KPI", "метрики", "показатели"], + "intents": ["METRICS"], + "question": "Выведи все KPI из отчёта по производству.", + "answer": "OEE линии: 85,2%; Тактовое время: 120 сек; Выработка на человека: 450 деталей/смену; Время переналадки: 35 мин; Процент брака: 2,1%." + } + ] + self._examples = static_examples + logger.info(f"Используются статические few-shot примеры (в количестве {len(self._examples)})") + return self._examples + + def select_examples(self, query: str, intent: str) -> List[Dict[str, Any]]: + """ + Выбирает наиболее релевантные примеры на основе запроса и намерения. + + Аргументы: + query: текст запроса пользователя. + intent: код намерения (FACT, PROCEDURE, ...). + + Возвращает: + Список словарей (примеры), не более max_examples. + """ + all_examples = self.load_examples() + if not all_examples: + return [] + + # Нормализуем запрос + query_lower = query.lower() + words = set(re.findall(r'\w+', query_lower)) + + # Вычисляем релевантность для каждого примера + scored = [] + for ex in all_examples: + # Проверка по намерению (если указано) + intent_match = (intent in ex.get('intents', [])) if ex.get('intents') else False + + # Проверка по ключевым словам + keywords = ex.get('keywords', []) + keyword_match = any(kw.lower() in query_lower for kw in keywords) if keywords else False + + # Количество совпадающих слов из запроса (дополнительно) + if keywords: + kw_set = set(kw.lower() for kw in keywords) + word_overlap = len(words.intersection(kw_set)) + else: + word_overlap = 0 + + # Суммарный балл + score = 0 + if intent_match: + score += 2 + if keyword_match: + score += 1 + score += word_overlap * 0.5 + + scored.append((score, ex)) + + # Сортируем по убыванию балла и берём top N + scored.sort(key=lambda x: x[0], reverse=True) + selected = [ex for _, ex in scored[:self.max_examples] if ex.get('question') and ex.get('answer')] + return selected + + def format_example(self, example: Dict[str, Any]) -> str: + """ + Форматирует один пример для вставки в промт. + """ + return f"Вопрос: {example['question']}\nОтвет: {example['answer']}" + + def build_prompt( + self, + query: str, + intent: str, + context: str, + history: List[Dict[str, str]], + system_prompt: Optional[str], + extra_instructions: Optional[str] = None, + ) -> str: + """ + Строит финальный промт по принципу «сэндвича». + + Аргументы: + query: текст запроса пользователя. + intent: код намерения. + context: извлечённый контекст из базы знаний. + history: история диалога (список сообщений с role и content). + system_prompt: системный промпт (роль, стиль). + extra_instructions: дополнительные инструкции (например, для CoT). + + Возвращает: + str: готовый промт для отправки в LLM. + """ + parts = [] + + # 1. Системный промпт (в самом начале) + if system_prompt: + parts.append(f"<|im_start|>system\n{system_prompt}\n<|im_end|>") + + # 2. История диалога (если есть) + if history: + for msg in history: + role = msg.get('role', 'user') + content = msg.get('content', '') + if role == 'system': + continue # системное уже добавлено + parts.append(f"<|im_start|>{role}\n{content}\n<|im_end|>") + + # 3. Ключевые инструкции (краткое напоминание о задаче) + instructions = "Ты — профессиональный методолог-аналитик. Отвечай строго на основе предоставленного контекста. Используй профессиональную терминологию. Если ответа нет в контексте, сообщи об этом." + if extra_instructions: + instructions += f"\n{extra_instructions}" + parts.append(f"<|im_start|>system\n{instructions}\n<|im_end|>") + + # 4. Few-shot примеры (после инструкций, до основного контекста) + examples = self.select_examples(query, intent) + if examples: + example_text = "\n\n".join(self.format_example(ex) for ex in examples) + parts.append(f"<|im_start|>system\nВот несколько примеров ответов на похожие вопросы:\n\n{example_text}\n<|im_end|>") + + # 5. Основной контекст (источники) + if context: + parts.append(f"<|im_start|>system\n[ДАННЫЕ ИЗ БАЗЫ ЗНАНИЙ]\n{context}\n<|im_end|>") + + # 6. Повторение ключевого вопроса и требований (в конце – важен для фокуса) + final_prompt = f"{query}\n\nПожалуйста, дай ответ на основе предоставленного контекста. Укажи источники в формате [Источник: название_файла]. Не используй LaTeX, формулы пиши текстом. Будь краток, но точен." + + if intent in ("CALCULATION", "PROCEDURE"): + final_prompt += "\n\nПокажи пошаговое решение перед итоговым ответом." + + parts.append(f"<|im_start|>user\n{final_prompt}\n<|im_end|>") + + # 7. Переход к ответу (ассистент) + parts.append("<|im_start|>assistant") + + return "\n".join(parts) \ No newline at end of file diff --git a/rag/query_processor.py b/rag/query_processor.py index a1f327b..fd5495c 100644 --- a/rag/query_processor.py +++ b/rag/query_processor.py @@ -9,13 +9,14 @@ import logging import re from typing import Optional, Dict, List, Any -from rag.services.giga_client import GigaClient -from rag.services.kb_service import KBService -from rag.functions.expand_query import expand_query -from rag.functions.rerank_context import rerank_context -from rag.functions.critique_answer import critique_answer -from rag.utils.text_utils import count_tokens -from rag.config_models import AppConfig +from .services.giga_client import GigaClient +from .services.kb_service import KBService +from .functions.expand_query import expand_query +from .functions.rerank_context import rerank_context +from .functions.critique_answer import critique_answer +from .utils.text_utils import count_tokens +from .prompt_builder import PromptBuilder +from .config_models import AppConfig logger = logging.getLogger(__name__) @@ -32,19 +33,11 @@ class QueryProcessor: config: AppConfig, default_prompts: Optional[Dict[str, str]] = None ): - """ - Инициализация процессора запросов. - - Аргументы: - giga: клиент GigaChat - kb: сервис базы знаний - config: объект конфигурации (AppConfig) - default_prompts: словарь промптов по умолчанию - """ self.giga = giga self.kb = kb self.config = config self.default_prompts = default_prompts or {} + self.prompt_builder = PromptBuilder(config) async def process( self, @@ -59,35 +52,20 @@ class QueryProcessor: ) -> Dict[str, Any]: """ Выполняет полный RAG-пайплайн для обычного запроса. - - Аргументы: - query: текст запроса пользователя - user_jid: JID пользователя - room_jid: JID комнаты (None для личного чата) - prompts: словарь промптов (expand, synthesis, critique, ...) - intent: код намерения (для выбора стратегии) - history: история диалога (уже сжатая, если нужно) - system_prompt: системный промпт - available_tokens_for_context: сколько токенов доступно для контекста (после вычета истории и промптов) - - Возвращает: - Словарь с ключами 'answer', 'context', 'sources', 'confidence' """ # 1. Расширение запроса - expand_prompt = prompts.get('expand', '') expanded = await expand_query( giga=self.giga, query=query, - prompt_text=expand_prompt, + prompt_text=prompts.get('expand', ''), bot_config=self.config ) search_query = expanded if expanded and expanded != query else query # 2. Поиск релевантного контекста в базе знаний - top_k = self.config.rag.default_top_k context = await self.kb.find_relevant_info( search_query, user_jid, room_jid, - top_k=top_k + top_k=getattr(self.config, 'rag_default_top_k', 30) ) logger.info(f"Найден контекст длиной {len(context)} символов (room={room_jid})") @@ -106,7 +84,7 @@ class QueryProcessor: context = "" # 4. Переранжирование контекста (если включено и контекст достаточно длинный) - rerank_min_length = self.config.rag.rerank_min_length + 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, @@ -116,34 +94,42 @@ class QueryProcessor: bot_config=self.config ) - # 5. Синтез ответа с добавлением цепочки рассуждений (CoT) для CALCULATION и PROCEDURE - synthesis_template = prompts.get('synthesis', '') - if not synthesis_template: - synthesis_template = "{context}\n\n{query}\n\nОтвет:" - + # 5. Формирование промта с помощью PromptBuilder (динамические few-shot, сэндвич) + extra_instructions = "" if intent in ("CALCULATION", "PROCEDURE"): - cot_instruction = ( - "\n\nПожалуйста, покажи пошаговое решение перед итоговым ответом. " + extra_instructions = ( + "Пожалуйста, покажи пошаговое решение перед итоговым ответом. " "Опиши каждый шаг вычислений или действий в логической последовательности. " "После всех шагов дай итоговый ответ." ) - synthesis_template += cot_instruction - full_query = synthesis_template.format(context=context, query=query) - logger.debug(f"Полный запрос к GigaChat (первые 500 символов): {full_query[:500]}") + # Если synthesis_template задан, добавим его в extra_instructions (но PromptBuilder уже использует стандартные инструкции) + synthesis_template = prompts.get('synthesis', '') + if synthesis_template and not extra_instructions: + # Если есть кастомный шаблон, используем его как дополнительную инструкцию + extra_instructions = synthesis_template.format(context=context, query=query) if '{context}' in synthesis_template else synthesis_template - # 6. Генерация ответа - temperature = self.config.ai.temperature - answer = await self.giga.chat( + prompt = self.prompt_builder.build_prompt( + query=query, + intent=intent, + context=context, history=history, - query=full_query, system_prompt=system_prompt, + extra_instructions=extra_instructions + ) + logger.debug(f"Сформированный промт (первые 500 символов): {prompt[:500]}") + + # 6. Генерация ответа (без отдельной передачи истории, она уже в промте) + answer = await self.giga.chat( + history=[], # история уже в промте + query=prompt, + system_prompt=None, # системный промпт тоже в промте file_id=None, - temperature=temperature + temperature=getattr(self.config, 'ai_temperature', 0.1) ) # 7. Самокритика (если включена) - if self.config.features.enable_self_critique and context: + if getattr(self.config, 'enable_self_critique', False) and context: critique_prompt = prompts.get('critique', '') if critique_prompt: logger.debug("Запуск самокритики") @@ -157,13 +143,13 @@ class QueryProcessor: ) if not is_ok: logger.warning("Ответ не прошёл самокритику, перегенерация") - full_query_retry = synthesis_template.format(context=context, query=query) + # Перегенерируем с тем же промтом answer = await self.giga.chat( - history=history, - query=full_query_retry, - system_prompt=system_prompt, + history=[], + query=prompt, + system_prompt=None, file_id=None, - temperature=temperature + temperature=getattr(self.config, 'ai_temperature', 0.1) ) # Повторная проверка после перегенерации if not await critique_answer( diff --git a/rag/utils/config_loader.py b/rag/utils/config_loader.py index c16252b..46790c3 100644 --- a/rag/utils/config_loader.py +++ b/rag/utils/config_loader.py @@ -105,6 +105,10 @@ def load_config(profile_dir: str) -> AppConfig: if 'auth' not in merged: merged['auth'] = {} + # Добавляем секцию fewshot, если её нет + if 'fewshot' not in merged: + merged['fewshot'] = {} + # Создаём экземпляр AppConfig try: config = AppConfig(**merged) diff --git a/template_bot_profile/data/fewshot_examples.json b/template_bot_profile/data/fewshot_examples.json new file mode 100644 index 0000000..b8cf47a --- /dev/null +++ b/template_bot_profile/data/fewshot_examples.json @@ -0,0 +1,26 @@ +[ + { + "keywords": ["OEE", "эффективность", "оборудование"], + "intents": ["PROCEDURE", "CALCULATION"], + "question": "Как рассчитать OEE для станка?", + "answer": "OEE = Доступность × Производительность × Качество. Доступность = (запланированное время - простои) / запланированное время. Производительность = (фактический выпуск / теоретический выпуск) × 100%. Качество = (годные изделия / общий выпуск) × 100%. Итоговый OEE = Доступность × Производительность × Качество." + }, + { + "keywords": ["время", "переналадка", "SMED"], + "intents": ["FACT"], + "question": "Какое время переналадки установлено для линии А?", + "answer": "Время переналадки линии А составляет 35 минут согласно регламенту." + }, + { + "keywords": ["5S", "рабочее место", "сортировка"], + "intents": ["PROCEDURE"], + "question": "Как внедрить 5S на рабочем месте?", + "answer": "1. Сортировка (Seiri) – удалить всё лишнее. 2. Соблюдение порядка (Seiton) – расположить инструменты рационально. 3. Сияние (Seiso) – поддерживать чистоту. 4. Стандартизация (Seiketsu) – создать стандарты. 5. Совершенствование (Shitsuke) – соблюдать дисциплину." + }, + { + "keywords": ["KPI", "метрики", "показатели"], + "intents": ["METRICS"], + "question": "Выведи все KPI из отчёта по производству.", + "answer": "OEE линии: 85,2%; Тактовое время: 120 сек; Выработка на человека: 450 деталей/смену; Время переналадки: 35 мин; Процент брака: 2,1%." + } +] \ No newline at end of file