LLM-as-Judge 완전 가이드 3편—에이전트 자동 평가 루프와 CI/CD 통합, Judge를 진짜 파이프라인에 박아라

 

1편에서 개념을 잡고, 2편에서 편향을 잡았다면, 3편은 실제 운영입니다. Judge를 실제 개발 파이프라인에 연결하지 않으면, 그건 그냥 일회성 스크립트입니다.

---

📌 핵심 요약
→ 2026년 현재 대부분의 팀은 Eval Level 0~1 — 수동 테스트가 전부
→ 목표: PR 올릴 때마다 Judge가 자동 실행되는 Level 3 파이프라인 구현
→ 에이전트 평가는 최종 결과 + 궤적(Trajectory) 함께 평가해야 함
→ 단계별 비용 최적화: 결정론적 체크 → 경량 Judge → 프론티어 Judge → 인간
→ Self-refinement 루프: Judge 피드백 → 에이전트 재시도 → 재평가 자동화
→ CoT 평가 함정: 에이전트가 CoT로 Judge를 속이는 현상 실증됨 (2026)
→ 프로덕션 모니터링: 실서비스 트래픽이 곧 다음 eval dataset
→ GitHub Actions 완성 코드 포함

---

실전1 — 에이전트 평가의 핵심: 결과가 아니라 궤적을 봐야 한다

단순 LLM 평가와 에이전트 평가의 결정적 차이는 여기서 납니다.

# 잘못된 에이전트 평가: 최종 결과만 본다
def naive_agent_eval(final_answer: str) -> float:
    # "답이 맞으면 OK" — 에이전트가 어떻게 거기 도달했는지 모름
    ...

# 올바른 에이전트 평가: 궤적(Trajectory) 전체를 본다
def trajectory_eval(trajectory: list[dict]) -> dict:
    """
    trajectory 구조 예시:
    [
        {"step": 1, "thought": "...", "tool": "web_search", "input": "...", "output": "..."},
        {"step": 2, "thought": "...", "tool": "calculator", "input": "...", "output": "..."},
        {"step": 3, "thought": "...", "action": "final_answer", "output": "42"},
    ]
    """
    ...

import anthropic, json

client = anthropic.Anthropic()

TRAJECTORY_JUDGE_PROMPT = """당신은 AI 에이전트의 실행 과정을 평가하는 전문가입니다.

[평가 대상]
목표: {goal}

실행 궤적:
{trajectory_str}

[평가 기준]
1. 도구 선택 적절성 (0~5): 각 단계에서 올바른 도구를 사용했는가?
2. 추론-행동 일관성 (0~5): thought와 실제 action이 일치하는가?
3. 불필요한 단계 여부 (0~5): 중복·우회 단계가 없는가?
4. 오류 복구 능력 (0~5): 실패 시 적절히 방향을 전환했는가?
5. 최종 답변 품질 (0~5): 목표를 달성했는가?

단계별 사고 후 반드시 JSON으로만 출력:
{{"tool_selection": N, "reasoning_consistency": N, "efficiency": N,
  "error_recovery": N, "final_quality": N, "total": N, "weakness": "가장 취약한 부분"}}"""

def evaluate_trajectory(goal: str, trajectory: list[dict]) -> dict:
    trajectory_str = ""
    for step in trajectory:
        trajectory_str += f"\n[Step {step['step']}]\n"
        trajectory_str += f"  생각: {step.get('thought', '-')}\n"
        trajectory_str += f"  도구: {step.get('tool', step.get('action', '-'))}\n"
        trajectory_str += f"  입력: {step.get('input', '-')}\n"
        trajectory_str += f"  결과: {step.get('output', '-')}\n"

    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=800,
        messages=[{
            "role": "user",
            "content": TRAJECTORY_JUDGE_PROMPT.format(
                goal=goal,
                trajectory_str=trajectory_str
            )
        }]
    )

    import re
    text = response.content[0].text
    match = re.search(r'\{.*\}', text, re.DOTALL)
    return json.loads(match.group()) if match else {}

# 테스트
sample_trajectory = [
    {"step": 1, "thought": "날씨 정보를 먼저 검색해야 한다", "tool": "web_search",
     "input": "서울 오늘 날씨", "output": "맑음, 22도"},
    {"step": 2, "thought": "검색 결과가 충분하다, 답변할 수 있다", "tool": "web_search",
     "input": "서울 오늘 날씨", "output": "맑음, 22도"},  # ← 중복 단계
    {"step": 3, "action": "final_answer", "output": "오늘 서울은 맑고 22도입니다"},
]

result = evaluate_trajectory("오늘 서울 날씨 알려줘", sample_trajectory)
print(json.dumps(result, ensure_ascii=False, indent=2))
# → efficiency 낮게 나옴 (중복 단계 감지)

→ 2025 arXiv 연구: 에이전트 실패의 17%는 단계 반복, 14%는 추론-행동 불일치
→ 최종 결과만 보면 이 두 가지 실패를 아예 탐지 못함
→ 궤적 평가는 단일 Judge로 한 번에 평가하지 말고 단계별로 나눠 평가할 것

---

실전2 — 비용 최적화: 4단계 레이어드 평가

모든 출력에 GPT-4/Claude를 갖다 대면 비용이 터집니다. 중요도에 따라 레이어를 나눕니다.

from anthropic import Anthropic
import json, re

client = Anthropic()

class LayeredEvalPipeline:
    """
    [Layer 1] 결정론적 체크   → 비용 거의 0, 100% 적용
    [Layer 2] 경량 휴리스틱  → 매우 저렴, 80% 적용
    [Layer 3] LLM Judge      → 중간 비용, 30% 적용
    [Layer 4] 인간 리뷰      → 최고 비용, 5% 적용
    """

    # ── Layer 1: 결정론적 체크 ─────────────────────────────
    def layer1_deterministic(self, output: str, schema: dict = None) -> dict:
        checks = {}

        # JSON 형식 검사
        if schema and schema.get("require_json"):
            try:
                json.loads(output)
                checks["json_valid"] = True
            except:
                checks["json_valid"] = False

        # 최소 길이 검사
        min_len = schema.get("min_length", 10) if schema else 10
        checks["length_ok"] = len(output.strip()) >= min_len

        # 금지어 검사
        banned = schema.get("banned_words", []) if schema else []
        checks["no_banned_words"] = not any(w in output.lower() for w in banned)

        passed = all(checks.values())
        return {"layer": 1, "passed": passed, "details": checks}

    # ── Layer 2: 휴리스틱 체크 ─────────────────────────────
    def layer2_heuristic(self, output: str, question: str) -> dict:
        checks = {}

        # 질문 핵심 키워드 포함 여부
        question_words = set(question.lower().split())
        output_words   = set(output.lower().split())
        overlap = len(question_words & output_words) / max(len(question_words), 1)
        checks["keyword_overlap"] = overlap >= 0.2

        # 과도한 길이 (Verbosity Bias 방어)
        checks["not_too_verbose"] = len(output) <= 2000

        # 반복 문장 감지
        sentences = [s.strip() for s in output.split('.') if s.strip()]
        checks["no_repetition"] = len(sentences) == len(set(sentences))

        passed = sum(checks.values()) >= 2  # 3개 중 2개 이상 통과
        return {"layer": 2, "passed": passed, "details": checks}

    # ── Layer 3: LLM Judge ─────────────────────────────────
    def layer3_llm_judge(self, output: str, question: str) -> dict:
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=200,
            messages=[{"role": "user", "content": f"""
다음 답변의 품질을 평가하세요. JSON만 출력:

질문: {question}
답변: {output}

{{"score": 0~10, "pass": true/false, "reason": "한 줄"}}
"""}]
        )
        text = response.content[0].text
        match = re.search(r'\{.*\}', text, re.DOTALL)
        result = json.loads(match.group()) if match else {"score": 5, "pass": True}
        result["layer"] = 3
        return result

    # ── 통합 실행 ──────────────────────────────────────────
    def run(self, question: str, output: str, schema: dict = None) -> dict:
        import random

        # Layer 1: 항상 실행
        r1 = self.layer1_deterministic(output, schema)
        if not r1["passed"]:
            return {"final": "FAIL", "layer_stopped": 1, "details": r1}

        # Layer 2: 항상 실행
        r2 = self.layer2_heuristic(output, question)
        if not r2["passed"]:
            return {"final": "FAIL", "layer_stopped": 2, "details": r2}

        # Layer 3: 30% 확률로 샘플링 (비용 제어)
        if random.random() < 0.30:
            r3 = self.layer3_llm_judge(output, question)
            if not r3["pass"]:
                return {"final": "HUMAN_REVIEW", "layer_stopped": 3, "details": r3}
            if r3["score"] < 4:
                return {"final": "FAIL", "layer_stopped": 3, "details": r3}

        return {"final": "PASS", "layer_stopped": None}

# 실행
pipeline = LayeredEvalPipeline()
result = pipeline.run(
    question="비동기 프로그래밍이란?",
    output="async/await를 사용해 I/O 대기 없이 여러 작업을 동시에 처리하는 방식입니다.",
    schema={"min_length": 20, "banned_words": ["모름", "잘 모르겠"]}
)
print(result)

→ Layer 1(결정론적): 비용 ≈ 0, 포맷 오류·금지어·빈 응답 필터링
→ Layer 2(휴리스틱): 비용 ≈ 0, 키워드 겹침·길이·반복 감지
→ Layer 3(LLM Judge): 30% 샘플링 → 비용 70% 절감하면서 품질 신호 확보
→ 전체 비용: 단일 LLM Judge 대비 약 1/5 수준

---

실전3 — Self-refinement 루프: Judge 피드백으로 에이전트 재시도

Judge 점수가 낮으면 피드백을 주고, 에이전트가 스스로 재답변하도록 자동화합니다.

import anthropic

client = anthropic.Anthropic()

def agent_with_self_refinement(question: str, max_retries: int = 2) -> dict:
    """
    Judge 피드백 기반 Self-refinement 루프
    실패 시 구체적 피드백을 담아 재시도 — 최대 max_retries회
    """
    history = []
    attempt = 0

    while attempt <= max_retries:
        # Step 1: 에이전트 답변 생성
        messages = [{"role": "user", "content": question}]

        # 이전 시도가 있으면 피드백 컨텍스트 추가
        if history:
            last = history[-1]
            feedback_msg = (
                f"이전 답변: {last['answer']}\n"
                f"Judge 피드백: {last['feedback']}\n"
                f"위 피드백을 반영해 더 나은 답변을 작성하세요."
            )
            messages = [
                {"role": "user", "content": question},
                {"role": "assistant", "content": last['answer']},
                {"role": "user", "content": feedback_msg}
            ]

        answer_resp = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=500,
            messages=messages
        )
        answer = answer_resp.content[0].text

        # Step 2: Judge가 답변 평가
        judge_resp = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=300,
            system="당신은 엄격한 기술 답변 평가자입니다.",
            messages=[{"role": "user", "content": f"""
질문: {question}
답변: {answer}

평가 기준: 정확성, 완결성, 예시 포함 여부

JSON만 출력:
{{"score": 0~10, "pass": true/false,
  "feedback": "구체적으로 부족한 부분과 개선 방향"}}
"""}]
        )

        import re, json
        text = judge_resp.content[0].text
        match = re.search(r'\{.*\}', text, re.DOTALL)
        judge_result = json.loads(match.group()) if match else {"score": 5, "pass": True}

        history.append({
            "attempt": attempt + 1,
            "answer": answer,
            "score": judge_result.get("score", 0),
            "feedback": judge_result.get("feedback", "")
        })

        # Judge 통과 → 루프 종료
        if judge_result.get("pass"):
            return {
                "final_answer": answer,
                "attempts": attempt + 1,
                "final_score": judge_result.get("score"),
                "history": history
            }

        attempt += 1

    # 최대 재시도 초과 → 마지막 답변 반환
    return {
        "final_answer": history[-1]["answer"],
        "attempts": attempt,
        "final_score": history[-1]["score"],
        "note": "최대 재시도 초과 — 인간 검토 권장",
        "history": history
    }

result = agent_with_self_refinement("Python 데코레이터의 실전 활용법을 예시와 함께 설명해줘")
print(f"시도 횟수: {result['attempts']}, 최종 점수: {result['final_score']}")

→ Self-refinement 핵심: "틀렸어" 가 아니라 "왜 틀렸고 뭘 고쳐야 해" 를 피드백으로
→ 재시도 상한선 필수 — 무한 루프 방지 (보통 2~3회가 최적)
→ 주의: CoT 평가 함정 — 에이전트가 CoT를 그럴듯하게 써서 Judge를 속이는 현상 (ICLR 2026 확인)
   ∟ 방어: CoT 내용이 아니라 실제 tool call 결과를 기준으로 평가
→ 점수 개선 없으면 조기 종료 → 추가 API 비용 낭비 방지

---

실전4 — GitHub Actions CI/CD 통합: PR마다 Judge 자동 실행

# .github/workflows/llm-eval.yml
name: LLM Judge Eval

on:
  pull_request:
    paths:
      - 'prompts/**'      # 프롬프트 파일 변경 시 트리거
      - 'src/agents/**'   # 에이전트 코드 변경 시 트리거

jobs:
  eval:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Python 설정
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - name: 의존성 설치
        run: pip install anthropic pytest

      - name: LLM Judge Eval 실행
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
        run: python eval/run_eval.py --threshold 7.0

      - name: 결과 PR 코멘트로 업로드
        uses: actions/github-script@v7
        if: always()
        with:
          script: |
            const fs = require('fs');
            const result = JSON.parse(fs.readFileSync('eval/result.json', 'utf8'));
            const icon = result.passed ? '✅' : '❌';
            const body = `## ${icon} LLM Judge 결과\n` +
              `- 평균 점수: **${result.avg_score}/10**\n` +
              `- 통과율: **${result.pass_rate}%**\n` +
              `- 실패 케이스: ${result.failed_cases.join(', ') || '없음'}\n\n` +
              (result.passed ? '품질 기준 통과 ✅' : '⚠️ 품질 기준 미달 — 머지 전 검토 필요');
            github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body
            });

# eval/run_eval.py
import anthropic, json, argparse, sys
from pathlib import Path

client = anthropic.Anthropic()

# eval 데이터셋: 질문-기대답변 쌍
EVAL_DATASET = [
    {
        "id": "basic_async",
        "question": "Python async/await를 언제 써야 하나요?",
        "expected_keywords": ["I/O", "비동기", "이벤트 루프"],
        "min_score": 7
    },
    {
        "id": "rag_failure",
        "question": "RAG 시스템에서 hallucination을 줄이는 방법은?",
        "expected_keywords": ["검색", "컨텍스트", "청킹"],
        "min_score": 7
    },
    # 실제 운영에서는 200개 이상 권장
]

def run_eval(threshold: float) -> dict:
    results = []

    for case in EVAL_DATASET:
        # 에이전트/LLM 실행
        answer_resp = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=300,
            messages=[{"role": "user", "content": case["question"]}]
        )
        answer = answer_resp.content[0].text

        # Judge 평가
        keywords_found = sum(1 for kw in case["expected_keywords"] if kw in answer)
        keyword_score = (keywords_found / len(case["expected_keywords"])) * 10

        judge_resp = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=100,
            messages=[{"role": "user", "content": f"""
질문: {case['question']}
답변: {answer}
점수(0~10, 숫자만):"""}]
        )

        try:
            llm_score = float(judge_resp.content[0].text.strip())
        except:
            llm_score = 5.0

        # 종합 점수: 키워드 40% + LLM Judge 60%
        final_score = keyword_score * 0.4 + llm_score * 0.6
        passed = final_score >= case["min_score"]

        results.append({
            "id": case["id"],
            "score": round(final_score, 2),
            "passed": passed
        })

    avg_score = sum(r["score"] for r in results) / len(results)
    pass_rate = sum(1 for r in results if r["passed"]) / len(results) * 100
    failed_cases = [r["id"] for r in results if not r["passed"]]
    overall_passed = avg_score >= threshold

    output = {
        "avg_score": round(avg_score, 2),
        "pass_rate": round(pass_rate, 1),
        "failed_cases": failed_cases,
        "passed": overall_passed,
        "details": results
    }

    Path("eval/result.json").write_text(json.dumps(output, ensure_ascii=False))
    print(json.dumps(output, ensure_ascii=False, indent=2))

    return output

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--threshold", type=float, default=7.0)
    args = parser.parse_args()

    result = run_eval(args.threshold)
    sys.exit(0 if result["passed"] else 1)  # CI 통과/실패 신호

→ PR 올릴 때마다 Judge 자동 실행 → 프롬프트 변경 품질 회귀 즉시 감지
→ sys.exit(1) 로 CI 실패 신호 → 기준 미달이면 머지 불가 설정 가능
→ eval dataset은 최소 50개 실패 케이스부터 시작, 버그 발견 시마다 추가
→ 2026 트렌드: eval 결과가 PR 코멘트로 자동 게시 → 팀 전체가 품질 인지

---

실전5 — 프로덕션 트래픽 → eval dataset 자동 수집

실서비스 로그가 가장 현실적인 eval dataset입니다.

import anthropic, json
from datetime import datetime
from pathlib import Path

client = anthropic.Anthropic()

class ProductionEvalCollector:
    """
    프로덕션 트래픽에서 eval dataset 자동 수집
    낮은 점수 / 사용자 thumbs-down → 자동으로 eval case에 추가
    """

    def __init__(self, dataset_path: str = "eval/production_cases.jsonl"):
        self.dataset_path = Path(dataset_path)
        self.dataset_path.parent.mkdir(exist_ok=True)

    def log_interaction(self, question: str, answer: str,
                        user_feedback: str = None):
        """실서비스 응답마다 호출 — 10% 샘플링"""
        import random
        if random.random() > 0.10:  # 10%만 샘플링
            return

        # 자동 Judge 점수
        judge_resp = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=50,
            messages=[{"role": "user", "content":
                f"질문: {question}\n답변: {answer}\n\n점수(0~10, 숫자만):"}]
        )
        try:
            score = float(judge_resp.content[0].text.strip())
        except:
            score = 5.0

        case = {
            "timestamp": datetime.now().isoformat(),
            "question": question,
            "answer": answer,
            "auto_score": score,
            "user_feedback": user_feedback,  # "thumbs_up" / "thumbs_down" / None
            # 낮은 점수 or 부정 피드백이면 인간 검토 플래그
            "needs_review": score < 5.0 or user_feedback == "thumbs_down"
        }

        # JSONL 형식으로 누적 저장
        with open(self.dataset_path, "a", encoding="utf-8") as f:
            f.write(json.dumps(case, ensure_ascii=False) + "\n")

    def export_regression_cases(self, min_cases: int = 50) -> list:
        """needs_review=True인 케이스만 뽑아 다음 eval dataset으로"""
        cases = []
        if not self.dataset_path.exists():
            return cases

        with open(self.dataset_path, encoding="utf-8") as f:
            for line in f:
                case = json.loads(line)
                if case.get("needs_review"):
                    cases.append(case)

        print(f"수집된 회귀 케이스: {len(cases)}개")
        return cases[:min_cases]

collector = ProductionEvalCollector()

# 실서비스 응답 로깅 예시
collector.log_interaction(
    question="Redis와 Memcached 차이는?",
    answer="Redis는 다양한 자료구조 지원, Memcached는 단순 키-값만 지원합니다.",
    user_feedback="thumbs_up"
)

regression_cases = collector.export_regression_cases()
print(f"다음 eval에 추가할 케이스: {len(regression_cases)}개")

→ 핵심 원칙: "실패 케이스가 발견될 때마다 dataset에 추가" → eval이 점점 강해짐
→ 사용자 thumbs-down은 Judge보다 신뢰도 높은 신호 — 우선 수집
→ 10% 샘플링으로 API 비용 제어하면서 데이터 품질 유지
→ Level 4 팀의 eval dataset은 100% 프로덕션 실패 사례에서 자라남

---

✅ 시리즈 3편 핵심 정리
✅ 에이전트는 최종 결과가 아닌 궤적(Trajectory) 전체를 평가해야 함
✅ 4단계 레이어드 평가로 단일 LLM Judge 대비 비용 1/5로 절감
✅ Self-refinement 루프: Judge 피드백 → 재시도 → 재평가 자동화
✅ GitHub Actions 통합: PR마다 Judge 자동 실행, 기준 미달 머지 차단
✅ 프로덕션 트래픽 10% 샘플링 → 실패 케이스 자동 수집 → eval dataset 성장

❌ CoT만 보고 평가하면 에이전트가 Judge 속일 수 있음 — tool call 결과 기준 평가 필수
❌ eval dataset 없이 Judge 단독 운용은 반쪽짜리 — 최소 50개 케이스부터
❌ Self-refinement 재시도 상한 없으면 비용 폭발 — 2~3회 제한 필수
❌ 2026년에도 대부분 팀은 Level 0~1 — 이 파이프라인 구축만으로 경쟁 우위

---

관련글

• LLM as a Judge 1편 — 개념과 첫 구현
• LLM as a Judge 2편 — 편향 제거와 Jury 패턴