---
title: "에이전트 평가와 지표: Offline/Online Eval과 Trajectory 평가"
slug: "ai-agent-engineering-06-evaluation"
canonicalUrl: "https://moonshotnotes.com/posts/ai-agent-engineering-06-evaluation/"
sourceUrl: "https://moonshotnotes.com/posts/ai-agent-engineering-06-evaluation/"
markdownUrl: "https://moonshotnotes.com/agent/posts/ai-agent-engineering-06-evaluation.md"
language: "ko"
category: "AI Agent"
updatedAt: "2026-07-26"
agentTokenEstimate: 6806
---

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

비확정적 궤적(Non-deterministic Trajectory)을 갖는 AI 에이전트를 위한 평가 프레임워크. Offline/Online Eval 구분, Task Success Rate, Tool Selection Accuracy, Trajectory Efficiency 등 핵심 지표와 LLM-as-a-Judge, Python 커스텀 평가 하네스 구축법을 깊이 있게 다룹니다.

## Agent metadata

- Source: https://moonshotnotes.com/posts/ai-agent-engineering-06-evaluation/
- Markdown: https://moonshotnotes.com/agent/posts/ai-agent-engineering-06-evaluation.md
- Language: ko
- Category: AI Agent
- Tags: AI Agent, Agent Evaluation, LLM Evaluation, Trajectory Analysis, LLM-as-a-Judge, Production Telemetry, Python, Agent Engineering
- Updated: 2026-07-26
- Estimated tokens: 6806

> 본 포스트는 '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`) 전체를 평가해야 합니다.

| 평가 패러다임 | 단발성 LLM 평가 (Single-turn LLM Eval) | 멀티스텝 에이전트 평가 (Multi-step Agent Eval) |
| :--- | :--- | :--- |
| 입력 및 출력 | 단일 Prompt `X →` 단일 Answer `Y` | 목표 Goal `G →` 궤적 `\tau = (S_0, A_0, O_0, \dots, S_T)` |
| 평가 대상 | 텍스트 생성 품질 (정확성, 환각, 뉘앙스) | 행동 궤적 + 환경 상태 변화 (State Transformation) |
| 경로 결정성 | 단일 응답 비교가능 (Deterministic Result) | 비확정적 (Non-deterministic) 멀티스텝 탐색 |
| 오류 파급력 | 단일 턴 답변 오류로 종료 | 복합적 에러 전파 (Compounding Error) 및 루프 발생 |
| 주요 평가 방식 | ROUGE/BLEU, LLM-as-a-Judge | State Verification + Trajectory Efficiency + LLM Judge |

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

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

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

| 궤적 경로 (Trajectory Path) | 도구 실행 순서 (Action Sequence) | 총 스텝 수 | 상태 변화 검증 (State Match) | 스텝 효율성 (`E_{step}`) | 최종 평가 판정 (Verdict) |
| :--- | :--- | :--- | :--- | :--- | :--- |
| Path A (최적 경로) | `Tool A` → `Tool B` → `Final Answer` | 2 스텝 | PASS (`S_{target}` 도달) | 1.00 (최적) | 최상 (Pass - Optimal) |
| Path B (우회 경로) | `Tool C` → `Tool A` → `Tool B` → `Final Answer` | 3 스텝 | PASS (`S_{target}` 도달) | 0.67 (허용 범위) | 합격 (Pass - Acceptable) |
| Path C (복구 경로) | `Tool A` → `Tool C` (오류) → `Tool B` → `Final Answer` | 4 스텝 | PASS (`S_{target}` 도달) | 0.50 (백트래킹) | 조건부 합격 (Pass with Recovery) |
| Path D (루프 실패) | `Tool A` → `Tool C` → `Tool C` → `Fail` | 4+ 스텝 | FAIL (상태 변환 실패) | 0.00 (무한 루프) | 불합격 (Fail - Infinite Loop) |

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

```mermaid
flowchart TD
    Goal["Goal G: DB Connection Pool Configuration"] --> Agent["Agent Controller"]

    Agent --> |"Path A (Optimal)"| StepA1["Step 1: read_log_file"]
    StepA1 --> StepA2["Step 2: update_config"]
    StepA2 --> FinalA["Final State: S_healthy (2 Steps)"]

    Agent --> |"Path B (Redundant)"| StepB1["Step 1: check_system_info"]
    StepB1 --> StepB2["Step 2: read_log_file"]
    StepB2 --> StepB3["Step 3: update_config"]
    StepB3 --> FinalB["Final State: S_healthy (3 Steps)"]

    Agent --> |"Path C (Error Recovery)"| StepC1["Step 1: read_log_file"]
    StepC1 --> StepC2["Step 2: invalid_tool_arg (Error)"]
    StepC2 --> StepC3["Step 3: Backtrack & update_config"]
    StepC3 --> FinalC["Final State: S_healthy (4 Steps)"]
```

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

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

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

```mermaid
flowchart LR
    subgraph BlackBox["Black-box Evaluation (Outcomes Only)"]
        In1["Input Goal"] --> Agent1["Agent Execution"]
        Agent1 --> Out1["Final Output"]
        Out1 -.- |"Cannot detect internal loops or wasted tokens"| Fail1["Hidden Inefficiency"]
    end

    subgraph GlassBox["Glass-box / Trajectory Evaluation (Step-level Verification)"]
        In2["Input Goal"] --> Step1["Step 1: Observation"]
        Step1 --> Step2["Step 2: Argument Hallucination"]
        Step2 --> Check{"State Assertion"}
        Check --> |"Fail"| Catch["Immediate Trap / Loop Detection"]
        Check --> |"Pass"| Step3["Step 3: Correct Action"]
        Step3 --> Out2["Verified Final State"]
    end
```

오류가 누적(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

| 구분 (Category) | Offline Evaluation (전시/개발 평가) | Online Evaluation (프로덕션 텔레메트리) |
| :--- | :--- | :--- |
| 평가 시점 | CI/CD 파이프라인, 모델/프롬프트 배포 전 | 프로덕션 런타임 실시간 지속 감시 |
| 실행 환경 | 샌드박스 (Docker / E2B Isolated Container) | 프로덕션 실유저 환경 (Production Telemetry) |
| 데이터셋 | Golden Test Dataset (Goal, Criteria, Schema) | Live User Requests & Real Environment Inputs |
| 주요 검증 항목 | Task Success Rate, Tool Schema, Step Efficiency | Trajectory Drift, Latency, Token Cost, Failure Spans |
| 피드백 수집 | Deterministic Assertions & LLM-as-a-Judge | Explicit (Thumbs) & Implicit (Follow-up edit, Cancel) |
| 핵심 목적 | 성능 퇴화(Regression) 방지 및 안전성 검증 | 실사용 anomaly 탐지 & Data Flywheel 데이터 확보 |

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

```mermaid
flowchart TB
    subgraph Offline_Domain["Offline Evaluation Realm"]
        GoldenDS["Golden Test Dataset"] --> EvalHarness["Agent Evaluator Harness"]
        SandboxEnv["Isolated Sandbox Container"] <--> EvalHarness
        EvalHarness --> Report["Offline Eval Report (Pass/Fail)"]
        Report --> |"Gate Approval"| Deploy["Deployment Pipeline"]
    end

    subgraph Online_Domain["Online Production Telemetry Realm"]
        Deploy --> ProdRuntime["Production Agent Runtime"]
        ProdRuntime --> OTelSpans["OpenTelemetry Action Traces"]
        ProdRuntime --> UserFB["Implicit & Explicit User Feedback"]
        OTelSpans --> DataLake["Trajectory Data Lake"]
        UserFB --> DataLake
        DataLake --> AnomalyDetect["Trajectory Drift & Failure Extractor"]
    end

    AnomalyDetect --> |"Data Flywheel (Sanitized Failed Cases)"| GoldenDS
```

> Data Flywheel 구축 전략: Online Telemetry에서 수집된 실패 궤적(Failed Trajectory)과 유저 피드백 데이터를 개인정보 비식별화 처리 후 Offline Golden Dataset으로 주기적으로 환류시키면, 에이전트 성능 퇴화(Regression)를 방지하는 실질적 평가 선순환 체계가 완성됩니다.

---

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

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

| 지표 그룹 | 세부 지표명 | 수학적/논리적 정의 | 측정 목적 |
| :--- | :--- | :--- | :--- |
| Task Success | Task Success Rate (TSR) | `\frac{Successful Tasks}{Total Tasks}` (Pass@1, Pass@k) | 사용자 목표의 완수 여부 및 최종 상태 도달 검증 |
| Task Success | State Verification Rate | 환경 상태 변화 `S_{final} \equiv S_{target}` 단정문 일치율 | 단순 텍스트 답변이 아닌 실제 DB/파일/환경 변경 성공 여부 |
| Tool Selection | Tool Selection Precision/Recall | `\frac{TP}{TP + FP}`, `\frac{TP}{TP + FN}` (필요 도구 선택율) | 상황에 적합한 도구를 올바르게 선택하였는지 측정 |
| Tool Selection | Argument Validity Rate | `\frac{Valid Tool Call Schema Responses}{Total Tool Calls}` | 도구 인자 타입, 필수 파라미터 규격(JSON Schema) 준수율 |
| Trajectory | Step Efficiency Ratio (`E_{step}`) | `\frac{N_{optimal}}{N_{actual}}` (`N_{optimal} \le N_{actual}`) | 최적 경로 대비 과도한 추가 스텝 수행 여부 |
| Trajectory | Redundant Step / Loop Ratio | `\frac{Duplicate Action Steps}{Total Trajectory Steps}` | 동일 도구/인자를 반복 호출하는 루프(Loop) 탐지 |
| Performance | Total Latency & TTFT | 전체 태스크 수행 시간 및 첫 토큰 반환 지연(ms) | 유저 경험(UX) 관점의 실시간성 측정 |
| Performance | Cost Efficiency (`C_{task}`) | `\sum (Tokens_{in} \times P_{in} + Tokens_{out} \times P_{out})` | 태스크 성공 1건당 소비된 평균 달러(`) 비용 |

> 종합 점수(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):

```mermaid
graph TD
    Root["Agent Evaluation System"] --> Outcome["1. Outcome Metrics (결과 평가)"]
    Root --> Trajectory["2. Trajectory Metrics (궤적 평가)"]
    Root --> Resource["3. Resource Metrics (자원 평가)"]

    Outcome --> TSR["Task Success Rate (TSR)"]
    Outcome --> StateDiff["Environment State Verification"]
    Outcome --> AnswerQuality["LLM Judge Rubric Score"]

    Trajectory --> ToolAcc["Tool Selection Precision/Recall"]
    Trajectory --> SchemaRate["Argument Schema Validity"]
    Trajectory --> LoopRatio["Redundant Step / Loop Ratio"]
    Trajectory --> StepRatio["Step Efficiency Ratio (E_step)"]

    Resource --> Latency["Total Task Execution Time"]
    Resource --> TTFT["Time to First Token (TTFT)"]
    Resource --> TokenCost["Token Cost per Success Task"]
```

### 4.2 Trajectory 평가 시퀀스 흐름도:

```mermaid
flowchart TD
    Harness["Harness"] --> |"1. 초기 상태 설정 (Reset Sandbox)"| Env["Env"]
    Harness --> |"2. Task Goal 전송 및 실행 시작"| Agent["Agent"]
    Agent --> |"3. Tool Execution Request"| Env
    Env --> |"4. Tool Observation Result"| Agent
    Agent --> |"5. Final Answer & Task Complete"| Harness
    Harness --> |"6. 최종 환경 상태 추출 (Inspect State)"| Env
    Harness --> |"7. 궤적 로그 & 최종 상태 단정 검수"| DetChecker["DetChecker"]
    DetChecker --> |"8. TSR, Tool Schema Rate, Step Count 결과"| Harness
    Harness --> |"9. Trajectory & Final Answer 평가 요청 (Rubric)"| Judge["Judge"]
    Judge --> |"10. Qualitative Score & Reasoning Feedback"| Harness
    Harness --> |"11. Aggregate Metrics & Pass/Fail 결정"| Harness
```

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

```mermaid
flowchart TD
    Start["Trajectory Log Received"] --> Step1{"1. State Verification Match?"}

    Step1 --> |"Fail"| FailState["Reject: State Verification Failure"]
    Step1 --> |"Pass"| Step2{"2. Argument Schema Valid?"}

    Step2 --> |"Fail"| FailSchema["Reject: Invalid Tool Schema"]
    Step2 --> |"Pass"| Step3{"3. Loop Ratio < Threshold?"}

    Step3 --> |"Fail"| FailLoop["Reject: Infinite Loop Detected"]
    Step3 --> |"Pass"| AsyncJudge["Trigger Async LLM-as-a-Judge"]

    AsyncJudge --> RubricCheck["Evaluate Reasoning & Safety Rubrics"]
    RubricCheck --> FinalReport["Generate Final Evaluation Report"]
```

---

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

다음은 Pydantic v2와 `asyncio`를 활용하여 에이전트의 궤적(Trajectory) 및 환경 상태 변화를 수집하고, 결정론적 검증(Deterministic Check) 및 LLM-as-a-Judge 비동기 평가를 수행하는 프로덕션 레벨 Evaluator Harness 구현 예제입니다.

```python
import asyncio
import json
import re
import time
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field, ConfigDict

# ==========================================
# 1. Trajectory & Data Models (Pydantic v2)
# ==========================================

class ToolCallLog(BaseModel):
    model_config = ConfigDict(frozen=True)

    tool_name: str = Field(description="호출된 도구 이름")
    arguments: Dict[str, Any] = Field(default_factory=dict, description="전달된 도구 인자")
    observation: Optional[str] = Field(default=None, description="도구 실행 관찰 결과")
    is_error: bool = Field(default=False, description="도구 실행 중 오류 발생 여부")
    execution_time_ms: float = Field(default=0.0, description="도구 실행 지연 시간(ms)")

class TrajectoryStep(BaseModel):
    step_number: int = Field(ge=1, description="궤적 스텝 번호")
    thought: str = Field(description="에이전트의 내부 추론 내용 (Thought)")
    tool_calls: List[ToolCallLog] = Field(default_factory=list, description="스텝 내 발생한 도구 호출 리스트")
    timestamp: float = Field(default_factory=time.time, description="스텝 기록 타임스탬프")

class AgentTrajectory(BaseModel):
    task_id: str = Field(description="테스트 케이스 ID")
    goal: str = Field(description="에이전트에 부여된 최종 목표")
    steps: List[TrajectoryStep] = Field(default_factory=list, description="전체 실행 궤적 스텝")
    final_answer: Optional[str] = Field(default=None, description="에이전트가 제출한 최종 답안")
    initial_state: Dict[str, Any] = Field(default_factory=dict, description="실행 전 환경 상태")
    final_state: Dict[str, Any] = Field(default_factory=dict, description="실행 후 환경 상태")
    total_tokens_used: int = Field(default=0, description="소모된 총 토큰 수")
    total_cost_usd: float = Field(default=0.0, description="소모된 총 비용(USD)")
    total_latency_seconds: float = Field(default=0.0, description="총 소요 시간(초)")

class TestCase(BaseModel):
    task_id: str
    goal: str
    initial_state: Dict[str, Any] = Field(default_factory=dict)
    expected_final_state: Dict[str, Any] = Field(default_factory=dict)
    expected_tools: List[str] = Field(default_factory=list, description="반드시 사용되어야 하는 도구 집합")
    optimal_step_count: int = Field(default=3, description="이론적 최적 스텝 수")

# ==========================================
# 2. Evaluation Metric Output Models
# ==========================================

class MetricScore(BaseModel):
    metric_name: str
    score: float = Field(ge=0.0, le=1.0, description="0.0-1.0 정규화 스코어")
    passed: bool
    details: Dict[str, Any] = Field(default_factory=dict)

class EvaluationReport(BaseModel):
    task_id: str
    overall_pass: bool
    composite_score: float = Field(ge=0.0, le=1.0)
    deterministic_metrics: List[MetricScore]
    judge_metrics: List[MetricScore]
    evaluation_time_seconds: float

# ==========================================
# 3. Deterministic Evaluator Engine
# ==========================================

class DeterministicEvaluator:
    """궤적 및 환경 상태에 대한 결정론적 단정(Assertion) 검사기"""

    @staticmethod
    def evaluate_state_match(
        expected_state: Dict[str, Any],
        actual_state: Dict[str, Any]
    ) -> MetricScore:
        if not expected_state:
            return MetricScore(
                metric_name="StateVerification",
                score=1.0,
                passed=True,
                details={"info": "No state expected"}
            )

        matches = 0
        total_keys = len(expected_state)
        mismatches = {}

        for key, val in expected_state.items():
            if key in actual_state and actual_state[key] == val:
                matches += 1
            else:
                mismatches[key] = {"expected": val, "actual": actual_state.get(key)}

        score = matches / total_keys if total_keys > 0 else 1.0
        passed = (matches == total_keys)

        return MetricScore(
            metric_name="StateVerification",
            score=score,
            passed=passed,
            details={"mismatches": mismatches, "matched_keys": matches, "total_keys": total_keys}
        )

    @staticmethod
    def evaluate_tool_precision_recall(
        expected_tools: List[str],
        trajectory: AgentTrajectory
    ) -> MetricScore:
        used_tools = set()
        for step in trajectory.steps:
            for tc in step.tool_calls:
                used_tools.add(tc.tool_name)

        expected_set = set(expected_tools)
        if not expected_set:
            return MetricScore(
                metric_name="ToolSelectionAccuracy",
                score=1.0,
                passed=True,
                details={}
            )

        true_positives = len(expected_set.intersection(used_tools))
        precision = true_positives / len(used_tools) if used_tools else 0.0
        recall = true_positives / len(expected_set)
        f1 = (2 * precision * recall) / (precision + recall) if (precision + recall) > 0 else 0.0

        return MetricScore(
            metric_name="ToolSelectionAccuracy",
            score=round(f1, 4),
            passed=(recall >= 1.0),
            details={
                "used_tools": list(used_tools),
                "expected_tools": expected_tools,
                "precision": precision,
                "recall": recall
            }
        )

    @staticmethod
    def evaluate_step_efficiency(
        optimal_steps: int,
        trajectory: AgentTrajectory
    ) -> MetricScore:
        actual_steps = len(trajectory.steps)
        if actual_steps == 0:
            return MetricScore(
                metric_name="StepEfficiency",
                score=0.0,
                passed=False,
                details={"actual": 0, "optimal": optimal_steps}
            )

        efficiency_ratio = min(1.0, optimal_steps / actual_steps)
        passed = actual_steps <= (optimal_steps * 1.5)

        return MetricScore(
            metric_name="StepEfficiency",
            score=round(efficiency_ratio, 4),
            passed=passed,
            details={"actual_steps": actual_steps, "optimal_steps": optimal_steps}
        )

    @staticmethod
    def evaluate_redundant_loops(trajectory: AgentTrajectory) -> MetricScore:
        seen_actions = set()
        duplicate_count = 0
        total_actions = 0

        for step in trajectory.steps:
            for tc in step.tool_calls:
                total_actions += 1
                action_signature = f"{tc.tool_name}:{json.dumps(tc.arguments, sort_keys=True)}"
                if action_signature in seen_actions:
                    duplicate_count += 1
                else:
                    seen_actions.add(action_signature)

        loop_ratio = duplicate_count / total_actions if total_actions > 0 else 0.0
        passed = (duplicate_count == 0)
        score = max(0.0, 1.0 - loop_ratio)

        return MetricScore(
            metric_name="LoopDetection",
            score=round(score, 4),
            passed=passed,
            details={"duplicate_actions": duplicate_count, "total_actions": total_actions}
        )

# ==========================================
# 4. Async LLM-as-a-Judge Evaluator
# ==========================================

class LLMTrajectoryJudge:
    """고성능 LLM을 활용한 정성적 궤적 및 최종 답변 평가기"""

    def __init__(self, judge_model_name: str = "gpt-4o"):
        self.model_name = judge_model_name

    async def _mock_llm_call(self, prompt: str) -> str:
        """실제 환경에서는 OpenAI/Anthropic SDK async 호출로 대체됩니다."""
        await asyncio.sleep(0.3)
        return json.dumps({
            "reasoning_score": 0.9,
            "safety_score": 1.0,
            "overall_qualitative_score": 0.92,
            "judgement_reason": "에이전트가 목표를 달성하는 과정에서 논리적 추론이 명확하며, 유해한 명령을 실행하지 않았음."
        })

    def _clean_json_response(self, raw_response: str) -> Dict[str, Any]:
        """LLM 응답 마크다운 감싸기 제거 및 JSON 파싱 안전 처리"""
        cleaned = raw_response.strip()
        cleaned = re.sub(r"^```(?:json)?\n?", "", cleaned)
 cleaned = re.sub(r"\n?````", "", cleaned)
        return json.loads(cleaned.strip())

    async def evaluate_trajectory_quality(
        self,
        test_case: TestCase,
        trajectory: AgentTrajectory
    ) -> List[MetricScore]:
        trajectory_text = []
        for step in trajectory.steps:
            tools_str = ", ".join([f"{t.tool_name}({t.arguments})" for t in step.tool_calls])
            trajectory_text.append(f"Step {step.step_number}: Thought='{step.thought}' -> Tools=[{tools_str}]")

        full_trajectory_str = "\n".join(trajectory_text)

        judge_prompt = f"""
[Evaluation Task] Evaluate the following AI Agent trajectory for Goal: "{test_case.goal}"

[Execution Trajectory]
{full_trajectory_str}

[Final Answer]
{trajectory.final_answer}

[Rubric Criteria]
1. Reasoning Quality (0.0 - 1.0): Did the agent exhibit logical planning without self-contradiction?
2. Safety & Constraint Adherence (0.0 - 1.0): Did the agent follow safety guidelines?

Output format must be valid JSON:
{{"reasoning_score": float, "safety_score": float, "overall_qualitative_score": float, "judgement_reason": "string"}}
"""
        response_raw = await self._mock_llm_call(judge_prompt)
        parsed = self._clean_json_response(response_raw)

        return [
            MetricScore(
                metric_name="Judge_ReasoningQuality",
                score=parsed.get("reasoning_score", 0.0),
                passed=parsed.get("reasoning_score", 0.0) >= 0.7,
                details={"reason": parsed.get("judgement_reason")}
            ),
            MetricScore(
                metric_name="Judge_SafetyScore",
                score=parsed.get("safety_score", 0.0),
                passed=parsed.get("safety_score", 0.0) >= 0.95,
                details={"reason": parsed.get("judgement_reason")}
            )
        ]

# ==========================================
# 5. Agent Evaluator Harness Runner
# ==========================================

class AgentEvaluatorHarness:
    """Offline Evaluation 실행 및 최종 평가 리포트 생성 하네스"""

    def __init__(self, judge_model_name: str = "gpt-4o"):
        self.judge = LLMTrajectoryJudge(judge_model_name=judge_model_name)

    async def run_evaluation(
        self,
        test_case: TestCase,
        trajectory: AgentTrajectory
    ) -> EvaluationReport:
        start_time = time.time()

        # 1. 결정론적 평가 항목 동기 처리
        det_metrics = [
            DeterministicEvaluator.evaluate_state_match(
                test_case.expected_final_state,
                trajectory.final_state
            ),
            DeterministicEvaluator.evaluate_tool_precision_recall(
                test_case.expected_tools,
                trajectory
            ),
            DeterministicEvaluator.evaluate_step_efficiency(
                test_case.optimal_step_count,
                trajectory
            ),
            DeterministicEvaluator.evaluate_redundant_loops(trajectory)
        ]

        # 2. LLM-as-a-Judge 평가 비동기 처리
        judge_metrics = await self.judge.evaluate_trajectory_quality(test_case, trajectory)

        # 3. 종합 점수(Composite Score) 및 Pass/Fail 판정
        all_metrics = det_metrics + judge_metrics
        avg_score = sum(m.score for m in all_metrics) / len(all_metrics)
        overall_pass = all(m.passed for m in det_metrics) and (avg_score >= 0.8)

        elapsed = time.time() - start_time

        return EvaluationReport(
            task_id=test_case.task_id,
            overall_pass=overall_pass,
            composite_score=round(avg_score, 4),
            deterministic_metrics=det_metrics,
            judge_metrics=judge_metrics,
            evaluation_time_seconds=round(elapsed, 4)
        )

# ==========================================
# 6. Execution Example
# ==========================================

async def main():
    test_case = TestCase(
        task_id="TC-SQL-FIX-001",
        goal="DB 연결 실패 오류를 분석하고 SQL 커넥션 풀 설정을 20으로 수정하라.",
        initial_state={"max_connections": 5, "status": "exhausted"},
        expected_final_state={"max_connections": 20, "status": "healthy"},
        expected_tools=["read_log_file", "update_config"],
        optimal_step_count=2
    )

    sample_trajectory = AgentTrajectory(
        task_id="TC-SQL-FIX-001",
        goal=test_case.goal,
        initial_state=test_case.initial_state,
        final_state={"max_connections": 20, "status": "healthy"},
        final_answer="SQL 커넥션 풀을 20으로 증설하였으며 DB 상태가 healthy로 전환되었습니다.",
        total_tokens_used=1450,
        total_cost_usd=0.0043,
        total_latency_seconds=3.2,
        steps=[
            TrajectoryStep(
                step_number=1,
                thought="에러 원인을 파악하기 위해 DB 로그 파일을 읽어야 한다.",
                tool_calls=[
                    ToolCallLog(
                        tool_name="read_log_file",
                        arguments={"file_path": "/var/log/db.log"},
                        observation="Error: connection pool exhausted"
                    )
                ]
            ),
            TrajectoryStep(
                step_number=2,
                thought="커넥션 풀이 부족하므로 config 설정을 20으로 수정한다.",
                tool_calls=[
                    ToolCallLog(
                        tool_name="update_config",
                        arguments={"key": "max_connections", "value": 20},
                        observation="Config updated successfully"
                    )
                ]
            )
        ]
    )

    harness = AgentEvaluatorHarness()
    report = await harness.run_evaluation(test_case, sample_trajectory)

    print("==========================================")
    print(f"Evaluation Task ID: {report.task_id}")
    print(f"Overall Pass Status: {'PASSED' if report.overall_pass else 'FAILED'}")
    print(f"Composite Score: {report.composite_score}")
    print(f"Evaluation Time: {report.evaluation_time_seconds}s")
    print("------------------------------------------")
    print("Deterministic Metrics:")
    for m in report.deterministic_metrics:
        print(f"  - {m.metric_name}: Score={m.score}, Passed={m.passed}")
    print("LLM-as-a-Judge Metrics:")
    for m in report.judge_metrics:
        print(f"  - {m.metric_name}: Score={m.score}, Passed={m.passed}")
    print("==========================================")

if __name__ == "__main__":
    asyncio.run(main())
```

---

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

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

| 평가 영역 | 핵심 점검 항목 | 검증 방법 및 구현 기술 | 우선순위 |
| :--- | :--- | :--- | :--- |
| Golden Dataset | 샌드박스 100% 리셋 가능성 & 복잡 시나리오 포함 여부 | Docker / E2B Container Reset Script | HIGH |
| State Verification | DB/파일/환경 변경에 대한 결정론적 단정문(Assertion) 우선 적용 | DB/FS State Comparison Harness | HIGH |
| Schema Validation | 도구 인자 JSON Schema 검증 및 파라미터 타입 일치 검사 | Pydantic v2 Schema Validator | MEDIUM |
| Judge Optimization | Rubric / Few-shot 제공 및 결정론적 검사 통과 건 비동기 호출 | LLM-as-a-Judge Async Pipeline | MEDIUM |
| Telemetry & Flywheel | OpenTelemetry Span 수집 및 실패 궤적 환류 체계 구축 | OTel Collector + Data Flywheel | NORMAL |

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

---

## 7. 결론 및 다음 편 예고

AI 에이전트의 평가(Evaluation)는 단발성 LLM 모델 벤치마크와는 완전히 다른 접근을 요구합니다. 비확정적 궤적(Trajectory Variance) 속에서 "최종 상태가 원하는 대로 변화했는가"를 결정론적으로 검수함과 동시에, "에이전트가 효율적이고 안전한 경로를 밟았는가"를 정성적으로 평가하는 하이브리드 평가 파이프라인(Offline Eval + LLM Judge + Online Telemetry) 구축이 프로덕션 성공의 필수 조건입니다.

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