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
This commit is contained in:
Markov Andrey
212
rag/prompt_builder.py
Normal file
212
rag/prompt_builder.py
Normal file
@@ -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)
|
||||
Reference in New Issue
Block a user