패턴과 안티 패턴
훅은 작은 이음새이며, 잘 쓰기도 나쁘게 쓰기도 쉽습니다. 이 페이지는 운영에서 견디는 형태와 물어뜯는 형태를 모았습니다.
패턴
감사 로그
재구성할 수 있을 만큼 충분히 담아 모든 결정을 기록하십시오. 상태, 질문, 답변, 모델, 라우팅 결정, 사용량, 지연 시간입니다.
import json
def audit(ctx):
for state, result in zip(ctx.states, ctx.results or []):
json.dump({
"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),
}, sys.stdout)
sys.stdout.write("\n")
laya.load("convaiinnovations/laya", on_predict_end=audit)훅은 호출당 한 번 실행되고, predict_batch 호출은 모든 상태를 담고 있으므로, 기록은 결정당
한 번 쓰입니다. ctx.states와 ctx.results는 인덱스로 정렬됩니다. ctx.usage와
ctx.elapsed_ms는 호출 전체의 합계이고, 각 결과는 자체 usage를 지닙니다.
로그 한 줄을 잃는 것이 요청을 실패시켜서는 안 된다면 너그럽게 만드십시오.
hooks_raise=False입니다. 감사 추적이 규정 준수 요건이라면 엄격하게 만드십시오.
PII 마스킹
마스킹은 토큰화 전인 on_predict_start에서 일어나야 합니다. 그렇지 않으면 모델이 이미 그
데이터를 본 것입니다.
import re
EMAIL = re.compile(r"\b[\w.+-]+@[\w-]+\.[\w.-]+\b")
def redact(ctx):
ctx.states = [
EMAIL.sub("[email]", s) if isinstance(s, str) else s
for s in ctx.states
]
laya.load("convaiinnovations/laya", on_predict_start=redact)마스킹 훅은 정책 훅입니다. hooks_raise=True를 유지하십시오. 조용히 고장 난 마스커는
데이터 유출이기 때문입니다.
캐싱
시작 훅이 캐시를 확인하고 ctx.skip(...)을 호출하며, end 훅이 캐시를 채웁니다. 히트 시
순전파는 건너뜁니다.
import hashlib, json
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
laya.load("convaiinnovations/laya", on_predict_start=read, on_predict_end=write)훅은 호출당 한 번 실행되므로, predict_batch에서는 상태 하나로 키를 잡는 것으로 충분하지
않습니다. ctx.skip()은 호출이 반환했을 모든 결과를 교체합니다. 동시에 서빙할 때는 잠금으로
캐시를 보호하십시오. Router에서는 캐시된 페이로드에도 여전히 routing 키가 붙으므로 반환
형태는 바뀌지 않습니다.
같은 쌍은 단일 LangChain 노드에서 그 노드의 hooks= 인자를 통해 작동하며,
이는 그 에이전트의 다른 모든 호출자가 보는 것을 바꾸지 않고 그래프의 뜨거운 단계 하나를
캐싱하는 방법입니다.
메트릭
ctx.model, ctx.usage, ctx.elapsed_ms에서 카운터와 히스토그램을 만듭니다. 너그럽게
유지하십시오.
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)
laya.load("convaiinnovations/laya", on_predict_end=metrics, hooks_raise=False)가드레일
정책 훅은 예외를 던져 요청을 차단합니다. hooks_raise=True(기본값)이면 차단이 호출자에게
도달하고, on_error와 on_predict_end가 여전히 실행되므로 감사 추적에 기록됩니다.
class Blocked(Exception):
pass
def guard(ctx):
if any("ssn" in str(state).lower() for state in ctx.states):
raise Blocked("possible PII in state")
laya.load("convaiinnovations/laya", on_predict_start=guard)상태 형태에 대고 테스트하십시오. ctx.states[0]만 읽는 가드는 단일 호출을 차단하고
predict_batch 호출이 나머지 모든 상태를 순전파에 통과시키도록 둡니다.
신뢰도 게이팅
end 훅이 낮은 신뢰도의 답변을 안전한 폴백으로 다시 쓰거나, 다운스트림 로직을 위해 주석을 답니다. 이는 거부가 아니라 결과 변경입니다.
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
laya.load("convaiinnovations/laya", on_predict_end=gate)호출의 상태당 하나의 딕셔너리를 담은 ctx.results를 통해 변경하십시오. 첫 번째만
게이팅하면 다른 모든 낮은 신뢰도 답변이 주석 없이 나갑니다.
라우팅 재정의
on_route는 ctx.decision을 교체해 특정 부류의 트래픽을 체크포인트에 고정할 수 있습니다.
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(hooks=[pin])모델 수명 주기
on_load와 on_evict는 체크포인트를 관찰합니다. 워밍업 로그, 메모리 회계, 해제 알림에
사용하십시오. Router 잠금 밖에서 실행되므로, 훅이 Router를 다시 호출할 수 있습니다.
class Lifecycle:
def on_load(self, ctx):
print("loaded", ctx.model)
def on_evict(self, ctx):
print("evicted", ctx.model)
Router(hooks=[Lifecycle()])멀티 테넌트 컨텍스트
훅 클로저에 캡처하거나 컨텍스트 로컬에서 읽어 테넌트 id를 전달하십시오. 잠금 없이 훅 객체에 요청별 상태를 저장하지 마십시오.
def make_audit(tenant):
def audit(ctx):
ship(tenant, ctx.run_id, ctx.results)
return audit
agent = laya.load("convaiinnovations/laya", on_predict_end=make_audit("acme"))조합
여러 종류의 훅이 자연스럽게 조합됩니다. 설치된 훅이 순서대로 먼저 실행됩니다.
agent = laya.load(
"convaiinnovations/laya",
hooks=[Metrics(), Guardrail()], # metrics first, then policy
on_predict_start=redact, # convenience callables appended after hooks
hooks_raise=True, # policy failures are fatal
)나중 훅이 이전 훅의 변경을 보므로, 순서를 의도적이고 문서화된 상태로 유지하십시오.
범위 한정 계측
에이전트를 재구성하는 대신, 필요한 코드에만 트레이서나 디버그 훅을 붙이십시오.
hooks_installed는 블록이 예외를 던져도 종료 시 이전 리스트를 복원합니다.
with agent.hooks_installed(DebugDump()):
agent.system_one(state, questions) # DebugDump only hereadd_hook/remove_hook은 블록 없이 같은 일을 하며, 프로세스와 같은 수명을 사는
트레이서에 적합합니다.
프로세스 전역 계측
모든 결정이 봐야 하는 트레이서나 메트릭 훅은 각 Agent와 Router에 전달하는 대신 한 번
등록할 수 있습니다. 기본값은 인스턴스 훅과 호출별 훅보다 먼저 실행됩니다.
from laya import BaseHook, hooks
class Metrics(BaseHook):
def on_predict_end(self, ctx):
record(ctx.model, ctx.elapsed_ms)
hooks.set_default_hooks(hooks=[Metrics()])이는 전역 상태이므로 의도적으로 범위를 잡으십시오. 시작 시 한 번 설정하고, 테스트에서는
clear_default_hooks()하여 한 테스트가 다음 테스트로 훅을 새게 하지 마십시오.
토큰 예산 조정
시작 훅은 한 호출의 토큰 예산을 올릴 수 있습니다. 예를 들어 질문에 옵션이 많아 기본 헤드 예산이 레이블을 붕괴시킬 때입니다. 네 가지 세부 사항이 그 훅이 도움이 되는지 조용히 호출을 더 나쁘게 만드는지를 결정합니다.
- 시작 훅의
ctx.head_max_len은 그 호출의 예산을 교체합니다. 그 전에 유효한 것은 호출자 자신의 호출별 값이거나ctx.agent.cfg의 체크포인트 기본값입니다. 따라서 그것과 비교해야 합니다. 그냥 숫자를 쓰면 호출자가 이미 설정한 예산을 낮출 수 있습니다. - 한 호출은 지닌 모든 질문에 답하므로, 어느 것이 먼저 오는지가 아니라 그중 가장 넓은 것에 맞춰 크기를 정하십시오.
- 옵션이 더 이상 헤드에 들어맞지 않으면,
laya/common.py가 각 옵션에max(4, (head_max_len - 16) // k)토큰을 줍니다. 따라서16 + 4 * k는 정확히 그 하한에 안착합니다. 모든 레이블이 다른 레이블과 공유하는 토큰으로 잘리며, 이것이 그 훅이 피하려고 작성된 붕괴입니다.16 + 8 * k는 레이블을 구분 가능하게 남깁니다. - 상태는
max_len - head_max_len - 8토큰을 받으므로, 넓힌 헤드는max_len도 함께 넓혀야 상태가 창을 잃지 않습니다.
def widen_for_high_cardinality(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 # 8 tokens per label, not the core's floor of 4
if need > head: # only ever widen, never lower
ctx.head_max_len = need
ctx.max_len = max(window, need + 8 + 64) # 8 reserved, then room for the state
agent = laya.load("convaiinnovations/laya", on_predict_start=widen_for_high_cardinality)이는 공유 에이전트 구성을 건드리지 않으므로 동시 호출이 영향을 받지 않습니다. 같은 손잡이는
호출별로도 사용할 수 있습니다.
agent.system_one(state, questions, head_max_len=512, max_len=1024)입니다.
넓히는 것은 공짜가 아닙니다. 더 긴 창은 더 큰 텐서를 뜻하고, 체크포인트는 512(laya)와
1,024 토큰에서 학습되었습니다. 그 이상에서는 예산을 늘리는 것보다
predict_shortlist로 후보를 좁히는 것이 낫습니다.
안티 패턴
블로킹 작업
훅은 호출 스레드에서 실행되고, laya.serve는 단일 추론 워커를 사용합니다. 잠들거나 네트워크
왕복을 기다리거나 input()을 호출하는 훅은 그 뒤의 다른 모든 요청을 지연시킵니다.
# bad: blocks the whole server
def audit(ctx):
requests.post("https://slow.example/decisions", json=..., timeout=30)
# better: enqueue, let a background worker ship it
def audit(ctx):
QUEUE.put_nowait(record(ctx))느린 작업을 꼭 해야 한다면, 최소한 훅 자체가 겹치지 않도록 hooks_concurrent=False로
설정하고 laya.serve를 큐 뒤에서 실행하십시오.
제어 흐름을 위해 end 훅에서 예외 던지기
on_predict_end는 추론 후에 실행됩니다. 거기서 예외를 던지면 계산된 결과를 버리고, 성공
경로에서는 호출자에게 드러납니다. 추론 비용을 치르기 전에 차단하려면 시작 훅을 사용하거나,
답을 바꾸려면 ctx.results를 다시 쓰십시오.
잠금 없는 공유 가변 상태
같은 훅 인스턴스가 여러 스레드에서 실행됩니다. self.counter += 1은 경합합니다.
# bad
class Count:
def __init__(self): self.n = 0
def on_predict_end(self, ctx): self.n += 1
# good
import threading
class Count:
def __init__(self):
self.n = 0
self._lock = threading.Lock()
def on_predict_end(self, ctx):
with self._lock:
self.n += 1조용한 실패
hooks_raise=False는 실패당 한 번 경고하지만, 모든 것을 스스로 잡는 훅은 진짜 문제를
숨깁니다.
# bad: no one will ever know the audit trail stopped
def audit(ctx):
try:
ship(record(ctx))
except Exception:
pass훅이 선택 사항이라면 hooks_raise=False가 처리하게 하고 경고를 지켜보십시오. 그렇지
않다면 예외를 던지게 두십시오.
컨텍스트 보유
ctx를 리스트에 추가하는 훅은 상태, 질문, 결과, 에이전트 전체를 살려 둡니다.
# bad: unbounded memory growth
SEEN = []
def audit(ctx):
SEEN.append(ctx)
# good: keep only what you need
SEEN = []
def audit(ctx):
SEEN.append((ctx.run_id, ctx.model, ctx.elapsed_ms))너무 늦은 마스킹
on_predict_end까지 가면 모델이 이미 상태를 토큰화했습니다. on_predict_start에서
마스킹하십시오.
호출별 훅 안의 질문별 로직
호출당 PredictContext 하나가 있고, 순전파 한 번이 모든 질문에 답합니다. 질문별 이벤트는
없습니다. on_predict_end 안에서 답변을 순회하고, 상태도 순회하십시오. 배치에서는 컨텍스트
하나가 호출의 모든 상태를 담습니다.
def flag(ctx):
for result in ctx.results or []:
for qid, answer in result["answers"].items():
if answer.get("confidence", 1.0) < 0.5:
alert(qid, ctx.run_id)재귀적 predict
agent.predict/system_one을 호출하는 훅은 훅을 다시 실행합니다. 깊이 가드가 없으면
재귀합니다.
# bad
def enrich(ctx):
ctx.results = [agent.predict(ctx.states[0], EXTRA_QUESTIONS)]
# good: guard, or use a separate agent with no hooks
def enrich(ctx):
if getattr(ctx, "_enriched", False):
return
ctx._enriched = True
ctx.results = [enricher.predict(state, EXTRA_QUESTIONS) for state in ctx.states]hooks=의 일반 호출 가능 객체
hooks=는 훅 객체를 받습니다. 맨 호출 가능 객체는 어느 이벤트를 위한 것인지 말하지 않으므로
거부됩니다. on_predict_start= / on_predict_end=를 사용하십시오.
# bad: TypeError
laya.load("convaiinnovations/laya", hooks=[lambda ctx: None])
# good
laya.load("convaiinnovations/laya", on_predict_end=lambda ctx: None)end 훅에 결과가 있다고 가정하기
실패 경로에서 ctx.results는 시작 훅이 설정하지 않은 한 None입니다. 항상 확인하십시오.
def audit(ctx):
if ctx.results is None:
log_failure(ctx.run_id, ctx.error)
return
log_success(ctx.run_id, ctx.results)순서 의존적 훅
다른 훅의 변경을 읽는 훅은 순서가 고정되지 않으면 취약합니다. 설치된 훅은 리스트 순서대로, 그다음 편의 호출 가능 객체가 실행됩니다. 결합이 있으면 문서화하거나, 결합된 훅들을 하나의 객체로 합치십시오.