ライフサイクル
ライフサイクル
このページは、各エントリポイントにおけるイベントの正確な順序です。図を 1 つだけ読むなら Router のものを読んでください。それが上位集合です。
- Agent.predict_batch
- Agent.system_one / predict
- Router.predict
- Router.predict_batch
- モデルのライフサイクル
- skip によるキャッシュ
- 空の入力
- 順序の規則
- 並行性
Agent.predict_batch
predict_batch が唯一の実装で、system_one と predict は 1 つの state でそれを呼びます。
predict_batch(states, questions, batch_size=..., hooks=..., ...)
│
├─ active = installed hooks + per-call hooks (installed first)
├─ ctx = PredictContext(states, questions, model=self.model_id, agent=self)
│
├─ try:
│ │
│ ├─ on_predict_start ─────────────────────────────┐
│ │ a hook may: │
│ │ • rewrite ctx.states / ctx.questions │
│ │ • set ctx.max_len / ctx.head_max_len │
│ │ • ctx.skip(results) ─────────────┐ │
│ │ • raise (aborts; see errors) │ │
│ │ │ │
│ ├─ if ctx.results is not None: ◄─────────┘ │ cache hit
│ │ skip tokenization and forward │
│ ├─ else: │
│ │ validate states is a list │
│ │ for each batch chunk: │
│ │ _encode_state ─► collate ─► _forward │
│ │ _decode_answers │
│ │ ctx.results = [...] │
│ │ │
│ └─ (any failure here) ──► except BaseException: │
│ ctx.error = exc │
│ on_error │
│ re-raise │
│ │
│ finally: │
│ ctx.elapsed_ms = now - ctx.started_at │
│ if ctx.results: ctx.usage = aggregate_usage(...)│
│ on_predict_end ─────────────────────────────────┘
│
└─ return ctx.results同じ ctx オブジェクトが start・error・end を通じて流れるので、run_id がそれらを相関させ、end フックは ctx.error を読めます。
Agent.system_one / predict
system_one(state, questions, hooks=..., ...)
└─ predict_batch([state], questions, hooks=..., ...)[0]つまり system_one はすべてのフックと同じライフサイクルを継承し、ctx.states == [state] になります。
Router.predict
Router.predict(state, questions, model=..., hooks=..., on_predict_start=..., on_predict_end=...)
│
├─ active = installed hooks + per-call hooks
│
├─ route(state, questions, ..., hooks=per-call, hooks_raise=...)
│ │
│ ├─ _route(...) detect script / language / workflow
│ ├─ on_route ──► ctx.decision a hook may replace the decision
│ └─ return ctx.decision
│
├─ load(decision["model"])
│ │
│ ├─ already resident? ──► return it
│ ├─ else build Agent(...) ──► on_load (after the Router lock is released)
│ └─ evict LRU checkpoints ──► on_evict (after the Router lock is released)
│
├─ ctx = PredictContext(states=[state], questions, decision, model=decision.model,
│ agent=agent, router=self)
├─ try:
│ ├─ on_predict_start
│ ├─ if ctx.results is None:
│ │ result = agent.system_one(ctx.states[0], ctx.questions)
│ │ └─ the Agent's own hooks run here (start / forward / end)
│ │ result["routing"] = decision
│ │ ctx.results = [result]
│ └─ else:
│ for each cached result: result.setdefault("routing", decision)
│ └─ (any failure) ──► except: on_error, re-raise
│ └─ finally: elapsed_ms, usage, on_predict_end
│
└─ return ctx.results[0]要点：
on_routeはモデルが読み込まれる前に走るので、フックはチェックポイントを固定し、別のものを読み込まずに済みます。- Router レベルの predict フックは呼び出し全体を包みます。それらは Agent へ転送されません。付属した Agent が独自のフックを持てばそれも走りますが、これは想定どおりです。
- Router レベルの
ctx.skip()でもroutingは付加されるので、戻り値の形は安定しています。
Router.predict_batch
各結果はそのリクエストに対して predict が返すものなので、Router レベルの predict フックはここでもリクエストごとに走ります。各リクエストが独自の PredictContext、run_id、elapsed_ms を持ちます。
Router.predict_batch(requests, batch_size=...)
│
├─ route_batch(requests) ──► on_route, once per request (no checkpoint loaded yet)
│
└─ for each checkpoint, in order of first appearance:
│
├─ load(checkpoint) ──► on_load / on_evict
├─ for each request of this checkpoint, in input order:
│ ctx = PredictContext(states=[state], questions, decision, model, agent, router,
│ max_len=request.get("max_len"),
│ head_max_len=request.get("head_max_len"))
│ on_predict_start a hook may redact, rewrite, set a token budget or skip
├─ group the requests left to infer by (questions, ctx.max_len, ctx.head_max_len)
│ agent.predict_batch(states, questions, ...) ──► one shared forward pass per group
│ result["routing"] = decision; ctx.results = [result]
├─ ctx.usage, once per request
├─ (any failure) ──► for every started request, in reverse input order:
│ ctx.error = exc, on_error, on_predict_end; then re-raise
└─ on_predict_end, once per request of this checkpoint, in reverse input order要点：
- あるチェックポイントのリクエストの start フックはすべて、それらの end フックより先に走ります。フォワードパスを共有するからです。したがって
on_predict_endで埋めるキャッシュは、同じチェックポイントグループ内の重複 state には応えられません。呼び出しをまたげば応えられます。 - 同じ理由で、リクエストは開始した順序の逆順で終わります。start で何かを設定し end で戻すフック（
contextvarsの値、OpenTelemetry のcontext.attach/detach）は、見つけた値まで巻き戻ります。 ctx.states、ctx.questions、トークン予算を置き換える start フックは、自分のリクエストだけを変えます。リクエストは start フックが走った後にフォワードパス用にグループ化されるからです。共有の questions dict をその場で変更するのは別物で、predictがすることでもありません。グループのどのリクエストも、そのすべての start フックが走るまで推論されないので、変更はその dict を共有するすべてのリクエストに届きます。start フックが先に走ったものも、呼び出し元も含みます。代わりに新しい dict をctx.questionsに代入してください。- チェックポイント名はグループ化の前に解決されるので、エイリアス（
"ml"）でリクエストを固定するon_routeフックはそのチェックポイントのフォワードパスを共有し、ctx.modelは解決後の名前になります。 - リクエストは独自の
max_len/head_max_lenを持てます。これはpredictが呼び出し引数として取るトークン予算のリクエストごとの形です。ctx.max_lenを設定する start フックはそれを上書きします。フックはコンテキストが構築された後に走るからです。異なる予算を要求するリクエストはフォワードパスを共有できないので、予算が混在するバッチは予算ごとに 1 回agent.predict_batchを呼びます。 - 開始されたリクエストはそれぞれちょうど 1 つの
on_predict_endを受け取ります。別のリクエストの end フックが例外を投げてもです。最初のそのようなエラーは、それらすべてが走った後に送出されます。 - チェックポイントグループが失敗すると、その開始されたすべてのリクエストがその例外で失敗します。それぞれが
ctx.errorにそれを設定されたon_errorを受け取り、次にon_predict_endを受け取ります。これにはキャッシュヒットと、質問グループがすでに走ったリクエストも含まれます。呼び出し元はそれらすべてについて結果ではなく例外を受け取り、ctx.errorが別のリクエストの失敗であることもありえます（1 つのリクエストで例外を投げる start フックはそのグループを失敗させます）。すでに完了したチェックポイントグループのリクエストは、[router.predict(...) for ...]の先行する呼び出しがそうだったように、結果とともに終わっています。
モデルのライフサイクル
on_load はチェックポイントが構築されたときに発火し、on_evict は解放されたときに発火します。どちらも Router の内部ロックが解放された後に走るので、フックは安全に Router へ呼び戻せます。
load("multilingual")
│
├─ [lock]
│ build Agent(...) (seconds: download + weights)
│ register in _agents / _order
│ evict LRU if over max_loaded ──► evicted = ["english"]
├─ [unlock]
├─ on_evict("english")
└─ on_load("multilingual")
unload("english")
├─ [lock] remove from _agents / _order
├─ [unlock]
└─ on_evict("english")attach(name, agent) は既存のエージェントを登録し、チェックポイントが構築されないので on_load を発火しません。
skip によるキャッシュ
on_predict_start
├─ cache hit? ctx.skip([cached_result])
│ └─ forward pass skipped
│ └─ on_predict_end still runs
│ └─ Router adds `routing` if missing
└─ cache miss? nothing
└─ forward pass runs
└─ on_predict_end can store the result動くキャッシュの実例は examples/hooks/cache.py にあります。
空の入力
監査がすべての呼び出しを見られるように、フックはそれでも発火します。
| 入力 | 終了時の ctx.results |
|---|---|
predict_batch([]) |
[] |
predict_batch(states, {})（質問なし） |
state ごとに 1 つの空の答えのペイロード |
system_one(state, {}) |
単一の空の答えのペイロード |
これらの場合はトークン化もフォワードパスも起きませんが、on_predict_start と on_predict_end は走ります。
順序の規則
- インストールされたフックは常に、呼び出しごとのフックより先に走ります。
- リスト内では、フックはリストの順序で走ります。
- 1 つのイベントについて、それを実装するすべてのフックが、その順序で、次のイベントの前に走ります。
on_errorは失敗経路でon_predict_endより先に走ります。- 単一の
loadが追い出しと構築の両方を行うとき、on_evictはon_loadより先に走ります。
installed: [A, B] per-call: [C]
on_predict_start: A, B, C
on_predict_end: A, B, C並行性
Agent と Router は多数のスレッドから安全に呼べます。各呼び出しが独自の PredictContext を作るので、コンテキストがリクエストをまたいで漏れることはありません。唯一の共有状態はフックオブジェクト自身なので、スレッドセーフでないフックは自分の状態を保護するか、hooks_concurrent=False でインストールする必要があります。
hooks_concurrent=True (default) hooks_concurrent=False
thread 1 ─┐ thread 1 ─┐
thread 2 ─┼─ hooks run in parallel thread 2 ─┼─ one hook at a time
thread 3 ─┘ thread 3 ─┘ (RLock)hooks_concurrent=False が直列化するのは各フックの呼び出しであり、呼び出し全体ではありません。2 つの呼び出しはイベントの合間で依然として交互に進めます。再入可能なロックを使うので、フックはデッドロックせずに同じ Agent/Router へ呼び戻せます。
関連項目
- エラー：フックが例外を投げたときにイベントごとに何が起きるか。
- パターンとアンチパターン：ライフサイクルをうまく使う方法。