Обзор инструментов для LLMOps

Локальный eval-контур: «LLM-judge» без внешних сервисов + отчёт

Цель. Поднять у себя минимальный контур оценки качества ответов (для RAG/FAQ): датасет из кейсов → генерация ответов вашей моделью → локальный «судья» (правила) → метрики и отчёт (CSV/Markdown). Всё самодостаточно, без зависимостей.

1) Что и зачем мы мерим

• pass@1 — доля кейсов, где ответ проходит рубрику.
• groundedness — доля числовых фактов в ответе, найденных в выданном контексте (числа «не с потолка»).
• citations_ok — есть ссылки вида [doc_id] и они принадлежат разрешённым источникам.
• must_include — обязательные слова/фразы по кейсу.

Результат — простые, но «острые» сигналы, которые поймают явные регрессы без сложных внешних инструментов.

2) Формат датасета (JSONL)

Создайте файл rag_cases.jsonl (5 кейсов для разгона):

{"id":"r1","question":"Сколько дней базового отпуска?", "context":[{"id":"d1","text":"Политика отпуска: базовый отпуск 28 дней. Регион: РФ."},{"id":"d2","text":"Дополнительный отпуск за переработки по согласованию."}],"rubric":{"must_include":["отпуск","дней"],"allow_cites":["d1","d2"]}}
{"id":"r2","question":"Оплачивается ли больничный?", "context":[{"id":"d3","text":"Больничный (лист временной нетрудоспособности) оплачивается по нормам."}],"rubric":{"must_include":["больничный"],"allow_cites":["d3"]}}
{"id":"r3","question":"Есть ли отпуск без содержания?", "context":[{"id":"d4","text":"FAQ: отпуск без сохранения заработной платы предоставляется по заявлению."}],"rubric":{"must_include":["без","содерж"],"allow_cites":["d4"]}}
{"id":"r4","question":"Какие правила командировок?", "context":[{"id":"d5","text":"Политика командировок: суточные, авиа/жд, отчёт. Изменений в 2024 нет."}],"rubric":{"must_include":["командиров"],"allow_cites":["d5"]}}
{"id":"r5","question":"Назовите все числа из политики отпуска.", "context":[{"id":"d1","text":"Политика отпуска: базовый отпуск 28 дней. Регион: РФ."},{"id":"d6","text":"Отпускные рассчитываются по среднему, коэффициенты 1.00."}],"rubric":{"must_include":["отпуск"],"allow_cites":["d1","d6"]}}

3) Каркас eval-харнеса (готов к запуску)

Сохраните как rag_eval.py. По умолчанию используется mock-генератор; для реальной модели замените функцию call_model().

# rag_eval.py
import json, re, csv
from dataclasses import dataclass
from pathlib import Path
from typing import List, Dict, Any, Tuple

DATA_FILE = Path("rag_cases.jsonl")

# ------------------ ВСТАВЬТЕ СВОЮ МОДЕЛЬ ЗДЕСЬ ------------------
def call_model(question: str, context: List[Dict[str,str]]) -> str:
    """
    Замените на реальный вызов (RAG-пайплайн):
    - сформируйте prompt с фрагментами context
    - верните ТЕКСТ ответа, включая цитаты вида [doc_id]
    По умолчанию — "умный" мок для демонстрации.
    """
    ctx_map = {c["id"]: c["text"] for c in context}
    q = question.lower()
    if "базов" in q and "отпуск" in q:
        return "Базовый отпуск составляет 28 дней [d1]."
    if "больнич" in q:
        return "Да, больничный оплачивается по нормам [d3]."
    if "без содерж" in q or "без содержания" in q:
        return "Да, предусмотрен отпуск без сохранения заработной платы [d4]."
    if "командиров" in q:
        return "Суточные, авиа/жд и обязательный отчёт, изменений в 2024 нет [d5]."
    if "все числа" in q or "числа" in q:
        return "Политика содержит числа 28 и 1.00 [d1][d6]."
    # "плохой" ответ без цитат/не из контекста
    return "Вероятно 30 дней. Уточните у HR."
# ----------------------------------------------------------------

# ------------------ JUDGE: правила-оценки ------------------
_Num = re.compile(r"\d+(?:[.,]\d+)?")

def extract_citations(answer: str) -> List[str]:
    return re.findall(r"\[([A-Za-z0-9_-]+)\]", answer)

def numbers_in_text(text: str) -> List[str]:
    return [x.replace(",", ".") for x in _Num.findall(text)]

def grounded_numbers(answer: str, context_texts: List[str]) -> Tuple[float, List[str]]:
    """Доля чисел из ответа, которые встречаются в объединённом контексте (как строки)."""
    nums_ans = numbers_in_text(answer)
    if not nums_ans: 
        return 1.0, []  # нет чисел — считаем ок
    blob = " ".join(context_texts)
    nums_ctx = set(numbers_in_text(blob))
    unsupported = [n for n in nums_ans if n not in nums_ctx]
    ok = (len(nums_ans) - len(unsupported)) / max(1, len(nums_ans))
    return ok, unsupported

def contains_all(answer: str, need: List[str]) -> bool:
    text = answer.lower()
    return all(any(n in text for n in (w if isinstance(w, list) else [w])) for w in need)

def citations_ok(answer: str, allow: List[str]) -> Tuple[bool, List[str], List[str]]:
    cites = extract_citations(answer)
    bad = [c for c in cites if c not in allow]
    return (len(cites) > 0 and not bad), cites, bad

# ------------------ Загрузка и прогон ------------------
@dataclass
class Case:
    id: str
    question: str
    context: List[Dict[str,str]]
    rubric: Dict[str, Any]

def load_cases(p: Path) -> List[Case]:
    out=[]
    for line in p.read_text(encoding="utf-8").splitlines():
        if not line.strip(): continue
        r = json.loads(line)
        out.append(Case(id=r["id"], question=r["question"], context=r["context"], rubric=r["rubric"]))
    return out

def judge_case(ans: str, case: Case) -> Dict[str, Any]:
    allow = case.rubric.get("allow_cites", [])
    must = case.rubric.get("must_include", [])
    ok_cites, cites, bad_cites = citations_ok(ans, allow)
    g_score, unsupported = grounded_numbers(ans, [c["text"] for c in case.context])
    ok_must = contains_all(ans, must)
    passed = ok_cites and ok_must and g_score >= 1.0  # все числа в ответе должны быть из контекста
    reasons=[]
    if not ok_cites: reasons.append("CITATIONS_FAIL")
    if not ok_must: reasons.append("MUST_FAIL")
    if g_score < 1.0: reasons.append(f"NUM_UNSUPPORTED:{','.join(unsupported[:3])}")
    return {
        "pass": passed,
        "citations_ok": ok_cites,
        "cites": cites,
        "bad_cites": bad_cites,
        "must_ok": ok_must,
        "groundedness": round(g_score, 3),
        "unsupported_numbers": unsupported,
        "reasons": reasons
    }

def run_eval(data_file: Path = DATA_FILE) -> Dict[str, Any]:
    cases = load_cases(data_file)
    rows=[]
    for cs in cases:
        ans = call_model(cs.question, cs.context)
        res = judge_case(ans, cs)
        rows.append({
            "id": cs.id,
            "pass": res["pass"],
            "groundedness": res["groundedness"],
            "cit_ok": res["citations_ok"],
            "bad_cites": ";".join(res["bad_cites"]),
            "reasons": ";".join(res["reasons"]),
            "answer": ans
        })
    pass_rate = sum(1 for r in rows if r["pass"]) / len(rows)
    return {"pass_rate": pass_rate, "rows": rows}

def save_reports(report: Dict[str,Any], csv_path="rag_eval_report.csv", md_path="rag_eval_report.md"):
    rows = report["rows"]
    # CSV
    with open(csv_path, "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
        w.writeheader(); w.writerows(rows)
    # Markdown (краткий)
    lines = [f"# RAG Eval Report", f"pass@1 = {report['pass_rate']:.2f}", "", "| id | pass | grounded | cit_ok | reasons |", "|---|---:|---:|---:|---|"]
    for r in rows:
        lines.append(f"| {r['id']} | {'✅' if r['pass'] else '❌'} | {r['groundedness']:.2f} | {'✅' if r['cit_ok'] else '❌'} | {r['reasons']} |")
    Path(md_path).write_text("\n".join(lines), encoding="utf-8")
    print(f"CSV → {csv_path}\nMD  → {md_path}")

if __name__ == "__main__":
    rpt = run_eval()
    print(f"pass@1 = {rpt['pass_rate']:.2f}")
    save_reports(rpt)

4) Как запускать

1. Сохраните два файла рядом: rag_cases.jsonl и rag_eval.py.
2. Запустите:

   python rag_eval.py

   Вы увидите pass@1 и получите два отчёта: rag_eval_report.csv и rag_eval_report.md.
3. Подключите свою модель: замените в rag_eval.py функцию call_model() на реальный вызов (ваш RAG/FAQ эндпоинт).

5) Автопроверки (микро-лаба)

• Ответы содержат цитаты [doc_id], и они входят в allow_cites кейса.
• Все числа ответа встречаются в контексте (иначе «галлюцинация» по числам).
• В ответе есть обязательные слова/части слов из must_include.

6) Практические подсказки

• Добавьте ещё один сигнал length_ok (например, ≤ 60 слов) и правило «если нет цитат — мягкий отказ».
• Ночной прогон на реальных логах (50–200 вопросов) даст стабильную базу для регресс-тестов.
• В отчёте храните версию промпта/модели и хеш датасета, чтобы сравнивать релизы честно.