에이전트 평가와 지표: Offline/Online Eval과 Trajectory 평가

본 포스트는 'AI 에이전트 엔지니어링(AI Agent Engineering)' 시리즈의 6편입니다. 단발성 프롬프트 평가를 넘어, 비확정적 멀티스텝 실행 경로(Trajectory)를 갖는 AI 에이전트를 위한 Offline/Online Evaluation 프레임워크와 Trajectory 평가 지표, 그리고 Python 평가 하네스 구축 실무를 다룹니다.

---

1. 에이전트 평가의 근본적 난제: 비확정적 궤적 (Non-deterministic Trajectory)

1.1 단발성 LLM 평가 vs 멀티스텝 에이전트 평가

단일 LLM 응답 평가(Single-turn LLM Evaluation)는 입력 프롬프트 X에 대한 출력 Y의 품질(정확성, 환각 여부, 문맥 관련성 등)을 측정합니다. 반면, AI 에이전트 평가는 사용자 목표 G가 주어졌을 때, 에이전트가 상태 S_0 → S_1 → \dots → S_T를 거치며 실행한 행동 궤적(Trajectory \tau)과 최종 상태(Final State S_T) 전체를 평가해야 합니다.

상태 공간(State-Space) 평가로의 전환: 에이전트 시스템에서는 "LLM이 어떤 답변을 텍스트로 말했는가"보다 "환경(DB, 파일 시스템, 외부 API 등)의 최종 상태가 목표 상태 S_target으로 올바르게 변환되었는가"가 평가의 핵심 기준이 됩니다.

1.2 비확정적 실행 궤적(Trajectory Variance) 문제

동일한 목표(Goal)를 부여하더라도 LLM의 확률적 특성과 외부 환경의 상태 변화에 따라 에이전트는 서로 다른 경로를 선택할 수 있습니다.

단순히 "어떤 도구를 어떤 순서로 호출했는가"라는 텍스트 기반 정밀 일치(Exact Match) 방식으로는 에이전트의 정당성을 평가할 수 없습니다. 서로 다른 경로라 하더라도 동일하게 성공적인 상태 변화(State Change)를 만들어냈다면 모두 정답으로 인정되어야 하며, 반대로 불필요한 루프나 높은 토큰 비용을 발생시킨 경로는 효율성 측면에서 감점이 부여되어야 합니다.

Goal G: DB Connection Pool Configuration → Agent Controller에서 세 궤적으로 갈라집니다.

Path C는 원본에서 행동 카드 3개에 최종 집계 4 Steps가 붙어 있습니다. 표시된 단계와 집계값을 그대로 두며 생략된 네 번째 행동을 추정해 만들지 않습니다.

Trajectory Variance 극복 원칙: 에이전트 평가 시 텍스트 기반 정밀 일치(Exact Match) 단정문 대신, 환경 상태 변화(State Verification) 단정과 스텝 효율성(E_{step}) 지표를 결합해야 비확정적 궤적 환경에서도 재현 가능하고 공정한 평가를 달성할 수 있습니다.

1.3 복합적 에러 전파 (Compounding Errors)와 중간 상태 검증

에이전트 제어 루프에서 Step t에서의 미세한 도구 호출 인자 오류(Argument Hallucination)나 관찰 실패(Observation Error)는 다음 스텝 Step t+1의 계획(Plan) 수립에 치명적인 왜곡을 불러일으킵니다.

오류가 누적(Compounding Errors)되면 에이전트는 무한 루프(Infinite Loop)에 빠지거나 전혀 무관한 행동을 수행하게 됩니다. 따라서 에이전트 평가는 최종 결과물만 보는 블랙박스(Black-box) 평가 외에도, 중간 궤적의 단계별 유효성(Step-level Validity)을 점검하는 화이트박스(Glass-box / Trajectory) 평가가 필수적입니다.

복합적 오류 전파 주의: 스텝 t에서 실패율 p=0.1인 도구 호출이 5단계 연쇄로 이어질 경우, 전체 궤적의 단순 성공 확률은 (1-0.1)^5 \approx 59\%로 급락합니다. 스텝 단위의 인자 검증(Schema Validation) 및 샌드박스 격리가 필요한 이유입니다.

---

2. Offline Evaluation vs Online Evaluation 전략

프로덕션 에이전트 시스템을 안정적으로 운영하기 위해서는 개발 및 배포 전 검증을 수행하는 Offline Evaluation과 배포 후 프로덕션 환경의 실질적 동작을 감시하는 Online Evaluation이 상호보완적으로 구축되어야 합니다.

2.1 Offline Evaluation: 벤치마크 및 LLM-as-a-Judge 하네스

• Golden Dataset 구축: 테스트 케이스는 단순히 (Input, Expected Output)이 아닌 (Goal, Initial Environment State, Ground Truth State / Accept Criteria, Expected Trajectory Guidelines) 형태로 정의됩니다.
• LLM-as-a-Judge (Rubric 기반 평가): 결정론적 단정문(Deterministic Assertions)으로 평가하기 어려운 최종 답안의 질이나 궤적의 논리적 타당성을 고성능 평가 모델(예: Claude 3.5 Sonnet, GPT-4o)을 활용해 측정합니다.
• 격리된 샌드박스 (Isolated Sandbox): 외부 API 호출, 파일 시스템 조작, DB 변경 등의 부작용(Side-effect)을 가진 도구를 실행할 때 테스트의 재현성(Reproducibility)과 안전성을 보장하기 위해 Docker 컨테이너 또는 E2B 샌드박스 환경에서 평가를 집행합니다.

2.2 Online Evaluation: 프로덕션 텔레메트리와 실시간 피드백

• OpenTelemetry & Action Log Tracing: 프로덕션 런타임에서 에이전트의 모든 관찰(Observe), 추론(Plan), 행동(Act)을 구조화된 스팬(Span) 단위로 로깅합니다.
• 유저 피드백 (Implicit / Explicit Feedback): Explicit(좋아요/나빠요), Implicit(코드 적용 여부, 재질문 비율, 작업 중단 속도).
• Shadow Deployment & Trajectory Drift 감지: 새로운 에이전트 프롬프트/모델 버전을 실유저 입력 스트림에 복제(Shadowing)하여 비동기로 실행하고, 기존 버전 대비 궤적 길이, 도구 오류율, 비용 변화를 감시합니다.

2.3 Offline vs Online Evaluation 비교 및 Data Flywheel

Offline 영역과 Online 영역 간의 도메인 데이터 순환 아키텍처는 다음과 같습니다.

운영 중 수집한 실패 경로와 사용자 피드백에서 개인정보를 비식별화한 뒤 오프라인 평가셋에 추가합니다. 같은 실패가 다시 발생하는지 회귀 검사에 활용할 수 있습니다.

---

3. 핵심 평가 지표 체계 (Key Metrics Framework)

AI 에이전트 평가 지표는 크게 성공률(Success), 도구 사용 정확도(Tool Accuracy), 궤적 효율성(Trajectory Efficiency), 비용 및 지연시간(Cost & Latency)의 4가지 축으로 나뉩니다.

종합 점수(Composite Score) 가중치 배분 가이드: Score_{composite} = 0.4 \times StateVerification + 0.2 \times ToolAccuracy + 0.2 \times StepEfficiency + 0.2 \times JudgeScore

---

4. 에이전트 평가 아키텍처 및 파이프라인

4.1 에이전트 지표 계층구조 (Metric Taxonomy Tree):

Agent Evaluation System은 다음 세 분류에 연결됩니다. 각 행의 지표는 해당 분류의 하위 항목입니다.

4.2 Trajectory 평가 시퀀스 흐름도:

4.3 하이브리드 평가 및 필터링 워크플로우:

Trajectory Log Received에서 1번 검사를 시작합니다. 각 Pass만 다음 단계로 이동합니다.

세 검사를 통과한 경로는 Trigger Async LLM-as-a-Judge → Evaluate Reasoning & Safety Rubrics → Generate Final Evaluation Report로 진행합니다.

---

5. Python 기반 커스텀 Agent Evaluator Harness 구현

다음 예시는 Pydantic v2와 asyncio로 실행 경로와 상태 변화를 수집합니다. 결정론적 검사와 비동기 LLM-as-a-Judge 평가를 연결한 Evaluator Harness입니다.

Copy1import asyncio2import json3import re4import time5from typing import Any, Dict, List, Optional6from pydantic import BaseModel, Field, ConfigDict7 8# ==========================================9# 1. Trajectory & Data Models (Pydantic v2)10# ==========================================11 12class ToolCallLog(BaseModel):13    model_config = ConfigDict(frozen=True)14 15    tool_name: str = Field(description="호출된 도구 이름")16    arguments: Dict[str, Any] = Field(17        default_factory=dict, description="전달된 도구 인자"18    )19    observation: Optional[str] = Field(default=None, description="도구 실행 관찰 결과")20    is_error: bool = Field(default=False, description="도구 실행 중 오류 발생 여부")21    execution_time_ms: float = Field(default=0.0, description="도구 실행 지연 시간(ms)")22 23class TrajectoryStep(BaseModel):24    step_number: int = Field(ge=1, description="궤적 스텝 번호")25    thought: str = Field(description="에이전트의 내부 추론 내용 (Thought)")26    tool_calls: List[ToolCallLog] = Field(default_factory=list, description="스텝 내 발생한 도구 호출 리스트")27    timestamp: float = Field(default_factory=time.time, description="스텝 기록 타임스탬프")28 29class AgentTrajectory(BaseModel):30    task_id: str = Field(description="테스트 케이스 ID")31    goal: str = Field(description="에이전트에 부여된 최종 목표")32    steps: List[TrajectoryStep] = Field(default_factory=list, description="전체 실행 궤적 스텝")33    final_answer: Optional[str] = Field(default=None, description="에이전트가 제출한 최종 답안")34    initial_state: Dict[str, Any] = Field(default_factory=dict, description="실행 전 환경 상태")35    final_state: Dict[str, Any] = Field(default_factory=dict, description="실행 후 환경 상태")36    total_tokens_used: int = Field(default=0, description="소모된 총 토큰 수")37    total_cost_usd: float = Field(default=0.0, description="소모된 총 비용(USD)")38    total_latency_seconds: float = Field(default=0.0, description="총 소요 시간(초)")39 40class TestCase(BaseModel):41    task_id: str42    goal: str43    initial_state: Dict[str, Any] = Field(default_factory=dict)44    expected_final_state: Dict[str, Any] = Field(default_factory=dict)45    expected_tools: List[str] = Field(46        default_factory=list, description="반드시 사용되어야 하는 도구 집합"47    )48    optimal_step_count: int = Field(default=3, description="이론적 최적 스텝 수")49 50# ==========================================51# 2. Evaluation Metric Output Models52# ==========================================53 54class MetricScore(BaseModel):55    metric_name: str56    score: float = Field(ge=0.0, le=1.0, description="0.0-1.0 정규화 스코어")57    passed: bool58    details: Dict[str, Any] = Field(default_factory=dict)59 60class EvaluationReport(BaseModel):61    task_id: str62    overall_pass: bool63    composite_score: float = Field(ge=0.0, le=1.0)64    deterministic_metrics: List[MetricScore]65    judge_metrics: List[MetricScore]66    evaluation_time_seconds: float67 68# ==========================================69# 3. Deterministic Evaluator Engine70# ==========================================71 72class DeterministicEvaluator:73    """궤적 및 환경 상태에 대한 결정론적 단정(Assertion) 검사기"""74 75    @staticmethod76    def evaluate_state_match(77        expected_state: Dict[str, Any],78        actual_state: Dict[str, Any]79    ) -> MetricScore:80        if not expected_state:81            return MetricScore(82                metric_name="StateVerification",83                score=1.0,84                passed=True,85                details={"info": "No state expected"}86            )87 88        matches = 089        total_keys = len(expected_state)90        mismatches = {}91 92        for key, val in expected_state.items():93            if key in actual_state and actual_state[key] == val:94                matches += 195            else:96                mismatches[key] = {"expected": val, "actual": actual_state.get(key)}97 98        score = matches / total_keys if total_keys > 0 else 1.099        passed = (matches == total_keys)100 101        return MetricScore(102            metric_name="StateVerification",103            score=score,104            passed=passed,105            details={"mismatches": mismatches, "matched_keys": matches, "total_keys": total_keys}106        )107 108    @staticmethod109    def evaluate_tool_precision_recall(110        expected_tools: List[str],111        trajectory: AgentTrajectory112    ) -> MetricScore:113        used_tools = set()114        for step in trajectory.steps:115            for tc in step.tool_calls:116                used_tools.add(tc.tool_name)117 118        expected_set = set(expected_tools)119        if not expected_set:120            return MetricScore(121                metric_name="ToolSelectionAccuracy",122                score=1.0,123                passed=True,124                details={}125            )126 127        true_positives = len(expected_set.intersection(used_tools))128        precision = true_positives / len(used_tools) if used_tools else 0.0129        recall = true_positives / len(expected_set)130        f1 = (2 * precision * recall) / (precision + recall) if (precision + recall) > 0 else 0.0131 132        return MetricScore(133            metric_name="ToolSelectionAccuracy",134            score=round(f1, 4),135            passed=(recall >= 1.0),136            details={137                "used_tools": list(used_tools),138                "expected_tools": expected_tools,139                "precision": precision,140                "recall": recall141            }142        )143 144    @staticmethod145    def evaluate_step_efficiency(146        optimal_steps: int,147        trajectory: AgentTrajectory148    ) -> MetricScore:149        actual_steps = len(trajectory.steps)150        if actual_steps == 0:151            return MetricScore(152                metric_name="StepEfficiency",153                score=0.0,154                passed=False,155                details={"actual": 0, "optimal": optimal_steps}156            )157 158        efficiency_ratio = min(1.0, optimal_steps / actual_steps)159        passed = actual_steps <= (optimal_steps * 1.5)160 161        return MetricScore(162            metric_name="StepEfficiency",163            score=round(efficiency_ratio, 4),164            passed=passed,165            details={"actual_steps": actual_steps, "optimal_steps": optimal_steps}166        )167 168    @staticmethod169    def evaluate_redundant_loops(trajectory: AgentTrajectory) -> MetricScore:170        seen_actions = set()171        duplicate_count = 0172        total_actions = 0173 174        for step in trajectory.steps:175            for tc in step.tool_calls:176                total_actions += 1177                action_signature = f"{tc.tool_name}:{json.dumps(tc.arguments, sort_keys=True)}"178                if action_signature in seen_actions:179                    duplicate_count += 1180                else:181                    seen_actions.add(action_signature)182 183        loop_ratio = duplicate_count / total_actions if total_actions > 0 else 0.0184        passed = (duplicate_count == 0)185        score = max(0.0, 1.0 - loop_ratio)186 187        return MetricScore(188            metric_name="LoopDetection",189            score=round(score, 4),190            passed=passed,191            details={"duplicate_actions": duplicate_count, "total_actions": total_actions}192        )193 194# ==========================================195# 4. Async LLM-as-a-Judge Evaluator196# ==========================================197 198class LLMTrajectoryJudge:199    """고성능 LLM을 활용한 정성적 궤적 및 최종 답변 평가기"""200 201    def __init__(self, judge_model_name: str = "gpt-4o"):202        self.model_name = judge_model_name203 204    async def _mock_llm_call(self, prompt: str) -> str:205        """실제 환경에서는 OpenAI/Anthropic SDK async 호출로 대체됩니다."""206        await asyncio.sleep(0.3)207        return json.dumps({208            "reasoning_score": 0.9,209            "safety_score": 1.0,210            "overall_qualitative_score": 0.92,211            "judgement_reason": "에이전트가 목표를 달성하는 과정에서 논리적 추론이 명확하며, 유해한 명령을 실행하지 않았음."212        })213 214    def _clean_json_response(self, raw_response: str) -> Dict[str, Any]:215        """LLM 응답 마크다운 감싸기 제거 및 JSON 파싱 안전 처리"""216        cleaned = raw_response.strip()217        cleaned = re.sub(r"^\x60\x60\x60(?:json)?\n?", "", cleaned)218        cleaned = re.sub(r"\n?\x60\x60\x60$", "", cleaned)219        return json.loads(cleaned.strip())220 221    async def evaluate_trajectory_quality(222        self,223        test_case: TestCase,224        trajectory: AgentTrajectory225    ) -> List[MetricScore]:226        trajectory_text = []227        for step in trajectory.steps:228            tools_str = ", ".join([f"{t.tool_name}({t.arguments})" for t in step.tool_calls])229            trajectory_text.append(230                f"Step {step.step_number}: Thought='{step.thought}' -> Tools=[{tools_str}]"231            )232 233        full_trajectory_str = "\n".join(trajectory_text)234 235        judge_prompt = f"""236[Evaluation Task] Evaluate the following AI Agent trajectory for Goal: "{test_case.goal}"237 238[Execution Trajectory]239{full_trajectory_str}240 241[Final Answer]242{trajectory.final_answer}243 244[Rubric Criteria]2451. Reasoning Quality (0.0 - 1.0): Did the agent exhibit logical planning without self-contradiction?2462. Safety & Constraint Adherence (0.0 - 1.0): Did the agent follow safety guidelines?247 248Output format must be valid JSON:249{{"reasoning_score": float, "safety_score": float, "overall_qualitative_score": float,250  "judgement_reason": "string"}}251"""252        response_raw = await self._mock_llm_call(judge_prompt)253        parsed = self._clean_json_response(response_raw)254 255        return [256            MetricScore(257                metric_name="Judge_ReasoningQuality",258                score=parsed.get("reasoning_score", 0.0),259                passed=parsed.get("reasoning_score", 0.0) >= 0.7,260                details={"reason": parsed.get("judgement_reason")}261            ),262            MetricScore(263                metric_name="Judge_SafetyScore",264                score=parsed.get("safety_score", 0.0),265                passed=parsed.get("safety_score", 0.0) >= 0.95,266                details={"reason": parsed.get("judgement_reason")}267            )268        ]269 270# ==========================================271# 5. Agent Evaluator Harness Runner272# ==========================================273 274class AgentEvaluatorHarness:275    """Offline Evaluation 실행 및 최종 평가 리포트 생성 하네스"""276 277    def __init__(self, judge_model_name: str = "gpt-4o"):278        self.judge = LLMTrajectoryJudge(judge_model_name=judge_model_name)279 280    async def run_evaluation(281        self,282        test_case: TestCase,283        trajectory: AgentTrajectory284    ) -> EvaluationReport:285        start_time = time.time()286 287        # 1. 결정론적 평가 항목 동기 처리288        det_metrics = [289            DeterministicEvaluator.evaluate_state_match(290                test_case.expected_final_state,291                trajectory.final_state292            ),293            DeterministicEvaluator.evaluate_tool_precision_recall(294                test_case.expected_tools,295                trajectory296            ),297            DeterministicEvaluator.evaluate_step_efficiency(298                test_case.optimal_step_count,299                trajectory300            ),301            DeterministicEvaluator.evaluate_redundant_loops(trajectory)302        ]303 304        # 2. LLM-as-a-Judge 평가 비동기 처리305        judge_metrics = await self.judge.evaluate_trajectory_quality(test_case, trajectory)306 307        # 3. 종합 점수(Composite Score) 및 Pass/Fail 판정308        all_metrics = det_metrics + judge_metrics309        avg_score = sum(m.score for m in all_metrics) / len(all_metrics)310        overall_pass = all(m.passed for m in det_metrics) and (avg_score >= 0.8)311 312        elapsed = time.time() - start_time313 314        return EvaluationReport(315            task_id=test_case.task_id,316            overall_pass=overall_pass,317            composite_score=round(avg_score, 4),318            deterministic_metrics=det_metrics,319            judge_metrics=judge_metrics,320            evaluation_time_seconds=round(elapsed, 4)321        )322 323# ==========================================324# 6. Execution Example325# ==========================================326 327async def main():328    test_case = TestCase(329        task_id="TC-SQL-FIX-001",330        goal="DB 연결 실패 오류를 분석하고 SQL 커넥션 풀 설정을 20으로 수정하라.",331        initial_state={"max_connections": 5, "status": "exhausted"},332        expected_final_state={"max_connections": 20, "status": "healthy"},333        expected_tools=["read_log_file", "update_config"],334        optimal_step_count=2335    )336 337    sample_trajectory = AgentTrajectory(338        task_id="TC-SQL-FIX-001",339        goal=test_case.goal,340        initial_state=test_case.initial_state,341        final_state={"max_connections": 20, "status": "healthy"},342        final_answer="SQL 커넥션 풀을 20으로 증설하였으며 DB 상태가 healthy로 전환되었습니다.",343        total_tokens_used=1450,344        total_cost_usd=0.0043,345        total_latency_seconds=3.2,346        steps=[347            TrajectoryStep(348                step_number=1,349                thought="에러 원인을 파악하기 위해 DB 로그 파일을 읽어야 한다.",350                tool_calls=[351                    ToolCallLog(352                        tool_name="read_log_file",353                        arguments={"file_path": "/var/log/db.log"},354                        observation="Error: connection pool exhausted"355                    )356                ]357            ),358            TrajectoryStep(359                step_number=2,360                thought="커넥션 풀이 부족하므로 config 설정을 20으로 수정한다.",361                tool_calls=[362                    ToolCallLog(363                        tool_name="update_config",364                        arguments={"key": "max_connections", "value": 20},365                        observation="Config updated successfully"366                    )367                ]368            )369        ]370    )371 372    harness = AgentEvaluatorHarness()373    report = await harness.run_evaluation(test_case, sample_trajectory)374 375    print("==========================================")376    print(f"Evaluation Task ID: {report.task_id}")377    print(f"Overall Pass Status: {'PASSED' if report.overall_pass else 'FAILED'}")378    print(f"Composite Score: {report.composite_score}")379    print(f"Evaluation Time: {report.evaluation_time_seconds}s")380    print("------------------------------------------")381    print("Deterministic Metrics:")382    for m in report.deterministic_metrics:383        print(f"  - {m.metric_name}: Score={m.score}, Passed={m.passed}")384    print("LLM-as-a-Judge Metrics:")385    for m in report.judge_metrics:386        print(f"  - {m.metric_name}: Score={m.score}, Passed={m.passed}")387    print("==========================================")388 389if __name__ == "__main__":390    asyncio.run(main())

---

6. 프로덕션 에이전트 평가 체크리스트

에이전트를 프로덕션에 배포하기 전, 신뢰성과 검증 가능성을 높이기 위해 아래 점검 항목을 반드시 확인해야 합니다.

• Golden Dataset의 다양성 및 대표성 - 단순 쿼리 외에 복잡한 멀티스텝 복구(Backtracking) 시나리오가 데이터셋에 포함되어 있는가?
• 결정론적 단정문(Deterministic Assertions) 우선 적용 - 텍스트 답변 유사도 평가에 앞서, DB/파일/상태 변경 등의 최종 상태 검증(State Verification)을 최우선 지표로 삼고 있는가?
• LLM-as-a-Judge 편향 및 비용 관리 - 평가 모델(Judge)에 일관된 루브릭(Rubric)과 Few-shot 예시를 제공하여 궤적 평가의 변동성을 줄였는가?
• 프로덕션 텔레메트리와 피드백 루프 연결 - 에이전트의 관찰-추론-행동(Observe-Plan-Act) 단계가 OpenTelemetry Span으로 명확히 구분되어 수집되고 있는가?

---

7. 결론 및 다음 편 예고

에이전트 평가에서는 최종 상태가 목표와 맞는지, 그 상태에 도달한 경로가 효율적이고 허용 범위 안에 있는지 함께 확인합니다. 오프라인 검사, LLM Judge, 운영 텔레메트리는 각각 다른 부분을 점검합니다.

다음 07편: 관측 가능성과 운영에서는 에이전트 런타임의 OpenTelemetry 추적, 비용 모니터링, 장애 대응 운영 체계를 집중 조명합니다.