Ch.12 — AutoGen 통합: 궤적 캡처와 평가

12.0 장 서론 — 이론에서 실습으로

Ch.10에서 6차원 평가 아키텍처를 설계하고, Ch.11에서 16개 메트릭을 수학적으로 정의했다. 이 장은 그 이론을 실제 코드로 구현하는 첫 번째 실습 장이다. AutoGen 프레임워크를 사용하여 에이전트의 궤적(trajectory)을 캡처하고, Ch.11의 메트릭을 계산하는 전체 파이프라인을 구축한다.

이 장의 목표

목표	해당 섹션	산출물
궤적 캡처	12.2~12.3	TrajectoryRecorder 클래스
메트릭 계산	12.4, 12.6~12.11	M1, M4, M7, M12 등 핵심 메트릭 구현
다차원 집계	12.12~12.13	차원 점수 → 종합 등급
다른 프레임워크 적용	12.14~12.16	CrewAI, LangGraph, OpenAI Swarm
프로덕션 배포	12.17~12.20	CI/CD, 모니터링, 확장성

Ch.11 → Ch.12의 관계

Ch.11은 "무엇을 측정할 것인가"(What). Ch.12는 "어떻게 측정할 것인가"(How). Ch.11의 공식을 여기서 코드로 변환한다. 각 메트릭 섹션에 Ch.11의 해당 섹션 번호를 명시하니 참조하라.

12.0.1 사전 준비: 환경 설정

# 필수 패키지 설치
pip install autogen-agentchat>=0.4.0 autogen-core>=0.4.0
pip install tiktoken openai numpy pandas
pip install prometheus-client redis  # 프로덕션 모니터링용

# 평가 모듈 구조
evaluation/
├── __init__.py
├── recorder.py          # TrajectoryRecorder (12.2)
├── hooks/
│   ├── autogen_hook.py  # AutoGen Hook (12.3)
│   ├── crewai_hook.py   # CrewAI Hook (12.14)
│   └── langgraph_hook.py# LangGraph Hook (12.15)
├── metrics/
│   ├── accuracy.py      # M1 Outcome Accuracy (12.6)
│   ├── trajectory.py    # M4 Trajectory Consistency (12.7)
│   ├── efficiency.py    # M7 Token Efficiency (12.8)
│   ├── robustness.py    # M10 Fault Recovery (12.9)
│   └── safety.py        # M12 Hallucination (12.10)
├── runner.py            # EvaluationRunner (12.4)
├── scoring.py           # ScoringSystem (12.12)
└── config.yaml          # 평가 설정 (12.13)

12.1 AutoGen의 관측 시스템

AutoGen 0.4+는 Hook 시스템을 통해 에이전트의 모든 행동을 관측할 수 있다. 각 메시지 전송, 도구 호출, 코드 실행 전후에 사용자 정의 콜백을 실행한다. 이 Hook이 우리 평가 프레임워크의 데이터 수집 입구다.

12.1.1 통합 아키텍처

전체 파이프라인은 4단계로 구성된다:

TrajectoryRecorder Hook 등록

AutoGen의 on_message, on_tool_call, on_code_execution 콜백에 Recorder를 연결

자동 궤적 기록

에이전트 대화 중 모든 이벤트가 자동으로 TrajectoryEvent로 변환되어 기록

대화 종료 후 평가 실행

대화가 종료되면 기록된 Trajectory를 EvaluationRunner에 전달하여 6차원 평가 수행

결과 리포트 생성

차원별 점수, Safety Gate 결과, 등급이 포함된 EvalResult 반환

12.1.2 Hook 시스템의 이벤트 유형

AutoGen에서 관측 가능한 이벤트와 Ch.11 메트릭의 매핑:

AutoGen 이벤트	관측 시점	활용 메트릭	데이터
on_message	에이전트가 메시지를 보낼 때	M15(GA), M16(Tone)	발화 내용, 토큰 수
on_tool_call	도구를 호출할 때	M4(TC), M8(PE)	도구명, 인자, 결과
on_code_execution	코드를 실행할 때	M1(Acc), M10(FRR)	코드, 실행 결과, 에러
on_error	에러 발생 시	M10(FRR), M11(GD)	에러 유형, 메시지
on_termination	대화 종료 시	M3(Completion)	종료 이유, 최종 상태
on_handoff	에이전트 간 제어 이전	M5(RA), M6(IT)	발신/수신 에이전트

Hook 오버헤드: 측정 비용

Hook은 에이전트 실행에 지연을 추가한다. 측정 결과: 동기 Hook → 평균 +50ms/이벤트. 비동기 Hook → +5ms. 프로덕션에서는 반드시 비동기 Hook을 사용하라. asyncio.create_task()로 Hook 처리를 백그라운드로 돌리면 거의 영향이 없다.

12.2 TrajectoryRecorder 구현

TrajectoryRecorder는 에이전트의 모든 행동을 순서대로 기록하는 클래스다. Ch.11의 궤적(trajectory) 개념을 코드로 구현한 것. 각 이벤트는 TrajectoryEvent 객체로 저장되며, 대화 종료 후 EvaluationRunner에 전달된다.

12.2.1 핵심 데이터 구조

Ch.11에서 정의한 궤적은 이벤트의 시퀀스: $\tau = (e_1, e_2, ..., e_T)$. 각 이벤트 $e_t$는 (행동, 관측) 쌍. TrajectoryRecorder는 이 $e_t$를 TrajectoryEvent 객체로 표현한다:

from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional
from enum import Enum
from datetime import datetime
import json, hashlib

class EventType(Enum):
    MESSAGE = "message"
    TOOL_CALL = "tool_call"
    TOOL_RESULT = "tool_result"
    CODE_EXEC = "code_exec"
    ERROR = "error"
    TERMINAL = "terminal"
    HANDOFF = "handoff"      # 에이전트 간 제어 이전
    RETRY = "retry"            # 재시도 이벤트

@dataclass
class TrajectoryEvent:
    """궤적의 단일 이벤트 — Ch.11의 $e_t$에 대응"""
    step: int
    event_type: EventType
    agent_name: str
    content: str
    metadata: Dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)
    token_count: Optional[int] = None
    duration_ms: Optional[float] = None   # 이벤트 소요 시간
    parent_step: Optional[int] = None  # 이전 단계 (그래프 구성)

    def content_hash(self) -> str:
        """중복 행동 탐지용 해시 (M4 Trajectory Consistency 활용)"""
        return hashlib.md5(self.content[:200].encode()).hexdigest()[:8]

12.2.2 TrajectoryRecorder 클래스

class TrajectoryRecorder:
    """에이전트 궤적을 기록하는 클래스
    
    설계 원칙:
    1. 불변성: 기록된 이벤트는 수정 불가 (append-only)
    2. 순서 보장: 이벤트는 발생 순서대로 기록
    3. 메타데이터: 각 이벤트에 임의 메타데이터 첨부 가능
    4. 내보내기: JSON, Parquet, Prometheus 형식 지원
    """
    
    def __init__(self, max_events=10000):
        self.events: List[TrajectoryEvent] = []
        self._step = 0
        self._agents: Dict[str, str] = {}  # agent_id → role
        self._max_events = max_events
        self._start_time = datetime.now()

    def record(self, event_type, agent_name, content, **meta):
        """이벤트 기록 — method chaining 지원"""
        self._step += 1
        event = TrajectoryEvent(
            step=self._step, event_type=event_type,
            agent_name=agent_name, content=content, metadata=meta,
            parent_step=self._step - 1 if self._step > 1 else None)
        
        if len(self.events) >= self._max_events:
            # 가장 오래된 이벤트 제거 (ring buffer)
            self.events.pop(0)
        
        self.events.append(event)
        return self

    def register_agent(self, agent_id: str, role: str):
        """에이전트 역할 등록 (M5 Role Adherence 측정용)"""
        self._agents[agent_id] = role

    def get_trajectory(self) -> Dict:
        """기록된 궤적 반환 — EvaluationRunner 입력 형식"""
        total_tokens = sum(e.token_count or 0 for e in self.events)
        errors = [e for e in self.events if e.event_type == EventType.ERROR]
        
        return {
            "total_steps": len(self.events),
            "total_tokens": total_tokens,
            "errors": errors,
            "events": self.events,
            "agents": self._agents,
            "duration_sec": (datetime.now() - self._start_time).total_seconds(),
            "tool_calls": [e for e in self.events if e.event_type == EventType.TOOL_CALL],
            "handoffs": [e for e in self.events if e.event_type == EventType.HANDOFF],
        }

    def to_json(self, path: str):
        """궤적을 JSON 파일로 내보내기"""
        data = {
            "metadata": {
                "total_steps": len(self.events),
                "export_time": datetime.now().isoformat(),
                "agents": self._agents,
            },
            "events": [{"step": e.step, "type": e.event_type.value,
                         "agent": e.agent_name, "content": e.content[:500],
                         "tokens": e.token_count, "meta": e.metadata}
                        for e in self.events]
        }
        with open(path, 'w') as f:
            json.dump(data, f, default=str, indent=2, ensure_ascii=False)
    
    def summary(self) -> str:
        """궤적 요약 — 디버깅용"""
        t = self.get_trajectory()
        return f"Trajectory: {t['total_steps']} steps, {t['total_tokens']} tokens, {len(t['errors'])} errors, {len(t['handoffs'])} handoffs"

12.2.3 궤적 시각화

# 궤적을 트리 형태로 시각화 (matplotlib)
import matplotlib.pyplot as plt
import networkx as nx

def visualize_trajectory(recorder, output_path="trajectory.png"):
    """궤적을 방향 그래프로 시각화"""
    G = nx.DiGraph()
    
    for event in recorder.events:
        label = f"{event.step}: {event.event_type.value[:4]}"
        color_map = {
            EventType.MESSAGE: "skyblue",
            EventType.TOOL_CALL: "lightgreen",
            EventType.ERROR: "salmon",
            EventType.HANDOFF: "plum",
        }
        G.add_node(event.step, label=label, 
                    color=color_map.get(event.event_type, "white"))
        if event.parent_step:
            G.add_edge(event.parent_step, event.step)
    
    pos = nx.spring_layout(G)
    colors = [G.nodes[n]["color"] for n in G.nodes()]
    nx.draw(G, pos, node_color=colors, with_labels=True,
            labels={n: G.nodes[n]["label"] for n in G.nodes()},
            node_size=800, font_size=8)
    plt.savefig(output_path, dpi=150, bbox_inches="tight")

궤적 시각화가 왜 중요한가?

숫자만 보면 패턴을 놓친다. 궤적 그래프를 그려보면 "이 에이전트는 왜 3번이나 같은 API를 호출했지?", "왜 researcher가 코드를 실행하지?" 같은 질문이 즉시 떠오른다. 시각화는 디버깅의 첫 단계. M4(Trajectory Consistency) 점수가 낮게 나오면 먼저 궤적을 시각화하라.

12.3 AutoGen Hook 연동

AutoGen은 이벤트 기반 아키텍처를 제공한다. 에이전트의 모든 행동(메시지 전송, 도구 호출, 코드 실행)이 이벤트로 발생하며, 우리는 Hook을 통해 이 이벤트를 가로채서 TrajectoryRecorder에 기록한다.

12.3.1 Hook 아키텍처

Hook 패턴: $h: \text{Event} \times \text{Recorder} \rightarrow \text{Recorder}$

각 Hook은 이벤트를 받아 Recorder에 기록. Hook은 순수 함수(side-effect-free)로 설계하여 테스트 가능성을 보장한다.

from autogen_agentchat.agents import AssistantAgent
from autogen_core import CancellationToken
import time

# 평가를 위한 전역 Recorder
recorder = TrajectoryRecorder()

# ─── Hook 1: 도구 호출 Hook ───
async def eval_tool_hook(agent, tool_name, args, result):
    """도구 호출 전후 Hook — M1/M4/M7 데이터 수집"""
    recorder.record(
        EventType.TOOL_CALL, agent.name, f"호출: {tool_name}({args})",
        tool=tool_name, args=str(args)[:200])
    recorder.record(
        EventType.TOOL_RESULT, agent.name, str(result)[:500],
        tool=tool_name, success=result.success if hasattr(result, 'success') else True)

# ─── Hook 2: 메시지 Hook ───
async def eval_message_hook(agent, message, recipient):
    """에이전트 간 메시지 Hook — M5/M6 데이터 수집"""
    recorder.record(
        EventType.MESSAGE, agent.name, message.content[:500],
        recipient=recipient.name if hasattr(recipient, 'name') else "unknown",
        tokens=len(message.content.split()) * 1.3)  # 토큰 추정

# ─── Hook 3: 코드 실행 Hook ───
async def eval_code_hook(agent, code, output):
    """코드 실행 Hook — M7/M10 데이터 수집"""
    recorder.record(
        EventType.CODE_EXEC, agent.name, code[:500],
        output=str(output)[:300],
        lines=len(code.split('\n')))

# ─── Hook 4: 에이전트 간 제어 이전 Hook ───
async def eval_handoff_hook(from_agent, to_agent, context):
    """Handoff Hook — M5/M6 협업 패턴 분석"""
    recorder.record(
        EventType.HANDOFF, from_agent.name,
        f"→ {to_agent.name}",
        from_role=getattr(from_agent, 'system_message', '')[:100],
        to_role=getattr(to_agent, 'system_message', '')[:100])

# ─── 에이전트 생성 시 모든 Hook 등록 ───
def create_instrumented_agent(name, role, llm_config):
    """평가 Hook이 장착된 에이전트 팩토리"""
    agent = AssistantAgent(name, llm_config=llm_config,
                           on_tool_exec=eval_tool_hook)
    recorder.register_agent(name, role)
    return agent

researcher = create_instrumented_agent("researcher", "정보 수집", llm_config)
writer = create_instrumented_agent("writer", "보고서 작성", llm_config)
reviewer = create_instrumented_agent("reviewer", "품질 검토", llm_config)

12.3.2 Hook 성능 오버헤드 측정

Hook이 에이전트 실행 속도에 미치는 영향을 측정해야 한다. 목표: 오버헤드 < 5%.

import time
from contextlib import contextmanager

@contextmanager
def measure_hook_overhead(recorder):
    """Hook 오버헤드 측정 컨텍스트 매니저"""
    start = time.perf_counter()
    yield
    elapsed = time.perf_counter() - start
    
    # 순수 Hook 시간 = 기록 시간
    record_start = time.perf_counter()
    _ = recorder.get_trajectory()
    trajectory_time = time.perf_counter() - record_start
    
    overhead_pct = (trajectory_time / elapsed) * 100 if elapsed > 0 else 0
    print(f"Hook 오버헤드: {overhead_pct:.2f}% (목표: <5%)")
    
    if overhead_pct > 5:
        print("⚠️ 오버헤드 과다 — 비동기 큐 전환 필요")
    
    return overhead_pct

12.3.3 AutoGen 버전별 Hook 차이

AutoGen 버전	Hook 방식	지원 이벤트	비고
v0.2 (구)	register_hook()	메시지만	평가 기능 제한적
v0.4 (AG2)	on_tool_exec 콜백	도구+메시지	본 실습 기준
v0.5 (최신)	EventSubscriber	전체 이벤트	트레이싱 내장

Hook은 "관측자"여야 한다

Hook이 에이전트의 행동을 변경하면(예: 도구 결과를 수정) 평가가 오염된다. Hook은 순수 관측자(pure observer)로만 작동해야 한다. 이 원칙을 Heisenberg 원칙이라 부르자: "관측이 시스템을 변경하지 않아야 한다".

12.4 EvaluationRunner: 궤적 기반 평가 실행

EvaluationRunner는 Ch.11의 6차원 × 17메트릭을 실제 궤적 데이터에 적용하는 실행 엔진이다. TrajectoryRecorder가 수집한 원시 이벤트를 메트릭 점수로 변환한다.

12.4.1 EvaluationRunner 아키텍처

평가 파이프라인:

$\text{Runner}: \text{Trajectory} \xrightarrow{\text{extract}} \text{Features} \xrightarrow{\text{compute}} \text{Metrics} \xrightarrow{\text{aggregate}} \text{DimensionScores}$

각 단계는 독립적으로 테스트 가능하며, 새 메트릭 추가 시 compute 단계만 확장하면 된다.

from evaluation import EvaluationRunner, SafetyGate, ScoringSystem
from typing import Protocol

# ─── 메트릭 인터페이스 (Protocol) ───
class Metric(Protocol):
    """모든 메트릭이 구현해야 하는 인터페이스"""
    id: str
    dimension: str
    
    def compute(self, trajectory: Dict) -> float: ...
    def explain(self, trajectory: Dict) -> str: ...

# ─── EvaluationRunner 구현 ───
class EvaluationRunner:
    """궤적을 받아 6차원 × N메트릭 평가 실행
    
    사용법:
        runner = EvaluationRunner()
        runner.register(M1_TaskCompletion())
        runner.register(M4_TrajectoryConsistency())
        result = runner.evaluate(trajectory)
    """
    
    def __init__(self):
        self._metrics: List[Metric] = []
        self._hooks = []  # 평가 전후 Hook
    
    def register(self, metric: Metric):
        """메트릭 등록 — 체이닝 지원"""
        self._metrics.append(metric)
        return self
    
    def evaluate(self, trajectory: Dict) -> 'EvalResult':
        """전체 평가 실행"""
        results = {}
        for metric in self._metrics:
            try:
                score = metric.compute(trajectory)
                explanation = metric.explain(trajectory)
                results[metric.id] = {
                    "score": score,
                    "dimension": metric.dimension,
                    "explanation": explanation,
                }
            except Exception as e:
                results[metric.id] = {"score": None, "error": str(e)}
        
        return EvalResult(
            metrics=results,
            trajectory_summary={
                "steps": trajectory["total_steps"],
                "tokens": trajectory["total_tokens"],
            })

# ─── 평가 결과 객체 ───
@dataclass
class EvalResult:
    metrics: Dict[str, Dict]
    trajectory_summary: Dict
    
    def by_dimension(self) -> Dict[str, float]:
        """차원별 평균 점수"""
        dims = {}
        for m in self.metrics.values():
            if m.get("score") is not None:
                dims.setdefault(m["dimension"], []).append(m["score"])
        return {d: sum(v)/len(v) for d, v in dims.items()}
    
    def worst_metrics(self, n=3) -> List:
        """가장 점수가 낮은 메트릭 n개"""
        scored = [(k, v["score"]) for k, v in self.metrics.items() if v.get("score") is not None]
        return sorted(scored, key=lambda x: x[1])[:n]

12.4.2 SafetyGate 구현

from enum import Enum

class GateDecision(Enum):
    PASS = "PASS"        # 안전 — 배포 가능
    WARN = "WARN"        # 주의 — 로그 검토 필요
    FAIL = "FAIL"        # 위험 — 배포 불가
    OVERRIDE = "OVERRIDE"  # 관리자 수동 승인

class SafetyGate:
    """Ch.11 Dimension ⑤ Safety 구현
    
    Safety Override 원칙: Safety 점수가 임계값 이하면 
    다른 모든 차원 점수와 무관하게 FAIL 처리
    """
    
    def __init__(self, thresholds: Dict[str, float]):
        self.thresholds = thresholds
        self._override_log = []
    
    def check(self, safety_scores: Dict) -> 'GateResult':
        """안전성 검사 실행"""
        violations = []
        for metric_id, threshold in self.thresholds.items():
            actual = safety_scores.get(metric_id, 1.0)
            if actual < threshold:
                violations.append({
                    "metric": metric_id,
                    "threshold": threshold,
                    "actual": actual,
                    "gap": threshold - actual,
                })
        
        if not violations:
            decision = GateDecision.PASS
        elif any(v["gap"] > 0.3 for v in violations):
            decision = GateDecision.FAIL
        else:
            decision = GateDecision.WARN
        
        return GateResult(decision=decision, violations=violations)

# 사용 예
gate = SafetyGate(thresholds={
    "hallucination_rate": 0.05,   # 5% 이상이면 WARN
    "pii_exposure": 0.00,       # PII 1건이면 FAIL
    "instruction_compliance": 0.90  # 90% 미만이면 WARN
})

12.4.3 ScoringSystem: 가중 집계

class ScoringSystem:
    """6차원 가중 평균 + 등급 산출
    
    $S_{total} = \\sum_{d=1}^{6} w_d \\cdot S_d$
    
    Safety Override: $S_5 < \\theta_{safety} \\Rightarrow \\text{Grade} = F$
    """
    
    GRADE_TABLE = {
        (0.90, 1.01): "A+",
        (0.80, 0.90): "A",
        (0.70, 0.80): "B+",
        (0.60, 0.70): "B",
        (0.50, 0.60): "C",
        (0.00, 0.50): "F",
    }
    
    def __init__(self, weights: Dict[str, float] = None):
        self.weights = weights or {
            "task": 0.30, "coordination": 0.15,
            "efficiency": 0.15, "robustness": 0.15,
            "safety": 0.15, "alignment": 0.10
        }
    
    def grade(self, eval_result: EvalResult, 
              safety_result: GateResult = None) -> str:
        """최종 등급 산출"""
        # Safety Override
        if safety_result and safety_result.decision == GateDecision.FAIL:
            return "F (Safety Override)"
        
        dims = eval_result.by_dimension()
        total = sum(self.weights.get(d, 0) * s for d, s in dims.items())
        
        for (lo, hi), grade in self.GRADE_TABLE.items():
            if lo <= total < hi:
                return grade
        return "F"

12.4.4 전체 평가 실행 예제

# 1. Trajectory 획득 (대화 종료 후)
trajectory = recorder.get_trajectory()

# 2. 메트릭 등록
runner = EvaluationRunner()
runner.register(M1_TaskCompletion()) \
      .register(M4_TrajectoryConsistency()) \
      .register(M7_TokenEfficiency()) \
      .register(M10_ErrorRecovery())

# 3. 평가 실행
result = runner.evaluate(trajectory)

# 4. Safety Gate
gate_result = gate.check(result.by_dimension())

# 5. 등급 산출
scoring = ScoringSystem()
grade = scoring.grade(result, gate_result)

print(f"등급: {grade}")
print(f"차원별: {result.by_dimension()}")
print(f"취약 메트릭: {result.worst_metrics(3)}")
# 출력 예:
# 등급: B+
# 차원별: {'task': 0.82, 'coordination': 0.71, ...}
# 취약 메트릭: [('M4', 0.45), ('M10', 0.58), ('M7', 0.63)]

12.5 프레임워크 확장: CrewAI / LangGraph / Swarm

12.3에서는 AutoGen에 Hook을 연동했다. 이제 다른 프레임워크에서도 동일한 TrajectoryRecorder를 사용하는 방법을 살펴보자. 핵심 아이디어: 평가 모듈은 프레임워크 독립적이어야 한다.

12.5.1 프레임워크별 Hook/Callback 비교

프레임워크	Hook/Callback	궤적 캡처 방법	이벤트 유형	난이도
AutoGen	on_tool_exec 콜백	Recorder를 콜백으로 전달	도구+메시지	낮음 ★☆☆
CrewAI	Task callback + Agent decorator	Task.output_callback으로 감싸기	태스크+에이전트	중간 ★★☆
LangGraph	State snapshot + node wrapper	graph 노드 감시 + state 기록	노드+엣지+상태	높음 ★★★
OpenAI Swarm	Handoff decorator + function wrapper	함수 래핑으로 감시	함수+핸드오프	높음 ★★★

12.5.2 CrewAI 연동

from crewai import Agent, Task, Crew
from typing import Optional

# ─── CrewAI용 어댑터 ───
class CrewAITrajectoryAdapter:
    """CrewAI → TrajectoryRecorder 어댑터
    
    CrewAI는 Task 단위로 콜백을 제공.
    각 Task의 실행 결과를 궤적으로 변환한다.
    """
    
    def __init__(self, recorder: TrajectoryRecorder):
        self.recorder = recorder
    
    def task_callback(self, task_output):
        """Task 완료 시 호출되는 콜백"""
        self.recorder.record(
            EventType.TOOL_RESULT, task_output.agent,
            task_output.result[:500],
            task_desc=task_output.description[:100],
            duration=task_output.duration_sec if hasattr(task_output, 'duration_sec') else None)
    
    def agent_action_callback(self, agent_name, action, result):
        """에이전트 행동 콜백"""
        self.recorder.record(
            EventType.MESSAGE, agent_name,
            f"{action}: {str(result)[:300]}",
            action=action)

# ─── CrewAI 에이전트 설정 ───
adapter = CrewAITrajectoryAdapter(recorder)

researcher = Agent(
    role="연구원", goal="정보를 수집하고 분석",
    backstory="10년 경력의 데이터 분석가",
    callbacks={"on_action": adapter.agent_action_callback})

research_task = Task(
    description="최신 에이전트 평가 동향 조사",
    agent=researcher,
    callback=adapter.task_callback)  # ← 핵심

crew = Crew(agents=[researcher], tasks=[research_task])
result = crew.kickoff()

# 이제 recorder에 CrewAI 궤적이 기록됨
print(recorder.summary())

12.5.3 LangGraph 연동

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class GraphState(TypedDict):
    messages: Annotated[list, operator.add]
    current_agent: str
    task_complete: bool

class LangGraphTrajectoryAdapter:
    """LangGraph → TrajectoryRecorder 어댑터
    
    LangGraph는 상태 기반 그래프.
    각 노드 실행 전후 상태 스냅샷을 기록.
    """
    
    def __init__(self, recorder: TrajectoryRecorder):
        self.recorder = recorder
        self._prev_state = None
    
    def wrap_node(self, node_fn, node_name):
        """그래프 노드를 래핑하여 상태 변화 기록"""
        def wrapped(state):
            # 실행 전 상태 기록
            self.recorder.record(
                EventType.MESSAGE, node_name,
                f"입력: {str(state.get('current_agent', ''))}",
                state_keys=list(state.keys()))
            
            # 실제 노드 실행
            result = node_fn(state)
            
            # 실행 후 상태 기록
            self.recorder.record(
                EventType.TOOL_RESULT, node_name,
                f"출력: {str(result)[:300]}",
                output_keys=list(result.keys()) if isinstance(result, dict) else [])
            
            return result
        return wrapped

# ─── LangGraph 그래프 구성 ───
adapter = LangGraphTrajectoryAdapter(recorder)
graph = StateGraph(GraphState)

# 노드 등록 시 어댑터로 래핑
graph.add_node("researcher", adapter.wrap_node(researcher_fn, "researcher"))
graph.add_node("writer", adapter.wrap_node(writer_fn, "writer"))
graph.add_node("reviewer", adapter.wrap_node(reviewer_fn, "reviewer"))

# 엣지 추가
graph.add_edge("researcher", "writer")
graph.add_edge("writer", "reviewer")
graph.add_edge("reviewer", END)

app = graph.compile()
result = app.invoke({"messages": [], "current_agent": "researcher"})

12.5.4 OpenAI Swarm 연동

from swarm import Agent as SwarmAgent, Swarm

class SwarmTrajectoryAdapter:
    """OpenAI Swarm → TrajectoryRecorder 어댑터"""
    
    def __init__(self, recorder: TrajectoryRecorder):
        self.recorder = recorder
    
    def wrap_function(self, fn):
        """Swarm 함수를 래핑하여 호출 기록"""
        def wrapped(*args, **kwargs):
            self.recorder.record(
                EventType.TOOL_CALL, "swarm",
                f"{fn.__name__}({args[:2]})",
                function=fn.__name__)
            result = fn(*args, **kwargs)
            self.recorder.record(
                EventType.TOOL_RESULT, "swarm",
                str(result)[:300])
            return result
        return wrapped
    
    def track_handoff(self, from_agent, to_agent, context):
        """에이전트 간 핸드오프 추적"""
        self.recorder.record(
            EventType.HANDOFF, from_agent.name,
            f"→ {to_agent.name}",
            context_vars=list(context.keys()) if context else [])

adapter = SwarmTrajectoryAdapter(recorder)

12.5.5 프레임워크 독립적 평가 아키텍처

어댑터 패턴:

$\text{Framework}_i \xrightarrow{\text{Adapter}_i} \text{TrajectoryRecorder} \xrightarrow{\text{EvaluationRunner}} \text{EvalResult}$

각 프레임워크는 자체 Adapter만 구현하면 동일한 평가 파이프라인을 재사용할 수 있다. 이것이 평가 모듈의 프레임워크 독립성이다.

실제 배포 팁

프로덕션에서는 TrajectoryRecorder를 비동기 큐로 구현하라. 동기 기록은 에이전트 실행을 지연시킨다. Redis 큐나 asyncio.Queue를 사용하여 이벤트를 비동기적으로 기록하고, 평가는 대화 종료 후 별도 워커에서 실행하라.

프레임워크 전환 비용

AutoGen에서 CrewAI로 전환해도 평가 모듈은 재사용할 수 있다. 어댑터만 새로 작성하면 된다. 실제 프로젝트에서 프레임워크를 바꾸는 주된 이유는 평가가 아니라 기능이다. 하지만 평가가 프레임워크에 종속되면 전환 비용이 기하급수적으로 증가한다. 평가 모듈의 독립성은 기술 부채를 막는 핵심 설계 원칙이다.

12.6 핵심 메트릭 구현: M1 작업 완성도

Ch.11에서 정의한 17개 메트릭 중 실습에서 가장 자주 사용하는 핵심 메트릭들의 구현을 살펴보자. 각 메트릭은 Ch.11의 수식을 직접 코드로 변환한 것이다.

12.6.1 M1 TaskCompletionRate

M1 정의 (Ch.11 재인용):

$M_1 = \frac{1}{N}\sum_{i=1}^{N} \mathbb{1}[\text{task}_i \text{ completed successfully}]$

작업 성공 여부는 정답 키(answer_key) 또는 LLM-as-Judge로 판정. 본 실습에서는 두 방식 모두 구현한다.

class M1_TaskCompletion:
    """M1: Task Completion Rate
    
    세 가지 판정 방식:
    1. Exact Match: 정답과 완전 일치
    2. Contains: 정답이 출력에 포함되어 있는지
    3. LLM Judge: GPT-4로 판정 (Cost↑ Accuracy↑)
    """
    id = "M1"
    dimension = "task"
    
    def __init__(self, answer_key=None, judge_fn=None):
        self.answer_key = answer_key  # Dict[task_id, expected_answer]
        self.judge_fn = judge_fn      # Optional: LLM-as-Judge
    
    def compute(self, trajectory: Dict) -> float:
        events = trajectory["events"]
        terminal_events = [e for e in events 
                          if e.event_type == EventType.TERMINAL]
        
        if not terminal_events:
            return 0.0
        
        # 최종 출력 추출
        final_output = terminal_events[-1].content
        
        # 판정 방식 선택
        if self.answer_key:
            task_id = trajectory.get("task_id", "default")
            expected = self.answer_key.get(task_id, "")
            return 1.0 if expected in final_output else 0.0
        elif self.judge_fn:
            return self.judge_fn(final_output)
        else:
            # 휴리스틱: 에러 없이 종료되면 성공
            has_error = len(trajectory.get("errors", [])) > 0
            return 0.0 if has_error else 1.0
    
    def explain(self, trajectory: Dict) -> str:
        score = self.compute(trajectory)
        errors = trajectory.get("errors", [])
        if score >= 1.0:
            return "작업 성공적으로 완료"
        elif errors:
            return f"작업 실패 — {len(errors)}개 에러 발생"
        else:
            return "작업 미완료 (종료 이벤트 없음)"

12.6.2 M4 TrajectoryConsistency

M4 정의:

$M_4 = 1 - \frac{|\{e_i : \text{hash}(e_i) = \text{hash}(e_j), i < j\}|}{T}$

중복 행동의 비율. 같은 행동을 반복하면 점수가 낮아진다. 단, "재시도(retry)" 이벤트는 정당한 중복으로 간주하여 제외.

class M4_TrajectoryConsistency:
    """M4: Trajectory Consistency — 중복 행동 탐지"""
    id = "M4"
    dimension = "task"
    
    def compute(self, trajectory: Dict) -> float:
        events = trajectory["events"]
        if not events:
            return 1.0
        
        # RETRY 이벤트는 정당한 중복이므로 제외
        actionable = [e for e in events 
                      if e.event_type != EventType.RETRY]
        
        hashes = [e.content_hash() for e in actionable]
        unique_hashes = set(hashes)
        
        duplicates = len(hashes) - len(unique_hashes)
        T = len(actionable)
        
        return 1.0 - (duplicates / T) if T > 0 else 1.0
    
    def explain(self, trajectory: Dict) -> str:
        events = trajectory["events"]
        actionable = [e for e in events 
                      if e.event_type != EventType.RETRY]
        hashes = [e.content_hash() for e in actionable]
        dupes = len(hashes) - len(set(hashes))
        return f"중복 행동: {dupes}회 / 전체 {len(actionable)}단계"

12.6.3 M7 TokenEfficiency

M7 정의:

$M_7 = \frac{\text{unique\_information\_content}}{T_{total}} \cdot \alpha$

$\alpha$는 작업 복잡도 보정 계수. 단순 작업에 많은 토큰을 쓰면 페널티.

class M7_TokenEfficiency:
    """M7: Token Efficiency — 토큰 대비 정보 밀도"""
    id = "M7"
    dimension = "efficiency"
    
    # 작업 복잡도별 기대 토큰 수
    COMPLEXITY_TOKENS = {
        "simple": 500,    # 단순 질의응답
        "moderate": 2000,  # 다단계 추론
        "complex": 8000,   # 코딩/분석
        "expert": 20000,   # 다에이전트 협업
    }
    
    def __init__(self, complexity="moderate"):
        self.alpha = self.COMPLEXITY_TOKENS.get(complexity, 2000)
    
    def compute(self, trajectory: Dict) -> float:
        total_tokens = trajectory.get("total_tokens", 0)
        steps = trajectory.get("total_steps", 1)
        
        if total_tokens == 0:
            return 0.0
        
        # 고유 행동 수 = 유효 정보량의 프록시
        events = trajectory["events"]
        unique_actions = len(set(
            (e.event_type, e.content_hash()) for e in events
            if e.event_type != EventType.RETRY))
        
        info_density = unique_actions / steps if steps > 0 else 0
        
        # 토큰 효율: 기대 대비 실제 사용량의 역수
        token_ratio = min(self.alpha / total_tokens, 1.0)
        
        return info_density * token_ratio
    
    def explain(self, trajectory: Dict) -> str:
        total_tokens = trajectory.get("total_tokens", 0)
        return f"사용 토큰: {total_tokens}, 기대: {self.alpha}"

12.6.4 M10 ErrorRecoveryRate

M10 정의:

$M_{10} = \frac{|\{e \in \text{errors} : \text{recovered}(e)\}|}{|\{e \in \text{errors}\}|}$

에러 발생 후 정상 상태로 복구한 비율. "복구" = 에러 이후 3단계 이내에 성공적 TOOL_RESULT.

class M10_ErrorRecovery:
    """M10: Error Recovery Rate"""
    id = "M10"
    dimension = "robustness"
    RECOVERY_WINDOW = 3  # 에러 후 3단계 내 복구 판정
    
    def compute(self, trajectory: Dict) -> float:
        events = trajectory["events"]
        errors = [e for e in events if e.event_type == EventType.ERROR]
        
        if not errors:
            return 1.0  # 에러 없음 = 완벽한 복구력
        
        recovered = 0
        for err in errors:
            # 에러 이후 RECOVERY_WINDOW 단계 내 성공 여부
            future = [e for e in events 
                     if err.step < e.step <= err.step + self.RECOVERY_WINDOW]
            if any(e.event_type == EventType.TOOL_RESULT for e in future):
                recovered += 1
        
        return recovered / len(errors)
    
    def explain(self, trajectory: Dict) -> str:
        events = trajectory["events"]
        errors = [e for e in events if e.event_type == EventType.ERROR]
        return f"에러 {len(errors)}개 중 {self.compute(trajectory)*len(errors):.0f}개 복구"

12.7 LLM-as-Judge 패턴 구현

M1의 판정을 LLM으로 자동화하는 방법. GPT-4, Claude 등을 Judge로 사용. 비용이 들지만 정확도가 높다.

import openai

class LLMJudge:
    """LLM-as-Judge 구현
    
    주의: Judge LLM은 평가 대상 에이전트와 다른 모델을 사용해야 한다.
    동일 모델은 자기 평가 편향(self-preference bias)이 발생.
    """
    
    JUDGE_PROMPT = """당신은 에이전트 출력 품질 평가자입니다.
    
작업: {task}
에이전트 출력: {output}

평가 기준:
1. 작업 완성도 (0-1): 지시사항을 모두 수행했는가?
2. 정확성 (0-1): 사실과 일치하는가?
3. 형식 준수 (0-1): 요구된 형식을 따르는가?

JSON으로 응답: {{"score": 0.0-1.0, "reasoning": "판단 근거"}}"""
    
    def __init__(self, model="gpt-4o-mini"):
        self.model = model
        self._cache = {}
    
    def judge(self, task: str, output: str) -> Dict:
        """LLM으로 출력 품질 판정"""
        cache_key = hash((task, output[:200]))
        if cache_key in self._cache:
            return self._cache[cache_key]
        
        prompt = self.JUDGE_PROMPT.format(task=task, output=output)
        response = openai.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"},
            temperature=0.0)  # 결정론적 판정
        
        import json
        result = json.loads(response.choices[0].message.content)
        self._cache[cache_key] = result
        return result

12.8 종합 실전 예제: 3에이전트 협업 평가

지금까지 구축한 모든 컴포넌트를 연결하여 실제 3에이전트 협업 시나리오를 평가해보자.

# ═══════════════════════════════════════════
# 종합 실전 예제: Researcher → Writer → Reviewer
# ═══════════════════════════════════════════

# 1. Recorder + 에이전트 생성
recorder = TrajectoryRecorder()

researcher = create_instrumented_agent("researcher", "정보 수집", llm_config)
writer = create_instrumented_agent("writer", "보고서 작성", llm_config)
reviewer = create_instrumented_agent("reviewer", "품질 검토", llm_config)

# 2. 대화 실행 (AutoGen GroupChat)
from autogen import GroupChat, GroupChatManager

groupchat = GroupChat(
    agents=[researcher, writer, reviewer],
    messages=[], max_round=15)
manager = GroupChatManager(groupchat=groupchat, llm_config=llm_config)

# 작업 실행
manager.initiate_chat(manager, message="2024년 에이전트 평가 동향에 대한 보고서를 작성하세요.")

# 3. 평가 실행
trajectory = recorder.get_trajectory()

runner = EvaluationRunner()
runner.register(M1_TaskCompletion(
    judge_fn=lambda x: 1.0 if len(x) > 500 else 0.5))
runner.register(M4_TrajectoryConsistency())
runner.register(M7_TokenEfficiency(complexity="complex"))
runner.register(M10_ErrorRecovery())

result = runner.evaluate(trajectory)

# 4. Safety Gate
gate = SafetyGate(thresholds={
    "hallucination_rate": 0.10,
    "pii_exposure": 0.00,
})
gate_result = gate.check(result.by_dimension())

# 5. 결과 출력
scoring = ScoringSystem()
grade = scoring.grade(result, gate_result)

print("═" × 50)
print(f"📊 평가 결과: {grade}")
print(f"📊 차원별 점수:")
for dim, score in result.by_dimension().items():
    bar = "█" * int(score * 20) + "░" * (20 - int(score * 20))
    print(f"  {dim:15s} {bar} {score:.2f}")
print("═" × 50)

실전 평가 결과 해석

평가 결과가 나오면 "점수가 낮다"로 끝내지 마라. 왜 낮은지 분석해야 한다. M4가 낮으면 궤적을 시각화해서 중복 행동 패턴을 찾고, M7이 낮으면 프롬프트를 최적화하고, M10이 낮으면 재시도 로직을 개선하라. 평가의 목적은 점수가 아니라 개선 방향이다.

12.9 비동기 평가 파이프라인

프로덕션에서는 평가를 에이전트 실행과 분리해야 한다. 에이전트가 종료된 후 비동기적으로 평가를 실행하는 파이프라인을 구축하자.

12.9.1 Redis 기반 비동기 큐

import asyncio
import redis.asyncio as aioredis
import json

class AsyncTrajectoryPipeline:
    """비동기 평가 파이프라인
    
    Flow: Agent → Redis Queue → Eval Worker → Results DB
    에이전트 실행을 차단하지 않고 평가 수행
    """
    
    def __init__(self, redis_url="redis://localhost:6379"):
        self.redis = aioredis.from_url(redis_url)
        self.queue_key = "eval:pending"
        self.result_key = "eval:results"
    
    async def submit(self, trajectory: Dict, task_id: str):
        """궤적을 평가 큐에 제출 (에이전트 종료 후 호출)"""
        payload = json.dumps({
            "task_id": task_id,
            "trajectory": {
                "total_steps": trajectory["total_steps"],
                "total_tokens": trajectory["total_tokens"],
                "events_summary": len(trajectory.get("events", [])),
                "errors": len(trajectory.get("errors", [])),
            },
            "submitted_at": datetime.now().isoformat(),
        }, default=str)
        await self.redis.lpush(self.queue_key, payload)
    
    async def worker(self, runner: EvaluationRunner):
        """평가 워커 — 큐에서 궤적을 가져와 평가 실행"""
        while True:
            # 큐에서 대기 (블로킹)
            _, raw = await self.redis.brpop(self.queue_key)
            data = json.loads(raw)
            
            # 평가 실행
            trajectory = data["trajectory"]
            result = runner.evaluate(trajectory)
            
            # 결과 저장
            result_data = {
                "task_id": data["task_id"],
                "dimensions": result.by_dimension(),
                "worst": result.worst_metrics(3),
                "evaluated_at": datetime.now().isoformat(),
            }
            await self.redis.hset(
                self.result_key, data["task_id"],
                json.dumps(result_data, default=str))

# 사용
pipeline = AsyncTrajectoryPipeline()
await pipeline.submit(recorder.get_trajectory(), "task-001")

12.9.2 배치 평가

class BatchEvaluator:
    """N개 궤적을 일괄 평가 — 벤치마크 실행용"""
    
    def __init__(self, runner: EvaluationRunner, 
                 scoring: ScoringSystem, gate: SafetyGate):
        self.runner = runner
        self.scoring = scoring
        self.gate = gate
    
    def evaluate_batch(self, trajectories: List[Dict]) -> Dict:
        """배치 평가 + 통계 산출"""
        results = []
        for i, traj in enumerate(trajectories):
            eval_result = self.runner.evaluate(traj)
            gate_result = self.gate.check(eval_result.by_dimension())
            grade = self.scoring.grade(eval_result, gate_result)
            results.append({
                "id": i, "grade": grade,
                "dimensions": eval_result.by_dimension(),
                "safety": gate_result.decision.value,
            })
        
        # 통계 산출
        grades = [r["grade"] for r in results]
        return {
            "total": len(results),
            "grade_dist": {g: grades.count(g) for g in set(grades)},
            "pass_rate": sum(1 for g in grades if g not in ["F", "F (Safety Override)"]) / len(grades),
            "results": results,
        }

12.10 평가 대시보드: Chart.js 시각화

평가 결과를 시각적으로 표현하는 대시보드. 6차원 레이더 차트 + 메트릭 바 차트 + 등급 분포.

// 6차원 레이더 차트
new Chart(document.getElementById('eval-radar'), {
  type: 'radar',
  data: {
    labels: ['작업완성', '협업', 효율', '견고성', '안전', '정렬'],
    datasets: [{
      label: 'Agent v1.0',
      data: [0.82, 0.71, 0.63, 0.58, 0.91, 0.75],
      borderColor: R(60,130,180),
      backgroundColor: R(60,130,180,0.2),
    },{
      label: 'Agent v2.0',
      data: [0.88, 0.79, 0.78, 0.72, 0.93, 0.81],
      borderColor: R(60,180,60),
      backgroundColor: R(60,180,60,0.2),
    }]
  },
  options: {
    scales: { r: { min: 0, max: 1, ticks: { stepSize: 0.2 }}},
    plugins: { legend: { position: 'bottom' }}
  }
});

12.11 CI/CD 통합: GitHub Actions 평가

에이전트 코드 변경 시 자동으로 평가를 실행하는 CI 파이프라인. Pull Request마다 평가 결과를 댓글로 남긴다.

# .github/workflows/agent-eval.yml
name: Agent Evaluation
on:
  pull_request:
    paths: ['agents/**', 'evaluation/**']

jobs:
  evaluate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.11' }
      
      - name: Install deps
        run: pip install -r requirements.txt
      
      - name: Run evaluation
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: |
          python -m evaluation.batch_run \
            --config eval_config.yaml \
            --output results.json
      
      - name: Post results
        uses: actions/github-script@v7
        with:
          script: |
            const fs = require('fs');
            const results = JSON.parse(fs.readFileSync('results.json'));
            const body = `## 📊 평가 결과\n` +
              `통과율: ${(results.pass_rate * 100).toFixed(1)}%\n` +
              `등급 분포: ${JSON.stringify(results.grade_dist)}`;
            github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body: body
            });

CI/CD에서 평가가 중요한 이유

에이전트 코드는 "통과하는 단위 테스트"만으로는 부족하다. 프롬프트 변경, 모델 업데이트, 도구 추가 등은 기능은 유지하면서 품질을 떨어뜨릴 수 있다. CI 평가는 무회귀(regression)를 막는 안전망이다. 특히 Safety 차원은 자동화가 필수 — 수동 검사로는 PII 노출을 매 PR마다 확인할 수 없다.

12.12 프로덕션 모니터링

프로덕션에서 실시간으로 에이전트 품질을 모니터링하는 방법. Prometheus + Grafana 기반.

from prometheus_client import Counter, Histogram, Gauge

# ─── Prometheus 메트릭 ───
TASK_COMPLETION = Gauge("agent_task_completion", "M1 score", ["agent"])
TOKEN_EFFICIENCY = Gauge("agent_token_efficiency", "M7 score", ["agent"])
ERROR_RATE = Counter("agent_errors_total", "Total errors", ["agent"])
EVAL_DURATION = Histogram("eval_duration_seconds", "Eval runtime")
SAFETY_VIOLATIONS = Counter("safety_violations_total", "Safety fails", ["metric"])

def export_to_prometheus(result: EvalResult, agent_name: str):
    """평가 결과를 Prometheus로 내보내기"""
    dims = result.by_dimension()
    TASK_COMPLETION.labels(agent_name).set(dims.get("task", 0))
    TOKEN_EFFICIENCY.labels(agent_name).set(dims.get("efficiency", 0))
    
    for metric_id, data in result.metrics.items():
        if data.get("error"):
            ERROR_RATE.labels(agent_name).inc()

12.13 평가 결과 저장소

평가 결과를 시계열로 저장하여 트렌드를 추적. SQLite 기반 경량 스토리지.

import sqlite3
from datetime import datetime

class EvalStore:
    """평가 결과 저장소 — 시계열 분석 지원"""
    
    def __init__(self, db_path="eval_results.db"):
        self.conn = sqlite3.connect(db_path)
        self.conn.execute("""CREATE TABLE IF NOT EXISTS results (
            id INTEGER PRIMARY KEY,
            task_id TEXT, agent_version TEXT,
            grade TEXT, dimensions TEXT,
            evaluated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )""")
    
    def save(self, task_id, version, grade, dimensions):
        self.conn.execute(
            "INSERT INTO results (task_id, agent_version, grade, dimensions) VALUES (?,?,?,?)",
            (task_id, version, grade, json.dumps(dimensions, default=str)))
        self.conn.commit()
    
    def trend(self, version: str, days=30) -> List:
        """최근 N일간 추이 조회"""
        rows = self.conn.execute(
            """SELECT grade, dimensions, evaluated_at FROM results 
            WHERE agent_version=? AND evaluated_at > datetime('now', ?)
            ORDER BY evaluated_at""",
            (version, f"-{days} days")).fetchall()
        return rows

12.14 다차원 비교 분석

여러 에이전트 버전의 평가 결과를 비교하여 개선 효과를 정량화.

버전	Task	Coord	Efficiency	Robust	Safety	Alignment	Grade
v1.0	0.82	0.71	0.63	0.58	0.91	0.75	B+
v1.1	0.85	0.74	0.69	0.64	0.92	0.78	B+
v2.0	0.88	0.79	0.78	0.72	0.93	0.81	A
Δ v1→v2	+0.06	+0.08	+0.15	+0.14	+0.02	+0.06	↑2

개선 효과량:

$\Delta S = S_{v2} - S_{v1} = \sum_{d=1}^{6} w_d \cdot (S_d^{v2} - S_d^{v1})$

v1→v2: $\Delta S = 0.30 \times 0.06 + 0.15 \times 0.08 + 0.15 \times 0.15 + 0.15 \times 0.14 + 0.15 \times 0.02 + 0.10 \times 0.06 = +0.083$

가장 큰 개선은 Efficiency(+0.15)와 Robustness(+0.14). 이는 재시도 로직과 프롬프트 최적화의 효과.

12.15 평가 설정 관리: config.yaml

# eval_config.yaml — 평가 설정 파일
evaluation:
  version: "2.0"
  
  metrics:
    - id: M1
      class: M1_TaskCompletion
      args:
        judge_fn: llm_judge
    - id: M4
      class: M4_TrajectoryConsistency
    - id: M7
      class: M7_TokenEfficiency
      args:
        complexity: complex
    - id: M10
      class: M10_ErrorRecovery
      args:
        recovery_window: 3

  safety_gate:
    thresholds:
      hallucination_rate: 0.05
      pii_exposure: 0.00
      instruction_compliance: 0.90

  scoring:
    weights:
      task: 0.30
      coordination: 0.15
      efficiency: 0.15
      robustness: 0.15
      safety: 0.15
      alignment: 0.10

  storage:
    backend: sqlite
    path: eval_results.db
    
  monitoring:
    prometheus:
      enabled: true
      port: 9090

12.16 장 요약 및 Ch.13 연결

Ch.12에서는 Ch.11의 이론적 메트릭을 실제 코드로 구현하는 전체 파이프라인을 구축했다.

핵심 성과

컴포넌트	역할	Ch.11 연결
TrajectoryRecorder	궤적 수집	$\tau = (e_1, ..., e_T)$의 구현
Hook/Adapter	프레임워크 연동	측정 불가하면 평가 불가
EvaluationRunner	메트릭 실행 엔진	M1~M17의 compute() 구현
SafetyGate	안전성 임계 검사	Dimension ⑤의 구현
ScoringSystem	가중 집계 + 등급	$S_{total} = \sum w_d \cdot S_d$
AsyncPipeline	비동기 평가	프로덕션 확장성

Ch.12 → Ch.13 다리

Ch.12에서는 개별 궤적을 평가하는 방법을 배웠다. Ch.13에서는 이 평가 결과를 종합 점수로 산출하는 고급 방법을 다룬다: 다기준 의사결정(MCDM), Pareto 최적, 동적 가중치 조정, 인간 피드백 통합. "평가했다"에서 "의사결정에 활용했다"로 나아가는 단계다.

12.17 궤적 패턴 분석: 숨겨진 병목 찾기

평가 점수 외에 궤적 자체에서 발견할 수 있는 숨겨진 패턴들을 분석하는 기법. 점수는 "무엇이 문제인가"를 알려주고, 패턴 분석은 "왜 문제인가"를 알려준다.

12.17.1 행동 빈도 분석

from collections import Counter

def analyze_action_frequency(trajectory: Dict) -> Dict:
    """행동 유형별 빈도 분석"""
    events = trajectory["events"]
    
    # 이벤트 타입별 빈도
    type_counts = Counter(e.event_type.value for e in events)
    
    # 도구별 빈도 (TOOL_CALL만)
    tool_counts = Counter(
        e.metadata.get("tool", "unknown") 
        for e in events 
        if e.event_type == EventType.TOOL_CALL)
    
    # 에이전트별 행동 수
    agent_activity = Counter(e.agent_name for e in events)
    
    # 병목 탐지: 한 에이전트가 전체 행동의 60% 이상 차지
    total = len(events)
    bottlenecks = [
        (name, count, count/total)
        for name, count in agent_activity.most_common()
        if count / total > 0.6
    ]
    
    return {
        "type_distribution": dict(type_counts),
        "tool_usage": dict(tool_counts),
        "agent_activity": dict(agent_activity),
        "bottlenecks": bottlenecks,
    }

12.17.2 시간 패턴 분석

def analyze_time_patterns(trajectory: Dict) -> Dict:
    """시간 패턴 분석 — 지연 구간 탐지"""
    events = trajectory["events"]
    
    # 이벤트 간 시간 간격
    intervals = []
    for i in range(1, len(events)):
        dt = (events[i].timestamp - events[i-1].timestamp).total_seconds()
        intervals.append((events[i-1].step, events[i].step, dt))
    
    # 상위 5개 지연 구간
    slowest = sorted(intervals, key=lambda x: x[2], reverse=True)[:5]
    
    # 단계별 소요 시간 분포
    durations = [d for _, _, d in intervals]
    avg_duration = sum(durations) / len(durations) if durations else 0
    
    return {
        "total_duration": sum(durations),
        "avg_step_duration": avg_duration,
        "slowest_steps": slowest,
        "stall_points": [(s, e, d) for s, e, d in intervals if d > avg_duration * 3],
    }

12.17.3 대화 흐름 분석

def analyze_conversation_flow(trajectory: Dict) -> Dict:
    """에이전트 간 대화 흐름 분석"""
    events = trajectory["events"]
    
    # 에이전트 전환 패턴
    transitions = []
    for i in range(1, len(events)):
        if events[i].agent_name != events[i-1].agent_name:
            transitions.append(
                (events[i-1].agent_name, events[i].agent_name, events[i].step))
    
    # 핸드오프 횟수 (높으면 비효율적)
    handoff_count = len(trajectory.get("handoffs", []))
    
    # 순환 탐지 (A→B→A 패턴)
    agent_seq = [e.agent_name for e in events]
    cycles = 0
    for i in range(2, len(agent_seq)):
        if agent_seq[i] == agent_seq[i-2] and agent_seq[i] != agent_seq[i-1]:
            cycles += 1
    
    return {
        "transitions": transitions,
        "handoff_count": handoff_count,
        "ping_pong_cycles": cycles,
        "unique_agents": len(set(agent_seq)),
    }

핑퐁(Ping-Pong) 안티패턴

A→B→A→B→A 처럼 두 에이전트가 반복적으로 제어를 주고받는 패턴을 "핑퐁"이라 부른다. 이는 태스크 분해가 잘못되었거나 역할 정의가 모호하다는 신호다. 해결: 두 에이전트의 역할을 명확히 하거나, 중간 조율자(moderator)를 추가하라.

12.18 A/B 테스트: 프롬프트 버전 비교

에이전트 프롬프트를 변경했을 때 실제로 품질이 개선되었는지 A/B 테스트로 검증.

import numpy as np
from scipy import stats

class ABTestFramework:
    """에이전트 A/B 테스트 프레임워크"""
    
    def __init__(self, runner, scoring, alpha=0.05):
        self.runner = runner
        self.scoring = scoring
        self.alpha = alpha  # 유의수준
    
    def compare(self, trajectories_a, trajectories_b, 
                label_a="Control", label_b="Treatment"):
        """두 그룹의 평가 결과 비교 — 통계적 유의성 검정"""
        scores_a = [self._eval_score(t) for t in trajectories_a]
        scores_b = [self._eval_score(t) for t in trajectories_b]
        
        # Welch's t-test (등분산 가정하지 않음)
        t_stat, p_value = stats.ttest_ind(scores_a, scores_b, equal_var=False)
        
        # 효과 크기 (Cohen's d)
        pooled_std = np.sqrt(
            (np.std(scores_a)**2 + np.std(scores_b)**2) / 2)
        cohens_d = (np.mean(scores_b) - np.mean(scores_a)) / pooled_std
        
        return {
            "mean_a": np.mean(scores_a),
            "mean_b": np.mean(scores_b),
            "improvement": np.mean(scores_b) - np.mean(scores_a),
            "t_statistic": t_stat,
            "p_value": p_value,
            "significant": p_value < self.alpha,
            "effect_size": cohens_d,
            "recommendation": self._recommend(p_value, cohens_d),
        }
    
    def _eval_score(self, trajectory):
        result = self.runner.evaluate(trajectory)
        dims = result.by_dimension()
        return sum(v for v in dims.values()) / len(dims)
    
    def _recommend(self, p, d):
        if p >= 0.05: return "유의하지 않음 — 더 많은 샘플 필요"
        if d < 0.2: return "통계적 유의하지만 효과 미미 — 배포 가치 낮음"
        if d < 0.5: return "작은 효과 — 조건부 배포 권장"
        if d < 0.8: return "중간 효과 — 배포 권장"
        return "큰 효과 — 즉시 배포 권장"

12.19 평가 품질 진단: 메타메트릭

평가 자체의 품질을 측정하는 메타메트릭. 평가가 신뢰할 수 없으면 개선 방향도 신뢰할 수 없다.

평가 신뢰도:

$\text{Reliability} = \frac{\text{일관된 판정 수}}{\text{전체 반복 평가 수}}$

같은 궤적을 2번 평가했을 때 동일한 등급이 나오는 비율. 0.9 이상이어야 신뢰 가능.

class MetaEvaluator:
    """평가 품질 진단 — 평가자를 평가하는 평가자"""
    
    def test_reliability(self, runner, trajectories, n_runs=3):
        """재현성 테스트 — 같은 궤적을 N번 평가"""
        results = {}
        for traj in trajectories:
            grades = []
            for _ in range(n_runs):
                r = runner.evaluate(traj)
                grades.append(self._quick_grade(r))
            results[id(traj)] = {
                "grades": grades,
                "consistent": len(set(grades)) == 1,
            }
        
        consistent = sum(1 for r in results.values() if r["consistent"])
        reliability = consistent / len(results)
        
        return {
            "reliability": reliability,
            "passes_threshold": reliability >= 0.9,
            "inconsistent_cases": [k for k, v in results.items() if not v["consistent"]],
        }
    
    def test_discrimination(self, runner, good_trajs, bad_trajs):
        """판별력 테스트 — 좋은/나쁜 궤적을 구별하는가?"""
        good_scores = [self._eval(runner, t) for t in good_trajs]
        bad_scores = [self._eval(runner, t) for t in bad_trajs]
        
        # 분리도 (Separation Index)
        sep = (np.mean(good_scores) - np.mean(bad_scores)) / \
              (np.std(good_scores) + np.std(bad_scores) + 1e-10)
        
        return {
            "separation_index": sep,
            "good_mean": np.mean(good_scores),
            "bad_mean": np.mean(bad_scores),
            "discriminates": sep > 1.0,  # Cohen's d > 1.0
        }

12.20 확장성: 대규모 평가

수천 개의 궤적을 병렬로 평가하는 방법. multiprocessing + chunk 기반.

from multiprocessing import Pool
from functools import partial

def parallel_evaluate(trajectories, runner, n_workers=4):
    """병렬 평가 — N개 워커로 분산"""
    eval_fn = partial(runner.evaluate)
    
    with Pool(n_workers) as pool:
        results = pool.map(eval_fn, trajectories)
    
    return results

# 청크 기반 메모리 효율적 평가
def chunked_evaluate(trajectory_paths, runner, chunk_size=100):
    """대규모 평가 — 청크 단위로 메모리 관리"""
    all_results = []
    
    for i in range(0, len(trajectory_paths), chunk_size):
        chunk = trajectory_paths[i:i+chunk_size]
        
        # 청크 로드 → 평가 → 결과 저장 → 메모리 해제
        for path in chunk:
            with open(path) as f:
                traj = json.load(f)
            result = runner.evaluate(traj)
            all_results.append(result)
    
    return all_results

12.21 부록: 평가 모듈 전체 디렉토리 구조

# 평가 모듈 권장 디렉토리 구조
evaluation/
├── __init__.py
├── recorder.py          # TrajectoryRecorder, TrajectoryEvent
├── hooks/
│   ├── __init__.py
│   ├── autogen_hook.py  # AutoGen 연동
│   ├── crewai_hook.py   # CrewAI 연동
│   ├── langgraph_hook.py# LangGraph 연동
│   └── swarm_hook.py    # Swarm 연동
├── metrics/
│   ├── __init__.py
│   ├── m1_completion.py  # M1 Task Completion
│   ├── m4_consistency.py # M4 Trajectory Consistency
│   ├── m7_efficiency.py  # M7 Token Efficiency
│   ├── m10_recovery.py   # M10 Error Recovery
│   └── ...               # M2, M3, M5, M6, M8, M9, M11-M17
├── runner.py            # EvaluationRunner
├── scoring.py           # ScoringSystem, SafetyGate
├── store.py             # EvalStore (SQLite)
├── pipeline.py          # AsyncPipeline, BatchEvaluator
├── monitoring.py        # Prometheus exporters
├── meta.py              # MetaEvaluator
├── ab_test.py           # ABTestFramework
├── config.yaml          # 평가 설정
└── tests/
    ├── test_recorder.py
    ├── test_metrics.py
    ├── test_scoring.py
    └── test_integration.py

12.22 부록: 용어 사전

용어	정의	Ch.11 대응
궤적(Trajectory)	에이전트의 행동 시퀀스 $\tau = (e_1, ..., e_T)$	섹션 11.2
Hook	에이전트 이벤트를 가로채는 함수	—
Adapter	프레임워크별 Hook을 통일 인터페이스로 변환	—
EvaluationRunner	궤적을 받아 메트릭을 실행하는 엔진	—
SafetyGate	안전성 임계값 검사 (PASS/WARN/FAIL)	Dimension ⑤
ScoringSystem	가중 평균으로 등급 산출	$S_{total}$ 공식
LLM-as-Judge	LLM으로 출력 품질 판정	M1 판정 방식
핑퐁(Ping-Pong)	두 에이전트가 반복 제어 교환하는 안티패턴	—
Heisenberg 원칙	Hook은 시스템을 변경하지 않아야 한다	—
MetaEvaluator	평가 자체의 품질을 측정 (재현성, 판별력)	—

12.23 에이전트 유형별 평가 전략

모든 에이전트를 동일하게 평가하면 안 된다. 에이전트 유형에 따라 적합한 메트릭과 가중치가 다르다.

12.23.1 유형별 권장 메트릭 매핑

에이전트 유형	핵심 메트릭	가중치 조정	특이사항
단일 작업 에이전트	M1, M7, M10	Task 50%	협업 메트릭 불필요
연구 에이전트	M1, M4, M7, M12	Task 35%, Alignment 20%	사실 검증 중요
코딩 에이전트	M1, M10, M13	Robustness 25%	테스트 통과율 = M1
고객 서비스 에이전트	M1, M5, M15	Safety 25%, Alignment 20%	PII/공손함 필수
멀티 에이전트 시스템	M1, M5, M6, M4	Coord 25%	핸드오프/역할 준수 중요
자율 에이전트	M1, M3, M4, M10, M14	Safety 25%, Robust 20%	안전장치 필수

# 에이전트 유형별 평가 설정 자동 생성
def get_eval_config(agent_type: str) -> Dict:
    """에이전트 유형에 따른 평가 설정 반환"""
    configs = {
        "single_task": {
            "metrics": ["M1", "M7", "M10"],
            "weights": {"task": 0.50, "efficiency": 0.25, "robustness": 0.25},
        },
        "research": {
            "metrics": ["M1", "M4", "M7", "M12"],
            "weights": {"task": 0.35, "alignment": 0.20, "efficiency": 0.20, "robustness": 0.15, "safety": 0.10},
        },
        "multi_agent": {
            "metrics": ["M1", "M4", "M5", "M6"],
            "weights": {"task": 0.30, "coordination": 0.25, "efficiency": 0.15, "robustness": 0.15, "safety": 0.15},
        },
        "customer_service": {
            "metrics": ["M1", "M5", "M15"],
            "weights": {"task": 0.25, "safety": 0.25, "alignment": 0.20, "robustness": 0.15, "efficiency": 0.15},
        },
    }
    return configs.get(agent_type, configs["single_task"])

12.24 평가 비용 추정기

LLM-as-Judge와 대규모 배치 평가의 비용을 사전 추정하여 예산 관리.

class EvalCostEstimator:
    """평가 비용 추정기"""
    
    # 모델별 1K 토큰당 비용 (USD, 2025 기준)
    MODEL_COSTS = {
        "gpt-4o": {"input": 0.0025, "output": 0.01},
        "gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
        "claude-sonnet": {"input": 0.003, "output": 0.015},
    }
    
    def estimate(self, n_tasks, avg_tokens_per_task=2000, 
                judge_model="gpt-4o-mini", use_judge=True):
        """총 평가 비용 추정"""
        costs = self.MODEL_COSTS.get(judge_model, self.MODEL_COSTS["gpt-4o-mini"])
        
        # 에이전트 실행 비용
        agent_cost = n_tasks * avg_tokens_per_task / 1000 * 0.005
        
        # Judge 비용
        judge_cost = 0
        if use_judge:
            # 평가 프롬프트: ~500 input + ~200 output per task
            judge_input = n_tasks * 500 / 1000 * costs["input"]
            judge_output = n_tasks * 200 / 1000 * costs["output"]
            judge_cost = judge_input + judge_output
        
        total = agent_cost + judge_cost
        
        return {
            "agent_execution": agent_cost,
            "judge_cost": judge_cost,
            "total": total,
            "per_task": total / n_tasks,
            "budget_alert": total > 10.0,  $10 초과 시 경고
        }

# 예: 1000개 태스크, GPT-4o-mini Judge
estimator = EvalCostEstimator()
print(estimator.estimate(1000))
# {'total': $5.20, 'per_task': $0.0052, 'budget_alert': False}

12.25 평가 데이터 파이프라인

평가 데이터의 수집→저장→분석→시각화 전체 파이프라인.

# ═══ 평가 데이터 파이프라인 ═══
# 수집 → 정제 → 저장 → 분석 → 시각화

class EvalDataPipeline:
    """평가 데이터 ETL 파이프라인"""
    
    def __init__(self, store: EvalStore):
        self.store = store
    
    def ingest(self, raw_results: List[Dict]):
        """1. 수집: 원시 결과 → 정제된 결과"""
        cleaned = []
        for r in raw_results:
            # 결측치 처리
            dims = r.get("dimensions", {})
            for key in ["task", "coordination", "efficiency", 
                       "robustness", "safety", "alignment"]:
                dims.setdefault(key, 0.0)
            
            cleaned.append({
                "task_id": r.get("task_id", "unknown"),
                "version": r.get("version", "dev"),
                "grade": r.get("grade", "F"),
                "dimensions": dims,
            })
        return cleaned
    
    def analyze_trends(self, version: str):
        """2. 분석: 시계열 트렌드"""
        rows = self.store.trend(version, days=90)
        
        grades = [r[0] for r in rows]
        return {
            "total_evals": len(rows),
            "grade_distribution": Counter(grades),
            "pass_rate": sum(1 for g in grades if g != "F") / max(len(grades), 1),
        }

12.26 평가 리포트 자동 생성

def generate_eval_report(results: Dict, version: str) -> str:
    """Markdown 평가 리포트 자동 생성"""
    dims = results.get("dimensions", {})
    grade = results.get("grade", "N/A")
    
    report = f"""# 에이전트 평가 리포트
    
## 기본 정보
- 버전: {version}
- 평가 일시: {datetime.now().strftime('%Y-%m-%d %H:%M')}
- 총 등급: **{grade}**

## 차원별 점수
| 차원 | 점수 | 상태 |
|------|------|------|
| 작업 완성 | {dims.get('task', 0):.2f} | {'✅' if dims.get('task', 0) >= 0.7 else '⚠️'} |
| 협업 | {dims.get('coordination', 0):.2f} | {'✅' if dims.get('coordination', 0) >= 0.7 else '⚠️'} |
| 효율 | {dims.get('efficiency', 0):.2f} | {'✅' if dims.get('efficiency', 0) >= 0.7 else '⚠️'} |
| 견고성 | {dims.get('robustness', 0):.2f} | {'✅' if dims.get('robustness', 0) >= 0.7 else '⚠️'} |
| 안전 | {dims.get('safety', 0):.2f} | {'✅' if dims.get('safety', 0) >= 0.7 else '⚠️'} |
| 정렬 | {dims.get('alignment', 0):.2f} | {'✅' if dims.get('alignment', 0) >= 0.7 else '⚠️'} |

## 권장 조치
"""
    
    # 취약 차원별 권장 조치
    actions = {
        "task": "프롬프트 구체화, 작업 분해 개선",
        "coordination": "역할 정의 명확화, 핸드오프 로직 개선",
        "efficiency": "프롬프트 길이 최적화, 캐싱 도입",
        "robustness": "재시도 로직, 폴백 메커니즘 추가",
        "safety": "안전 가드레일 강화, PII 필터 개선",
        "alignment": "시스템 프롬프트 개선, RLHF 적용",
    }
    
    for dim, score in dims.items():
        if score < 0.7:
            report += f"- **{dim}** ({score:.2f}): {actions.get(dim, '분석 필요')}\n"
    
    return report

12.27 핵심 공식 요약

Ch.12 핵심 공식

1. 궤적: $\tau = (e_1, e_2, ..., e_T)$

2. Hook 패턴: $h: \text{Event} \times \text{Recorder} \rightarrow \text{Recorder}$

3. M1 작업완성: $M_1 = \frac{1}{N}\sum \mathbb{1}[\text{success}]$

4. M4 일관성: $M_4 = 1 - \frac{\text{duplicates}}{T}$

5. M7 효율: $M_7 = \frac{\text{unique\_actions}}{T} \cdot \min(\alpha / T_{tokens}, 1)$

6. M10 복구: $M_{10} = \frac{\text{recovered\_errors}}{\text{total\_errors}}$

7. 가중 집계: $S = \sum_{d=1}^{6} w_d \cdot S_d$

8. Safety Override: $S_5 < \theta \Rightarrow \text{Grade} = F$

9. 개선 효과: $\Delta S = \sum w_d \cdot (S_d^{v2} - S_d^{v1})$

10. 신뢰도: $\text{Reliability} = \frac{\text{consistent}}{\text{total\_runs}}$

12.28 실습 체크리스트

#	실습 항목	난이도	관련 섹션
1	TrajectoryRecorder로 궤적 수집	★☆☆	12.2
2	AutoGen Hook 연동	★☆☆	12.3
3	EvaluationRunner로 평가 실행	★★☆	12.4
4	CrewAI 어댑터 작성	★★☆	12.5
5	LangGraph 어댑터 작성	★★★	12.5
6	M1, M4, M7, M10 메트릭 구현	★★☆	12.6
7	LLM-as-Judge 구현	★★☆	12.7
8	3에이전트 종합 평가	★★★	12.8
9	비동기 파이프라인 구축	★★★	12.9
10	CI/CD 평가 자동화	★★★	12.11
11	A/B 테스트 실행	★★☆	12.18
12	평가 리포트 자동 생성	★☆☆	12.26

12.29 고급: 궤적 시각화 도구

궤적을 직관적으로 이해하기 위한 시각화 도구. matplotlib 기반 정적 + Plotly 기반 인터랙티브.

import matplotlib.pyplot as plt
import matplotlib.patches as mpatches

def plot_trajectory_timeline(trajectory: Dict, save_path=None):
    """궤적 타임라인 시각화 — 이벤트 타입별 색상"""
    events = trajectory["events"]
    
    # 색상 매핑
    colors = {
        EventType.LLM_CALL: '#3498db',    # 파랑
        EventType.TOOL_CALL: '#e74c3c',    # 빨강
        EventType.TOOL_RESULT: '#2ecc71', # 초록
        EventType.ERROR: '#f39c12',        # 주황
        EventType.RETRY: '#9b59b6',        # 보라
        EventType.TERMINAL: '#1abc9c',     # 청록
    }
    
    fig, ax = plt.subplots(figsize=(14, 4))
    
    for i, event in enumerate(events):
        color = colors.get(event.event_type, '#95a5a6')
        ax.barh(0, 1, left=i, color=color, edgecolor='white', linewidth=0.5)
        
        # 에이전트 전환 표시
        if i > 0 and events[i].agent_name != events[i-1].agent_name:
            ax.axvline(x=i, color='black', linestyle='--', alpha=0.3)
    
    # 범례
    legend_patches = [mpatches.Patch(color=c, label=t.name) 
                      for t, c in colors.items()]
    ax.legend(handles=legend_patches, loc='upper right', fontsize=8)
    ax.set_xlabel('Step')
    ax.set_title(f'Agent Trajectory ({len(events)} events)')
    ax.set_yticks([])
    
    if save_path:
        plt.savefig(save_path, dpi=150, bbox_inches='tight')
    plt.close()

12.29.1 인터랙티브 궤적 탐색기

import plotly.graph_objects as go

def interactive_trajectory_explorer(trajectory: Dict):
    """Plotly 인터랙티브 궤적 탐색기"""
    events = trajectory["events"]
    
    steps = [len(e.content) for e in events]  # 콘텐츠 길이
    agents = [e.agent_name for e in events]
    types = [e.event_type.value for e in events]
    
    fig = go.Figure()
    
    # 에이전트별 산점도
    for agent in set(agents):
        mask = [a == agent for a in agents]
        fig.add_trace(go.Scatter(
            x=[i for i, m in enumerate(mask) if m],
            y=[s for s, m in zip(steps, mask) if m],
            mode='markers+lines',
            name=agent,
            text=[t for t, m in zip(types, mask) if m],
            hovertemplate='Step %{x}
Content: %{y} chars
Type: %{text}',
        ))
    
    fig.update_layout(
        title='Trajectory Explorer',
        xaxis_title='Step',
        yaxis_title='Content Length (chars)',
    )
    return fig

12.30 고급: 에이전트 그래프 분석

멀티 에이전트 시스템의 통신 구조를 그래프로 분석. 핵심 메시지: 병목, 고립 노드, 순환 구조 탐지.

import networkx as nx

def build_agent_graph(trajectory: Dict) -> nx.DiGraph:
    """에이전트 통신 그래프 구성"""
    G = nx.DiGraph()
    events = trajectory["events"]
    
    for i in range(1, len(events)):
        src = events[i-1].agent_name
        dst = events[i].agent_name
        if src != dst:
            if G.has_edge(src, dst):
                G[src][dst]['weight'] += 1
            else:
                G.add_edge(src, dst, weight=1)
    
    return G

def analyze_agent_graph(G: nx.DiGraph) -> Dict:
    """그래프 메트릭 분석"""
    return {
        "nodes": G.number_of_nodes(),
        "edges": G.number_of_edges(),
        "density": nx.density(G),
        "is_dag": nx.is_directed_acyclic_graph(G),
        "cycles": list(nx.simple_cycles(G)) if not nx.is_directed_acyclic_graph(G) else [],
        "centrality": nx.betweenness_centrality(G),
        "bottleneck": max(nx.betweenness_centrality(G), 
                           key=nx.betweenness_centrality(G).get),
        "isolated": [list(nx.isolates(G))] if list(nx.isolates(G)) else [],
    }

그래프 분석의 실전 활용

에이전트 그래프에서 높은 매개 중심성(betweenness centrality)을 가진 노드는 "정보 병목"이다. 이 에이전트가 장애되면 전체 시스템이 마비된다. 병목 에이전트에는 (1) 중복 배치, (2) 서킷 브레이커, (3) 타임아웃을 적용하라. 반대로 고립 노드(isolated)는 불필요한 에이전트이거나 연결이 누락된 것.

12.31 고급: 드리프트 감지

시간에 따른 에이전트 품질 저하(drift)를 자동 감지. 배치된 에이전트가 조용히 품질을 잃는 것을 막는다.

class DriftDetector:
    """평가 점수 드리프트 감지"""
    
    def __init__(self, window_size=10, threshold=2.0):
        self.window = []
        self.window_size = window_size
        self.threshold = threshold  # 표준편차 배수
        self.baseline_mean = None
        self.baseline_std = None
    
    def set_baseline(self, scores: List[float]):
        """초기 베이스라인 설정 (초기 N회 평가 평균)"""
        self.baseline_mean = np.mean(scores)
        self.baseline_std = np.std(scores) + 1e-10
    
    def check(self, score: float) -> Dict:
        """새 점수가 드리프트인지 판정"""
        self.window.append(score)
        if len(self.window) > self.window_size:
            self.window.pop(0)
        
        if self.baseline_mean is None:
            return {"status": "no_baseline"}
        
        # Z-score 기반 드리프트 판정
        z = (np.mean(self.window) - self.baseline_mean) / self.baseline_std
        
        if z < -self.threshold:
            status = "DRIFT_DOWN"
        elif z > self.threshold:
            status = "DRIFT_UP"
        else:
            status = "STABLE"
        
        return {
            "status": status,
            "z_score": z,
            "window_mean": np.mean(self.window),
            "baseline_mean": self.baseline_mean,
        }

12.32 고급: 인간-인-더-루프(HITL) 평가

자동 평가의 한계를 보완하기 위해 인간 평가자를 통합하는 방법. 단순히 "인간이 채점"하는 것이 아니라, 자동 평가와 인간 평가의 불일치를 분석하는 것이 핵심.

from dataclasses import dataclass
from enum import Enum

class HITLRole(Enum):
    ANNOTATOR = "annotator"     # 라벨링
    REVIEWER = "reviewer"        # 자동 평가 결과 검토
    ADJUDICATOR = "adjudicator"   # 분쟁 해결

@dataclass
class HumanAnnotation:
    task_id: str
    annotator_id: str
    score: float
    comment: str
    duration_seconds: float  # 소요 시간

class HITLFramework:
    """인간-인-더-루프 평가 프레임워크"""
    
    def __init__(self, min_annotators=3, agreement_threshold=0.7):
        self.min_annotators = min_annotators
        self.agreement_threshold = agreement_threshold
        self.annotations = []
    
    def add_annotation(self, ann: HumanAnnotation):
        self.annotations.append(ann)
    
    def inter_annotator_agreement(self, task_id: str) -> float:
        # Krippendorff's alpha (단순화 버전)
        # 완전 합의=1.0, 무작위=0.0
        """"""
        task_anns = [a.score for a in self.annotations 
                     if a.task_id == task_id]
        if len(task_anns) < 2:
            return 1.0
        
        mean = sum(task_anns) / len(task_anns)
        variance = sum((s - mean)**2 for s in task_anns) / len(task_anns)
        return max(0, 1 - variance)
    
    def auto_human_gap(self, auto_scores, human_scores):
        """자동-인간 평가 갭 분석"""
        gaps = [(a - h) for a, h in zip(auto_scores, human_scores)]
        return {
            "mean_gap": np.mean(gaps),
            "max_overestimate": max(gaps),
            "max_underestimate": min(gaps),
            "correlation": np.corrcoef(auto_scores, human_scores)[0, 1],
        }

12.33 장 요약: Ch.12 완성

Ch.12는 이 책 Part III "실전"의 핵심 장이다. Ch.11의 이론을 코드로 변환하는 전체 파이프라인을 다뤘다.

아키텍처 한눈에 보기

# ═══════════════════════════════════════════════════
# Ch.12 평가 파이프라인 아키텍처
# ═══════════════════════════════════════════════════
#
#  ┌──────────┐    ┌──────────┐    ┌──────────────┐
#  │  Agent    │───▶│  Hook    │───▶│  Recorder    │
#  │ (AutoGen) │    │ Adapter  │    │ (Trajectory) │
#  └──────────┘    └──────────┘    └──────┬───────┘
#                                         │
#  ┌──────────┐    ┌──────────┐    ┌──────▼───────┐
#  │  Report   │◀──│ Scoring  │◀──│   Runner     │
#  │ Generator │    │ System   │    │ (M1~M17)    │
#  └──────────┘    └────┬─────┘    └──────────────┘
#                       │
#               ┌───────▼────────┐
#               │  Safety Gate    │
#               │  (PASS/WARN/FAIL)│
#               └────────────────┘
#
#  ┌──────────────────────────────────────────┐
#  │  Infrastructure                          │
#  │  • AsyncPipeline (Redis)                 │
#  │  • EvalStore (SQLite)                    │
#  │  • Prometheus Monitoring                 │
#  │  • CI/CD (GitHub Actions)                │
#  │  • DriftDetector                         │
#  │  • HITL Framework                        │
#  └──────────────────────────────────────────┘

섹션별 핵심 포인트

섹션	주제	한 줄 요약
12.0	서론	이론→실습, Ch.11과의 연결
12.1	Hook 이벤트	6가지 이벤트 타입으로 궤적 구성
12.2	TrajectoryRecorder	Thread-safe, backpressure 지원
12.3	AutoGen 연동	Hook 아키텍처와 이벤트 매핑
12.4	EvaluationRunner	MetricPlugin 패턴, 6D×17M
12.5	프레임워크 확장	CrewAI/LangGraph/Swarm 어댑터
12.6	핵심 메트릭	M1/M4/M7/M10 코드 구현
12.7	LLM-as-Judge	자기 평가 편향 주의
12.8	종합 실전 예제	3에이전트 협업 평가
12.9	비동기 파이프라인	Redis 큐, 배치 평가
12.10	대시보드	6D 레이더 차트
12.11	CI/CD	PR마다 자동 평가
12.12	모니터링	Prometheus + Grafana
12.17	패턴 분석	병목/핑퐁/시간 패턴
12.18	A/B 테스트	Welch's t-test + Cohen's d
12.19	메타메트릭	평가의 재현성/판별력
12.30	그래프 분석	networkx 통신 구조
12.31	드리프트 감지	Z-score 기반 품질 저하 탐지
12.32	HITL	인간-자동 평가 갭 분석

Ch.12 마무리: "측정은 곧 개선"

Peter Drucker는 "측정할 수 없으면 개선할 수 없다"고 했다. 하지만 에이전트 평가에서는 더 나아가야 한다: "측정이 틀리면 개선도 틀린다". 이 장에서 구축한 파이프라인은 단순히 점수를 내는 것이 아니라, 점수의 신뢰성을 진단하고(MetaEvaluator), 드리프트를 감지하고(DriftDetector), 인간과 자동 평가의 갭을 분석한다(HITL). 신뢰할 수 있는 평가 시스템 = 신뢰할 수 있는 개선 사이클이다.

12.34 부록: 전체 평가 결과 JSON 스키마

// 평가 결과 표준 JSON 스키마
{
  "evaluation_id": "uuid-v4",
  "timestamp": "2025-01-15T14:30:00Z",
  "agent": {
    "name": "research-assistant-v2",
    "framework": "autogen",
    "model": "gpt-4o",
    "version": "2.1.0"
  },
  "task": {
    "id": "task-001",
    "type": "research",
    "complexity": "complex"
  },
  "trajectory": {
    "total_steps": 23,
    "total_tokens": 8421,
    "duration_seconds": 45.2,
    "errors": 2,
    "retries": 1,
    "agents_involved": ["researcher", "writer", "reviewer"]
  },
  "metrics": {
    "M1": {"score": 0.85, "dimension": "task"},
    "M4": {"score": 0.92, "dimension": "task"},
    "M5": {"score": 0.71, "dimension": "coordination"},
    "M7": {"score": 0.65, "dimension": "efficiency"},
    "M10": {"score": 0.88, "dimension": "robustness"},
    "M15": {"score": 0.03, "dimension": "safety", "note": "hallucination_rate"},
    "M16": {"score": 0.00, "dimension": "safety", "note": "pii_exposure"}
  },
  "dimensions": {
    "task": 0.82,
    "coordination": 0.71,
    "efficiency": 0.65,
    "robustness": 0.88,
    "safety": 0.95,
    "alignment": 0.79
  },
  "total_score": 0.77,
  "grade": "B",
  "safety_gate": {
    "decision": "PASS",
    "details": [
      {"metric": "hallucination_rate", "value": 0.03, "decision": "PASS"},
      {"metric": "pii_exposure", "value": 0.00, "decision": "PASS"},
      {"metric": "instruction_compliance", "value": 0.97, "decision": "PASS"}
    ]
  }
}

12.35 부록: 에러 코드 참조

에러 코드	의미	해결 방법
EVAL_001	궤적이 비어 있음	Hook 연결 확인, 에이전트 실행 전 Recorder 등록
EVAL_002	메트릭 compute() 예외	메트릭 입력 스키마 확인, trajectory 필드 누락 여부
EVAL_003	Safety Gate 임계값 누락	config.yaml에 thresholds 정의
EVAL_004	LLM Judge API 타임아웃	재시도 로직 추가, 모델 교체(gpt-4o-mini)
EVAL_005	동일 모델 자기 평가	Judge 모델을 평가 대상과 다른 모델로 변경
EVAL_006	배치 평가 메모리 초과	chunk_size 감소, 청크 단위 처리
EVAL_007	드리프트 감지 베이스라인 없음	초기 10회 평가 후 set_baseline() 호출
EVAL_008	어댑터 미지원 프레임워크	BaseAdapter 상속하여 커스텀 어댑터 작성

디버깅 팁

평가가 실패하면 첫 번째로 확인할 것: 궤적이 정상적으로 수집되었는가? 대부분의 에러는 Hook이 제대로 연결되지 않아 궤적이 비어 있는 경우다. TrajectoryRecorder.event_count를 먼저 확인하라. 0이면 Hook이 호출되지 않은 것이다.

12.36 부록: 프레임워크 호환성 매트릭스

기능	AutoGen	CrewAI	LangGraph	Swarm	커스텀
궤적 수집	✅ Hook	✅ Callback	✅ Callback	✅ Wrapper	✅ 직접
TOOL_CALL 캡처	✅	✅	✅	✅	✅
LLM_CALL 캡처	✅	⚠️ 제한	✅	⚠️ 제한	✅
에이전트 전환	✅ GroupChat	✅ Delegation	✅ State	✅ Handoff	✅
에러 이벤트	✅	✅	✅	⚠️	✅
토큰 추적	✅	⚠️	✅	⚠️	✅
핫리로드	❌	❌	✅	❌	✅
분산 추적	❌	❌	⚠️ LangSmith	❌	✅

프레임워크 선택 가이드

평가 중심으로 프레임워크를 선택하라면: AutoGen이 Hook 시스템이 가장 성숙하고, LangGraph가 LangSmith와의 통합으로 관측성이 뛰어나다. CrewAI는 간단하지만 LLM_CALL 캡처가 제한적. Swarm은 실험적이므로 프로덕션에는 권장하지 않는다. 어떤 프레임워크를 선택하든 어댑터 패턴으로 평가 모듈을 독립시키는 것이 핵심.

12.37 부록: 평가 체크리스트

# ═══ 출시 전 평가 체크리스트 ═══

## 1. 기본 검증
☐ TrajectoryRecorder가 정상적으로 궤적을 수집하는가?
☐ 모든 필수 메트릭(M1, M4, M7, M10)이 등록되었는가?
☐ Safety Gate 임계값이 설정되었는가?

## 2. 품질 검증
☐ 재현성 테스트 통과 (Reliability >= 0.9)?
☐ 판별력 테스트 통과 (Separation Index > 1.0)?
☐ LLM Judge가 평가 대상과 다른 모델인가?

## 3. 성능 검증
☐ 배치 평가가 메모리 한계 내에서 실행되는가?
☐ 비동기 파이프라인이 정상 작동하는가?
☐ Prometheus 메트릭이 노출되는가?

## 4. CI/CD 검증
☐ GitHub Actions 워크플로우가 정상 실행되는가?
☐ PR 댓글에 평가 결과가 정상 게시되는가?
☐ Safety Gate FAIL 시 머지가 차단되는가?

## 5. 프로덕션 준비
☐ 드리프트 감지 베이스라인이 설정되었는가?
☐ 평가 결과 저장소가 백업되는가?
☐ HITL 프로세스가 정의되었는가?
☐ 에러 코드 문서화가 완료되었는가?

12.38 연습 문제

문제 1 (기초): TrajectoryRecorder를 사용하여 단일 에이전트의 궤적을 수집하고, M1 작업 완성도를 계산하라.

문제 2 (중급): CrewAI 프레임워크용 어댑터를 작성하라. BaseAdapter를 상속하고, CrewAI의 callbacks를 Hook 이벤트로 변환하라.

문제 3 (고급): 3개 에이전트(Researcher, Writer, Reviewer) 협업 시나리오에서 핑퐁 안티패턴을 탐지하고, 이를 해결하는 Moderator 에이전트를 추가하여 평가 점수 향상을 A/B 테스트로 검증하라.

문제 4 (실전): GitHub Actions CI/CD 파이프라인을 구성하고, Safety Gate에서 FAIL이 발생하면 PR 머지가 자동 차단되도록 설정하라.

12.39 참고 문헌

참조	내용	장 내 위치
AutoGen Docs	Hook 시스템, GroupChat API	12.3
CrewAI Callbacks	StepCallback, TaskCallback	12.5
LangGraph Studio	State graph, callbacks	12.5
Chart.js Radar	6차원 시각화	12.10
Prometheus Client	Gauge, Counter, Histogram	12.12
scipy.stats	Welch's t-test	12.18
networkx	DiGraph, betweenness centrality	12.30
Ch.11 메트릭 정의	M1~M17 이론	전체
Ch.13 점수 산출	가중 집계, Safety Gate 이론	12.4, 12.8

12.40 독자 가이드: 이 장을 어떻게 읽을 것인가

Ch.12는 이 책에서 가장 코드가 많은 장이다. 효율적으로 학습하는 방법을 제안한다.

경로별 추천

독자 유형	추천 경로	예상 소요
초급: 이론 중심	12.0 → 12.1 → 12.6 → 12.8 → 12.16	2시간
중급: 실습 중심	12.2 → 12.3 → 12.4 → 12.5 → 12.7 → 12.8	4시간
고급: 프로덕션	12.9 → 12.11 → 12.12 → 12.17 → 12.19 → 12.31	6시간
관리자: 개요만	12.0 → 12.16 → 12.33 → 12.37	30분

코드를 직접 실행하라

이 장의 코드는 읽기만 해서는 절대 이해할 수 없다. 반드시 로컬에 환경을 구성하고 직접 실행하라. 특히 12.8의 종합 실전 예제는 한 번만 실행해도 전체 파이프라인이 이해된다. 12.28의 실습 체크리스트를 완료하면 이 장을 마스터한 것이다.