Паттерны и антипаттерны

Хуки — небольшая точка расширения, и применять их хорошо или плохо одинаково легко. Эта страница собирает приёмы, которые выдерживают проверку в продакшене, и те, что кусаются.

• Паттерны
  • Журнал аудита
  • Маскирование PII
  • Кэширование
  • Метрики
  • Ограничители
  • Gating по уверенности
  • Переопределение маршрутизации
  • Жизненный цикл модели
  • Мультитенантный контекст
  • Композиция
  • Локальная инструментация
  • Инструментация в масштабе процесса
  • Настройка бюджета токенов
• Антипаттерны
  • Блокирующая работа
  • Исключения из конечных хуков для управления потоком
  • Общее изменяемое состояние без блокировки
  • Тихий сбой
  • Удержание контекстов
  • Маскирование слишком поздно
  • Логика по вопросам в хуке на вызов
  • Рекурсивный predict
  • Простые callable в hooks=
  • Предположение, что результаты есть в конечных хуках
  • Хуки, зависящие от порядка

Паттерны

Журнал аудита

Записывайте каждое решение с данными, достаточными для его восстановления: состояние, вопросы, ответы, модель, решение о маршрутизации, использование и задержку.

import json

def audit(ctx):
    for state, result in zip(ctx.states, ctx.results or []):
        json.dump({
            "run_id": ctx.run_id,
            "model": ctx.model,
            "state": state,
            "routing": result.get("routing"),
            "answers": result["answers"],
            "usage": result.get("usage"),
            "call_usage": ctx.usage,
            "call_elapsed_ms": round(ctx.elapsed_ms or 0.0, 3),
        }, sys.stdout)
        sys.stdout.write("\n")

laya.load("convaiinnovations/laya", on_predict_end=audit)

Хук срабатывает один раз на вызов, а вызов predict_batch несёт в себе все состояния, поэтому запись пишется на каждое решение: ctx.states и ctx.results выровнены по индексу. ctx.usage и ctx.elapsed_ms — это итоги по всему вызову; каждый результат несёт свой usage.

Сделайте его нестрогим, если потеря строки журнала не должна приводить к сбою запроса: hooks_raise=False. Сделайте его строгим, если журнал аудита — требование соответствия.

Маскирование PII

Маскирование должно происходить в on_predict_start, до токенизации, иначе модель уже увидела данные.

import re
EMAIL = re.compile(r"\b[\w.+-]+@[\w-]+\.[\w.-]+\b")

def redact(ctx):
    ctx.states = [
        EMAIL.sub("[email]", s) if isinstance(s, str) else s
        for s in ctx.states
    ]

laya.load("convaiinnovations/laya", on_predict_start=redact)

Хук маскирования — это хук политики: оставьте hooks_raise=True, потому что молча сломавшийся механизм маскирования — это утечка данных.

Кэширование

Стартовый хук проверяет кэш и вызывает ctx.skip(...); конечный хук его заполняет. При попадании прямой проход пропускается.

import hashlib, json

CACHE = {}

def key(ctx, index):
    # Not sort_keys=True: criteria order is positional, so two orders are two questions,
    # and the checkpoint and token budget change the answer too.
    payload = json.dumps([ctx.states[index], ctx.questions, ctx.model,
                          ctx.max_len, ctx.head_max_len], default=str)
    return hashlib.sha256(payload.encode()).hexdigest()

def read(ctx):
    hits = [CACHE.get(key(ctx, i)) for i in range(len(ctx.states))]
    if all(hit is not None for hit in hits):
        ctx.skip(hits)   # one per state: skip replaces the whole call

def write(ctx):
    for i, result in enumerate(ctx.results or []):
        CACHE[key(ctx, i)] = result

laya.load("convaiinnovations/laya", on_predict_start=read, on_predict_end=write)

Хуки срабатывают один раз на вызов, поэтому ключа по одному состоянию недостаточно в predict_batch: ctx.skip() заменяет каждый результат, который вернул бы вызов. Защищайте кэш блокировкой при конкурентном обслуживании. В Router кэшированный payload всё равно получает ключ routing, поэтому форма возвращаемого значения не меняется.

Та же пара работает на отдельном узле LangChain через его аргумент hooks=, — это способ закэшировать один горячий шаг в графе, не меняя того, что видит любой другой вызывающий этого агента.

Метрики

Счётчики и гистограммы из ctx.model, ctx.usage и ctx.elapsed_ms. Держите хук нестрогим.

COUNTS, LATENCIES = {}, []

def metrics(ctx):
    COUNTS[ctx.model] = COUNTS.get(ctx.model, 0) + 1
    if ctx.elapsed_ms is not None:
        LATENCIES.append(ctx.elapsed_ms)

laya.load("convaiinnovations/laya", on_predict_end=metrics, hooks_raise=False)

Ограничители

Хук политики возбуждает исключение, чтобы заблокировать запрос. hooks_raise=True (значение по умолчанию) позволяет блокировке дойти до вызывающего; on_error и on_predict_end всё равно выполняются, поэтому журнал аудита это фиксирует.

class Blocked(Exception):
    pass

def guard(ctx):
    if any("ssn" in str(state).lower() for state in ctx.states):
        raise Blocked("possible PII in state")

laya.load("convaiinnovations/laya", on_predict_start=guard)

Проверяйте его на форме состояния. Ограничитель, который читает только ctx.states[0], блокирует один вызов и позволяет вызову predict_batch пропустить все остальные состояния через прямой проход.

Gating по уверенности

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

def gate(ctx):
    for result in ctx.results or []:
        answer = result["answers"].get("dept")
        if answer and answer["confidence"] < 0.6:
            answer["choice"] = "human-review"
            answer["gated"] = True

laya.load("convaiinnovations/laya", on_predict_end=gate)

Меняйте через ctx.results, который хранит по одному dict на каждое состояние вызова: gating только первого отправит все остальные ответы с низкой уверенностью без аннотации.

Переопределение маршрутизации

on_route может заменить ctx.decision, чтобы закрепить чекпойнт за классом трафика.

from laya.router import RouteDecision

def pin(ctx):
    if "refund" in str(ctx.states[0]).lower():
        ctx.decision = RouteDecision(
            model="typed-decisions",
            repo="convaiinnovations/laya/typed-decisions",
            reason="refund workflow",
            detection=None,
            workflow=None,
        )

Router(hooks=[pin])

Жизненный цикл модели

on_load и on_evict наблюдают за чекпойнтами. Используйте их для логов прогрева, учёта памяти или оповещений о вытеснении. Они выполняются вне блокировки Router, поэтому хук может снова вызвать Router.

class Lifecycle:
    def on_load(self, ctx):
        print("loaded", ctx.model)

    def on_evict(self, ctx):
        print("evicted", ctx.model)

Router(hooks=[Lifecycle()])

Мультитенантный контекст

Проведите идентификатор тенанта, захватив его в замыкании хука или прочитав из context-local. Не храните состояние на уровне запроса в объекте хука без блокировки.

def make_audit(tenant):
    def audit(ctx):
        ship(tenant, ctx.run_id, ctx.results)
    return audit

agent = laya.load("convaiinnovations/laya", on_predict_end=make_audit("acme"))

Композиция

Несколько хуков разных типов компонуются естественно; установленные хуки выполняются первыми, по порядку.

agent = laya.load(
    "convaiinnovations/laya",
    hooks=[Metrics(), Guardrail()],     # metrics first, then policy
    on_predict_start=redact,            # convenience callables appended after hooks
    hooks_raise=True,                   # policy failures are fatal
)

Держите порядок осознанным и задокументированным, потому что последующий хук видит мутации предыдущего.

Локальная инструментация

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

with agent.hooks_installed(DebugDump()):
    agent.system_one(state, questions)   # DebugDump only here

add_hook/remove_hook делают то же без блока — для трассировщика, живущего столько же, сколько процесс.

Инструментация в масштабе процесса

Хук трассировки или метрик, который должно видеть каждое решение, можно зарегистрировать один раз, а не передавать в каждый Agent и Router. Значения по умолчанию выполняются перед хуками экземпляра и вызова.

from laya import BaseHook, hooks

class Metrics(BaseHook):
    def on_predict_end(self, ctx):
        record(ctx.model, ctx.elapsed_ms)

hooks.set_default_hooks(hooks=[Metrics()])

Это глобальное состояние, поэтому задавайте его осознанно: установите один раз при запуске, и вызывайте clear_default_hooks() в тестах, чтобы один тест не протёк хуком в следующий.

Настройка бюджета токенов

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

• ctx.head_max_len стартового хука заменяет бюджет вызова. До него действует собственное значение вызывающего на вызов или значение по умолчанию чекпойнта в ctx.agent.cfg — так что сравнивайте с этим: запись простого числа может понизить бюджет, который вызывающий уже задал.
• Один вызов отвечает на все несомые им вопросы, поэтому рассчитывайте размер по самому широкому из них, а не по тому, который случится первым.
• Как только варианты перестают помещаться в head, laya/common.py даёт каждому из них max(4, (head_max_len - 16) // k) токенов. Поэтому 16 + 4 * k попадает ровно на этот минимум: каждая метка всё ещё обрезается до токенов, общих с остальными, — это и есть схлопывание, ради избегания которого писался хук. 16 + 8 * k оставляет их различимыми.
• Состоянию достаётся max_len - head_max_len - 8 токенов, поэтому расширенный head должен расширить и max_len, иначе состояние теряет своё окно.

def widen_for_high_cardinality(ctx):
    k = max((len(q.get("criteria", {}) or {}) for q in ctx.questions.values()), default=0)
    if k < 50:
        return
    cfg = getattr(ctx.agent, "cfg", None) or {}
    head = ctx.head_max_len if ctx.head_max_len is not None else cfg.get("head_max_len", 192)
    window = ctx.max_len if ctx.max_len is not None else cfg.get("max_len", 512)
    need = 16 + 8 * k                          # 8 tokens per label, not the core's floor of 4
    if need > head:                            # only ever widen, never lower
        ctx.head_max_len = need
        ctx.max_len = max(window, need + 8 + 64)   # 8 reserved, then room for the state

agent = laya.load("convaiinnovations/laya", on_predict_start=widen_for_high_cardinality)

Это не трогает общую конфигурацию агента, поэтому конкурентные вызовы не затронуты. Те же ручки доступны на вызов: agent.system_one(state, questions, head_max_len=512, max_len=1024).

Расширение не бесплатно: более длинное окно означает больший тензор, а чекпойнты обучались на 512 (laya) и 1,024 токенах. За этим пределом сужение кандидатов с помощью predict_shortlist выигрывает у растягивания бюджета.

Антипаттерны

Блокирующая работа

Хуки выполняются в вызывающем потоке, а laya.serve использует единственный воркер инференса. Хук, который спит, ждёт сетевого обмена или вызывает input(), задерживает все остальные запросы за собой.

# bad: blocks the whole server
def audit(ctx):
    requests.post("https://slow.example/decisions", json=..., timeout=30)

# better: enqueue, let a background worker ship it
def audit(ctx):
    QUEUE.put_nowait(record(ctx))

Если приходится делать медленную работу, задайте hooks_concurrent=False, чтобы хотя бы сам хук не перекрывался, и запускайте laya.serve за очередью.

Исключения из конечных хуков для управления потоком

on_predict_end выполняется после инференса. Исключение там выбрасывает уже вычисленный результат и на успешном пути всплывает до вызывающего. Используйте стартовый хук, чтобы заблокировать до оплаты инференса, или перепишите ctx.results, чтобы изменить ответ.

Общее изменяемое состояние без блокировки

Один и тот же экземпляр хука выполняется во многих потоках. self.counter += 1 порождает гонку.

# bad
class Count:
    def __init__(self): self.n = 0
    def on_predict_end(self, ctx): self.n += 1

# good
import threading
class Count:
    def __init__(self):
        self.n = 0
        self._lock = threading.Lock()
    def on_predict_end(self, ctx):
        with self._lock:
            self.n += 1

Тихий сбой

hooks_raise=False предупреждает один раз на сбой, но хук, который сам ловит всё, скрывает реальные проблемы.

# bad: no one will ever know the audit trail stopped
def audit(ctx):
    try:
        ship(record(ctx))
    except Exception:
        pass

Если хук необязателен, позвольте hooks_raise=False обработать это и следите за предупреждениями. Если нет — пусть возбуждает исключение.

Удержание контекстов

Хук, который добавляет ctx в список, удерживает живыми всё состояние, вопросы, результаты и агента.

# bad: unbounded memory growth
SEEN = []
def audit(ctx):
    SEEN.append(ctx)

# good: keep only what you need
SEEN = []
def audit(ctx):
    SEEN.append((ctx.run_id, ctx.model, ctx.elapsed_ms))

Маскирование слишком поздно

К моменту on_predict_end модель уже токенизировала состояние. Маскируйте в on_predict_start.

Логика по вопросам в хуке на вызов

На каждый вызов приходится один PredictContext, и один прямой проход отвечает на все вопросы. Событий по отдельным вопросам нет. Перебирайте ответы внутри on_predict_end, а также перебирайте состояния: в батче один контекст несёт все состояния вызова.

def flag(ctx):
    for result in ctx.results or []:
        for qid, answer in result["answers"].items():
            if answer.get("confidence", 1.0) < 0.5:
                alert(qid, ctx.run_id)

Рекурсивный predict

Хук, который вызывает agent.predict/system_one, запускает хуки заново. Без защиты по глубине это уходит в рекурсию.

# bad
def enrich(ctx):
    ctx.results = [agent.predict(ctx.states[0], EXTRA_QUESTIONS)]

# good: guard, or use a separate agent with no hooks
def enrich(ctx):
    if getattr(ctx, "_enriched", False):
        return
    ctx._enriched = True
    ctx.results = [enricher.predict(state, EXTRA_QUESTIONS) for state in ctx.states]

Простые callable в hooks=

hooks= принимает объекты хуков; голый callable не говорит, для какого события он, поэтому он отклоняется. Используйте on_predict_start= / on_predict_end=.

# bad: TypeError
laya.load("convaiinnovations/laya", hooks=[lambda ctx: None])

# good
laya.load("convaiinnovations/laya", on_predict_end=lambda ctx: None)

Предположение, что результаты есть в конечных хуках

На пути сбоя ctx.results равно None, если только стартовый хук его не установил. Всегда проверяйте.

def audit(ctx):
    if ctx.results is None:
        log_failure(ctx.run_id, ctx.error)
        return
    log_success(ctx.run_id, ctx.results)

Хуки, зависящие от порядка

Хук, который читает мутацию из другого хука, хрупок, если порядок не зафиксирован. Установленные хуки выполняются в порядке списка, затем callable для удобства; документируйте любую связанность или объедините связанные хуки в один объект.

См. также

• Ошибки: матрица сбоев, стоящая за несколькими из этих антипаттернов.
• Примеры: более полные версии паттернов выше.