218 lines
11 KiB
Python
218 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) -> None:
|
||
"""
|
||
Инициализация.
|
||
|
||
Аргументы:
|
||
config: объект AppConfig (содержит пути и настройки few-shot).
|
||
"""
|
||
self.config: AppConfig = config
|
||
self.max_examples: int = config.fewshot.max_examples
|
||
examples_path: Path = Path(config.fewshot.examples_file)
|
||
if not examples_path.is_absolute():
|
||
examples_path = config.profile_dir / examples_path
|
||
self.examples_file: Path = examples_path
|
||
self._examples: Optional[List[Dict[str, Any]]] = 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: Any = 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: List[Dict[str, Any]] = [
|
||
{
|
||
"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: List[Dict[str, Any]] = self.load_examples()
|
||
if not all_examples:
|
||
return []
|
||
|
||
# Нормализуем запрос
|
||
query_lower: str = query.lower()
|
||
words: set = set(re.findall(r'\w+', query_lower))
|
||
|
||
# Вычисляем релевантность для каждого примера
|
||
scored: List[tuple] = []
|
||
for ex in all_examples:
|
||
# Проверка по намерению (если указано)
|
||
intent_match: bool = (intent in ex.get('intents', [])) if ex.get('intents') else False
|
||
|
||
# Проверка по ключевым словам
|
||
keywords: List[str] = ex.get('keywords', [])
|
||
keyword_match: bool = any(kw.lower() in query_lower for kw in keywords) if keywords else False
|
||
|
||
# Количество совпадающих слов из запроса (дополнительно)
|
||
if keywords:
|
||
kw_set: set = set(kw.lower() for kw in keywords)
|
||
word_overlap: int = len(words.intersection(kw_set))
|
||
else:
|
||
word_overlap = 0
|
||
|
||
# Суммарный балл
|
||
score: float = 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: List[Dict[str, Any]] = [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: List[str] = []
|
||
|
||
# 1. Системный промпт (в самом начале)
|
||
if system_prompt:
|
||
parts.append(f"<|im_start|>system\n{system_prompt}\n<|im_end|>")
|
||
|
||
# 2. История диалога (если есть)
|
||
if history:
|
||
for msg in history:
|
||
role: str = msg.get('role', 'user')
|
||
content: str = msg.get('content', '')
|
||
if role == 'system':
|
||
continue # системное уже добавлено
|
||
parts.append(f"<|im_start|>{role}\n{content}\n<|im_end|>")
|
||
|
||
# 3. Ключевые инструкции (краткое напоминание о задаче)
|
||
instructions: str = "Ты — профессиональный методолог-аналитик. Отвечай строго на основе предоставленного контекста. Используй профессиональную терминологию. Если ответа нет в контексте, сообщи об этом."
|
||
if extra_instructions:
|
||
instructions += f"\n{extra_instructions}"
|
||
parts.append(f"<|im_start|>system\n{instructions}\n<|im_end|>")
|
||
|
||
# 4. Few-shot примеры (после инструкций, до основного контекста)
|
||
examples: List[Dict[str, Any]] = self.select_examples(query, intent)
|
||
if examples:
|
||
example_text: str = "\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: str = (
|
||
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) |