329 lines
15 KiB
Python
329 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
"""Прогон корпуса качества суммаризации по всем уровням AI.
|
||
|
||
Для каждого уровня (`min`/`medium`/`max`, матрица `backend/services/ai_tiers.TIERS`,
|
||
ADR-004 `docs/architecture/adr/004-ai-tier-matrix.md`) скрипт:
|
||
1. проверяет ДОСТУПНОСТЬ уровня на текущей машине — GPU (для `max`) и живой
|
||
LLM-сервер с ИМЕННО той моделью, что предписана `TierSpec` (сверка через
|
||
`/v1/models` официального llama.cpp сервера); недоступный
|
||
уровень корректно пропускается с человекочитаемой причиной, скрипт не падает;
|
||
2. для доступного уровня создаёт плагин `QwenLocal` штатной Factory
|
||
(`backend/core/plugins/factory.py::create_summarizer`) из конфигурации
|
||
`TierSpec.summarizer` — так же, как это делает прод-код воркера
|
||
суммаризации, никакой отдельной логики вызова LLM в этом скрипте нет;
|
||
3. прогоняет через `Summarizer.summarize()` весь корпус
|
||
(`workers/summarizer/eval/corpus/*.txt`, формат — как штатный вход
|
||
`summarize`: строки `[Имя MM:SS] текст`) и пишет результаты в
|
||
`workers/summarizer/eval/results/<уровень>/<имя-транскрипта>.md` плюс
|
||
метаданные прогона `_run_meta.json` в тот же каталог.
|
||
|
||
Промпты (`workers/summarizer/prompts/summary_map_ru.txt`,
|
||
`summary_reduce_ru.txt`) — ШТАТНЫЕ, скрипт их не читает и не редактирует
|
||
напрямую: только передаёт `QwenLocal` абсолютный путь к каталогу с ними
|
||
(`git diff --stat workers/summarizer/prompts/` остаётся пуст).
|
||
|
||
Запуск (из корня репозитория, с поднятым `docker compose --profile llm up -d`):
|
||
|
||
cd backend && uv run python ../workers/summarizer/eval/run_tiers.py \\
|
||
--base-url http://localhost:8080/v1
|
||
|
||
`--base-url` переопределяет адрес LLM-сервера для ВСЕХ уровней — на dev-машине
|
||
вне docker-сети штатные `http://llm:8080/v1`/`http://llm-gpu:8080/v1`
|
||
(`backend/services/ai_tiers.py`) не резолвятся; без флага скрипт предполагает
|
||
запуск изнутри docker-сети (например, `docker compose exec worker ...`), где
|
||
эти хосты резолвятся штатно.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import shutil
|
||
import sys
|
||
import time
|
||
from dataclasses import dataclass
|
||
from datetime import UTC, datetime
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import httpx
|
||
|
||
# --- Настройка sys.path: скрипт живёт вне пакета `backend`, но использует
|
||
# его модули (`core.plugins.factory`, `services.ai_tiers`) напрямую, как это
|
||
# делает прод-код `workers/` (см. докстринг `workers/celery_app.py`: пакеты
|
||
# backend и workers делят один Python-процесс/venv). Вставляем `backend/` в
|
||
# sys.path здесь же, а не через PYTHONPATH — чтобы скрипт можно было
|
||
# запускать одной командой без дополнительной настройки окружения.
|
||
_EVAL_DIR = Path(__file__).resolve().parent
|
||
_REPO_ROOT = _EVAL_DIR.parents[2]
|
||
_BACKEND_DIR = _REPO_ROOT / "backend"
|
||
_PROMPTS_DIR = _REPO_ROOT / "workers" / "summarizer" / "prompts"
|
||
if str(_BACKEND_DIR) not in sys.path:
|
||
sys.path.insert(0, str(_BACKEND_DIR))
|
||
|
||
from core.plugins.factory import create_summarizer # noqa: E402
|
||
from services.ai_tiers import TIERS, TierSpec # noqa: E402
|
||
|
||
|
||
@dataclass
|
||
class TierAvailability:
|
||
"""Результат проверки доступности уровня AI на текущей машине."""
|
||
|
||
available: bool
|
||
reason: str | None
|
||
base_url: str
|
||
|
||
|
||
def _has_nvidia_gpu() -> bool:
|
||
"""Простой детект NVIDIA GPU для eval-скрипта: наличие `nvidia-smi` в PATH.
|
||
|
||
Не использует `backend.core.config.Settings.hw_*` (детект инсталлятора по
|
||
`.env`) — скрипт запускается вне контекста инстанса, ему достаточно
|
||
факта присутствия GPU-инструментария на машине, где он выполняется.
|
||
"""
|
||
return shutil.which("nvidia-smi") is not None
|
||
|
||
|
||
def _expected_gguf_name(spec: TierSpec) -> str:
|
||
"""Имя файла GGUF-модели, ожидаемой для уровня (из `TierSpec.model_files`)."""
|
||
for path in spec.model_files:
|
||
if path.endswith(".gguf"):
|
||
return Path(path).name
|
||
raise ValueError(f"В model_files уровня нет .gguf файла: {spec.model_files!r}")
|
||
|
||
|
||
def _check_tier_availability(
|
||
level: str,
|
||
spec: TierSpec,
|
||
base_url_override: str | None,
|
||
timeout_s: float,
|
||
) -> TierAvailability:
|
||
"""Проверить доступность уровня: GPU (если требуется) + живой сервер с нужной моделью.
|
||
|
||
Проверка модели — через `GET /v1/models` (OpenAI-совместимый эндпоинт
|
||
llama.cpp server, официальная документация `tools/server/README.md`):
|
||
без неё сервер мог бы ответить на `/health`, но с ДРУГОЙ загруженной
|
||
моделью (`min` и `medium` в проде используют один и тот же compose-сервис
|
||
`llm` на разных пресетах — на этой машине единовременно поднят только
|
||
один из них), и результат прогона был бы приписан не тому уровню.
|
||
"""
|
||
if spec.requires_gpu and not _has_nvidia_gpu():
|
||
return TierAvailability(
|
||
available=False,
|
||
reason=(
|
||
"требуется NVIDIA GPU (nvidia-smi не найден в PATH) — недоступно на "
|
||
"этой машине; прогон уровня — ручной шаг на GPU-хосте"
|
||
),
|
||
base_url="",
|
||
)
|
||
|
||
base_url = base_url_override or spec.summarizer.options.get("base_url", "")
|
||
root = base_url.removesuffix("/v1").rstrip("/")
|
||
health_url = f"{root}/health"
|
||
models_url = f"{root}/v1/models"
|
||
|
||
try:
|
||
health_resp = httpx.get(health_url, timeout=timeout_s)
|
||
except httpx.HTTPError as exc:
|
||
return TierAvailability(
|
||
available=False,
|
||
reason=f"LLM-сервер недоступен по {health_url}: {exc}",
|
||
base_url=base_url,
|
||
)
|
||
if health_resp.status_code != 200:
|
||
return TierAvailability(
|
||
available=False,
|
||
reason=(
|
||
f"LLM-сервер не готов ({health_url} -> {health_resp.status_code}: "
|
||
f"{health_resp.text.strip()})"
|
||
),
|
||
base_url=base_url,
|
||
)
|
||
|
||
expected = _expected_gguf_name(spec)
|
||
try:
|
||
models_resp = httpx.get(models_url, timeout=timeout_s)
|
||
models_resp.raise_for_status()
|
||
data = models_resp.json()["data"]
|
||
loaded = Path(data[0]["id"]).name if data else "<пусто>"
|
||
except (httpx.HTTPError, KeyError, IndexError, ValueError) as exc:
|
||
return TierAvailability(
|
||
available=False,
|
||
reason=f"не удалось получить {models_url}: {exc}",
|
||
base_url=base_url,
|
||
)
|
||
|
||
if loaded != expected:
|
||
return TierAvailability(
|
||
available=False,
|
||
reason=(
|
||
f"на {base_url} загружена другая модель ({loaded}), для уровня "
|
||
f"{level!r} ожидалась {expected} — поднимите отдельный сервер "
|
||
f"с нужной моделью (LLM_MODEL_FILE в .env)"
|
||
),
|
||
base_url=base_url,
|
||
)
|
||
|
||
return TierAvailability(available=True, reason=None, base_url=base_url)
|
||
|
||
|
||
def _run_corpus(
|
||
level: str,
|
||
spec: TierSpec,
|
||
base_url: str,
|
||
out_dir: Path,
|
||
corpus_files: list[Path],
|
||
tokenizer_override: str | None,
|
||
) -> dict[str, Any]:
|
||
"""Прогнать весь корпус через `QwenLocal` уровня `level`, вернуть метаданные прогона."""
|
||
options = dict(spec.summarizer.options)
|
||
options["base_url"] = base_url
|
||
options["prompts_dir"] = str(_PROMPTS_DIR)
|
||
if tokenizer_override:
|
||
options["tokenizer_path"] = tokenizer_override
|
||
cfg = spec.summarizer.model_copy(update={"options": options})
|
||
|
||
level_dir = out_dir / level
|
||
level_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
run_meta: dict[str, Any] = {
|
||
"level": level,
|
||
"model": cfg.model,
|
||
"base_url": base_url,
|
||
"chunk_minutes": cfg.chunk_minutes,
|
||
"temperature": options.get("temperature"),
|
||
"max_tokens_map": options.get("max_tokens_map"),
|
||
"max_tokens_reduce": options.get("max_tokens_reduce"),
|
||
"tokenizer_path": options.get("tokenizer_path"),
|
||
"started_at": datetime.now(UTC).isoformat(),
|
||
"files": [],
|
||
}
|
||
|
||
for corpus_file in corpus_files:
|
||
# Новый инстанс плагина на файл — так же, как `create_summarizer`
|
||
# используется в проде: один плагин на одну задачу суммаризации
|
||
# (см. докстринг `QwenLocal.summarize`).
|
||
summarizer = create_summarizer(cfg)
|
||
transcript = corpus_file.read_text(encoding="utf-8")
|
||
|
||
started = time.monotonic()
|
||
status = "ok"
|
||
error_text: str | None = None
|
||
summary = ""
|
||
try:
|
||
summary = summarizer.summarize(transcript)
|
||
except Exception as exc: # noqa: BLE001 — сбой LLM (см. LlmUnavailableError) не должен прервать прогон корпуса
|
||
status = "error"
|
||
error_text = f"{type(exc).__name__}: {exc}"
|
||
elapsed_s = time.monotonic() - started
|
||
|
||
out_path = level_dir / f"{corpus_file.stem}.md"
|
||
header = (
|
||
f"<!-- уровень: {level} | модель: {cfg.model} | источник: {corpus_file.name} "
|
||
f"| статус: {status} | время: {elapsed_s:.1f}с -->\n\n"
|
||
)
|
||
body = summary if status == "ok" else f"ОШИБКА: {error_text}\n"
|
||
out_path.write_text(header + body, encoding="utf-8")
|
||
|
||
run_meta["files"].append(
|
||
{
|
||
"source": corpus_file.name,
|
||
"status": status,
|
||
"elapsed_s": round(elapsed_s, 1),
|
||
"error": error_text,
|
||
"result_file": str(out_path.relative_to(_REPO_ROOT)),
|
||
}
|
||
)
|
||
print(f" [{level}] {corpus_file.name}: {status} за {elapsed_s:.1f}с -> {out_path}")
|
||
|
||
run_meta["finished_at"] = datetime.now(UTC).isoformat()
|
||
(level_dir / "_run_meta.json").write_text(
|
||
json.dumps(run_meta, ensure_ascii=False, indent=2), encoding="utf-8"
|
||
)
|
||
return run_meta
|
||
|
||
|
||
def main() -> int:
|
||
"""Точка входа CLI: разобрать аргументы, прогнать доступные уровни, напечатать сводку."""
|
||
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0] if __doc__ else "")
|
||
parser.add_argument(
|
||
"--levels",
|
||
default="min,medium,max",
|
||
help="Уровни через запятую (по умолчанию все три из ADR-004)",
|
||
)
|
||
parser.add_argument(
|
||
"--corpus-dir",
|
||
default=str(_EVAL_DIR / "corpus"),
|
||
help="Каталог с *.txt транскриптами (формат входа summarize)",
|
||
)
|
||
parser.add_argument(
|
||
"--out-dir",
|
||
default=str(_EVAL_DIR / "results"),
|
||
help="Каталог для результатов (подкаталог на уровень)",
|
||
)
|
||
parser.add_argument(
|
||
"--base-url",
|
||
default=None,
|
||
help=(
|
||
"Переопределить base_url LLM-сервера для всех уровней "
|
||
"(например, http://localhost:8080/v1 при запуске с хоста вне docker-сети)"
|
||
),
|
||
)
|
||
parser.add_argument(
|
||
"--tokenizer-path",
|
||
default=None,
|
||
help="Переопределить путь к tokenizer.json (по умолчанию — путь из TierSpec)",
|
||
)
|
||
parser.add_argument(
|
||
"--health-timeout",
|
||
type=float,
|
||
default=5.0,
|
||
help="Таймаут проверки доступности сервера (секунды)",
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
levels = [lvl.strip() for lvl in args.levels.split(",") if lvl.strip()]
|
||
corpus_dir = Path(args.corpus_dir)
|
||
out_dir = Path(args.out_dir)
|
||
|
||
corpus_files = sorted(corpus_dir.glob("*.txt"))
|
||
if not corpus_files:
|
||
print(f"В {corpus_dir} не найдено ни одного .txt транскрипта", file=sys.stderr)
|
||
return 1
|
||
|
||
summary_lines: list[str] = []
|
||
for level in levels:
|
||
spec = TIERS.get(level)
|
||
if spec is None:
|
||
print(
|
||
f"Неизвестный уровень {level!r} (ожидались min/medium/max), пропуск",
|
||
file=sys.stderr,
|
||
)
|
||
continue
|
||
|
||
availability = _check_tier_availability(level, spec, args.base_url, args.health_timeout)
|
||
if not availability.available:
|
||
print(f"[{level}] ПРОПУСК: {availability.reason}")
|
||
summary_lines.append(f"{level}: пропущен — {availability.reason}")
|
||
continue
|
||
|
||
print(
|
||
f"[{level}] доступен: base_url={availability.base_url}, модель={spec.summarizer.model}"
|
||
)
|
||
run_meta = _run_corpus(
|
||
level, spec, availability.base_url, out_dir, corpus_files, args.tokenizer_path
|
||
)
|
||
ok_count = sum(1 for f in run_meta["files"] if f["status"] == "ok")
|
||
summary_lines.append(
|
||
f"{level}: {ok_count}/{len(run_meta['files'])} успешно, результаты в {out_dir / level}"
|
||
)
|
||
|
||
print("\nИтог прогона:")
|
||
for line in summary_lines:
|
||
print(f" - {line}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|