Реализация динамической генерации промптов

Реализация динамической генерации промптов

Динамическая генерация промптов — построение промпта в runtime на основе контекста запроса: пользовательских данных, результатов поиска, истории взаимодействий. Промпт — не статический текст, а артефакт, собираемый по правилам. Повышает релевантность и точность ответов LLM.

Контекстно-зависимый промпт

from openai import OpenAI
from dataclasses import dataclass
from typing import Optional
import json

client = OpenAI()

@dataclass
class UserContext:
    user_id: str
    role: str           # "admin", "manager", "employee"
    department: str
    language: str       # "ru", "en"
    expertise_level: str  # "novice", "intermediate", "expert"

class DynamicPromptBuilder:

    def build_system_prompt(self, context: UserContext) -> str:
        """Строит system prompt под конкретного пользователя"""

        parts = [f"Ты — корпоративный ассистент."]

        # Адаптация к уровню экспертизы
        if context.expertise_level == "novice":
            parts.append("Объясняй понятно, избегай технических терминов, используй аналогии.")
        elif context.expertise_level == "expert":
            parts.append("Используй технические термины без объяснений. Фокусируйся на деталях и edge cases.")

        # Адаптация к роли
        role_context = {
            "admin": "Пользователь — системный администратор. Отвечай на технические вопросы развёрнуто.",
            "manager": "Пользователь — руководитель. Акцентируй бизнес-последствия, не технические детали.",
            "employee": "Пользователь — рядовой сотрудник. Давай пошаговые инструкции.",
        }
        if context.role in role_context:
            parts.append(role_context[context.role])

        # Язык ответа
        if context.language == "en":
            parts.append("Always respond in English.")

        return " ".join(parts)

    def build_user_prompt(
        self,
        question: str,
        retrieved_docs: list[dict] = None,
        conversation_history: list[dict] = None,
        user_context: UserContext = None,
    ) -> str:

        parts = []

        # Добавляем релевантные документы
        if retrieved_docs:
            docs_text = "\n\n".join([
                f"[{doc['title']}]:\n{doc['content'][:500]}"
                for doc in retrieved_docs[:3]
            ])
            parts.append(f"Релевантные документы:\n{docs_text}")

        # Краткая история (последние 2 обмена)
        if conversation_history and len(conversation_history) > 2:
            recent = conversation_history[-4:]  # 2 пары user/assistant
            history_text = "\n".join([
                f"{'Пользователь' if m['role'] == 'user' else 'Ассистент'}: {m['content'][:200]}"
                for m in recent
            ])
            parts.append(f"Контекст диалога:\n{history_text}")

        parts.append(f"Вопрос: {question}")

        return "\n\n".join(parts)

Промпт из шаблона + данных

class DataDrivenPromptGenerator:

    def generate_report_prompt(
        self,
        metrics: dict,
        period: str,
        audience: str,
        focus_areas: list[str] = None,
    ) -> str:

        # Определяем фокус на основе метрик
        anomalies = self.detect_anomalies(metrics)
        trend = self.calculate_trend(metrics)

        prompt = f"""Создай отчёт за период: {period}
Аудитория: {audience}

Метрики:
{self.format_metrics(metrics)}

"""
        if anomalies:
            prompt += f"Аномалии (требуют объяснения):\n{json.dumps(anomalies, ensure_ascii=False)}\n\n"

        if focus_areas:
            prompt += f"Сфокусируйся на: {', '.join(focus_areas)}\n\n"

        prompt += f"Общий тренд: {trend}\n\n"

        # Формат зависит от аудитории
        format_instructions = {
            "ceo": "Формат: executive summary 3-4 предложения + bullet points. Без технических деталей.",
            "finance": "Формат: таблица ключевых метрик + интерпретация отклонений. С цифрами.",
            "team": "Формат: что сделано + что не сделано + следующие шаги.",
        }
        prompt += format_instructions.get(audience, "Формат: структурированный markdown.")

        return prompt

    def detect_anomalies(self, metrics: dict) -> list[dict]:
        anomalies = []
        for key, values in metrics.items():
            if isinstance(values, list) and len(values) > 1:
                last = values[-1]
                prev = values[-2]
                if prev > 0 and abs(last - prev) / prev > 0.2:  # Изменение > 20%
                    anomalies.append({
                        "metric": key,
                        "change_pct": round((last - prev) / prev * 100, 1),
                    })
        return anomalies

Промпт-компилятор с валидацией

class PromptCompiler:
    """Компилирует промпт из компонентов с валидацией"""

    MAX_CONTEXT_TOKENS = 60000
    CHARS_PER_TOKEN = 4  # Приблизительно

    def compile(
        self,
        components: list[dict],  # [{"name": "...", "content": "...", "required": bool, "priority": int}]
        query: str,
    ) -> str:

        # Сортируем по приоритету
        sorted_components = sorted(components, key=lambda x: x.get("priority", 5))

        compiled_parts = []
        current_tokens = len(query) // self.CHARS_PER_TOKEN

        for component in sorted_components:
            content = component["content"]
            content_tokens = len(content) // self.CHARS_PER_TOKEN

            if current_tokens + content_tokens > self.MAX_CONTEXT_TOKENS:
                if component.get("required"):
                    # Обрезаем если обязательный
                    max_chars = (self.MAX_CONTEXT_TOKENS - current_tokens) * self.CHARS_PER_TOKEN
                    content = content[:max_chars] + "...[обрезано]"
                else:
                    # Пропускаем если опциональный
                    continue

            compiled_parts.append(f"## {component['name']}\n{content}")
            current_tokens += content_tokens

        compiled_parts.append(f"## Запрос\n{query}")
        return "\n\n".join(compiled_parts)

Практический кейс: персонализированный ассистент

Задача: корпоративный ассистент для 500 сотрудников разных отделов. Один статический промпт давал нерелевантные ответы для разных ролей.

Динамический промпт:

• При каждом запросе: профиль пользователя из LDAP → адаптация роли и уровня
• RAG: поиск по базе знаний → включение 3 релевантных фрагментов
• История: последние 4 сообщения → контекст диалога

Результат: оценка релевантности ответов: 61% (статический) → 84% (динамический).

Сроки

• Базовый динамический промпт (роль + контекст): 2–3 дня
• Интеграция с RAG и историей: 1 неделя
• Полная система с валидацией токенов: 2 недели