jevlaya
Русский
Документация

Трассировка

Все hooks одного вызова получают один и тот же PredictContext.run_id, поэтому tracer может сопоставить события начала, конца и ошибки (и любые span’ы, которые он открывает) без собственного учёта.

run_id

  • Строка uuid4().hex, создаваемая один раз на публичный вызов (predict_batch, system_one, Router.predict, ONNXAgent.system_one). Router.predict_batch вместо этого создаёт по одной на запрос — ту run_id, которая была бы у вызова Router.predict для этого запроса.
  • Её разделяют все hooks этого вызова, включая on_error и on_predict_end.
  • Не глобальная и не сохраняется: она идентифицирует вызов внутри процесса. Кладите её в логи и исходящие payload’ы, чтобы сопоставлять данные между системами.
  • Различна для каждого вызова, поэтому два вызова никогда не совпадают.
call A:  run_id=1a2b...  start ──┐
                                 ├─ end
                          error ─┘
call B:  run_id=9f8e...  start ─── end

Минимальный tracer

Держите отображение из run_id в span, открытый вами при старте, и закрывайте его при завершении или ошибке.

import time

class Tracer:
    def __init__(self):
        self.spans = {}

    def on_predict_start(self, ctx):
        self.spans[ctx.run_id] = {
            "model": ctx.model,
            "started_at": ctx.started_at,
        }

    def on_predict_end(self, ctx):
        span = self.spans.pop(ctx.run_id, None)
        if span is None:
            return
        emit_span(
            name="laya.predict",
            run_id=ctx.run_id,
            model=ctx.model,
            duration_ms=ctx.elapsed_ms,
            usage=ctx.usage,
            ok=ctx.error is None,
        )

    def on_error(self, ctx):
        # on_error runs before on_predict_end; leaving the span for on_predict_end is fine,
        # or close it here if you prefer.
        pass

agent = laya.load("convaiinnovations/laya", hooks=[Tracer()])

Поскольку on_predict_end выполняется всегда, это естественное место для закрытия span’а, и он может видеть ctx.error на пути сбоя.

Время жизни span

on_predict_start ──► open span (run_id, model, started_at)
      │
      ├─ inference
      │
on_error ──► record ctx.error on the span
      │
on_predict_end ──► close span (elapsed_ms, usage, ok)

Span ошибок

on_predict_end выполняется на пути сбоя с установленным ctx.error, поэтому одно место закрытия покрывает оба случая:

def on_predict_end(self, ctx):
    span = self.spans.pop(ctx.run_id, None)
    if span is None:
        return
    if ctx.error is not None:
        span["status"] = "error"
        span["error_type"] = type(ctx.error).__name__
        span["error_message"] = str(ctx.error)
    span["duration_ms"] = ctx.elapsed_ms
    emit(span)

Если вы реализуете только on_error, помните, что он срабатывает до on_predict_end; не закрывайте span в обоих местах, иначе посчитаете дважды.

OpenTelemetry

Пример examples/hooks/otel.py записывает счётчики и гистограмму. Для настоящих span’ов управляйте API OTel из tracer’а. Hooks синхронны, поэтому используйте синхронный экспортёр (или ставьте в очередь и экспортируйте из воркера).

from opentelemetry import trace

tracer = trace.get_tracer("laya")

class OTelHooks:
    def __init__(self):
        self.spans = {}

    def on_predict_start(self, ctx):
        span = tracer.start_span("laya.predict", attributes={"laya.run_id": ctx.run_id, "laya.model": ctx.model})
        self.spans[ctx.run_id] = span

    def on_predict_end(self, ctx):
        span = self.spans.pop(ctx.run_id, None)
        if span is None:
            return
        if ctx.usage:
            span.set_attribute("laya.input_tokens", ctx.usage["input_tokens"])
        if ctx.error is not None:
            span.record_exception(ctx.error)
            span.set_status(trace.Status(trace.StatusCode.ERROR))
        span.end()

laya.load("convaiinnovations/laya", hooks=[OTelHooks()], hooks_raise=False)

Установите hooks_raise=False, чтобы сбой tracer’а никогда не приводил к отказу запроса.

Распределённое распространение

run_id — обычная строка, поэтому включайте её во всё, что покидает процесс: строки логов, payload, отправляемый в сервис аудита, или HTTP-заголовок, если решение запускает нисходящий вызов.

def audit(ctx):
    requests.post(
        "https://audit.example/decisions",
        json=record(ctx),
        headers={"X-Laya-Run-Id": ctx.run_id},
        timeout=2,
    )

Помните, что блокирующий вызов вроде этого останавливает вызывающий поток; вместо этого ставьте его в очередь, когда обслуживаете параллельно. Смотрите паттерны.

Вложенные вызовы

Hook, который снова вызывает predict, начинает новый вызов с новой run_id. Родитель и потомок независимы, если только вы не свяжете их сами. Захватите id родителя и передайте его дальше:

def enrich(ctx):
    for state in ctx.states:                     # a hook sees every state of the call
        child = enricher.predict(state, EXTRA_QUESTIONS)
        record_child_span(parent_run_id=ctx.run_id, child_run_id=child.get("run_id"))

Одна родительская run_id, один дочерний вызов на состояние. Тело, читающее ctx.states[0], связывает первое решение батча и молча отбрасывает остальные.

Защищайтесь от рекурсии (смотрите антипаттерны); самая простая защита — отдельный агент enricher без hooks.

Смотрите также