Введение в LLMOps

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

Содержание курса

Трейс одной сессии: trace_id/span_id в JSON-логах

Цель. Научиться логировать «сквозной» запрос с trace_id и иерархией span_id → быстро видеть, что делали retriever, llm.generate и postprocess, сколько это заняло и чем закончилось.

1) Мини-теория

trace_id — общий идентификатор для всех шагов одного запроса.
span — отдельная операция (например, retriever), имеет span_id, parent_span_id, start/end, атрибуты и статус.
Что логируем: события span_start/span_end в структурном JSON: имя, атрибуты, задержка, статус.

2) Готовый код (без внешних зависимостей)

Скрипт ниже реализует простый трассер на contextvars. Он печатает JSON-строки (подходит для последующей загрузки в ClickHouse/ELK/BigQuery).

# trace_demo.py
from __future__ import annotations
import json, time, uuid, random, contextvars
from dataclasses import dataclass
from typing import Any, Dict, List, Optional

# --- контекст запроса (сквозной) ---
trace_id_var = contextvars.ContextVar("trace_id", default="")
span_stack_var = contextvars.ContextVar("span_stack", default=[])

LOGS: List[Dict[str, Any]] = []  # буфер для автопроверок

def now_iso() -> str:
    import datetime as dt
    return dt.datetime.utcnow().isoformat(timespec="milliseconds") + "Z"

def log(event: str, **fields):
    rec = {"ts": now_iso(), "trace_id": trace_id_var.get(""), "event": event, **fields}
    line = json.dumps(rec, ensure_ascii=False)
    print(line)
    LOGS.append(rec)

# --- примитивный span-контекст ---
@dataclass
class _SpanCtx:
    span_id: str
    parent_span_id: Optional[str]
    name: str
    t0: float

class span:
    def __init__(self, name: str, **attrs):
        self.name = name
        self.attrs = attrs
        self.ctx: Optional[_SpanCtx] = None
    def __enter__(self):
        stack = list(span_stack_var.get())
        parent = stack[-1].span_id if stack else None
        self.ctx = _SpanCtx(span_id=uuid.uuid4().hex[:16], parent_span_id=parent, name=self.name, t0=time.perf_counter())
        span_stack_var.set(stack + [self.ctx])
        log("span_start", name=self.name, span_id=self.ctx.span_id, parent_span_id=self.ctx.parent_span_id, **self.attrs)
        return self
    def __exit__(self, exc_type, exc, tb):
        assert self.ctx is not None
        elapsed_ms = int((time.perf_counter() - self.ctx.t0) * 1000)
        # pop
        stack = list(span_stack_var.get())
        span_stack_var.set(stack[:-1])
        status = "ERROR" if exc_type else "OK"
        log("span_end", name=self.name, span_id=self.ctx.span_id, parent_span_id=self.ctx.parent_span_id, latency_ms=elapsed_ms, status=status)
        # не подавляем исключения
        return False

# --- «бизнес»-функции пайплайна (симуляция) ---
def retriever(query: str, k=3) -> List[str]:
    with span("retriever", k=k):
        time.sleep(random.uniform(0.01, 0.03))
        ids = ["d1", "d3", "d5"][:k]
        log("retriever.result", retrieved_ids=",".join(ids), k=k)
        return ids

def llm_generate(query: str, context_ids: List[str], model="demo-pro", prompt_version="v1") -> str:
    with span("llm.generate", model=model, prompt_version=prompt_version):
        toks = random.randint(80, 140)
        time.sleep(0.005 * toks / 10)  # имитация работы
        txt = f"Ответ из модели {model} с цитатами [{context_ids[0]}]."
        log("llm.tokens", tokens_out=toks, model=model)
        return txt

def postprocess(answer: str) -> str:
    with span("postprocess.validate", schema="faq_v1"):
        time.sleep(random.uniform(0.002, 0.006))
        ok = answer.endswith(".")
        log("validator.result", ok=ok)
        return answer if ok else (answer + ".")

# --- обработчик запроса ---
def handle_request(user_query: str, prompt_version="v1") -> str:
    # установить trace_id и корневой span
    trace_id_var.set(uuid.uuid4().hex)
    with span("request", user_query=user_query):
        ids = retriever(user_query, k=3)
        ans = llm_generate(user_query, ids, model="demo-pro", prompt_version=prompt_version)
        final = postprocess(ans)
        log("request.done", bytes=len(final.encode("utf-8")))
        return final

# --- автопроверки шага ---
def _self_test():
    LOGS.clear()
    out = handle_request("Сколько дней базового отпуска?")
    # 1) один trace_id у всех записей
    tids = {r["trace_id"] for r in LOGS}
    assert len(tids) == 1 and next(iter(tids)), "Должен быть единый непустой trace_id"
    # 2) есть 4 спана (request, retriever, llm.generate, postprocess.validate)
    names = [r["name"] for r in LOGS if r["event"] == "span_end"]
    for need in ("request", "retriever", "llm.generate", "postprocess.validate"):
        assert need in names, f"Нет завершающего спана {need}"
    # 3) все span_end имеют latency_ms > 0
    assert all((r.get("latency_ms", 1) > 0) for r in LOGS if r["event"] == "span_end")
    # 4) у дочерних спанов parent_span_id совпадает с span_id корня
    root_span = next(r for r in LOGS if r["event"]=="span_start" and r["name"]=="request")
    children = [r for r in LOGS if r.get("parent_span_id")==root_span["span_id"] and r.get("name") in ("retriever","llm.generate","postprocess.validate") and r["event"]=="span_start"]
    assert len(children) == 3, "Все дочерние спаны должны ссылаться на корень"
    print("\nАвтотесты пройдены ✓")

if __name__ == "__main__":
    _self_test()
    # пример: можно вызвать несколько раз
    # handle_request("Оплачивается ли больничный?")

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

Сохраните код в файл trace_demo.py.
Запустите:
```
python trace_demo.py
```
Скрипт выполнит автопроверки и напечатает JSON-логи с событиями span_start/span_end и тех. событиями (retriever.result, llm.tokens, validator.result, request.done).

4) Что проверить (микро-лаба)

У всех строк один и тот же trace_id (одна сессия).
Есть четыре спана: request → retriever → llm.generate → postprocess.validate (у дочерних верный parent_span_id).
У всех span_end есть поле latency_ms и статус OK.

5) Мини-чеклист для продакшна

Структурные логи (JSON), не свободный текст.
Обязательные поля: trace_id, span_id, parent_span_id, name, latency_ms, status, а также model, prompt_version, tokens_out для LLM-спана.
Сэмплинг: 100% ошибок + хвостовые по latency/cost; остальное — долями.
PII: перед записью редактируйте e-mail/телефоны; не логируйте секреты/ключи.

6) (Опционально) Апгрейд до OpenTelemetry

Когда будете готовы, замените кастомные спаны на OTel (opentelemetry-sdk + ConsoleSpanExporter или OTLP-экспорт в ваш бэкенд). Имена спанов и атрибуты сохраняйте такими же, чтобы не ломать анализ.