7장: 감사 로깅과 컴플라이언스

from pydantic import BaseModel
from datetime import datetime
from typing import Any
 
class AuditEntry(BaseModel):
    """감사 로그 항목"""
    # 식별
    entry_id: str
    workflow_id: str
    trace_id: str  # 분산 추적 ID
    span_id: str
 
    # 시간
    timestamp: datetime
    duration_ms: int | None = None
 
    # 행위자
    actor_type: str  # agent, human, system
    actor_id: str
 
    # 행동
    action_type: str  # llm_call, tool_call, decision, human_review
    action_detail: dict
 
    # 결과
    outcome: str  # success, failure, escalated
    outcome_detail: dict | None = None
 
    # 컨텍스트
    input_summary: str
    output_summary: str
    reasoning: str | None = None  # 결정 근거
    confidence: float | None = None
    alternatives_considered: list[dict] | None = None
 
class LLMCallAudit(BaseModel):
    """LLM 호출 감사 기록"""
    model: str
    provider: str
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int
    temperature: float
    system_prompt_hash: str  # 전문 대신 해시
    user_prompt_summary: str
    response_summary: str
    latency_ms: int
    cost_usd: float
 
class ToolCallAudit(BaseModel):
    """도구 호출 감사 기록"""
    tool_name: str
    tool_version: str
    input_params: dict
    output_data: dict
    side_effects: list[str]  # 외부 시스템 변경 사항
    latency_ms: int
    idempotency_key: str

import hashlib
import json
 
class ImmutableAuditLog:
    """체이닝 해시를 사용한 불변 감사 로그"""
 
    def __init__(self, store: AuditStore):
        self.store = store
 
    async def append(self, entry: AuditEntry) -> str:
        """감사 항목 추가 (체이닝 해시로 무결성 보장)"""
        # 이전 항목의 해시를 가져와 체이닝
        previous_hash = await self.store.get_latest_hash(entry.workflow_id)
 
        # 현재 항목의 해시 계산 (이전 해시 포함)
        entry_data = entry.model_dump_json()
        chain_input = f"{previous_hash}:{entry_data}"
        entry_hash = hashlib.sha256(chain_input.encode()).hexdigest()
 
        # 저장 (append-only 테이블)
        await self.store.insert(
            entry=entry,
            entry_hash=entry_hash,
            previous_hash=previous_hash,
        )
 
        return entry_hash
 
    async def verify_integrity(self, workflow_id: str) -> bool:
        """체이닝 해시를 검증하여 로그 무결성 확인"""
        entries = await self.store.get_all(workflow_id)
        previous_hash = "genesis"
 
        for entry, stored_hash in entries:
            entry_data = entry.model_dump_json()
            chain_input = f"{previous_hash}:{entry_data}"
            computed_hash = hashlib.sha256(chain_input.encode()).hexdigest()
 
            if computed_hash != stored_hash:
                return False
 
            previous_hash = stored_hash
 
        return True

-- 감사 로그 테이블
CREATE TABLE audit_log (
    entry_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    workflow_id UUID NOT NULL,
    trace_id VARCHAR(64) NOT NULL,
    timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    actor_type VARCHAR(20) NOT NULL,
    actor_id VARCHAR(100) NOT NULL,
    action_type VARCHAR(50) NOT NULL,
    action_detail JSONB NOT NULL,
    outcome VARCHAR(20) NOT NULL,
    reasoning TEXT,
    entry_hash VARCHAR(64) NOT NULL,
    previous_hash VARCHAR(64) NOT NULL
);
 
-- UPDATE, DELETE 차단 트리거
CREATE OR REPLACE FUNCTION prevent_audit_modification()
RETURNS TRIGGER AS $$
BEGIN
    RAISE EXCEPTION 'Audit log entries cannot be modified or deleted';
END;
$$ LANGUAGE plpgsql;
 
CREATE TRIGGER audit_immutable_update
    BEFORE UPDATE ON audit_log
    FOR EACH ROW EXECUTE FUNCTION prevent_audit_modification();
 
CREATE TRIGGER audit_immutable_delete
    BEFORE DELETE ON audit_log
    FOR EACH ROW EXECUTE FUNCTION prevent_audit_modification();
 
-- 인덱스
CREATE INDEX idx_audit_workflow ON audit_log (workflow_id, timestamp);
CREATE INDEX idx_audit_trace ON audit_log (trace_id);
CREATE INDEX idx_audit_actor ON audit_log (actor_id, timestamp);

@dataclass
class ComplianceRequirement:
    regulation: str
    requirement: str
    check_function: Callable
    severity: str  # critical, high, medium
 
class ComplianceChecker:
    def __init__(self, requirements: list[ComplianceRequirement]):
        self.requirements = requirements
 
    async def check_workflow(self, workflow_id: str) -> ComplianceReport:
        """워크플로우의 컴플라이언스 점검"""
        audit_entries = await self.audit_store.get_all(workflow_id)
        violations = []
 
        for req in self.requirements:
            result = await req.check_function(audit_entries)
            if not result.passed:
                violations.append(ComplianceViolation(
                    regulation=req.regulation,
                    requirement=req.requirement,
                    severity=req.severity,
                    detail=result.detail,
                ))
 
        return ComplianceReport(
            workflow_id=workflow_id,
            checked_at=datetime.utcnow(),
            total_requirements=len(self.requirements),
            violations=violations,
            passed=len([v for v in violations if v.severity == "critical"]) == 0,
        )
 
# 금융 서비스 컴플라이언스 요구사항 예시
financial_requirements = [
    ComplianceRequirement(
        regulation="전자금융감독규정",
        requirement="모든 금융 거래 결정에 대한 감사 추적 존재",
        check_function=check_financial_audit_trail,
        severity="critical",
    ),
    ComplianceRequirement(
        regulation="전자금융감독규정",
        requirement="고위험 거래에 대한 사람 승인 기록",
        check_function=check_human_approval_for_high_risk,
        severity="critical",
    ),
    ComplianceRequirement(
        regulation="개인정보보호법",
        requirement="개인정보 접근 기록 및 최소 수집 준수",
        check_function=check_pii_access_logging,
        severity="high",
    ),
]

@dataclass
class DecisionExplanation:
    """에이전트 결정의 설명"""
    decision: str
    reasoning_steps: list[str]
    evidence: list[Evidence]
    alternatives: list[Alternative]
    confidence: float
    limitations: list[str]
 
@dataclass
class Evidence:
    source: str  # 참조한 문서/데이터
    relevance: float
    excerpt: str  # 핵심 발췌
 
@dataclass
class Alternative:
    option: str
    reason_not_chosen: str
    estimated_confidence: float
 
class ExplainableAgent:
    async def decide_with_explanation(
        self, context: dict
    ) -> tuple[str, DecisionExplanation]:
        """결정과 함께 설명을 생성"""
        # 에이전트의 추론 과정을 구조화하여 기록
        response = await self.llm.invoke(
            messages=[
                {"role": "system", "content": EXPLAINABLE_SYSTEM_PROMPT},
                {"role": "user", "content": json.dumps(context)},
            ],
            response_format=DecisionWithExplanation,
        )
 
        explanation = DecisionExplanation(
            decision=response.decision,
            reasoning_steps=response.reasoning_steps,
            evidence=[
                Evidence(source=e.source, relevance=e.relevance, excerpt=e.excerpt)
                for e in response.evidence
            ],
            alternatives=[
                Alternative(
                    option=a.option,
                    reason_not_chosen=a.reason,
                    estimated_confidence=a.confidence,
                )
                for a in response.alternatives
            ],
            confidence=response.confidence,
            limitations=response.limitations,
        )
 
        # 감사 로그에 설명 기록
        await self.audit_log.append(AuditEntry(
            action_type="decision",
            reasoning=json.dumps(explanation.__dict__, default=str),
            confidence=explanation.confidence,
        ))
 
        return response.decision, explanation

@dataclass
class ReproductionContext:
    """결정 재현을 위한 컨텍스트"""
    model_id: str
    model_version: str
    temperature: float
    seed: int | None
    system_prompt_version: str
    system_prompt_content: str
    input_data: dict
    tool_responses: dict[str, Any]  # 도구 호출 결과 스냅샷
    timestamp: datetime
 
class ReproductionManager:
    async def capture(self, workflow_id: str, node: str) -> ReproductionContext:
        """현재 실행 컨텍스트를 캡처"""
        # ... 현재 상태 캡처 ...
 
    async def replay(self, context: ReproductionContext) -> ReplayResult:
        """캡처된 컨텍스트로 결정을 재현"""
        # 도구 호출을 실제 실행 대신 캡처된 결과로 대체
        mock_tools = MockToolSet(context.tool_responses)
 
        response = await self.llm.invoke(
            model=context.model_id,
            messages=[
                {"role": "system", "content": context.system_prompt_content},
                {"role": "user", "content": json.dumps(context.input_data)},
            ],
            temperature=context.temperature,
            seed=context.seed,
            tools=mock_tools,
        )
 
        return ReplayResult(
            original_timestamp=context.timestamp,
            replay_timestamp=datetime.utcnow(),
            output=response,
        )

from opentelemetry import trace
from opentelemetry.trace import StatusCode
 
tracer = trace.get_tracer("agentic-workflow")
 
class InstrumentedWorkflowRunner:
    async def run_node(self, node: str, state: dict) -> dict:
        with tracer.start_as_current_span(
            f"workflow.node.{node}",
            attributes={
                "workflow.id": state["workflow_id"],
                "workflow.node": node,
                "workflow.phase": state["phase"],
            },
        ) as span:
            try:
                result = await self.execute_node(node, state)
                span.set_status(StatusCode.OK)
                span.set_attribute("workflow.node.output_size", len(str(result)))
                return result
            except Exception as e:
                span.set_status(StatusCode.ERROR, str(e))
                span.record_exception(e)
                raise
 
    async def call_llm(self, messages: list, model: str) -> str:
        with tracer.start_as_current_span(
            "llm.call",
            attributes={
                "llm.model": model,
                "llm.input_tokens": self._count_tokens(messages),
            },
        ) as span:
            response = await self.llm.invoke(messages, model=model)
            span.set_attribute("llm.output_tokens", response.usage.completion_tokens)
            span.set_attribute("llm.total_tokens", response.usage.total_tokens)
            span.set_attribute("llm.latency_ms", response.latency_ms)
            return response.content

@dataclass
class RetentionPolicy:
    hot_storage_days: int    # 빠른 조회 가능 (PostgreSQL)
    warm_storage_days: int   # 조회 가능하나 느림 (S3 Standard)
    cold_storage_days: int   # 아카이브 (S3 Glacier)
    total_retention_days: int
 
# 산업별 보존 정책 예시
RETENTION_POLICIES = {
    "financial": RetentionPolicy(
        hot_storage_days=90,
        warm_storage_days=365,
        cold_storage_days=2555,  # 7년
        total_retention_days=2555,
    ),
    "healthcare": RetentionPolicy(
        hot_storage_days=90,
        warm_storage_days=365,
        cold_storage_days=3650,  # 10년
        total_retention_days=3650,
    ),
    "general": RetentionPolicy(
        hot_storage_days=30,
        warm_storage_days=180,
        cold_storage_days=365,
        total_retention_days=365,
    ),
}