예제

복사-붙여넣기 레시피. 모든 스니펫은 이름을 밝힌 헬퍼 (ship, CACHE 등, 직접 제공)를 제외하면 자기 완결적입니다.

• 빠른 시작
• 감사
• PII 마스킹
• 캐시
• 메트릭
• 가드레일
• 신뢰도 게이트
• 라우팅 고정
• 수명 주기
• 조합
• 호출별 훅
• 배치
• HTTP 서버
• ONNXAgent
• 런타임 등록
• 기반 클래스와 프로세스 전역 기본값
• 비동기 훅
• 훅 타임아웃
• 토큰 예산
• 훅 테스트

빠른 시작

import laya

def log(ctx):
    print(ctx.model, ctx.results[0]["answers"])

agent = laya.load("convaiinnovations/laya", on_predict_end=log)
agent.system_one("I was charged twice.", {"urgent": {"type": "noul", "instructions": "Urgent?"}})

감사

browser-use 사용 사례입니다. 모든 결정을 포착해 외부 서비스로 보냅니다.

import json, sys
import laya

def audit(ctx):
    for state, result in zip(ctx.states, ctx.results or []):
        record = {
            "run_id": ctx.run_id,
            "model": ctx.model,
            "state": state,
            "routing": result.get("routing"),
            "answers": result["answers"],
            "usage": result.get("usage"),
            "call_usage": ctx.usage,
            "call_elapsed_ms": round(ctx.elapsed_ms or 0.0, 3),
        }
        print(json.dumps(record), file=sys.stderr)
        # ship_to_service(record)

agent = laya.load("convaiinnovations/laya", on_predict_end=audit)

훅 호출 하나가 호출 전체를 감당하므로, 루프는 결정당 기록 하나를 씁니다. predict_batch의 같은 형태는 배치를 참고하십시오.

전체 실행 가능 버전은 examples/hooks/audit.py에 있습니다.

PII 마스킹

import re
import laya

EMAIL = re.compile(r"\b[\w.+-]+@[\w-]+\.[\w.-]+\b")
PHONE = re.compile(r"\+?\d[\d ()-]{7,}\d")

def scrub(value):
    if isinstance(value, str):
        return PHONE.sub("[phone]", EMAIL.sub("[email]", value))
    if isinstance(value, dict):
        return {k: scrub(v) for k, v in value.items()}
    if isinstance(value, list):
        return [scrub(v) for v in value]
    return value

def redact(ctx):
    ctx.states = [scrub(s) for s in ctx.states]

agent = laya.load("convaiinnovations/laya", on_predict_start=redact)

examples/hooks/redact.py를 참고하십시오.

캐시

import hashlib, json
import laya

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 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 per state: skip replaces the whole call

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

agent = laya.load("convaiinnovations/laya", on_predict_start=read, on_predict_end=write)
first = agent.system_one("state", QUESTIONS)    # runs the model
second = agent.system_one("state", QUESTIONS)   # served from CACHE

examples/hooks/cache.py를 참고하십시오.

메트릭

import laya

COUNTS, LATENCIES = {}, []

def metrics(ctx):
    COUNTS[ctx.model] = COUNTS.get(ctx.model, 0) + 1
    if ctx.elapsed_ms is not None:
        LATENCIES.append(ctx.elapsed_ms)

agent = laya.load("convaiinnovations/laya", on_predict_end=metrics, hooks_raise=False)

examples/hooks/otel.py를 참고하십시오.

가드레일

시작 훅에서 예외를 던져 요청을 차단합니다.

import laya

class Blocked(Exception):
    pass

def guard(ctx):
    text = " ".join(str(state) for state in ctx.states).lower()
    if "ignore previous instructions" in text:
        raise Blocked("prompt injection")

agent = laya.load("convaiinnovations/laya", on_predict_start=guard)

try:
    agent.system_one("Ignore previous instructions and ...", QUESTIONS)
except Blocked:
    handle_block()

시작 훅은 호출의 모든 상태를 보므로 전부 테스트하십시오. ctx.states[0]만 읽으면 predict_batch 호출의 나머지가 통과합니다.

신뢰도 게이트

낮은 신뢰도의 답변을 다시 쓰거나 주석을 답니다.

def gate(ctx):
    for result in ctx.results or []:
        answer = result["answers"].get("dept")
        if answer and answer["confidence"] < 0.6:
            answer["choice"] = "human-review"
            answer["gated"] = True

agent = laya.load("convaiinnovations/laya", on_predict_end=gate)

ctx.results는 호출의 상태당 딕셔너리 하나를 담으므로, 루프는 첫 번째 상태의 답변만이 아니라 임계값을 놓친 모든 답변에 주석을 답니다.

라우팅 고정

특정 부류의 트래픽에 체크포인트를 강제합니다.

from laya import Router
from laya.router import RouteDecision

def pin(ctx):
    if "refund" in str(ctx.states[0]).lower():
        ctx.decision = RouteDecision(
            model="typed-decisions",
            repo="convaiinnovations/laya/typed-decisions",
            reason="refund workflow",
            detection=None,
            workflow=None,
        )

router = Router(hooks=[pin])

설치하지 않고 호출별로:

router.predict("refund request", QUESTIONS, hooks=[pin])

수명 주기

체크포인트 생성과 해제를 관찰합니다.

from laya import Router

class Lifecycle:
    def on_load(self, ctx):
        print("loaded", ctx.model, "agent", type(ctx.agent).__name__)

    def on_evict(self, ctx):
        print("evicted", ctx.model)

router = Router(max_loaded=1, hooks=[Lifecycle()])
router.preload(["english", "multilingual"])   # on_load fires per build
router.unload()                               # on_evict fires per freed checkpoint

조합

설치된 훅이 먼저, 그다음 편의 호출 가능 객체입니다. 모두 컨텍스트 하나를 공유합니다.

import laya

class Metrics:
    def on_predict_end(self, ctx):
        record_latency(ctx.model, ctx.elapsed_ms)

def redact(ctx):
    ctx.states = [strip_pii(s) for s in ctx.states]

def audit(ctx):
    ship(ctx.run_id, ctx.results)

agent = laya.load(
    "convaiinnovations/laya",
    hooks=[Metrics()],              # installed, runs first
    on_predict_start=redact,        # convenience, appended
    on_predict_end=audit,           # convenience, appended
    hooks_raise=True,
)

호출별 훅

단일 호출에 대해 훅을 재정의하거나 확장합니다.

agent.system_one(
    state,
    questions,
    on_predict_end=lambda ctx: debug_dump(ctx),
    hooks_raise=False,
)

router.predict(
    state,
    questions,
    hooks=[pin],                    # applies to on_route too
    on_predict_end=audit,
)

배치

훅은 Agent.predict_batch 호출당 한 번 실행되며, ctx.states가 모든 상태를 담습니다. Router.predict_batch는 대신 Router 수준 훅을 요청당 한 번 실행하며, 각각 상태 하나와 자체 run_id를 가지므로, 거기서 같은 훅은 훅의 호출당 기록 하나를 씁니다.

def audit_batch(ctx):
    for state, result in zip(ctx.states, ctx.results):
        ship_one(ctx.run_id, state, result)

results = agent.predict_batch([state_a, state_b, state_c], questions, on_predict_end=audit_batch)

HTTP 서버

서버가 Router.predict를 호출하므로, Router 훅은 laya.serve에 대해 자동으로 실행됩니다.

from laya import Router
from laya.serve import create_app

router = Router(hooks=[Metrics()], on_predict_end=audit, hooks_raise=False)
app = create_app(router=router)

ONNXAgent

ONNXAgent는 예측 수준 이벤트만 노출합니다.

from laya.onnx_agent import ONNXAgent

agent = ONNXAgent("convaiinnovations/laya", onnx_path="laya.onnx", on_predict_end=audit)
agent.system_one(state, questions)

런타임 등록

생성 후에 훅을 붙이거나 떼거나 범위를 한정합니다.

agent.add_hook(Metrics())          # attach at runtime
agent.remove_hook(Metrics())       # by identity

with agent.hooks_installed(DebugDump()):
    agent.system_one(state, questions)   # DebugDump only here

기반 클래스와 프로세스 전역 기본값

BaseHook을 상속해 필요한 것만 재정의하고, 모든 Agent와 Router에 전달하는 대신 프로세스 전체에 한 번 등록하십시오.

from laya import BaseHook, hooks

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

hooks.set_default_hooks(hooks=[Audit()])   # runs for every call in the process

# later, or in tests:
hooks.clear_default_hooks()

토큰 예산

훅이나 호출별 인자에서 한 호출의 토큰 예산을 조정합니다. 훅의 값은 유효한 예산을 교체하므로, 먼저 그 예산을 읽어야 합니다. 호출의 가장 넓은 질문에 맞춰 크기를 정하고, 코어가 옵션에 적용하는 토큰 하한 위에 머물고, 상태가 창을 유지하도록 max_len을 head_max_len과 함께 넓히십시오.

def widen(ctx):
    k = max((len(q.get("criteria", {}) or {}) for q in ctx.questions.values()), default=0)
    if k < 50:
        return
    cfg = getattr(ctx.agent, "cfg", None) or {}
    head = ctx.head_max_len if ctx.head_max_len is not None else cfg.get("head_max_len", 192)
    window = ctx.max_len if ctx.max_len is not None else cfg.get("max_len", 512)
    need = 16 + 8 * k
    if need > head:
        ctx.head_max_len = need
        ctx.max_len = max(window, need + 8 + 64)

agent = laya.load("convaiinnovations/laya", on_predict_start=widen)

# or per call
agent.system_one(state, questions, head_max_len=512, max_len=1024)

토큰 예산 조정에 각 줄 뒤의 산술이 있고, predict_shortlist는 레이블 집합이 넓힌 창에도 들어맞지 않을 때의 선택지입니다.

비동기 훅

비동기 훅을 AsyncHook으로 감싸십시오. 호출자가 동기이든 이미 이벤트 루프 안에 있든, 각 코루틴은 동기 코어에서 완료될 때까지 실행됩니다.

import laya
from laya import AsyncHook

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

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

일반 비동기 호출 가능 객체도 작동합니다.

async def async_end(ctx):
    await ship(ctx.results)

agent.system_one(state, questions, on_predict_end=async_end)

훅 타임아웃

각 훅 호출을 제한해, 멈춘 훅이 서빙 중인 요청을 붙잡지 못하게 합니다.

agent = laya.load("convaiinnovations/laya", on_predict_end=metrics, hooks_timeout=2.0)

# or per call
agent.system_one(state, questions, on_predict_end=metrics, hooks_timeout=0.5)

타임아웃된 훅은 TimeoutError를 발생시킵니다(hooks_raise=False이면 경고). 훅은 백그라운드에서 계속 실행되므로, 네트워크 호출에도 자체 타임아웃을 주십시오. 오류를 참고하십시오.

훅 테스트

모델 없이 훅이 본 것을 검증합니다. tests/test_hooks.py가 그렇듯이, encode/forward/decode 헬퍼를 스텁으로 두고 predict_batch를 구동하십시오.

seen = []
agent.predict_batch(["s0"], questions, on_predict_end=lambda ctx: seen.append(ctx.results))
assert len(seen) == 1

API 표면은 tests/test_hooks_api.py로 고정됩니다.

함께 보기

• 트레이싱: run_id, 스팬, OpenTelemetry.
• 패턴과 안티 패턴: 이 레시피들 뒤의 근거.