Referência da API

Tudo nesta página é importável de laya (os nomes comuns) ou de laya.hooks (a superfície inteira).

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

• PredictContext
• Protocolo de hook
• Tipos de conveniência
• Superfície de configuração
• Cargas úteis dos eventos
• Validação
• Auxiliares avançados

PredictContext

Um PredictContext é criado uma vez por chamada pública e passado a todo hook daquela chamada. Router.predict_batch é a única exceção: ele cria um por solicitação, como uma chamada Router.predict para aquela solicitação faria, então seus hooks no nível do Router disparam uma vez por solicitação com um estado cada. Ele é mutável: os hooks podem reescrever states, questions e results, e on_route pode reescrever decision. Ele usa igualdade por identidade (eq=False), então um contexto é hasheável e dois contextos nunca são iguais.

@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)

Encurta a inferência. Chamado de on_predict_start, ele define ctx.results para que a passada direta seja pulada; on_predict_end ainda roda e os resultados fornecidos são retornados.

A chave tem que cobrir tudo do que a resposta depende, e o hook tem que cobrir tudo que a chamada carrega: os hooks disparam uma vez por chamada, e predict_batch os chama com todos os estados de uma vez.

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 executa este bloco, examples/hooks/cache.py, e os blocos de cache de docs/hooks/patterns.md e docs/hooks/examples.md, e verifica as quatro chaves da mesma forma, então uma página não consegue ensinar uma chave da qual o exemplo já se afastou.

No Router, uma carga útil pulada ganha uma chave routing adicionada (sem sobrescrever uma que já tenha), então Router.predict mantém seu formato de retorno documentado.

Protocolo de hook

Hook é um typing.Protocol. Implemente qualquer subconjunto dos métodos; o resto é pulado.

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: ...

Um hook é livre para definir atributos e métodos extras; apenas os seis nomes de evento são consultados. Se um hook define um dos seis como algo não chamável, a configuração falha rápido (veja Validação).

BaseHook

BaseHook é a contraparte concreta do protocolo: uma classe com um corpo no-op para todo evento. Herde dela e sobrescreva apenas os eventos de que você precisa.

from laya import BaseHook

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

Hook é melhor quando você quer tipagem estrutural (qualquer objeto com os métodos certos); BaseHook é melhor quando você quer uma base explícita para herdar e chamar super().

Tipos de conveniência

PredictHook = Callable[[PredictContext], None]

PredictHook é o tipo de um callable simples usado com on_predict_start= / on_predict_end=. Passe um único callable ou uma sequência deles; cada um é envolvido em um hook mínimo.

Registro em tempo de execução

Todo runtime mistura HookRegistry, então os hooks podem ser adicionados, removidos ou ter escopo definido depois da construção. A mutação é thread-safe; uma chamada lê um snapshot da lista, então adicionar ou remover um hook nunca perturba uma chamada em andamento.

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 aceita os mesmos objetos que hooks= (não callables simples). hooks_installed aceita qualquer número de objetos de hook ou sequências e restaura a lista anterior na saída, inclusive quando o bloco lança.

Padrões para todo o processo

laya.hooks mantém um pequeno registro para todo o processo, então um tracer, hook de métricas ou marcador de tenant não precisa ser passado por fio em cada Agent e Router. Os padrões rodam primeiro, depois os hooks instalados na instância, depois os hooks por chamada.

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)

Os padrões se aplicam a todo evento, incluindo os eventos do ciclo de vida do Router on_load e on_evict. O registro é lido no momento da chamada, então hooks definidos depois de um Agent ou Router ser construído ainda se aplicam. Não há opt-out por instância; chame clear_default_hooks() para desligar o conjunto para todo o processo.

Hooks assíncronos

Um evento pode ser uma corrotina. Envolva o hook em AsyncHook e seus métodos async def rodam até o fim no núcleo síncrono:

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())])

Um callable async simples passado para on_predict_start= / on_predict_end= também funciona, porque dispatch roda qualquer awaitable que um hook retorne.

Onde a corrotina roda:

• Se a thread que chama não tem um loop em execução, ela roda com asyncio.run.
• Se já tem um (um chamador dentro de uma função async), ela roda em um loop de fundo dedicado, para que a thread que chama possa bloquear sem deadlock. Passe AsyncHook(hook, loop=...) para canalizar para um loop específico; ele precisa estar rodando, e não pode ser o próprio loop da thread que chama. Ambos são verificados: um loop parado e o próprio loop do chamador lançam ValueError em vez de bloquear para sempre.

Um hook sem métodos async não é afetado.

Superfície de configuração

Todo ponto de entrada aceita os mesmos parâmetros de hook. hooks aceita um objeto ou uma sequência de objetos; on_predict_start / on_predict_end aceitam um callable ou uma sequência.

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 e hooks_timeout em um método por chamada usam None por padrão, ou seja, “usar o valor da instância”.

• hooks_concurrent é apenas no nível da instância.

• Em predict_long os hooks envolvem a inferência que responde ao estado, que para um documento que precisa de várias janelas é o único predict_batch compartilhado sobre elas: on_predict_start dispara uma vez, e ctx.states guarda os textos das janelas decodificadas em ordem de varredura em vez do estado do chamador, que foi tokenizado para produzi-las. states é mutável a partir de um hook de início, então a varredura que chega à inferência não precisa ser a divisão que predict_long calculou. O que muda é o que a resposta pode afirmar:

  o que o hook de início fez	usage["windows"]	answer["window"]
  respondeu com ctx.skip([result])	0	ausente – nenhuma janela o pontuou
  deixou a varredura como foi construída	N	presente – index, token_start/token_end nomeiam o trecho decisivo
  substituiu a varredura, de qualquer forma	os estados que foram pontuados	ausente – os offsets descrevem as janelas de predict_long, não o texto que foi pontuado

  usage["windows"] é total em todo caminho, incluindo aquele em que um estado já cabia em uma única janela, então uma resposta em cache nunca é lida como uma janela que o modelo leu.

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

• O hooks= por chamada em route, route_batch, predict e predict_batch se aplica à chamada inteira, incluindo on_route. Em predict_batch a lista é composta do mesmo jeito que em predict (primeiro os hooks instalados, depois a lista por chamada; None e [] não adicionam nada) e roda uma vez por solicitação.
• route() é público: chamá-lo despacha on_route com os hooks instalados mais quaisquer hooks por chamada.

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 não tem Router, então expõe apenas os eventos no nível da predição.

Cargas úteis dos eventos

Quais campos são preenchidos, por evento e runtime:

Detalhes de tempo:

• on_predict_end vê results no caminho de sucesso. No caminho de falha results é None, a menos que um hook de início os tenha definido via skip(), e usage também é None por consequência (ele é derivado de results); elapsed_ms é sempre definido.
• on_error roda antes do bloco finally que calcula elapsed_ms e usage, então ambos são None ali. Leia tempo e uso de on_predict_end em vez disso.
• run_id é sempre preenchido.

Validação

A configuração é validada quando os hooks são normalizados, o que acontece na construção para hooks instalados e no momento da chamada para hooks por chamada. Estes lançam TypeError:

hooks= não aceita callables simples, porque um callable puro não diz para qual evento ele serve. Use on_predict_start= / on_predict_end= para esses.

Auxiliares avançados

Estes são usados internamente e são estáveis, mas a maioria dos usuários não precisa deles.

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 achata um objeto/sequência hooks e os dois callables em uma única lista ordenada. dispatch chama event em todo hook que o implementa, aplicando a política de raise, o lock e o timeout, e roda o resultado de um hook se ele for awaitable. run_coroutine_sync roda um awaitable até o fim a partir de código síncrono, no loop do chamador se ele estiver livre, ou em um loop de fundo se o chamador já tiver um. aggregate_usage soma os blocos de uso por estado.

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)

Veja também

• Ciclo de vida: quando cada evento roda, com fluxogramas.
• Erros: a matriz de falhas e as regras de encadeamento.
• Padrões e antipadrões: como estruturar bem os hooks.
• Exemplos: receitas para cada caso de uso.