Проблема: почему тестирование LLM сводит с ума
Вы только что реализовали красивую систему вызова функций для вашего LLM-агента. Он должен анализировать запросы пользователей, определять нужные инструменты и вызывать их с правильными параметрами. Вы запускаете тест: "Забронируй столик в ресторане на 4 человека на 19:00". Первый запуск — идеально. Второй — параметры перепутаны. Третий — модель решила, что нужно вызвать совсем другую функцию.
Ключевое отличие: Традиционное тестирование проверяет правильность (получили ли мы ожидаемый результат). Тестирование LLM проверяет стабильность (насколько последовательно мы получаем корректный результат).
Это классическая проблема недетерминизма в LLM. В отличие от традиционного ПО, где один и тот же вход всегда даёт одинаковый выход, языковые модели генерируют разные ответы даже при одинаковых промптах и параметрах. Особенно критично это становится при работе с функциональными вызовами (tool calling), где ошибка может привести к реальным последствиям: неправильным бронированиям, ошибочным транзакциям или некорректным данным в базе.
Стратегия: что мы на самом деле тестируем
Прежде чем писать тесты, нужно понять, какие метрики нас интересуют:
- Стабильность выбора функции — насколько последовательно модель выбирает правильный инструмент
- Консистентность параметров — насколько стабильно заполняются аргументы функции
- Семантическая корректность — соответствуют ли выбранные функции и параметры намерению пользователя
- Стоимость тестирования — сколько денег уходит на прогон тестов (особенно важно для облачных моделей)
- Латентность — как быстро модель принимает решение о вызове функции
💡
В нашей предыдущей статье "Обзор лучших LLM с поддержкой Tool Calling" мы рассматривали различные модели, которые можно использовать для функциональных вызовов. Тестирование поможет выбрать наиболее стабильную из них для ваших задач.
Шаг 1: Генерация тестовых данных
Первая ошибка начинающих — тестировать на 5-10 примерах. Для недетерминированных систем нужны сотни тестовых сценариев. Вот как их генерировать:
1 Определите домен и вариации
Если ваш агент работает с ресторанами, определите все возможные типы запросов:
test_domains = {
'booking': [
'Забронируй столик на 2 персоны на 19:00',
'Нужен столик на 4 человека в пятницу вечером',
'Хочу забронировать место в ресторане на завтра в 20:30',
# 50+ вариаций
],
'menu': [
'Какие вегетарианские блюда есть в меню?',
'Покажи десерты с фото',
'Есть ли детское меню?',
# 30+ вариаций
],
'reviews': [
'Покажи отзывы о ресторане',
'Какая средняя оценка у этого места?',
# 20+ вариаций
]
}2 Добавьте краевые случаи
Краевые случаи — это запросы, которые могут сломать вашу систему:
edge_cases = [
# Неполная информация
'Забронируй столик', # нет времени и количества
'На завтра', # нет типа действия
# Противоречивые запросы
'Забронируй столик на -1 человека', # некорректное количество
'На 25:00', # некорректное время
# Мульти-интенты
'Забронируй столик и покажи меню', # два действия сразу
# Абстрактные запросы
'Я голоден, что делать?', # неявный интент
# Опечатки и сленг
'Забронируй сталик', # опечатка
'Столик на вечерок', # сленг
]3 Используйте LLM для генерации тестов
Лучший способ сгенерировать разнообразные тесты — использовать саму LLM:
import openai
from typing import List
def generate_test_cases(
function_descriptions: List[str],
num_cases: int = 100
) -> List[str]:
"""
Генерирует тестовые запросы для заданных функций
"""
prompt = f"""Ты — генератор тестовых данных для LLM-агента.
У агента есть следующие функции:
{chr(10).join(function_descriptions)}
Сгенерируй {num_cases} разнообразных пользовательских запросов,
которые могут привести к вызову этих функций.
Запросы должны быть:
1. Естественными и разнообразными
2. Включать краевые случаи (10%)
3. Иметь разную сложность
4. Содержать опечатки и разговорную речь (15%)
5. Включать неполные запросы (10%)
Верни только список запросов, каждый с новой строки.
"""
response = openai.ChatCompletion.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
temperature=0.8, # Добавляем разнообразия
)
return response.choices[0].message.content.split('\n')Важно: Сохраняйте сгенерированные тесты в версионируемом хранилище. Они станут вашим регрессионным тест-сьютом. При изменении промптов или функций вы сможете проверить, не ухудшилась ли стабильность.
Шаг 2: Архитектура тестовой системы
Тестирование LLM требует особой архитектуры. Вот минимальная рабочая система:
import json
import asyncio
from dataclasses import dataclass
from typing import Dict, Any, List, Optional
from datetime import datetime
import statistics
@dataclass
class TestResult:
"""Результат одного теста"""
test_id: str
user_query: str
expected_function: Optional[str] # Ожидаемая функция
expected_params: Optional[Dict] # Ожидаемые параметры
actual_function: Optional[str] # Фактически вызванная функция
actual_params: Optional[Dict] # Фактические параметры
is_correct: bool # Корректность вызова
confidence: float # Уверенность модели (если доступно)
latency_ms: float # Время ответа
cost: float # Стоимость запроса
timestamp: datetime # Время выполнения
raw_response: Dict[str, Any] # Полный ответ LLM
class LLMTestSuite:
"""Основной класс для тестирования LLM"""
def __init__(self, llm_client, functions: List[Dict]):
self.llm = llm_client
self.functions = functions
self.results = []
async def run_single_test(
self,
user_query: str,
expected_function: Optional[str] = None,
expected_params: Optional[Dict] = None
) -> TestResult:
"""Запускает один тест"""
start_time = datetime.now()
try:
# Вызов LLM с функциями
response = await self.llm.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": user_query}],
functions=self.functions,
function_call="auto",
)
latency = (datetime.now() - start_time).total_seconds() * 1000
# Извлечение вызова функции
message = response.choices[0].message
called_function = None
called_params = None
if hasattr(message, 'function_call') and message.function_call:
called_function = message.function_call.name
called_params = json.loads(message.function_call.arguments)
# Проверка корректности
is_correct = True
if expected_function:
is_correct = (called_function == expected_function)
if expected_params and called_params:
# Сравнение параметров с допуском
is_correct = is_correct and self._compare_params(
expected_params,
called_params
)
return TestResult(
test_id=f"test_{len(self.results)}",
user_query=user_query,
expected_function=expected_function,
expected_params=expected_params,
actual_function=called_function,
actual_params=called_params,
is_correct=is_correct,
confidence=getattr(message, 'confidence', 0.0),
latency_ms=latency,
cost=self._calculate_cost(response),
timestamp=datetime.now(),
raw_response=response.dict()
)
except Exception as e:
# Обработка ошибок
return TestResult(
test_id=f"test_{len(self.results)}_error",
user_query=user_query,
expected_function=expected_function,
expected_params=expected_params,
actual_function=None,
actual_params=None,
is_correct=False,
confidence=0.0,
latency_ms=(datetime.now() - start_time).total_seconds() * 1000,
cost=0.0,
timestamp=datetime.now(),
raw_response={"error": str(e)}
)
def _compare_params(self, expected: Dict, actual: Dict) -> bool:
"""Сравнивает параметры с семантической толерантностью"""
# Здесь реализуйте логику сравнения
# Например, нормализация времени, чисел и т.д.
return expected == actual # Упрощённая версия
def _calculate_cost(self, response) -> float:
"""Рассчитывает стоимость запроса"""
# Реализуйте расчёт на основе токенов
return 0.01 # ПримерШаг 3: Метрики и анализ стабильности
Самый важный этап — правильные метрики. Точность (accuracy) здесь недостаточна.
| Метрика | Формула/Описание | Целевое значение |
|---|---|---|
| Стабильность выбора функции | Процент одинаковых выборов при N запусках | >95% (для критичных функций) |
| Консистентность параметров | Стандартное отклонение значений параметров | < 5% от диапазона |
| Успешность парсинга | % запросов, где функция вызвана корректно | >90% |
| Стоимость ошибки | Средняя стоимость неправильных вызовов | Минимизировать |
| Латентность P95 | 95-й перцентиль времени ответа | < 2 секунды |
4 Реализация анализа стабильности
class StabilityAnalyzer:
"""Анализатор стабильности LLM"""
def __init__(self, test_results: List[TestResult]):
self.results = test_results
def calculate_stability_metrics(self) -> Dict[str, Any]:
"""Рассчитывает все метрики стабильности"""
metrics = {
'function_selection_stability': self._calculate_function_stability(),
'parameter_consistency': self._calculate_parameter_consistency(),
'parsing_success_rate': self._calculate_parsing_rate(),
'cost_analysis': self._calculate_cost_analysis(),
'latency_metrics': self._calculate_latency_metrics(),
'error_analysis': self._analyze_errors(),
}
return metrics
def _calculate_function_stability(self) -> Dict:
"""Рассчитывает стабильность выбора функции"""
# Группируем результаты по запросам
query_groups = {}
for result in self.results:
if result.user_query not in query_groups:
query_groups[result.user_query] = []
query_groups[result.user_query].append(result.actual_function)
# Для каждого запроса считаем консистентность
consistency_scores = []
for query, functions in query_groups.items():
if len(functions) > 1:
# Сколько раз была выбрана самая популярная функция
most_common = max(set(functions), key=functions.count)
consistency = functions.count(most_common) / len(functions)
consistency_scores.append(consistency)
return {
'average_consistency': statistics.mean(consistency_scores) if consistency_scores else 1.0,
'min_consistency': min(consistency_scores) if consistency_scores else 1.0,
'problematic_queries': self._find_problematic_queries(query_groups),
}
def _calculate_parameter_consistency(self) -> Dict:
"""Анализирует консистентность параметров"""
# Группируем по функциям и параметрам
param_variations = {}
for result in self.results:
if result.actual_function and result.actual_params:
key = f"{result.user_query}->{result.actual_function}"
if key not in param_variations:
param_variations[key] = []
param_variations[key].append(result.actual_params)
# Анализируем вариативность каждого параметра
param_stability = {}
for key, param_list in param_variations.items():
if len(param_list) > 1:
# Для каждого параметра считаем вариативность
param_names = set()
for params in param_list:
param_names.update(params.keys())
for param_name in param_names:
values = [p.get(param_name) for p in param_list if param_name in p]
if len(values) > 1:
# Простой анализ: одинаковы ли значения
unique_values = len(set(str(v) for v in values))
stability = 1.0 / unique_values # 1.0 если все одинаковые
if key not in param_stability:
param_stability[key] = {}
param_stability[key][param_name] = {
'stability': stability,
'unique_values': unique_values,
'values': values[:5], # Первые 5 значений
}
return param_stability
def _calculate_parsing_rate(self) -> float:
"""Процент успешных парсингов"""
successful = sum(1 for r in self.results if r.actual_function is not None)
return successful / len(self.results) if self.results else 0.0
def _calculate_cost_analysis(self) -> Dict:
"""Анализ стоимости"""
total_cost = sum(r.cost for r in self.results)
error_cost = sum(r.cost for r in self.results if not r.is_correct)
return {
'total_cost': total_cost,
'error_cost': error_cost,
'error_cost_percentage': (error_cost / total_cost * 100) if total_cost > 0 else 0,
'cost_per_query': total_cost / len(self.results) if self.results else 0,
}
def _calculate_latency_metrics(self) -> Dict:
"""Метрики латентности"""
latencies = [r.latency_ms for r in self.results]
if not latencies:
return {}
return {
'p50': statistics.median(latencies),
'p95': sorted(latencies)[int(len(latencies) * 0.95)],
'p99': sorted(latencies)[int(len(latencies) * 0.99)],
'average': statistics.mean(latencies),
'std_dev': statistics.stdev(latencies) if len(latencies) > 1 else 0,
}
def _analyze_errors(self) -> List[Dict]:
"""Анализ ошибок"""
errors = []
for result in self.results:
if not result.is_correct:
error_info = {
'query': result.user_query,
'expected': {
'function': result.expected_function,
'params': result.expected_params,
},
'actual': {
'function': result.actual_function,
'params': result.actual_params,
},
'confidence': result.confidence,
'latency': result.latency_ms,
}
errors.append(error_info)
return errorsШаг 4: Автоматизация и CI/CD
Тестирование LLM должно быть частью вашего пайплайна разработки. Вот как это организовать:
Интеграция с pytest
# test_llm_functions.py
import pytest
import asyncio
from typing import Dict, List
class TestLLMFunctions:
"""Тесты для функциональных вызовов LLM"""
@pytest.fixture(scope="module")
def test_suite(self):
"""Инициализация тестового набора"""
return LLMTestSuite(
llm_client=get_llm_client(),
functions=load_functions()
)
@pytest.mark.asyncio
@pytest.mark.parametrize("query,expected_function,expected_params", [
("Забронируй столик на 2 человека на 19:00", "book_table", {"guests": 2, "time": "19:00"}),
("Покажи меню ресторана", "get_menu", {}),
("Какие отзывы у этого места?", "get_reviews", {}),
])
async def test_function_calling(
self,
test_suite,
query: str,
expected_function: str,
expected_params: Dict
):
"""Тестирует вызов конкретной функции"""
# Запускаем тест несколько раз для проверки стабильности
results = []
for _ in range(5): # 5 запусков для проверки стабильности
result = await test_suite.run_single_test(
query,
expected_function,
expected_params
)
results.append(result)
# Проверяем, что хотя бы 4 из 5 запусков корректны
correct_count = sum(1 for r in results if r.is_correct)
assert correct_count >= 4, f"Стабильность ниже 80%: {correct_count}/5"
# Проверяем латентность
avg_latency = sum(r.latency_ms for r in results) / len(results)
assert avg_latency < 2000, f"Средняя латентность {avg_latency}ms превышает 2 секунды"
@pytest.mark.asyncio
async def test_function_stability(self, test_suite):
"""Тест на стабильность выбора функции"""
# Используем один запрос, запускаем 10 раз
query = "Забронируй столик на 4 человека в пятницу вечером"
results = []
for _ in range(10):
result = await test_suite.run_single_test(query)
results.append(result)
# Анализируем стабильность
functions_called = [r.actual_function for r in results if r.actual_function]
if functions_called:
most_common = max(set(functions_called), key=functions_called.count)
consistency = functions_called.count(most_common) / len(functions_called)
assert consistency >= 0.9, f"Стабильность выбора функции {consistency*100}% < 90%"
@pytest.mark.asyncio
async def test_cost_control(self, test_suite):
"""Тест на контроль стоимости"""
# Запускаем 100 тестов и проверяем стоимость
test_queries = load_test_queries()[:100] # Первые 100 запросов
total_cost = 0
for query in test_queries:
result = await test_suite.run_single_test(query)
total_cost += result.cost
# Проверяем, что стоимость не превышает бюджет
expected_max_cost = 5.0 # $5 за 100 запросов
assert total_cost <= expected_max_cost, f"Стоимость {total_cost} превышает бюджет {expected_max_cost}"
# GitHub Actions workflow для автоматического тестирования
# .github/workflows/test-llm.yml
yaml_content = """
name: LLM Tests
on:
push:
branches: [ main, develop ]
pull_request:
branches: [ main ]
schedule:
- cron: '0 0 * * *' # Ежедневно в полночь
jobs:
test-llm:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.10'
- name: Install dependencies
run: |
pip install -r requirements.txt
pip install pytest pytest-asyncio
- name: Run LLM tests
env:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
run: |
pytest test_llm_functions.py -v --tb=short
- name: Generate stability report
run: |
python generate_stability_report.py
- name: Upload test results
uses: actions/upload-artifact@v3
with:
name: llm-test-results
path: test_results/
retention-days: 30
"""💡
Для более глубокого понимания проблем с недетерминизмом рекомендую статью "Interpretation Drift: почему ваша LLM сегодня отвечает иначе, чем вчера", где мы разбираем причины изменения поведения моделей со временем.
Шаг 5: Продвинутые техники
Семантическое сравнение параметров
Прямое сравнение параметров часто не работает. "19:00" и "7 вечера" — это одно и то же время для человека, но разные строки для компьютера.
from datetime import datetime
import re
from typing import Any, Dict
def normalize_parameters(params: Dict[str, Any]) -> Dict[str, Any]:
"""Нормализует параметры для сравнения"""
normalized = {}
for key, value in params.items():
if isinstance(value, str):
# Нормализация времени
if 'time' in key.lower() or 'hour' in key.lower():
normalized[key] = normalize_time(value)
# Нормализация чисел
elif value.isdigit():
normalized[key] = int(value)
# Нормализация дат
elif looks_like_date(value):
normalized[key] = normalize_date(value)
else:
normalized[key] = value.lower().strip()
else:
normalized[key] = value
return normalized
def normalize_time(time_str: str) -> str:
"""Приводит время к формату HH:MM"""
try:
# Пытаемся распарсить различные форматы
formats = ['%H:%M', '%I:%M %p', '%I %p', '%H часов', '%H:00']
for fmt in formats:
try:
dt = datetime.strptime(time_str.lower(), fmt)
return dt.strftime('%H:%M')
except ValueError:
continue
# Если не получилось, возвращаем как есть
return time_str
except:
return time_str
def compare_params_semantically(expected: Dict, actual: Dict, tolerance: float = 0.1) -> bool:
"""Сравнивает параметры с семантической толерантностью"""
expected_norm = normalize_parameters(expected)
actual_norm = normalize_parameters(actual)
# Сравниваем ключи
if set(expected_norm.keys()) != set(actual_norm.keys()):
return False
# Сравниваем значения
for key in expected_norm.keys():
exp_val = expected_norm[key]
act_val = actual_norm[key]
if isinstance(exp_val, (int, float)) and isinstance(act_val, (int, float)):
# Числовое сравнение с допуском
if abs(exp_val - act_val) > tolerance * max(abs(exp_val), abs(act_val)):
return False
elif isinstance(exp_val, str) and isinstance(act_val, str):
# Строковое сравнение (можно добавить semantic similarity)
if exp_val != act_val:
# Проверяем синонимы
if not are_synonyms(exp_val, act_val):
return False
else:
# Точное сравнение для остальных типов
if exp_val != act_val:
return False
return TrueА/Б тестирование промптов
Разные промпты могут давать разную стабильность. Автоматизируйте поиск оптимального промпта:
class PromptABTester:
"""А/Б тестирование промптов"""
def __init__(self, llm_client, functions: List[Dict]):
self.llm = llm_client
self.functions = functions
async def test_prompt_variants(self, test_queries: List[str], prompt_variants: List[str]) -> Dict:
"""Тестирует разные варианты промптов"""
results = {}
for prompt in prompt_variants:
print(f"Testing prompt: {prompt[:50]}...")
variant_results = []
for query in test_queries[:50]: # Тестируем на 50 запросах
full_prompt = f"{prompt}\n\nUser: {query}"
result = await self.llm.chat.completions.create(
model="gpt-4",
messages=[{"role": "system", "content": full_prompt}],
functions=self.functions,
)
variant_results.append(result)
# Анализируем результаты
analyzer = StabilityAnalyzer(variant_results)
metrics = analyzer.calculate_stability_metrics()
results[prompt] = {
'metrics': metrics,
'sample_results': variant_results[:5],
}
return results
# Примеры промптов для тестирования
prompt_variants = [
"Ты — помощник по бронированию ресторанов. Анализируй запросы пользователей и вызывай соответствующие функции.",
"Ты — AI-ассистент. Твоя задача — понимать запросы пользователей и вызывать подходящие функции. Будь точным и последовательным.",
"Система: Ты обрабатываешь запросы пользователей. Определи интент и вызови соответствующую функцию. Если не уверен — уточни.",
# Добавьте больше вариантов
]Распространённые ошибки и как их избежать
| Ошибка | Последствия | Решение |
|---|---|---|
| Тестирование только на "идеальных" запросах | Система работает в вакууме, ломается на реальных данных | Добавляйте краевые случаи, опечатки, неполные запросы |
| Один запуск на тест | Не видите недетерминизм, ложное чувство стабильности | Запускайте каждый тест 5-10 раз, считайте консистентность |
| Игнорирование стоимости тестов | Тесты становятся слишком дорогими, запускаются редко | Используйте лимиты, кэширование, локальные модели для разработки |
| Бинарная оценка (правильно/неправильно) | Теряете нюансы, не видите постепенной деградации | Вводите оценку от 0 до 1, учитывайте частичную правильность |
| Тестирование без контекста | Модель работает иначе в диалоге, чем на единичных запросах | Тестируйте в контексте диалога, с историей сообщений |
Практический пример: полный пайплайн тестирования
Вот как выглядит end-to-end пайплайн тестирования для продакшн-системы:
# complete_test_pipeline.py
import asyncio
import json
from datetime import datetime, timedelta
from typing import Dict, List
import pandas as pd
class LLMTestingPipeline:
"""Полный пайплайн тестирования LLM"""
def __init__(self, config: Dict):
self.config = config
self.test_suite = LLMTestSuite(
llm_client=self._get_llm_client(),
functions=self._load_functions()
)
async def run_full_pipeline(self):
"""Запускает полный цикл тестирования"""
print("🚀 Starting LLM testing pipeline...")
# 1. Загрузка тестовых данных
test_queries = self._load_test_queries()
print(f"📊 Loaded {len(test_queries)} test queries")
# 2. Запуск тестов
print("⚡ Running tests...")
all_results = []
# Батчинг для эффективности
batch_size = self.config.get('batch_size', 10)
for i in range(0, len(test_queries), batch_size):
batch = test_queries[i:i + batch_size]
# Запускаем параллельно
tasks = [
self.test_suite.run_single_test(query)
for query in batch
]
batch_results = await asyncio.gather(*tasks, return_exceptions=True)
# Фильтруем успешные результаты
for result in batch_results:
if isinstance(result, TestResult):
all_results.append(result)
else:
print(f"❌ Error in batch: {result}")
print(f"✅ Processed {min(i + batch_size, len(test_queries))}/{len(test_queries)} queries")
# 3. Анализ результатов
print("📈 Analyzing results...")
analyzer = StabilityAnalyzer(all_results)
metrics = analyzer.calculate_stability_metrics()
# 4. Генерация отчёта
report = self._generate_report(metrics, all_results)
# 5. Сохранение результатов
self._save_results(all_results, metrics, report)
# 6. Проверка качества
passed = self._check_quality_gates(metrics)
print(f"🎯 Pipeline completed. Quality gates: {"PASSED" if passed else "FAILED"}")
return {
'passed': passed,
'metrics': metrics,
'report': report,
'total_tests': len(all_results),
}
def _generate_report(self, metrics: Dict, results: List[TestResult]) -> Dict:
"""Генерирует детальный отчёт"""
report = {
'timestamp': datetime.now().isoformat(),
'summary': {
'total_tests': len(results),
'success_rate': sum(1 for r in results if r.is_correct) / len(results),
'average_latency': metrics['latency_metrics'].get('average', 0),
'total_cost': metrics['cost_analysis'].get('total_cost', 0),
},
'stability_analysis': metrics['function_selection_stability'],
'problematic_areas': self._identify_problematic_areas(results),
'recommendations': self._generate_recommendations(metrics),
'trend_analysis': self._analyze_trends(),
}
return report
def _check_quality_gates(self, metrics: Dict) -> bool:
"""Проверяет, проходим ли мы quality gates"""
gates = self.config.get('quality_gates', {})
# Проверяем стабильность
stability = metrics['function_selection_stability']['average_consistency']
if stability < gates.get('min_stability', 0.9):
print(f"❌ Stability gate failed: {stability} < {gates.get('min_stability', 0.9)}")
return False
# Проверяем успешность парсинга
parsing_rate = metrics['parsing_success_rate']
if parsing_rate < gates.get('min_parsing_rate', 0.85):
print(f"❌ Parsing rate gate failed: {parsing_rate} < {gates.get('min_parsing_rate', 0.85)}")
return False
# Проверяем латентность
latency_p95 = metrics['latency_metrics'].get('p95', 0)
if latency_p95 > gates.get('max_latency_p95', 3000):
print(f"❌ Latency gate failed: {latency_p95} > {gates.get('max_latency_p95', 3000)}")
return False
# Проверяем стоимость
error_cost_percentage = metrics['cost_analysis'].get('error_cost_percentage', 0)
if error_cost_percentage > gates.get('max_error_cost_percentage', 20):
print(f"❌ Cost gate failed: {error_cost_percentage}% > {gates.get('max_error_cost_percentage', 20)}%")
return False
return True
def _identify_problematic_areas(self, results: List[TestResult]) -> List[Dict]:
"""Идентифицирует проблемные области"""
errors_by_type = {}
for result in results:
if not result.is_correct:
error_type = self._classify_error(result)
if error_type not in errors_by_type:
errors_by_type[error_type] = []
errors_by_type[error_type].append(result)
problematic_areas = []
for error_type, error_results in errors_by_type.items():
percentage = len(error_results) / len(results) * 100
if percentage > 1: # Больше 1% ошибок этого типа
problematic_areas.append({
'error_type': error_type,
'count': len(error_results),
'percentage': percentage,
'examples': [
{
'query': r.user_query,
'expected': r.expected_function,
'actual': r.actual_function,
}
for r in error_results[:3] # Первые 3 примера
]
})
return problematic_areas
def _classify_error(self, result: TestResult) -> str:
"""Классифицирует тип ошибки"""
if result.actual_function is None:
return "no_function_called"
elif result.actual_function != result.expected_function:
return "wrong_function"
elif not self._compare_params(result.expected_params, result.actual_params):
return "wrong_parameters"
else:
return "unknown_error"
def _generate_recommendations(self, metrics: Dict) -> List[str]:
"""Генерирует рекомендации по улучшению"""
recommendations = []
# Рекомендации по стабильности
stability = metrics['function_selection_stability']['average_consistency']
if stability < 0.95:
recommendations.append(
f"Стабильность выбора функции ({stability*100:.1f}%) ниже целевой 95%. "
"Рассмотрите уточнение промптов или добавление примеров few-shot."
)
# Рекомендации по стоимости
error_cost = metrics['cost_analysis'].get('error_cost_percentage', 0)
if error_cost > 15:
recommendations.append(
f"Стоимость ошибок ({error_cost:.1f}%) высока. "
"Добавьте валидацию перед вызовом функций или используйте более дешёвую модель для простых запросов."
)
# Рекомендации по латентности
latency_p95 = metrics['latency_metrics'].get('p95', 0)
if latency_p95 > 2000:
recommendations.append(
f"P95 латентность ({latency_p95:.0f}ms) превышает 2 секунды. "
"Рассмотрите кэширование, оптимизацию промптов или использование более быстрой модели."
)
return recommendations
def _analyze_trends(self) -> Dict:
"""Анализирует тренды по историческим данным"""
# Загружаем исторические результаты
try:
historical_data = self._load_historical_results()
if len(historical_data) < 2:
return {"message": "Недостаточно исторических данных для анализа трендов"}
# Анализируем изменения метрик
latest = historical_data[-1]
previous = historical_data[-2]
trends = {
'stability_change': latest['stability'] - previous['stability'],
'latency_change': latest['latency_p95'] - previous['latency_p95'],
'cost_change': latest['cost_per_query'] - previous['cost_per_query'],
'success_rate_change': latest['success_rate'] - previous['success_rate'],
}
return trends
except FileNotFoundError:
return {"message": "Исторические данные не найдены"}
def _save_results(self, results: List[TestResult], metrics: Dict, report: Dict):
"""Сохраняет результаты тестирования"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
# Сохраняем сырые результаты
results_dicts = [
{
'test_id': r.test_id,
'query': r.user_query,
'is_correct': r.is_correct,
'latency_ms': r.latency_ms,
'cost': r.cost,
'timestamp': r.timestamp.isoformat(),
}
for r in results
]
with open(f"test_results/results_{timestamp}.json", 'w') as f:
json.dump(results_dicts, f, indent=2)
# Сохраняем метрики
with open(f"test_results/metrics_{timestamp}.json", 'w') as f:
json.dump(metrics, f, indent=2)
# Сохраняем отчёт
with open(f"test_results/report_{timestamp}.json", 'w') as f:
json.dump(report, f, indent=2)
# Обновляем исторические данные
self._update_historical_data(metrics, report)
print(f"💾 Results saved to test_results/ directory")
# Конфигурация пайплайна
config = {
'batch_size': 20,
'quality_gates': {
'min_stability': 0.9,
'min_parsing_rate': 0.85,
'max_latency_p95': 2500, # 2.5 секунды
'max_error_cost_percentage': 15,
},
'test_queries_path': 'data/test_queries.json',
'functions_path': 'data/functions.json',
}
# Запуск пайплайна
async def main():
pipeline = LLMTestingPipeline(config)
results = await pipeline.run_full_pipeline()
print(json.dumps(results, indent=2))
if __name__ == "__main__":
asyncio.run(main())Заключение: тестируйте умно, а не много
Тестирование недетерминированных LLM — это не поиск идеальной точности, а управление стабильностью. Ключевые выводы:
- Тестируйте стабильность, а не точность — запускайте каждый тест многократно, измеряйте консистентность
- Генерируйте разнообразные тестовые данные — включайте краевые случаи, опечатки, неполные запросы
- Автоматизируйте полностью — интегрируйте в CI/CD, генерируйте отчёты, отслеживайте тренды
- Контролируйте стоимость — используйте лимиты, кэширование, локальные модели для разработки
- Анализируйте ошибки системно — классифицируйте типы ошибок, находите паттерны, улучшайте промпты
Помните: идеальной стабильности (100%) у LLM достичь невозможно. Ваша цель — не perfection, а predictability. Знать, что в 95% случаев система работает корректно, и понимать, в каких 5% она может ошибиться — это уже огромный шаг к надёжному продакшн-приложению.
💡
Для более глубокого погружения в тему промптов и их тестирования рекомендую ознакомиться с нашей коллекцией промптов для тестирования LLM, где вы найдёте готовые шаблоны для различных сценариев тестирования.
Начните с малого: реализуйте базовую систему тестирования из этой статьи, запустите её на своих данных, и постепенно добавляйте сложность. Уже через неделю регулярного тестирования вы заметите, насколько стабильнее стала работать ваша система вызова функций.