API リファレンス

API リファレンス

このページのものはすべて laya（よく使う名前）または laya.hooks（全表面）からインポートできます。

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

• PredictContext
• Hook プロトコル
• 便利な型
• 設定の表面
• イベントのペイロード
• 検証
• 高度なヘルパー

PredictContext

PredictContext は公開呼び出しごとに 1 つ作られ、その呼び出しのすべてのフックへ渡されます。唯一の例外は Router.predict_batch で、それはそのリクエストに対する Router.predict 呼び出しがそうするように、リクエストごとに 1 つ作るので、Router レベルのフックはリクエストごとに、それぞれ 1 つの state で発火します。これは可変です。フックは states、questions、results を書き換えられ、on_route は decision を書き換えられます。同一性等価（eq=False）を使うので、コンテキストはハッシュ可能で、2 つのコンテキストが等しくなることはありません。

@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 はそれでも走り、渡された結果が返ります。

キーは答えが依存するすべてを覆う必要があり、フックは呼び出しが運ぶすべてを覆う必要があります。フックは呼び出しごとに 1 回発火し、predict_batch はすべての state を一度に渡してそれを呼ぶからです。

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 のキャッシュブロックを exec し、4 つすべてが同じキーになることを表明します。だからあるページが、例がとっくに捨てたキーを教えることはありません。

Router では、スキップされたペイロードに 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: ...

フックは追加の属性やメソッドを自由に定義できます。参照されるのは 6 つのイベント名だけです。フックが 6 つのいずれかを callable でないものとして定義すると、設定は即座に失敗します（検証を参照）。

BaseHook

BaseHook はプロトコルに対する具象の対応物です。すべてのイベントに no-op の本体を持つクラスです。これをサブクラス化し、必要なイベントだけを上書きしてください。

from laya import BaseHook

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

構造的型付けが欲しい（正しいメソッドを持つ任意のオブジェクト）ときは Hook が最善で、サブクラス化して super() を呼ぶ明示的な基底が欲しいときは BaseHook が最善です。

便利な型

PredictHook = Callable[[PredictContext], None]

PredictHook は on_predict_start= / on_predict_end= で使う素の callable の型です。単一の callable かそのシーケンスを渡します。各々は最小限のフックに包まれます。

実行時登録

すべてのランタイムが HookRegistry をミックスインするので、フックは構築後に追加・削除・スコープできます。変更はスレッドセーフで、呼び出しはリストのスナップショットを読むので、フックの追加や削除が飛行中の呼び出しを乱すことはありません。

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 は任意個のフックオブジェクトやシーケンスを取り、ブロックが例外を投げた場合も含めて、終了時に以前のリストを復元します。

プロセス全体の既定値

laya.hooks は小さなプロセス全体のレジストリを保つので、tracer、メトリクスフック、テナントタガーをすべての Agent と Router に通す必要がありません。既定は最初に走り、次にインスタンスにインストールされたフック、その後に呼び出しごとのフックが走ります。

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 を含むすべてのイベントに適用されます。レジストリは呼び出し時に読まれるので、Agent や Router が構築された後に設定したフックも依然として適用されます。インスタンスごとのオプトアウトはありません。プロセス全体のセットを切るには clear_default_hooks() を呼んでください。

非同期フック

イベントはコルーチンでもかまいません。フックを 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())])

on_predict_start= / on_predict_end= に渡す素の非同期 callable も動きます。dispatch はフックが返す awaitable を実行するからです。

コルーチンがどこで走るか：

• 呼び出しスレッドに走っているループがなければ、asyncio.run で走ります。
• すでにある場合（非同期関数内の呼び出し元）、専用のバックグラウンドループで走るので、呼び出しスレッドはデッドロックせずにブロックできます。特定のループへ流すには AsyncHook(hook, loop=...) を渡します。それは走っていて、呼び出しスレッド自身のループであってはなりません。どちらも検査されます。停止したループと呼び出し元自身のループは、永遠にブロックする代わりにそれぞれ ValueError を投げます。

async メソッドを持たないフックは影響を受けません。

設定の表面

すべてのエントリポイントが同じフックパラメータを受け取ります。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 では、フックは state に答える推論を包みます。複数の窓を必要とする文書では、それはそれらに対する単一の共有 predict_batch です。on_predict_start は 1 回発火し、ctx.states は呼び出し元の state（それを生み出すためにトークン化されたもの）ではなく、デコードされた窓テキストをスキャン順に保持します。states は start フックから可変なので、推論に到達するスキャンは predict_long が計算した分割である必要はありません。変わるのは答えが主張できることです。

  start フックがしたこと	usage["windows"]	answer["window"]
  ctx.skip([result]) で答えた	0	なし —— どの窓も採点していない
  構築されたままのスキャンを残した	N	あり —— index、token_start/token_end が決定した span を名指す
  スキャンを何らかの形で置き換えた	採点された state	なし —— オフセットは predict_long の窓を記述し、採点されたテキストではない

  usage["windows"] はすべての経路で合計で、state がすでに単一の窓に収まった場合も含むので、キャッシュされた答えがモデルの読んだ窓として読まれることはありません。

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

• route と predict の呼び出しごとの hooks= は、on_route を含む呼び出し全体に適用されます。
• route() は公開です。それを呼ぶと、インストールされたフックと任意の呼び出しごとの hooks で on_route を dispatch します。

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 レベルのイベントだけを公開します。

イベントのペイロード

イベントとランタイムごとに、どのフィールドが埋まるか：

タイミングの詳細：

• on_predict_end は成功経路で results を見ます。失敗経路では、start フックが skip() で設定していない限り results は None で、usage も同様に None です（results から導出されるため）。elapsed_ms は常に設定されます。
• on_error は elapsed_ms と usage を計算する finally ブロックの前に走るので、そこではどちらも None です。タイミングと usage は代わりに on_predict_end から読んでください。
• run_id は常に埋まります。

検証

設定はフックが正規化されるときに検証され、それはインストールされたフックでは構築時に、呼び出しごとのフックでは呼び出し時に行われます。これらは TypeError を投げます。

hooks= は素の callable を受け取りません。素の callable はどのイベント向けかを述べないからです。それらには on_predict_start= / on_predict_end= を使ってください。

高度なヘルパー

これらは内部で使われ、安定していますが、ほとんどの利用者には必要ありません。

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 のオブジェクト/シーケンスと 2 つの callable を 1 つの順序付きリストへ平坦化します。dispatch はそれを実装するすべてのフックで event を呼び、raise の方針・ロック・タイムアウトを適用し、フックの結果が awaitable ならそれを実行します。run_coroutine_sync は awaitable を同期コードから完走させ、呼び出し元のループが空いていればそこで、すでに持っていればバックグラウンドループで走らせます。aggregate_usage は state ごとの 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)

関連項目

• ライフサイクル：各イベントがいつ走るか（フローチャート付き）。
• エラー：失敗行列と連鎖の規則。
• パターンとアンチパターン：フックをうまく構造化する方法。
• 例：すべてのユースケース向けのレシピ。