# -*- 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)