AI 에이전트 모니터링·평가 체계 구축: 프로덕션에서 조용히 망가지는 품질을 DeepEval·Langfuse로 잡아내는 법

from deepeval import evaluate
from deepeval.metrics import (
    AnswerRelevancyMetric,
    HallucinationMetric,
    ToolCorrectnessMetric,
)
from deepeval.test_case import LLMTestCase, ToolCall
 
test_case = LLMTestCase(
    input="서울 날씨 알려줘",
    actual_output="서울은 현재 맑고 기온은 22도입니다.",
    context=["weather_api 호출 결과: {'city': 'Seoul', 'temp': 22, 'condition': 'clear'}"],
    tools_called=[
        ToolCall(name="weather_api", input_parameters={"city": "Seoul"})
    ],
    expected_tools=[
        ToolCall(name="weather_api", input_parameters={"city": "Seoul"})
    ],
)
 
metrics = [
    AnswerRelevancyMetric(threshold=0.7, model="gpt-4o"),
    HallucinationMetric(threshold=0.3, model="gpt-4o"),
    ToolCorrectnessMetric(),  # LLM API 비용 없이 결정론적으로 평가
]
 
evaluate(test_cases=[test_case], metrics=metrics)

from langfuse.decorators import observe, langfuse_context
from langfuse.openai import openai  # OpenAI 클라이언트 래핑
 
@observe()  # 루트 스팬: 이 함수 전체가 하나의 트레이스로 기록됨
async def run_agent(user_query: str) -> str:
    langfuse_context.update_current_trace(
        name="customer-support-agent",
        user_id="user-123",
        tags=["production", "v2.1"],
    )
 
    # 도구 스팬: 외부 도구 호출을 별도로 기록
    with langfuse_context.span("tool:search_kb"):
        kb_results = await search_knowledge_base(user_query)
 
    # 추론 스팬: LLM 호출 시 토큰·비용이 자동으로 기록됨
    response = openai.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "당신은 고객 지원 에이전트입니다."},
            {"role": "user", "content": user_query},
            {"role": "assistant", "content": f"검색 결과: {kb_results}"},
        ],
    )
 
    return response.choices[0].message.content

# test_agent_eval.py
import pytest
from deepeval import assert_test
from deepeval.metrics import (
    AnswerRelevancyMetric,
    HallucinationMetric,
    GEval,
)
from deepeval.test_case import LLMTestCase, ToolCall
 
# --- 실제 에이전트 모듈로 교체하세요 ---
async def run_customer_support_agent(query: str) -> str:
    return "30일 이내 구매 시 영수증 지참하면 전액 환불 가능합니다."
# ----------------------------------------
 
# 골든 데이터셋: 핵심 시나리오를 입력-컨텍스트-기대출력 쌍으로 정의
TEST_CASES = [
    {
        "input": "환불 정책이 어떻게 되나요?",
        "context": ["30일 이내 영수증 지참 시 전액 환불 가능"],
        "expected_output": "30일 이내 구매 시 환불 가능",
    },
    {
        "input": "주문 취소는 어떻게 하나요?",
        "context": ["배송 시작 전 앱에서 직접 취소 가능, 배송 후는 반품 절차 진행"],
        "expected_output": "배송 시작 전 앱에서 취소 가능",
    },
]
 
@pytest.mark.parametrize("case", TEST_CASES)
@pytest.mark.asyncio
async def test_customer_support_agent(case):
    actual_output = await run_customer_support_agent(case["input"])
 
    test_case = LLMTestCase(
        input=case["input"],
        actual_output=actual_output,
        context=case["context"],
        expected_output=case["expected_output"],
    )
 
    # 커스텀 평가 기준도 자연어로 정의할 수 있는데, 처음 짰을 때 꽤 놀랐습니다
    policy_adherence = GEval(
        name="PolicyAdherence",
        criteria="응답이 제공된 컨텍스트의 정책을 정확히 반영하는가",
        evaluation_params=["actual_output", "context"],
        threshold=0.8,
    )
 
    assert_test(
        test_case,
        metrics=[
            AnswerRelevancyMetric(threshold=0.7, model="gpt-4o-mini"),
            HallucinationMetric(threshold=0.2, model="gpt-4o-mini"),
            policy_adherence,
        ],
    )

# .github/workflows/agent-eval.yml
name: Agent Evaluation
 
on: [pull_request]
 
jobs:
  evaluate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - name: Install dependencies
        run: pip install deepeval langfuse openai pytest pytest-asyncio
      - name: Run agent evaluation
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
          DEEPEVAL_API_KEY: ${{ secrets.DEEPEVAL_API_KEY }}
        run: deepeval test run test_agent_eval.py

from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context
import asyncio
import json
 
langfuse = Langfuse()
 
@observe()
async def run_agent_with_online_eval(user_id: str, query: str) -> str:
    # --- 실제 에이전트 호출로 교체하세요 ---
    result = f"'{query}'에 대한 고객 지원 응답입니다."
    # ----------------------------------------
 
    trace_id = langfuse_context.get_current_trace_id()
 
    # 사용자 응답 지연 없이 백그라운드에서 품질 평가 실행
    asyncio.create_task(
        run_online_evaluation(trace_id, query, result)
    )
 
    return result
 
async def run_online_evaluation(trace_id: str, query: str, response: str):
    """백그라운드에서 LLM-as-a-Judge 평가 실행"""
    from openai import AsyncOpenAI
 
    client = AsyncOpenAI()
 
    eval_prompt = f"""
    사용자 질문: {query}
    에이전트 응답: {response}
 
    다음 기준으로 응답 품질을 0.0~1.0 사이 점수로만 답해주세요:
    - 질문에 대한 관련성 (0.4)
    - 정확성과 사실 기반 (0.4)
    - 간결하고 명확한 설명 (0.2)
 
    JSON 형식: {{"score": 0.85, "reason": "간략 이유"}}
    """
 
    eval_response = await client.chat.completions.create(
        model="gpt-4o-mini",  # 비용 절감을 위해 미니 모델 활용
        messages=[{"role": "user", "content": eval_prompt}],
        response_format={"type": "json_object"},
    )
 
    result = json.loads(eval_response.choices[0].message.content)
 
    langfuse.score(
        trace_id=trace_id,
        name="online-quality-score",
        value=result["score"],
        comment=result["reason"],
    )
 
    if result["score"] < 0.6:
        await send_alert(trace_id, result["score"], result["reason"])

from deepeval.test_case import LLMTestCase, ToolCall
from deepeval.metrics import ToolCorrectnessMetric
 
# 고객 지원 에이전트의 도구 호출 시나리오
tool_test_cases = [
    LLMTestCase(
        input="지난달 주문 내역 조회해줘",
        actual_output="지난달 주문 3건을 찾았습니다...",
        tools_called=[
            ToolCall(
                name="get_orders",
                input_parameters={
                    "user_id": "user-123",
                    "date_range": "last_month",
                },
            )
        ],
        expected_tools=[
            ToolCall(
                name="get_orders",
                input_parameters={
                    "user_id": "user-123",
                    "date_range": "last_month",
                },
            )
        ],
    ),
    LLMTestCase(
        input="이 제품 재고 있어?",
        actual_output="현재 재고가 50개 있습니다.",
        tools_called=[
            ToolCall(
                name="check_inventory",
                input_parameters={"product_id": "PROD-456"},
            )
        ],
        expected_tools=[
            ToolCall(
                name="check_inventory",
                input_parameters={"product_id": "PROD-456"},
            )
        ],
    ),
]
 
# ToolCorrectnessMetric은 LLM API 비용 없이 결정론적으로 평가
metric = ToolCorrectnessMetric()
for tc in tool_test_cases:
    metric.measure(tc)
    print(f"입력: {tc.input}")
    print(f"도구 정확도: {metric.score:.2f} - {metric.reason}\n")