Справочник по API

Всё на этой странице можно импортировать из laya (общие имена) или из laya.hooks (вся поверхность).

from laya import PredictContext, PredictHook, Hook
from laya.hooks import HOOK_EVENTS, normalise_hooks, dispatch, aggregate_usage

• PredictContext
• Протокол hook
• Удобные типы
• Поверхность конфигурации
• Payload событий
• Валидация
• Продвинутые helpers

PredictContext

PredictContext создаётся один раз на каждый публичный вызов и передаётся каждому hook этого вызова. Router.predict_batch — единственное исключение: он создаёт по одному на запрос, как это сделал бы вызов Router.predict для этого запроса, поэтому его hooks уровня Router срабатывают один раз на запрос с одним состоянием каждый. Он изменяемый: hooks могут переписывать states, questions и results, а on_route может переписывать decision. Он использует равенство по идентичности (eq=False), поэтому контекст хэшируем и два контекста никогда не равны.

@dataclass(eq=False)
class PredictContext:
    states: List[Any]
    questions: Dict[str, Any]
    run_id: str = <uuid4 hex>
    results: Optional[List[Dict[str, Any]]] = None
    decision: Optional[Dict[str, Any]] = None
    model: Optional[str] = None
    agent: Any = None
    router: Any = None
    max_len: Optional[int] = None
    head_max_len: Optional[int] = None
    usage: Optional[Dict[str, int]] = None
    started_at: float = <perf_counter()>
    elapsed_ms: Optional[float] = None
    error: Optional[BaseException] = None

PredictContext.skip(results)

Закорачивает инференс. Вызванный из on_predict_start, он задаёт ctx.results, так что прямой проход пропускается; on_predict_end всё равно выполняется, и переданные результаты возвращаются.

Ключ должен покрывать всё, от чего зависит ответ, а hook — всё, что несёт вызов: hooks срабатывают один раз на вызов, а predict_batch вызывает их сразу со всеми состояниями.

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 cache_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 entry per state, same shape as predict_batch's return

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

laya.load("convaiinnovations/laya", on_predict_start=cache_read, on_predict_end=cache_write)

tests/test_hooks_api.py выполняет этот блок, examples/hooks/cache.py и блоки кэширования из docs/hooks/patterns.md и docs/hooks/examples.md, и проверяет все четыре ключа одинаково, поэтому страница не может учить ключу, от которого пример уже отошёл.

На Router в пропущенный payload добавляется ключ routing (не перезаписывая уже имеющийся), поэтому Router.predict сохраняет свою документированную форму возврата.

Протокол hook

Hook — это typing.Protocol. Реализуйте любое подмножество методов; остальные пропускаются.

class Hook(Protocol):
    def on_predict_start(self, ctx: PredictContext) -> None: ...
    def on_predict_end(self, ctx: PredictContext) -> None: ...
    def on_route(self, ctx: PredictContext) -> None: ...
    def on_load(self, ctx: PredictContext) -> None: ...
    def on_evict(self, ctx: PredictContext) -> None: ...
    def on_error(self, ctx: PredictContext) -> None: ...

Hook может свободно определять дополнительные атрибуты и методы; учитываются только шесть имён событий. Если hook определяет одно из шести как невызываемое, конфигурация падает сразу (см. Валидация).

BaseHook

BaseHook — конкретный двойник протокола: класс с пустым телом для каждого события. Создайте подкласс и переопределите только нужные события.

from laya import BaseHook

class Audit(BaseHook):
    def on_predict_end(self, ctx):
        ship(ctx.run_id, ctx.results)

Hook лучше, когда нужна структурная типизация (любой объект с нужными методами); BaseHook лучше, когда нужна явная база для наследования и вызова super().

Удобные типы

PredictHook = Callable[[PredictContext], None]

PredictHook — это тип обычного callable-объекта, используемого с on_predict_start= / on_predict_end=. Передайте один callable-объект или последовательность; каждый оборачивается в минимальный hook.

Регистрация во время выполнения

Каждая среда выполнения подмешивает HookRegistry, поэтому hooks можно добавлять, удалять или ограничивать по области после создания. Изменение потокобезопасно; вызов читает снимок списка, поэтому добавление или удаление hook никогда не мешает выполняющемуся вызову.

agent.add_hook(tracer)              # one hook or a sequence; returns self for chaining
agent.remove_hook(tracer)           # by identity; True if it was installed

with agent.hooks_installed(debug):  # installed for the block, removed on exit
    agent.system_one(state, questions)

add_hook принимает те же объекты, что и hooks= (не обычные callable-объекты). hooks_installed принимает любое число объектов hook или последовательностей и при выходе восстанавливает предыдущий список, в том числе когда блок вызывает исключение.

Значения по умолчанию для всего процесса

laya.hooks хранит небольшой реестр для всего процесса, поэтому tracer, hook метрик или разметчик tenant’ов не нужно протягивать через каждый Agent и Router. Значения по умолчанию выполняются первыми, затем hooks, установленные на экземпляре, затем hooks на вызов.

from laya import hooks

hooks.set_default_hooks(hooks=[Tracer()])          # replaces the set, accepts the hooks= arguments
hooks.add_default_hook(Metrics())                  # appends
hooks.clear_default_hooks()                        # removes everything
hooks.default_hooks()                              # a copy of the current list

hooks.compose_hooks(agent.hooks)                   # defaults + installed (advanced)

Значения по умолчанию применяются к каждому событию, включая события жизненного цикла Router on_load и on_evict. Реестр читается во время вызова, поэтому hooks, заданные после создания Agent или Router, всё равно применяются. Нет отказа для отдельного экземпляра; вызовите clear_default_hooks(), чтобы отключить набор для всего процесса.

Асинхронные hooks

Событие может быть корутиной. Оберните hook в AsyncHook, и его методы async def выполняются до завершения в синхронном ядре:

from laya import AsyncHook

class Remote:
    async def on_predict_end(self, ctx):
        await ship(ctx.results)

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

Обычный асинхронный callable-объект, переданный в on_predict_start= / on_predict_end=, тоже работает, потому что dispatch выполняет любой awaitable, который вернёт hook.

Где выполняется корутина:

• Если у вызывающего потока нет запущенного цикла, она выполняется через asyncio.run.
• Если он уже есть (вызывающий внутри асинхронной функции), она выполняется в выделенном фоновом цикле, поэтому вызывающий поток может блокироваться без взаимной блокировки. Передайте AsyncHook(hook, loop=...), чтобы направить её в конкретный цикл; он должен быть запущен и не должен быть собственным циклом вызывающего потока. Проверяются оба: остановленный цикл и собственный цикл вызывающего каждый вызывает ValueError вместо бесконечной блокировки.

Hook без async-методов не затрагивается.

Поверхность конфигурации

Каждая точка входа принимает одни и те же параметры hook. hooks принимает объект или последовательность объектов; on_predict_start / on_predict_end принимают callable-объект или последовательность.

Agent

Agent(
    model_id_or_path="convaiinnovations/laya",
    device=None, token=None, subfolder=None, fast=False, compile=False,
    hooks=None, on_predict_start=None, on_predict_end=None,
    hooks_raise=True, hooks_concurrent=True, hooks_timeout=None,
)

load(..., hooks=None, on_predict_start=None, on_predict_end=None,
     hooks_raise=True, hooks_concurrent=True, hooks_timeout=None)

agent.predict_batch(states, questions, batch_size=None,
                    hooks=None, on_predict_start=None, on_predict_end=None, hooks_raise=None,
                    hooks_timeout=None, max_len=None, head_max_len=None, sort_by_length=False)

agent.system_one(state, questions,
                 hooks=None, on_predict_start=None, on_predict_end=None, hooks_raise=None,
                 hooks_timeout=None, max_len=None, head_max_len=None)

agent.predict_long(state, questions, window=None, stride=None, aggregate="auto",
                   batch_size=None, lang=None,
                   hooks=None, on_predict_start=None, on_predict_end=None, hooks_raise=None,
                   hooks_timeout=None)

agent.predict(...)          # alias of system_one

• hooks_raise и hooks_timeout в методе на вызов по умолчанию равны None, что означает «использовать значение экземпляра».

• hooks_concurrent существует только на уровне экземпляра.

• В predict_long hooks оборачивают инференс, который отвечает на состояние; для документа, требующего нескольких окон, это единственный общий predict_batch поверх них: on_predict_start срабатывает один раз, и ctx.states содержит декодированные тексты окон в порядке сканирования, а не состояние вызывающего, которое было токенизировано, чтобы их создать. states изменяем из hook начала, поэтому сканирование, доходящее до инференса, не обязано быть разбиением, которое вычислил predict_long. Меняется то, что может утверждать ответ:

  что сделал hook начала	usage["windows"]	answer["window"]
  ответил через ctx.skip([result])	0	отсутствует — ни одно окно его не оценивало
  оставил сканирование как было построено	N	присутствует — index, token_start/token_end называют решающий участок
  заменил сканирование любым способом	состояния, которые были оценены	отсутствует — смещения описывают окна predict_long, а не текст, который был оценён

  usage["windows"] — это сумма по каждому пути, включая тот, где состояние уже помещалось в одно окно, поэтому кэшированный ответ никогда не читается как окно, которое прочитала модель.

Router

Router(
    models=None, device=None, token=None, max_loaded=2, default="english",
    auto_task_detection=False, standalone_repos=False, preload=False, lang_guess=None,
    hooks=None, on_predict_start=None, on_predict_end=None,
    hooks_raise=True, hooks_concurrent=True,
)

router.route(state, questions=None, model=None, task=None, lang=None, lang_guess=None,
             hooks=None, hooks_raise=None)

router.predict(state, questions, model=None, task=None, lang=None, lang_guess=None,
               hooks=None, on_predict_start=None, on_predict_end=None, hooks_raise=None,
               hooks_timeout=None, max_len=None, head_max_len=None)

router.predict_batch(requests, batch_size=None, hooks_timeout=None, min_confidence=None,
                     sort_by_length=False, hooks=None, on_predict_start=None, on_predict_end=None,
                     hooks_raise=None)

router.system_one(...)      # alias of predict
router.load(name)           # builds on first use; fires on_load
router.preload(names=None)  # builds several; fires on_load per build
router.unload(name=None)    # frees one or all; fires on_evict
router.attach(name, agent)  # registers an existing agent; does not fire on_load
router.loaded               # list of resident checkpoint names

• hooks= на вызов в route, route_batch, predict и predict_batch применяется ко всему вызову, включая on_route. В predict_batch список составляется так же, как в predict (сначала установленные hooks, затем список на вызов; None и [] ничего не добавляют), и выполняется один раз на запрос.
• route() публичный: его вызов диспетчеризует on_route с установленными hooks плюс любые hooks на вызов.

ONNXAgent

ONNXAgent(model_id_or_path, onnx_path="laya.onnx", subfolder=None,
          hooks=None, on_predict_start=None, on_predict_end=None,
          hooks_raise=True, hooks_concurrent=True)

onnx_agent.system_one(state, questions,
                      hooks=None, on_predict_start=None, on_predict_end=None, hooks_raise=None,
                      hooks_timeout=None, max_len=None, head_max_len=None)

onnx_agent.predict(...)     # alias of system_one

У ONNXAgent нет Router, поэтому он предоставляет только события уровня predict.

Payload событий

Какие поля заполнены, по событиям и средам выполнения:

Детали по времени:

• on_predict_end видит results на пути успеха. На пути сбоя results равно None, если hook начала не задал их через skip(), и usage поэтому тоже None (оно выводится из results); elapsed_ms всегда задано.
• on_error выполняется до блока finally, который вычисляет elapsed_ms и usage, поэтому там оба равны None. Читайте время и usage из on_predict_end вместо этого.
• run_id всегда заполнено.

Валидация

Конфигурация проверяется, когда hooks нормализуются, что происходит при создании для установленных hooks и во время вызова для hooks на вызов. Следующее вызывает TypeError:

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

Продвинутые helpers

Они используются внутри и стабильны, но большинству пользователей не нужны.

HOOK_EVENTS          # tuple of the six event names, in dispatch order
normalise_hooks(hooks=None, on_predict_start=None, on_predict_end=None) -> list
dispatch(hooks, event, ctx, *, raise_errors=True, lock=None) -> None
aggregate_usage(results) -> {"input_tokens": int, "output_tokens": int}

dispatch(hooks, event, ctx, *, raise_errors=True, lock=None, timeout=None)
run_coroutine_sync(coro, loop=None)

normalise_hooks разворачивает объект/последовательность hooks и два callable-объекта в один упорядоченный список. dispatch вызывает event на каждом hook, который его реализует, применяя политику raise, блокировку и тайм-аут, и выполняет результат hook, если он awaitable. run_coroutine_sync выполняет awaitable до завершения из синхронного кода, в цикле вызывающего, если он свободен, или в фоновом цикле, если у вызывающего уже есть свой. aggregate_usage суммирует блоки usage по состояниям.

from laya.hooks import normalise_hooks, dispatch, PredictContext

hooks = normalise_hooks(on_predict_start=[log, redact])
ctx = PredictContext(states=["..."], questions={...})
dispatch(hooks, "on_predict_start", ctx)

См. также

• Жизненный цикл: когда выполняется каждое событие, со схемами потока.
• Ошибки: матрица сбоев и правила цепочки.
• Паттерны и антипаттерны: как правильно структурировать hooks.
• Примеры: рецепты для каждого случая использования.