67d86d6c1d
- /rag/config_models.py - /rag/utils/config_loader.py - /rag/prompt_builder.py - /rag/query_processor.py - /template_bot_profile/data/fewshot_examples.json
212 lines
11 KiB
Python
212 lines
11 KiB
Python
# -*- 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) |