C.W.K.
Stream
Lesson 04 of 05 · published

LLM 액션을 위한 audit trail

~12 min · audit, logging, compliance

Level 0Observer
0 XP0/64 lessons0/13 achievements
0/150 XP to next level150 XP to go0% complete

Audit trail이 기록하는 거

각 에이전트 run에 대해 — 누가 trigger했는지(user id, system trigger), 언제, 어떤 프롬프트, 어떤 args로 어떤 도구 호출, 어떤 도구 결과 반환, 에이전트가 어떤 final 답 생성, human이 어떤 permission 결정 내림. 합쳐서 몇 달 후 '뭐 일어났는지' 답.

Hooks가 너 audit hook(literally)

가장 단순 구현 — PostToolUse hook이 도구 호출당 structured 라인 append, plus UserPromptSubmit hook이 trigger와 Stop의 final summary. 각 라인이 session id 운반해서 timeline 엮음.

보존 vs 비용

Audit 로그 성장. 보존 정책 미리 결정 — regulatory나 비즈니스 need 기반 30/90/365일. Cold-storage tier(S3 Glacier, GCS Coldline)이 multi-year 보존 싸게; hot-storage 쿼리는 last N day만 target.

원칙: Audit trail 없는 액션이 fingerprint 없는 사고. Trail 먼저 빌드; cool 기능 나중.

Code

Structured audit logger·python
import json, time, uuid
from claude_agent_sdk import ClaudeAgentOptions

class Audit:
    def __init__(self, path: str, run_id: str | None = None):
        self.path = path
        self.run_id = run_id or str(uuid.uuid4())

    def emit(self, event_type: str, **fields):
        line = {"ts": time.time(), "run_id": self.run_id, "event": event_type, **fields}
        with open(self.path, "a") as f:
            f.write(json.dumps(line) + "\n")

async def make_hooks(audit: Audit):
    async def on_user_prompt(context):
        audit.emit("user_prompt", text_hash=hash(context.prompt))
        return HookOutput(allow=True)

    async def on_tool_post(context):
        audit.emit(
            "tool_use",
            tool=context.tool_name,
            input_keys=sorted(context.tool_input.keys()),
            duration_ms=context.duration_ms,
            error=context.is_error,
        )

    return {"UserPromptSubmit": [on_user_prompt], "PostToolUse": [on_tool_post]}
이벤트를 세션 timeline에 엮기·python
import json
from collections import defaultdict

def session_timeline(audit_path: str):
    runs = defaultdict(list)
    for line in open(audit_path):
        ev = json.loads(line)
        runs[ev["run_id"]].append(ev)
    for run_id, events in runs.items():
        events.sort(key=lambda e: e["ts"])
        print(f"\n== run {run_id} ({len(events)} events) ==")
        for e in events:
            print(f"  {e['ts']:.3f}  {e['event']}")

External links

Exercise

에이전트 하나에 audit logger 패턴 구현. 세션 run, 그다음 'last hour의 tool 호출 몇, tool name별' 답하는 쿼리 작성. 로그가 싸게 답 verify.
Hint
로그가 text-only면 JSONL로 switch — 포맷이 structured 아니면 모든 observability 질문이 더 어려워.

Progress

Progress is local-only — sign in to sync across devices.
이 페이지에서 버그를 발견하셨거나 피드백이 있으세요?문제 신고

댓글 0

🔔 답글 알림 (로그인 필요)
로그인댓글을 남기려면 로그인해 주세요.

아직 댓글이 없어요. 첫 댓글을 남겨보세요.