LLM Prompt Management: Управление шаблонами промптов в продакшене
prompt-managementllm-opstemplatesversioningtesting
Проблема управления промптами
Когда вы переходите от экспериментов к продакшену, промпты становятся таким же критическим компонентом системы, как и код. Но в отличие от кода, промпты:
- Меняются часто и непредсказуемо
- Зависят от данных и моделей
- Трудно тестируются
- Не имеют системы версионирования
Шаблонизация промптов
Jinja2 шаблоны
from jinja2 import Environment, FileSystemLoader
from typing import Dict, Any, Optional
import json
import hashlib
class PromptTemplate:
"""Шаблон промпта с поддержкой Jinja2"""
def __init__(self, name: str, template: str, variables: list):
self.name = name
self.template = template
self.variables = variables
self.env = Environment(
block_start_string='{%%',
block_end_string='%}',
variable_start_string='%%',
variable_end_string='%%'
)
self.jinja_template = self.env.from_string(template)
def render(self, **kwargs) -> str:
"""Рендеринг шаблона с данными"""
missing = set(self.variables) - set(kwargs.keys())
if missing:
raise ValueError(f"Missing required variables: {missing}")
return self.jinja_template.render(**kwargs)
def hash(self) -> str:
"""Хеш шаблона для версионирования"""
return hashlib.md5(self.template.encode()).hexdigest()[:8]
# Пример шаблона
classification_template = """
%%system%%
You are a sentiment analysis assistant. Analyze the sentiment of the following text.
%%end%%
%%user%%
Text: {{text}}
Language: {{language}}
Return your analysis in this JSON format:
{
"sentiment": "positive|negative|neutral",
"confidence": 0.0-1.0,
"reasoning": "brief explanation"
}
%%end%%
"""
template = PromptTemplate(
name="sentiment_analysis",
template=classification_template,
variables=["text", "language"]
)
prompt = template.render(
text="This product is absolutely amazing!",
language="en"
)
Файловая структура промптов
prompts/
├── v1/
│ ├── sentiment.yaml
│ ├── classification.yaml
│ └── extraction.yaml
├── v2/
│ ├── sentiment.yaml
│ └── classification.yaml
├── templates/
│ ├── system.jinja
│ ├── user.jinja
│ └── assistant.jinja
└── registry.yaml
YAML определение шаблона
# prompts/v2/sentiment.yaml
name: sentiment_analysis
version: "2.0"
template: |
%%system%%
You are a sentiment analysis assistant. Analyze the sentiment of the following text.
%%end%%
%%user%%
Text: {{text}}
Language: {{language}}
Return your analysis in this JSON format:
{
"sentiment": "positive|negative|neutral",
"confidence": 0.0-1.0,
"reasoning": "brief explanation"
}
%%end%%
variables:
- name: text
type: string
required: true
- name: language
type: string
required: true
default: "en"
metadata:
author: "team-ml"
created: "2026-05-01"
model: "Qwen2.5-7B-Instruct"
expected_tokens: 50
Registry и версионирование
import yaml
import json
from pathlib import Path
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from datetime import datetime
import copy
@dataclass
class PromptVersion:
"""Версия промпта"""
name: str
version: str
template: str
variables: List[Dict]
metadata: Dict = field(default_factory=dict)
created_at: str = field(default_factory=lambda: datetime.now().isoformat())
hash: str = ""
def __post_init__(self):
if not self.hash:
self.hash = hashlib.md5(self.template.encode()).hexdigest()[:8]
class PromptRegistry:
"""Registry для управления версиями промптов"""
def __init__(self, registry_path: str = "prompts/registry.yaml"):
self.registry_path = Path(registry_path)
self._registry = self._load_registry()
def _load_registry(self) -> Dict:
"""Загрузка registry из файла"""
if self.registry_path.exists():
with open(self.registry_path) as f:
return yaml.safe_load(f) or {}
return {"prompts": {}, "versions": {}}
def register_prompt(
self,
name: str,
version: str,
template: str,
variables: List[Dict],
metadata: Dict = None
) -> PromptVersion:
"""Регистрация новой версии промпта"""
version_obj = PromptVersion(
name=name,
version=version,
template=template,
variables=variables,
metadata=metadata or {}
)
if name not in self._registry["prompts"]:
self._registry["prompts"][name] = []
self._registry["prompts"][name].append({
"version": version,
"hash": version_obj.hash,
"created_at": version_obj.created_at,
"metadata": version_obj.metadata
})
self._registry["versions"][f"{name}:{version}"] = version_obj
self._save_registry()
return version_obj
def get_latest(self, name: str) -> Optional[PromptVersion]:
"""Получить последнюю версию промпта"""
if name not in self._registry["prompts"]:
return None
versions = self._registry["prompts"][name]
latest = max(versions, key=lambda v: v["created_at"])
key = f"{name}:{latest['version']}"
return self._registry["versions"].get(key)
def get_version(self, name: str, version: str) -> Optional[PromptVersion]:
"""Получить конкретную версию промпта"""
key = f"{name}:{version}"
return self._registry["versions"].get(key)
def list_prompts(self) -> List[str]:
"""Список всех промптов"""
return list(self._registry["prompts"].keys())
def list_versions(self, name: str) -> List[Dict]:
"""Список версий промпта"""
return self._registry["prompts"].get(name, [])
def _save_registry(self):
"""Сохранение registry в файл"""
with open(self.registry_path, 'w') as f:
yaml.dump(self._registry, f, default_flow_style=False)
# Использование
registry = PromptRegistry()
registry.register_prompt(
name="sentiment_analysis",
version="1.0",
template="...",
variables=[{"name": "text", "type": "string"}],
metadata={"author": "team-ml"}
)
Тестирование промптов
Unit тесты
import unittest
from typing import List, Dict
import json
class PromptTestRunner:
"""Тестирование промптов"""
def __init__(self, template: PromptTemplate):
self.template = template
self.tests: List[Dict] = []
def add_test(
self,
name: str,
inputs: Dict,
expected_patterns: List[str] = None,
expected_not_patterns: List[str] = None,
expected_json_keys: List[str] = None,
max_tokens: int = None
):
"""Добавление теста"""
self.tests.append({
"name": name,
"inputs": inputs,
"expected_patterns": expected_patterns or [],
"expected_not_patterns": expected_not_patterns or [],
"expected_json_keys": expected_json_keys,
"max_tokens": max_tokens
})
def run_all(self) -> Dict:
"""Запуск всех тестов"""
results = []
all_passed = True
for test in self.tests:
passed = True
errors = []
# Рендеринг
try:
rendered = self.template.render(**test["inputs"])
except Exception as e:
results.append({
"test": test["name"],
"passed": False,
"error": str(e)
})
all_passed = False
continue
# Проверка паттернов
for pattern in test["expected_patterns"]:
if pattern not in rendered:
errors.append(f"Missing pattern: {pattern}")
passed = False
# Проверка отсутствия паттернов
for pattern in test["expected_not_patterns"]:
if pattern in rendered:
errors.append(f"Unexpected pattern found: {pattern}")
passed = False
# Проверка JSON структуры
if test["expected_json_keys"]:
try:
# Извлекаем JSON из промпта (упрощённо)
json_start = rendered.find('{')
if json_start >= 0:
json_str = rendered[json_start:]
parsed = json.loads(json_str)
for key in test["expected_json_keys"]:
if key not in parsed:
errors.append(f"Missing JSON key: {key}")
passed = False
except json.JSONDecodeError as e:
errors.append(f"Invalid JSON: {str(e)}")
passed = False
# Проверка токенов
if test["max_tokens"]:
token_count = len(rendered.split()) * 1.3 # approximate
if token_count > test["max_tokens"]:
errors.append(f"Too many tokens: {token_count:.0f} > {test['max_tokens']}")
passed = False
results.append({
"test": test["name"],
"passed": passed,
"errors": errors,
"rendered_preview": rendered[:200]
})
if not passed:
all_passed = False
return {
"total": len(self.tests),
"passed": sum(1 for r in results if r["passed"]),
"failed": sum(1 for r in results if not r["passed"]),
"results": results,
"all_passed": all_passed
}
# Пример использования
template = PromptTemplate(
name="sentiment",
template=classification_template,
variables=["text", "language"]
)
runner = PromptTestRunner(template)
runner.add_test(
name="basic_positive",
inputs={"text": "I love this!", "language": "en"},
expected_patterns=["I love this!"],
expected_json_keys=["sentiment", "confidence", "reasoning"]
)
runner.add_test(
name="missing_variable",
inputs={"text": "Hello"},
expected_patterns=[],
expected_not_patterns=[]
)
results = runner.run_all()
print(f"Passed: {results['passed']}/{results['total']}")
A/B тестирование промптов
import random
from collections import defaultdict
class PromptABTester:
"""A/B тестирование промптов"""
def __init__(self, registry: PromptRegistry):
self.registry = registry
self.results = defaultdict(lambda: {"scores": [], "count": 0})
def evaluate_prompt(
self,
prompt_key: str,
rendered_prompt: str,
model_response: str,
ground_truth: str = None,
human_score: float = None
):
"""Оценка промпта"""
score = 0.0
# Автоматическая оценка
if ground_truth:
score = self._compute_similarity(model_response, ground_truth)
# Сохранение результата
self.results[prompt_key]["scores"].append(score)
self.results[prompt_key]["count"] += 1
def _compute_similarity(self, predicted: str, ground_truth: str) -> float:
"""Вычисление similarity с ground truth"""
# Можно использовать BLEU, ROUGE, или embedding similarity
pred_words = set(predicted.lower().split())
truth_words = set(ground_truth.lower().split())
if not pred_words or not truth_words:
return 0.0
intersection = pred_words & truth_words
return len(intersection) / len(pred_words | truth_words)
def get_leaderboard(self) -> List[Dict]:
"""Таблица лидеров"""
leaderboard = []
for key, data in self.results.items():
scores = data["scores"]
avg_score = sum(scores) / len(scores) if scores else 0
leaderboard.append({
"prompt": key,
"samples": data["count"],
"avg_score": avg_score,
"min_score": min(scores) if scores else 0,
"max_score": max(scores) if scores else 0
})
leaderboard.sort(key=lambda x: x["avg_score"], reverse=True)
return leaderboard
def is_significant(
self,
prompt_a: str,
prompt_b: str,
alpha: float = 0.05
) -> bool:
"""Проверка статистической значимости (упрощённо)"""
scores_a = self.results[prompt_a]["scores"]
scores_b = self.results[prompt_b]["scores"]
if not scores_a or not scores_b:
return False
mean_a = sum(scores_a) / len(scores_a)
mean_b = sum(scores_b) / len(scores_b)
# Простая проверка разницы
diff = abs(mean_a - mean_b)
pooled_std = self._pooled_std(scores_a, scores_b)
if pooled_std == 0:
return False
# Cohen's d
effect_size = diff / pooled_std
return effect_size > 0.5 # medium effect
def _pooled_std(self, scores_a: List[float], scores_b: List[float]) -> float:
"""Пулы стандартное отклонение"""
n_a, n_b = len(scores_a), len(scores_b)
if n_a < 2 or n_b < 2:
return 1.0
mean_a = sum(scores_a) / n_a
mean_b = sum(scores_b) / n_b
var_a = sum((x - mean_a) ** 2 for x in scores_a) / (n_a - 1)
var_b = sum((x - mean_b) ** 2 for x in scores_b) / (n_b - 1)
pooled_var = ((n_a - 1) * var_a + (n_b - 1) * var_b) / (n_a + n_b - 2)
return pooled_var ** 0.5
Продвинутые техники
Dynamic Few-Shot Prompting
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np
class DynamicFewShot:
"""Динамические примеры в промпте"""
def __init__(self, examples: List[Dict], embedder=None):
"""
examples: [{"input": "...", "output": "..."}]
"""
self.examples = examples
self.embedder = embedder
self.example_embeddings = None
if self.embedder:
self._compute_embeddings()
def _compute_embeddings(self):
"""Вычисление эмбеддингов примеров"""
texts = [ex["input"] for ex in self.examples]
self.example_embeddings = self.embedder.embed_documents(texts)
def get_few_shot_examples(self, query: str, n: int = 3) -> List[Dict]:
"""Получение наиболее релевантных примеров"""
if self.example_embeddings is None:
# Если нет embedder, берём случайные
return random.sample(self.examples, min(n, len(self.examples)))
query_embedding = self.embedder.embed_query(query)
# Вычисление similarity
sims = cosine_similarity(
query_embedding.reshape(1, -1),
self.example_embeddings
)[0]
# Top-k
top_indices = np.argsort(sims)[-n:][::-1]
return [self.examples[i] for i in top_indices]
def build_prompt(self, query: str, template: str, n_examples: int = 3) -> str:
"""Построение промпта с динамическими примерами"""
examples = self.get_few_shot_examples(query, n_examples)
examples_text = "\n\n".join([
f"Input: {ex['input']}\nOutput: {ex['output']}"
for ex in examples
])
return template.replace("%%FEW_SHOT%%", examples_text).replace("%%QUERY%%", query)
# Пример
examples = [
{"input": "I love this product!", "output": '{"sentiment": "positive", "confidence": 0.95}'},
{"input": "Terrible experience.", "output": '{"sentiment": "negative", "confidence": 0.88}'},
{"input": "It is okay.", "output": '{"sentiment": "neutral", "confidence": 0.72}'}
]
few_shot = DynamicFewShot(examples, embedder=embedding_service)
prompt = few_shot.build_prompt("This is the best thing ever!", n_examples=2)
Prompt Chaining
from typing import Callable, Any
import asyncio
class PromptChain:
"""Цепочка промптов"""
def __init__(self, name: str):
self.name = name
self.steps: List[Dict] = []
def add_step(
self,
name: str,
template: PromptTemplate,
parser: Callable[[str], Any] = None,
condition: Callable[[Dict], bool] = None
):
"""Добавление шага в цепочку"""
self.steps.append({
"name": name,
"template": template,
"parser": parser,
"condition": condition
})
async def execute(self, initial_data: Dict) -> Dict:
"""Выполнение цепочки"""
context = initial_data.copy()
for i, step in enumerate(self.steps):
# Проверка условия
if step["condition"] and not step["condition"](context):
context[f"step_{i}_skipped"] = True
continue
# Рендеринг промпта
prompt = step["template"].render(**context)
# Вызов модели (асинхронно)
response = await call_llm(prompt)
# Парсинг ответа
if step["parser"]:
context[step["name"]] = step["parser"](response)
else:
context[step["name"]] = response
return context
async def call_llm(prompt: str) -> str:
"""Вызов LLM"""
# Реализация зависит от вашего LLM сервера
pass
# Пример цепочки для анализа
chain = PromptChain("document_analysis")
chain.add_step(
"extract_entities",
entity_template,
parser=lambda x: json.loads(x)
)
chain.add_step(
"classify_topics",
topic_template,
parser=lambda x: json.loads(x),
condition=lambda ctx: "extract_entities" in ctx
)
chain.add_step(
"generate_summary",
summary_template,
condition=lambda ctx: "classify_topics" in ctx
)
result = await chain.execute({"document": "..."})
Prompt Caching
import hashlib
import time
from functools import lru_cache
from typing import Dict, Any
class PromptCache:
"""Кэш для рендеренных промптов"""
def __init__(self, max_size: int = 1000, ttl: int = 3600):
self.max_size = max_size
self.ttl = ttl
self._cache: Dict[str, tuple] = {}
def _make_key(self, template_hash: str, inputs: Dict) -> str:
"""Создание ключа кэша"""
inputs_str = json.dumps(inputs, sort_keys=True)
return hashlib.md5(f"{template_hash}:{inputs_str}".encode()).hexdigest()
def get(self, template_hash: str, inputs: Dict) -> Optional[str]:
"""Получение из кэша"""
key = self._make_key(template_hash, inputs)
if key not in self._cache:
return None
cached_time, cached_value = self._cache[key]
# Проверка TTL
if time.time() - cached_time > self.ttl:
del self._cache[key]
return None
return cached_value
def set(self, template_hash: str, inputs: Dict, value: str):
"""Сохранение в кэш"""
if len(self._cache) >= self.max_size:
# Удаляем самый старый
oldest_key = min(self._cache, key=lambda k: self._cache[k][0])
del self._cache[oldest_key]
key = self._make_key(template_hash, inputs)
self._cache[key] = (time.time(), value)
def stats(self) -> Dict:
"""Статистика кэша"""
return {
"size": len(self._cache),
"max_size": self.max_size,
"hit_rate": self._calculate_hit_rate()
}
# Декоратор для кэширования
def cached_prompt(cache: PromptCache):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
template_hash = args[0].hash if hasattr(args[0], 'hash') else id(args[0])
cached = cache.get(template_hash, kwargs)
if cached:
return cached
result = func(*args, **kwargs)
cache.set(template_hash, kwargs, result)
return result
return wrapper
return decorator
Мониторинг промптов
Логирование
import logging
from datetime import datetime
class PromptLogger:
"""Логирование промптов для мониторинга"""
def __init__(self, log_path: str = "prompts.log"):
self.logger = logging.getLogger("prompt_logger")
self.logger.setLevel(logging.INFO)
handler = logging.FileHandler(log_path)
formatter = logging.Formatter('%(asctime)s - %(message)s')
handler.setFormatter(formatter)
self.logger.addHandler(handler)
def log_prompt(
self,
template_name: str,
version: str,
inputs: Dict,
rendered: str,
model_response: str = None,
latency_ms: float = None,
tokens_used: int = None
):
"""Логирование промпта"""
log_entry = {
"timestamp": datetime.now().isoformat(),
"template": template_name,
"version": version,
"inputs": {k: str(v)[:100] for k, v in inputs.items()},
"rendered_length": len(rendered),
"rendered_tokens": len(rendered.split()),
}
if model_response:
log_entry["response_length"] = len(model_response)
log_entry["response_tokens"] = len(model_response.split())
if latency_ms:
log_entry["latency_ms"] = latency_ms
if tokens_used:
log_entry["tokens_used"] = tokens_used
self.logger.info(json.dumps(log_entry))
# Мониторинг аномалий
class PromptAnomalyDetector:
"""Обнаружение аномалий в промптах"""
def __init__(self, baseline_token_count: int, baseline_length: int):
self.baseline_tokens = baseline_token_count
self.baseline_length = baseline_length
self.threshold = 2.0 # standard deviations
def check_anomaly(self, token_count: int, length: int) -> Dict:
"""Проверка на аномалию"""
token_deviation = abs(token_count - self.baseline_tokens) / max(self.baseline_tokens, 1)
length_deviation = abs(length - self.baseline_length) / max(self.baseline_length, 1)
is_anomaly = token_deviation > self.threshold or length_deviation > self.threshold
return {
"is_anomaly": is_anomaly,
"token_deviation": token_deviation,
"length_deviation": length_deviation,
"token_count": token_count,
"length": length
}
Инструменты и экосистема
| Инструмент | Описание | Ссылка |
|---|---|---|
| Promptfoo | Тестирование промптов | github.com/anthropics/promptfoo |
| LangChain Prompt Management | Встроенный management | langchain.com |
| PromptLayer | Трекинг промптов | promptlayer.com |
| Braintrust | Оценка промптов | braintrustdata.com |
| Weights & Biases | Мониторинг экспериментов | wandb.ai |
Заключение
Управление промптами — это не просто шаблонизация. Это система с версионированием, тестированием, A/B тестами и мониторингом. Хороший prompt management значительно повышает надёжность LLM-приложений в продакшене.