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는 공개 호출마다 한 번 생성되어 그 호출의 모든 훅에 전달됩니다. 유일한 예외는 Router.predict_batch입니다. 이 메서드는 그 요청에 대한 Router.predict 호출이 그랬을 것처럼 요청마다 하나를 생성하므로, Router 수준 훅은 각각 상태 하나를 담은 채 요청마다 한 번 실행됩니다. 이 컨텍스트는 가변입니다. 훅이 states, questions, results를 다시 쓸 수 있고, on_route가 decision을 다시 쓸 수 있습니다. 항등(identity) 동등성 (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는 여전히 실행되고, 제공된 결과가 반환됩니다.

키는 답이 의존하는 모든 것을 포괄해야 하고, 훅은 호출이 지니는 모든 것을 포괄해야 합니다. 훅은 호출당 한 번 실행되고, 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에서는 건너뛴 페이로드에 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: ...

훅은 추가 속성과 메서드를 자유롭게 정의할 수 있습니다. 참조되는 것은 여섯 개 이벤트 이름뿐입니다. 여섯 개 중 하나를 호출 불가능한 것으로 정의하면 구성이 즉시 실패합니다 (검증 참고).

BaseHook

BaseHook은 프로토콜의 구체적인 대응물입니다. 모든 이벤트에 대해 본문이 비어 있는 클래스입니다. 이를 상속하고 필요한 이벤트만 재정의하십시오.

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=에 사용되는 일반 호출 가능 객체의 타입입니다. 호출 가능 객체 하나 또는 그 시퀀스를 전달하면, 각각이 최소한의 훅으로 감싸집니다.

런타임 등록

모든 런타임은 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=와 같은 객체를 받습니다(일반 호출 가능 객체는 안 됨). hooks_installed는 임의 개수의 훅 객체나 시퀀스를 받아 블록이 예외를 던져도 종료 시 이전 리스트를 복원합니다.

프로세스 전역 기본값

laya.hooks는 작은 프로세스 전역 레지스트리를 유지하므로, 트레이서나 메트릭 훅, 테넌트 태거를 모든 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=에 전달한 일반 비동기 호출 가능 객체도 작동합니다. dispatch가 훅이 반환하는 모든 어웨이터블을 실행하기 때문입니다.

코루틴이 실행되는 위치:

• 호출 스레드에 실행 중인 루프가 없으면 asyncio.run으로 실행됩니다.
• 이미 하나가 있으면(비동기 함수 내부의 호출자), 전용 백그라운드 루프에서 실행되므로 호출 스레드가 교착 없이 블로킹할 수 있습니다. 특정 루프로 유입하려면 AsyncHook(hook, loop=...)를 전달하십시오. 그 루프는 실행 중이어야 하고 호출 스레드 자신의 루프가 아니어야 합니다. 둘 다 검사합니다. 정지한 루프와 호출자 자신의 루프는 영원히 블로킹하는 대신 각각 ValueError를 발생시킵니다.

async 메서드가 없는 훅은 영향을 받지 않습니다.

구성 표면

모든 진입점은 같은 훅 매개변수를 받습니다. hooks는 객체 또는 객체 시퀀스를 받고, on_predict_start / on_predict_end는 호출 가능 객체 또는 시퀀스를 받습니다.

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에서 훅은 상태에 답하는 추론을 감쌉니다. 여러 창이 필요한 문서의 경우 이는 그 창들에 대한 단일 공유 predict_batch입니다. on_predict_start는 한 번 실행되고, ctx.states는 호출자의 상태가 아니라 스캔 순서대로 디코딩된 창 텍스트를 담습니다. 호출자의 상태는 그 창들을 만들기 위해 토큰화된 것입니다. states는 시작 훅에서 가변이므로, 추론에 도달하는 스캔이 predict_long이 계산한 분할과 같을 필요가 없습니다. 달라지는 것은 답이 주장할 수 있는 것입니다.

  시작 훅이 한 일	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

• route, route_batch, predict, predict_batch의 호출별 hooks=는 on_route를 포함해 호출 전체에 적용됩니다. predict_batch에서는 목록이 predict와 같은 방식으로 구성되며(설치된 훅 먼저, 그다음 호출별 목록. None과 []는 아무것도 추가하지 않습니다), 요청마다 한 번씩 실행됩니다.
• route()는 공개입니다. 호출하면 설치된 훅과 호출별 hooks를 합쳐 on_route를 디스패치합니다.

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가 없으므로 예측 수준 이벤트만 노출합니다.

이벤트 페이로드

이벤트와 런타임별로 어떤 필드가 채워지는지는 다음과 같습니다.

타이밍 세부 사항:

• on_predict_end는 성공 경로에서 results를 봅니다. 실패 경로에서 results는 시작 훅이 skip()으로 설정하지 않은 한 None이고, 따라서 usage도 None입니다(results에서 파생되기 때문). elapsed_ms는 항상 설정됩니다.
• on_error는 elapsed_ms와 usage를 계산하는 finally 블록보다 먼저 실행되므로, 그곳에서는 둘 다 None입니다. 타이밍과 사용량은 대신 on_predict_end에서 읽으십시오.
• run_id는 항상 채워집니다.

검증

구성은 훅이 정규화될 때 검증됩니다. 설치된 훅은 생성 시점에, 호출별 훅은 호출 시점에 이루어집니다. 다음은 TypeError를 발생시킵니다.

hooks=는 일반 호출 가능 객체를 받지 않습니다. 맨 호출 가능 객체는 어느 이벤트를 위한 것인지 말하지 않기 때문입니다. 그런 것에는 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 객체/시퀀스와 두 호출 가능 객체를 하나의 순서 있는 리스트로 평탄화합니다. dispatch는 이를 구현한 모든 훅에서 event를 호출하며, raise 정책, 잠금, 타임아웃을 적용하고, 훅의 결과가 어웨이터블이면 실행합니다. run_coroutine_sync는 동기 코드에서 어웨이터블을 완료될 때까지 실행하며, 호출자의 루프가 비어 있으면 그 루프에서, 호출자에게 이미 루프가 있으면 백그라운드 루프에서 실행합니다. aggregate_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)

함께 보기

• 수명 주기: 각 이벤트가 실행되는 시점, 순서도 포함.
• 오류: 실패 행렬과 연결 규칙.
• 패턴과 안티 패턴: 훅을 잘 구조화하는 방법.
• 예제: 모든 사용 사례를 위한 레시피.