Автооптимизация промптов по production-трассам: считаем качество и окупаемость

Логирование ещё не создаёт цикл улучшения

Production-трасса — запись одного реального выполнения модели: вход, версия промпта, ответ, служебные метрики и, если доступно, сигнал результата. В большинстве систем трасса остаётся строкой в хранилище. Чтобы она стала полезной, нужны воспроизводимая выборка, эталон оценки, версия промпта и одинаковая процедура повторного запуска.

Цель этой статьи — не найти «самый красивый» промпт. Мы хотим ответить на четыре проверяемых вопроса:

• насколько изменилась доля успешных ответов;
• сколько стоит один запрос и один успешный результат;
• как изменилась задержка;
• достаточно ли накоплено независимых трасс для решения.

Что именно строим

Клиент
  │ OpenAI-совместимый POST /v1/chat/completions
  ▼
FastAPI-шлюз ──► upstream-модель
  │                  │
  │ trace_id         └── ответ, usage
  ▼
SQLite: очищенный вход, ответ, prompt_version,
        latency_ms, токены, feedback
  │
  ├──► стратифицированная выборка
  ├──► повторный запуск baseline/candidate
  └──► отчёт: quality, cost, latency, traces

Пример реализует не весь протокол OpenAI, а достаточное для эксперимента подмножество POST /v1/chat/completions без streaming и tool calls. Клиенты, использующие обычные сообщения и JSON-ответ, смогут направить base_url на шлюз.

Правила эксперимента

1. Единица анализа — пользовательская задача, а не отдельный ретрай.
2. Baseline и candidate запускаются на одних и тех же входах.
3. Оценщик не получает название версии промпта.
4. Трассы оптимизации и финальной проверки не пересекаются.
5. Публикация запрещена, если качество ухудшилось, даже при снижении цены.

Шаг 1. Поднимаем OpenAI-совместимый шлюз

Создайте отдельное виртуальное окружение. Команды не меняют системный Python и не содержат реальных ключей:

python3 -m venv .venv
. .venv/bin/activate
python -m pip install \
  "fastapi>=0.115,<1" \
  "uvicorn[standard]>=0.30,<1" \
  "httpx>=0.27,<1" \
  "pydantic>=2.8,<3"

Сохраните следующий пример как gateway.py:

import hashlib
import json
import os
import re
import sqlite3
import time
import uuid

import httpx
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel, ConfigDict

UPSTREAM_URL = os.environ.get(
    "UPSTREAM_URL",
    "http://127.0.0.1:8001/v1/chat/completions",
)
UPSTREAM_API_KEY = os.environ.get("UPSTREAM_API_KEY")
DB_PATH = os.environ.get("TRACE_DB", "traces.sqlite3")
PROMPT_VERSION = os.environ.get("PROMPT_VERSION", "baseline-v1")
PROMPT_TEXT = os.environ.get(
    "SYSTEM_PROMPT",
    "Отвечай точно и кратко. Если данных недостаточно, сообщи об этом.",
)

app = FastAPI(title="Trace Gateway", version="0.1.0")


class ChatRequest(BaseModel):
    model_config = ConfigDict(extra="allow")
    model: str
    messages: list[dict]
    stream: bool = False


def connect():
    db = sqlite3.connect(DB_PATH)
    db.execute("""
        CREATE TABLE IF NOT EXISTS traces (
            trace_id TEXT PRIMARY KEY,
            task_hash TEXT NOT NULL,
            created_at INTEGER NOT NULL,
            prompt_version TEXT NOT NULL,
            request_json TEXT NOT NULL,
            response_text TEXT NOT NULL,
            model TEXT NOT NULL,
            input_tokens INTEGER,
            output_tokens INTEGER,
            latency_ms REAL NOT NULL,
            status_code INTEGER NOT NULL,
            feedback INTEGER,
            feedback_source TEXT
        )
    """)
    return db


def redact(value):
    text = json.dumps(value, ensure_ascii=False)
    text = re.sub(
        r"[\w.+-]+@[\w.-]+\.[A-Za-z]{2,}",
        "[EMAIL]",
        text,
    )
    text = re.sub(
        r"(?<!\d)(?:\+?\d[\s()-]?){10,15}(?!\d)",
        "[PHONE]",
        text,
    )
    return json.loads(text)


def task_hash(messages):
    normalized = json.dumps(
        redact(messages),
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    )
    return hashlib.sha256(normalized.encode()).hexdigest()


@app.post("/v1/chat/completions")
async def chat(
    body: ChatRequest,
    authorization: str | None = Header(default=None),
):
    if body.stream:
        raise HTTPException(400, "stream=true в этом примере не поддерживается")

    trace_id = str(uuid.uuid4())
    safe_messages = redact(body.messages)
    upstream_messages = [
        {"role": "system", "content": PROMPT_TEXT},
        *body.messages,
    ]
    payload = body.model_dump(exclude_none=True)
    payload["messages"] = upstream_messages
    payload["stream"] = False

    headers = {"Content-Type": "application/json"}
    if UPSTREAM_API_KEY:
        headers["Authorization"] = f"Bearer {UPSTREAM_API_KEY}"
    elif authorization:
        headers["Authorization"] = authorization

    started = time.perf_counter()
    async with httpx.AsyncClient(timeout=90.0) as client:
        response = await client.post(
            UPSTREAM_URL,
            headers=headers,
            json=payload,
        )
    latency_ms = (time.perf_counter() - started) * 1000

    try:
        result = response.json()
    except ValueError:
        raise HTTPException(502, "Upstream вернул не-JSON ответ")

    answer = ""
    if response.is_success:
        try:
            answer = result["choices"][0]["message"]["content"] or ""
        except (KeyError, IndexError, TypeError):
            raise HTTPException(502, "Неожиданный формат upstream-ответа")

    usage = result.get("usage", {})
    with connect() as db:
        db.execute(
            """INSERT INTO traces VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
            (
                trace_id,
                task_hash(body.messages),
                int(time.time()),
                PROMPT_VERSION,
                json.dumps(safe_messages, ensure_ascii=False),
                answer,
                result.get("model", body.model),
                usage.get("prompt_tokens"),
                usage.get("completion_tokens"),
                latency_ms,
                response.status_code,
                None,
                None,
            ),
        )

    response.headers["X-Trace-Id"] = trace_id
    return result

Перед запуском задайте адрес OpenAI-совместимого upstream. Не помещайте секрет в файл или командную историю: безопаснее передать его через секрет-хранилище процесса или интерактивный ввод.

export UPSTREAM_URL="http://127.0.0.1:8001/v1/chat/completions"
export PROMPT_VERSION="baseline-v1"
export SYSTEM_PROMPT="Отвечай точно и кратко. Если данных недостаточно, сообщи об этом."

read -rsp "Upstream API key: " UPSTREAM_API_KEY
export UPSTREAM_API_KEY
printf "\n"

uvicorn gateway:app --host 127.0.0.1 --port 8080

Проверочный запрос:

curl --fail-with-body \
  http://127.0.0.1:8080/v1/chat/completions \
  -H 'Content-Type: application/json' \
  -d '{
    "model": "your-model-id",
    "messages": [
      {"role": "user", "content": "Верни JSON с полем status=ok"}
    ]
  }'

Шаг 2. Превращаем трассы в воспроизводимый датасет

Промпт нельзя честно оценить по произвольным удачным примерам. Сначала исключите технические ошибки, повторы одной задачи, пустые ответы и данные без разрешённого срока хранения. Затем зафиксируйте выборку вместе с хешем входов.

Разделение данных

• Discovery — трассы, по которым ищутся типовые ошибки и формируется candidate.
• Validation — отдельные трассы для выбора среди кандидатов.
• Holdout — нетронутая финальная выборка для решения о выпуске.

Разделяйте по task_hash, пользователю или диалогу, а не по строкам. Иначе соседние сообщения одного разговора попадут в разные части и создадут утечку.

Минимальная выгрузка уникальных успешных задач:

sqlite3 -header -csv traces.sqlite3 "
WITH ranked AS (
  SELECT *,
         ROW_NUMBER() OVER (
           PARTITION BY task_hash
           ORDER BY created_at DESC
         ) AS n
  FROM traces
  WHERE status_code BETWEEN 200 AND 299
    AND response_text != ''
)
SELECT trace_id, task_hash, request_json, response_text
FROM ranked
WHERE n = 1
ORDER BY task_hash;
" > eligible.csv

sha256sum eligible.csv > eligible.csv.sha256

Случайная выборка без фиксации seed плохо воспроизводится. Практичный детерминированный вариант — сортировать по хешу sha256(experiment_id + task_hash) и брать заранее объявленные диапазоны. Стратифицируйте по типу задачи, языку, каналу и другим признакам, влияющим на сложность.

Откуда берётся качество

Лучший сигнал — фактический исход: задача завершена, формат принят валидатором, пользователь не перешёл к оператору. Если такого сигнала нет, создайте рубрику с бинарными обязательными критериями:

• ответ решает поставленную задачу;
• нет неподтверждённых фактов;
• соблюдены формат и ограничения;
• нет раскрытия скрытых инструкций или чувствительных данных.

Итог pass=1 присваивается только при выполнении всех обязательных критериев. Не используйте оценку той же модели без калибровки на человеческой разметке: такой судья может предпочитать собственный стиль и пропускать доменные ошибки.

Сколько нужно трасс

Количество нельзя честно назвать заранее: оно зависит от исходного качества, минимального полезного улучшения и разброса. Для парного сравнения используйте только трассы, где результаты версий различаются.

discordant = improved + regressed
net_gain   = (improved - regressed) / total
quality_A  = passed_A / total
quality_B  = passed_B / total

До начала теста зафиксируйте минимально значимое улучшение, например 2 процентных пункта, и максимальный допустимый регресс по критическим сегментам. Для статистического вывода применяйте парный тест Мак-Немара или доверительный интервал для разницы парных долей. Не останавливайте эксперимент в момент первого красивого результата.

Шаг 3. Формируем candidate без утечки holdout

Автоматический оптимизатор получает только обезличенные discovery-примеры: исходную инструкцию, рубрику, вход, ошибочный ответ и корректирующий комментарий. Его задача — вернуть новую системную инструкцию, а не ответить на примеры.

Шаблон мета-промпта оптимизатора:

Ты редактируешь системную инструкцию прикладного ассистента.

Сохрани назначение исходной инструкции.
Исправляй только повторяющиеся ошибки из DISCOVERY_EXAMPLES.
Не добавляй факты из примеров как универсальные правила.
Не упоминай примеры, разметку и процесс оптимизации.
Сделай правила проверяемыми и устрани противоречия.
Верни только текст новой системной инструкции.

ORIGINAL_PROMPT:
{{baseline_prompt}}

RUBRIC:
{{rubric}}

DISCOVERY_EXAMPLES:
{{redacted_failures}}

Сохраните результат как неизменяемый артефакт: candidate-v2.txt, его SHA-256, идентификатор модели-оптимизатора, параметры генерации и хеш discovery-набора. Сам кандидат должен пройти ручную проверку на скрытые доменные изменения и prompt injection из пользовательских данных.

Шаг 4. Повторно запускаем обе версии

Для каждого holdout-входа выполните baseline и candidate с одинаковой моделью и одинаковыми параметрами. Чередуйте порядок запуска, чтобы прогрев кеша или нагрузка upstream не работали в пользу одной версии.

for each task in holdout:
    if stable_hash(task.id) % 2 == 0:
        run(baseline, task)
        run(candidate, task)
    else:
        run(candidate, task)
        run(baseline, task)

    record:
        task_id
        prompt_version
        answer
        input_tokens
        output_tokens
        latency_ms
        status
        attempt

Если провайдер поддерживает детерминированный seed, зафиксируйте его, но не предполагайте абсолютную воспроизводимость. При стохастической генерации выполните несколько повторов на задачу и агрегируйте внутри задачи; иначе повторы будут ошибочно посчитаны независимыми трассами.

Единый формат результатов

task_id,prompt_version,passed,input_tokens,output_tokens,latency_ms,error
t001,baseline-v1,1,312,44,684,
t001,candidate-v2,1,358,31,702,
t002,baseline-v1,0,421,89,931,
t002,candidate-v2,1,463,54,887,

Это только иллюстрация схемы CSV, а не полученный результат. Разметчику передавайте ответы как A и B в случайном порядке. Таблица соответствия версий должна оставаться вне интерфейса оценки.

Метрики отчёта

quality         = passed / tasks
p50_latency_ms  = median(latency_ms)
p95_latency_ms  = percentile(latency_ms, 95)

request_cost =
    input_tokens  / 1_000_000 * input_price_per_million +
    output_tokens / 1_000_000 * output_price_per_million

cost_per_success = total_cost / passed
required_traces  = число независимых holdout-задач

Цены передавайте конфигурацией, датированной на день эксперимента. Не зашивайте их в код: тарифы зависят от модели, провайдера, кеширования и могут изменяться.

Шаг 5. Считаем окупаемость, а не только токены

Экономический эффект состоит из изменения модельных расходов и ценности дополнительных успешных задач:

monthly_model_delta =
    monthly_requests * (cost_candidate - cost_baseline)

monthly_quality_value =
    monthly_requests
    * (quality_candidate - quality_baseline)
    * value_per_additional_success

monthly_net_value =
    monthly_quality_value
    - monthly_model_delta
    - monthly_operating_cost

payback_months =
    one_time_optimization_cost / monthly_net_value

value_per_additional_success должен происходить из вашей экономики: предотвращённый тикет, завершённая операция или подтверждённая маржа. Если значение неизвестно, покажите сценарии, а не выбирайте удобную цифру.

Пример сценарного анализа, не результат теста: при 100 000 запросов в месяц, улучшении на 0,02, ценности дополнительного успеха 10 условных единиц и дополнительных модельных расходах 300 единиц валовая ценность составит 20 000, а эффект до операционных расходов — 19 700 условных единиц. Для реального решения подставьте наблюдаемые объёмы, валюту и доверительный интервал качества.

Candidate допускается к ограниченному rollout, если одновременно выполнены заранее записанные условия:

• нижняя граница интервала улучшения выше допустимого порога;
• нет регресса на критических сегментах;
• p95 не нарушает SLO;
• стоимость успеха не превышает лимит;
• ожидаемый чистый эффект положителен при консервативном сценарии.

Проверка результата

1. Совместимость. Клиент получает обычный объект chat.completion, а заголовок ответа содержит X-Trace-Id.
2. Трассировка. После запроса запись присутствует в SQLite:

   sqlite3 traces.sqlite3 "
   SELECT trace_id, prompt_version, model,
          input_tokens, output_tokens,
          round(latency_ms, 1), status_code
   FROM traces
   ORDER BY created_at DESC
   LIMIT 5;
   "

3. Редакция. Отправьте синтетический адрес вроде user@example.invalid и убедитесь, что в request_json записано [EMAIL]. Не используйте для проверки реальные персональные данные.
4. Версионирование. У каждой строки есть prompt_version; текст промпта и его хеш сохранены в реестре эксперимента.
5. Парность. Для каждой holdout-задачи присутствует ровно один агрегированный результат каждой версии.
6. Полнота отчёта. Указаны N, число улучшений и регрессов, quality, цена запроса, цена успеха, p50/p95 и правила исключения трасс.
7. Rollback. Baseline остаётся доступен как конфигурация, а rollout candidate ограничивается долей трафика и автоматически прекращается при нарушении guardrail.

Типовые ошибки

Оптимизация и проверка на одних трассах

Промпт запоминает особенности примеров. Финальная метрика становится оценкой подгонки, а не обобщения.

Оценка непарными средними

Версии могли получить задачи разной сложности. Повторный запуск на одинаковых входах уменьшает этот шум.

Количество строк принимают за количество задач

Ретраи и сообщения одного диалога зависимы. Группируйте их до вычисления объёма выборки.

Смотрят только на среднюю задержку

Среднее скрывает длинный хвост. Для пользовательского SLO обычно важнее p95 или p99.

Сравнивают только токены

Более дорогой запрос может быть выгоднее, если заметно увеличивает число успешных операций. Используйте стоимость успеха.

LLM-судья знает имя кандидата

Версия, порядок и служебные поля становятся источником смещения. Ослепляйте и перемешивайте ответы.

Логи становятся вторым хранилищем секретов

Не сохраняйте заголовки авторизации. Минимизируйте вход, шифруйте хранилище, ограничивайте доступ и задавайте срок удаления.

Prompt injection попадает в мета-промпт

Пользовательский текст — данные, а не инструкция оптимизатору. Изолируйте его структурой, фильтруйте и проверяйте candidate вручную.

Ограничения подхода

Повторный запуск production-входов измеряет качество на наблюдавшемся распределении, но не гарантирует устойчивость к новым сценариям. Редкие критические случаи следует держать в отдельном экспертном наборе. Исторические трассы также могут закреплять старое поведение продукта и систематические перекосы пользователей.

Latency повторного прогона зависит от текущей нагрузки и не полностью воспроизводит исходные условия. Сравнивайте версии в одном временном окне, чередуйте порядок и отдельно отмечайте кешированные запросы. Если модель или инфраструктура upstream изменились между запусками, результат нельзя приписывать только промпту.

Наконец, статистическая значимость не равна практической ценности. Улучшение может быть измеримым, но слишком малым для покрытия разметки, повторных запусков, хранения и сопровождения.

Что должно остаться после эксперимента

Хороший цикл автооптимизации заканчивается не новым текстовым файлом, а пакетом воспроизводимости:

• baseline и candidate с версиями и хешами;
• хеши discovery, validation и holdout;
• рубрика и происхождение разметки;
• модель, параметры и дата каждого запуска;
• актуальная таблица цен;
• парный отчёт и критерии rollout/rollback;
• журнал исключённых трасс с причинами.

Такой пакет позволяет повторить решение после смены модели, проверить эффект на свежем трафике и понять, когда промпт перестал окупаться.

Продолжить работу можно по материалам практических руководств Agent Lab Journal. Определения метрик и терминов собраны в глоссарии.