LLM 에이전트 출력 검증: 할루시네이션이 JSON 스키마를 통과하는 이유와 3레이어 방어 설계

import json
from pydantic import BaseModel
from anthropic import Anthropic
 
class AnalysisResult(BaseModel):
    summary: str
    confidence: float
    sources: list[str]
 
client = Anthropic()
 
# 시스템 프롬프트에 JSON 스키마 직접 주입
schema_json = json.dumps(AnalysisResult.model_json_schema(), indent=2)
system_prompt = f"""항상 다음 JSON 스키마를 정확히 따르는 응답만 반환하세요:
{schema_json}
 
JSON 객체 외의 텍스트는 절대 포함하지 마세요."""
 
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    system=system_prompt,
    messages=[{"role": "user", "content": "최신 AI 트렌드를 분석해줘"}]
)

┌─────────────────────────────────────────────┐
│  레이어 3: 규칙 검증                          │  ← 도메인 정책, 외부 사실 검증 API
├─────────────────────────────────────────────┤
│  레이어 2: 의미 검증                          │  ← LLM-as-Judge, 다중 검증자 표결
├─────────────────────────────────────────────┤
│  레이어 1: 구조 검증                          │  ← Pydantic, JSON Schema, 재시도 루프
└─────────────────────────────────────────────┘

# pydantic-ai >= 0.0.30 기준
from pydantic import BaseModel, field_validator
from pydantic_ai import Agent
 
class ResearchSummary(BaseModel):
    title: str
    key_points: list[str]
    confidence: float
 
    @field_validator("confidence")
    @classmethod
    def check_range(cls, v: float) -> float:
        if not 0.0 <= v <= 1.0:
            raise ValueError("confidence must be between 0 and 1")
        return v
 
# 모듈 레벨에서 초기화 — 함수 안에서 매 호출마다 생성하면 불필요한 초기화 비용 반복 발생
research_agent = Agent(
    "anthropic:claude-sonnet-4-6",
    result_type=ResearchSummary,   # 스키마 자동 주입 + 유효성 검사
    retries=3,                     # 검증 실패 시 실패 이유를 담아 자동 재질의
)
 
async def summarize(topic: str) -> ResearchSummary:
    result = await research_agent.run(topic)
    return result.output

from pydantic_ai import Agent
 
# 모델을 분리하는 것이 핵심 — 같은 모델은 같은 편향을 공유합니다
generator = Agent("anthropic:claude-sonnet-4-6")
critic = Agent("anthropic:claude-opus-4-8")   # Judge는 더 강력한 모델 권장
 
async def reflection_loop(task: str, max_rounds: int = 3) -> str:
    draft = (await generator.run(task)).output
 
    for _ in range(max_rounds):
        critique = await critic.run(
            f"다음 초안을 아래 항목별로 검토해줘:\n"
            f"① 사실 오류 ② 논리 비약 ③ 누락된 근거\n\n"
            f"초안:\n{draft}\n\n"
            f"문제가 없으면 'APPROVED'를, 있으면 구체적 수정 지시를 반환해줘."
        )
 
        if "APPROVED" in critique.output:
            break
 
        draft = (await generator.run(
            f"원래 과제: {task}\n\n"
            f"이전 초안의 문제점:\n{critique.output}\n\n"
            f"위 문제를 반영해서 개선된 버전을 작성해줘."
        )).output
 
    return draft

from pydantic import BaseModel
from pydantic_ai import Agent
 
class JudgeVerdict(BaseModel):
    score: int    # 0-10
    reason: str
    passed: bool
 
class OutputValidationError(Exception):
    pass
 
# 싱글턴으로 관리 — 함수 안에서 매번 생성하면 초기화 비용이 반복 발생
judge_agent = Agent("anthropic:claude-opus-4-8", result_type=JudgeVerdict)
main_agent = Agent("anthropic:claude-sonnet-4-6")
 
async def judge_output(output: str, criteria: str) -> JudgeVerdict:
    verdict = await judge_agent.run(
        f"다음 출력을 '{criteria}' 기준으로 평가해줘.\n\n출력:\n{output}"
    )
    return verdict.output
 
async def validated_pipeline(task: str) -> str:
    result = await main_agent.run(task)
 
    verdict = await judge_output(
        result.output,
        criteria="사실 정확성, 논리적 완결성, 주제 적합성"
    )
 
    if not verdict.passed or verdict.score < 7:
        raise OutputValidationError(
            f"품질 기준 미달 (score={verdict.score}): {verdict.reason}"
        )
 
    return result.output

import asyncio
from pydantic import BaseModel
from pydantic_ai import Agent
 
class VerifierResult(BaseModel):
    refuted: bool
    reason: str
 
# 편향 다양성을 위해 다른 모델/공급자 조합 — 이것이 MAV의 실질적 가치
VERIFIER_MODELS = [
    "anthropic:claude-sonnet-4-6",
    "anthropic:claude-haiku-4-5",   # 같은 공급자라도 모델 특성이 다름
    "openai:gpt-4o-mini",           # 공급자를 달리하면 편향 다양성 극대화
]
 
async def multi_verifier_vote(claim: str) -> bool:
    async def verify(model: str) -> bool:
        verifier = Agent(model, result_type=VerifierResult)
        result = await verifier.run(
            f"다음 주장에서 반박할 수 있는 오류나 사실 불일치가 있으면 "
            f"refuted=True로 답해줘. 확실하지 않으면 refuted=True가 안전한 선택입니다.\n\n"
            f"주장: {claim}"
        )
        return result.output.refuted
 
    votes = await asyncio.gather(*[verify(model) for model in VERIFIER_MODELS])
    refuted_count = sum(votes)
 
    # 과반 미만이 반박 → 유효한 출력으로 판정
    return refuted_count < (len(VERIFIER_MODELS) / 2)

from dataclasses import dataclass
import httpx
 
class PolicyViolationError(Exception):
    pass
 
@dataclass
class MedicalSummary:
    diagnosis: str
    recommended_dosage_mg: float
    patient_age: int
 
# 도메인 정책 규칙 — LLM이 아닌 코드로 결정론적으로 검증
def validate_medical_rules(summary: MedicalSummary) -> None:
    # 규칙 1: 소아 환자(18세 미만)에게 성인 용량 상한 초과 금지
    if summary.patient_age < 18 and summary.recommended_dosage_mg > 500:
        raise PolicyViolationError(
            f"소아 환자에게 {summary.recommended_dosage_mg}mg 처방은 정책 위반"
        )
 
    # 규칙 2: 외부 데이터베이스로 진단명 실재 여부 확인
    if not _verify_diagnosis_exists(summary.diagnosis):
        raise PolicyViolationError(
            f"'{summary.diagnosis}'는 검증된 질병 데이터베이스에 없는 진단명"
        )
 
def _verify_diagnosis_exists(diagnosis: str) -> bool:
    # ICD-10 공개 API 조회
    response = httpx.get(
        "https://clinicaltables.nlm.nih.gov/api/icd10cm/v3/search",
        params={"terms": diagnosis, "maxList": 1}
    )
    return response.json()[0] > 0