LLM 평가와 프로덕션 모니터링

4. MLflow Evaluation — mlflow.genai.evaluate()

4-1. 기본 개념

mlflow.genai.evaluate() 는 LLM 애플리케이션의 출력을 Scorer (평가자) 를 사용하여 자동으로 채점합니다. 평가 결과는 MLflow Experiment 에 기록되어 버전 간 비교가 가능합니다.

평가 데이터셋 (inputs + expected_outputs)
        ↓
    앱/모델 실행 → 실제 출력 생성
        ↓
    Scorer 실행 → 각 항목 채점 (0.0 ~ 1.0 또는 True/False)
        ↓
    MLflow Experiment 에 평가 Run 기록

4-2. 기본 Scorer 종류

Scorer	설명	주요 판정 기준
Guidelines	사용자 정의 지침 준수 여부	프롬프트 기반 LLM-as-a-judge
Correctness	예상 정답과의 일치도	의미적 동등성 (Semantic Equivalence)
Safety	유해 콘텐츠 포함 여부	폭력, 혐오, 개인정보 등
RetrievalGroundedness	검색 결과에 근거한 답변인지	환각 (Hallucination) 감지

4-3. mlflow.genai.evaluate() 사용 예시

import mlflow
import mlflow.genai
from mlflow.genai.scorers import Guidelines, Correctness, Safety, RetrievalGroundedness
import pandas as pd

# 평가 데이터셋 준비
eval_dataset = pd.DataFrame([
    {
        "inputs": {"question": "Databricks의 Unity Catalog란 무엇인가요?"},
        "expected_outputs": "Unity Catalog는 Databricks의 통합 거버넌스 솔루션입니다.",
    },
    {
        "inputs": {"question": "Delta Lake의 ACID 트랜잭션을 설명해주세요."},
        "expected_outputs": "Delta Lake는 ACID 트랜잭션을 지원하여 데이터 일관성을 보장합니다.",
    },
])

# 평가할 앱 함수 정의 (이미 배포된 모델이나 로컬 함수 모두 가능)
def my_rag_app(inputs):
    question = inputs["question"]
    # 실제 앱 호출 로직
    answer = call_rag_pipeline(question)
    return {"answer": answer}

# 평가 실행
with mlflow.start_run(run_name="rag_eval_v1"):
    results = mlflow.genai.evaluate(
        data=eval_dataset,
        predict_fn=my_rag_app,
        scorers=[
            Guidelines(guidelines="답변은 한국어로 작성되어야 합니다."),
            Correctness(),
            Safety(),
            RetrievalGroundedness(),
        ],
    )

print(results.metrics)
# {'correctness/v1/mean': 0.85, 'safety/v1/mean': 1.0,
#  'retrieval_groundedness/v1/mean': 0.78, ...}

참고: mlflow.genai.evaluate() 는 MLflow 2.14 이상에서 지원합니다. 기존 mlflow.evaluate() 와 API 구조가 다릅니다.

5. 커스텀 Scorer 작성

기본 Scorer 로 커버되지 않는 도메인 특화 평가 기준은 @scorer 데코레이터를 사용하여 직접 정의합니다.

5-1. @scorer 데코레이터 기본 구조

from mlflow.genai.scorers import scorer
from mlflow.entities.assessment import Assessment

@scorer
def korean_language_check(inputs, outputs) -> Assessment:
    """응답이 한국어로만 작성되었는지 검사하는 커스텀 Scorer"""
    answer = outputs.get("answer", "")

    # 한글 문자 비율 계산
    korean_chars = sum(1 for c in answer if '\uAC00' <= c <= '\uD7A3')
    total_chars = len([c for c in answer if c.strip()])

    if total_chars == 0:
        return Assessment(name="korean_language_check", value=False, rationale="빈 응답")

    korean_ratio = korean_chars / total_chars
    passed = korean_ratio >= 0.5  # 50% 이상 한글이면 통과

    return Assessment(
        name="korean_language_check",
        value=passed,
        rationale=f"한글 비율: {korean_ratio:.1%}",
    )

5-2. LLM-as-a-Judge 패턴 커스텀 Scorer

import openai
from mlflow.genai.scorers import scorer
from mlflow.entities.assessment import Assessment

@scorer
def domain_relevance(inputs, outputs) -> Assessment:
    """응답이 Databricks/데이터 엔지니어링 도메인에 관련되는지 판단"""
    question = inputs.get("question", "")
    answer = outputs.get("answer", "")

    prompt = f"""다음 질문과 답변이 데이터 엔지니어링 또는 Databricks 플랫폼과 관련이 있는지 판단하세요.

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

판단 기준:
- PASS: Databricks, Apache Spark, Delta Lake, MLflow, Unity Catalog 등과 명확히 관련
- FAIL: 위 주제와 무관하거나 일반적인 잡담 수준의 답변

반드시 JSON 형식으로만 응답하세요:
{{"result": "PASS" or "FAIL", "reason": "판단 이유 한 문장"}}"""

    response = openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},
    )

    import json
    verdict = json.loads(response.choices[0].message.content)

    return Assessment(
        name="domain_relevance",
        value=(verdict["result"] == "PASS"),
        rationale=verdict["reason"],
    )

5-3. 커스텀 Scorer 등록 및 실행

# 여러 커스텀 Scorer 를 함께 사용하기
results = mlflow.genai.evaluate(
    data=eval_dataset,
    predict_fn=my_rag_app,
    scorers=[
        Correctness(),           # 기본 Scorer
        korean_language_check,   # 커스텀: 언어 확인
        domain_relevance,        # 커스텀: 도메인 연관성
    ],
)

6. 평가 데이터셋 구축

좋은 평가는 좋은 데이터셋에서 시작합니다. 평가 데이터셋은 크게 세 가지 방법으로 구축합니다.

6-1. Trace에서 데이터셋 추출

프로덕션 Trace 에서 고품질 사례를 자동으로 수집합니다.

import mlflow

# 사용자 피드백이 좋은 Trace 를 평가 데이터로 수집
good_traces = mlflow.search_traces(
    experiment_ids=["123456789"],
    filter_string="tag.user_score >= '4' AND status = 'OK'",
    max_results=500,
)

# 평가 데이터셋 형식으로 변환
eval_records = []
for _, row in good_traces.iterrows():
    request = row["request"]
    response = row["response"]
    eval_records.append({
        "inputs": request,
        "expected_outputs": response,  # 좋은 응답을 Gold Answer로 사용
        "source_trace_id": row["trace_id"],
    })

import pandas as pd
eval_df = pd.DataFrame(eval_records)
print(f"수집된 평가 데이터: {len(eval_df)}건")

6-2. 수동 레이블링 워크플로우

자동 수집만으로는 엣지 케이스를 충분히 커버하기 어렵습니다. 도메인 전문가가 직접 레이블링하는 과정이 필요합니다.

# 검토가 필요한 Trace 에 태그 붙이기
def flag_for_review(trace_id: str, reason: str):
    mlflow.set_trace_tag(trace_id, "review_status", "pending")
    mlflow.set_trace_tag(trace_id, "review_reason", reason)

# 낮은 점수를 받은 Trace 를 검토 대상으로 표시
low_score_traces = mlflow.search_traces(
    experiment_ids=["123456789"],
    filter_string="tag.user_score <= '2'",
)

for _, row in low_score_traces.iterrows():
    flag_for_review(row["trace_id"], "낮은 사용자 평점 - 전문가 검토 필요")

6-3. Gold Set 관리

Gold Set 은 정확성이 검증된 핵심 평가 데이터셋입니다. Spark DataFrame 또는 Delta Table 로 관리하면 버전 관리와 공유가 용이합니다.

from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

# Gold Set을 Delta Table로 저장
eval_df_spark = spark.createDataFrame(eval_df)
eval_df_spark.write.format("delta").mode("overwrite").saveAsTable(
    "ml_catalog.evaluation.rag_gold_set_v1"
)

# CI/CD 파이프라인에서 Gold Set 로드 후 평가 실행
gold_set = spark.table("ml_catalog.evaluation.rag_gold_set_v1").toPandas()
results = mlflow.genai.evaluate(
    data=gold_set,
    predict_fn=my_rag_app,
    scorers=[Correctness(), Safety()],
)

7. 프로덕션 모니터링

7-1. Trace 기반 품질 대시보드

Unity Catalog 시스템 테이블 (system.mlflow.traces) 을 Databricks SQL 로 쿼리하여 품질 대시보드를 구축합니다.

-- 일별 핵심 품질 지표 집계
SELECT
    DATE(TIMESTAMP_MILLIS(timestamp_ms))          AS day,
    COUNT(*)                                       AS total_traces,
    SUM(CASE WHEN status = 'ERROR' THEN 1 ELSE 0 END)   AS error_count,
    SUM(CASE WHEN status = 'ERROR' THEN 1 ELSE 0 END)
        * 100.0 / COUNT(*)                         AS error_rate_pct,
    AVG(execution_time_ms)                         AS avg_latency_ms,
    PERCENTILE(execution_time_ms, 0.95)            AS p95_latency_ms,
    AVG(CAST(tags['user_score'] AS DOUBLE))        AS avg_user_score
FROM system.mlflow.traces
WHERE experiment_id = '123456789'
    AND timestamp_ms >= UNIX_MILLIS(CURRENT_DATE() - INTERVAL 30 DAYS)
GROUP BY DATE(TIMESTAMP_MILLIS(timestamp_ms))
ORDER BY day DESC;

7-2. 드리프트 감지

이동 평균 (Moving Average) 비교로 품질 드리프트를 조기에 감지합니다.

import mlflow
import pandas as pd
import numpy as np

traces_df = mlflow.search_traces(
    experiment_ids=["123456789"],
    max_results=2000,
    order_by=["timestamp_ms ASC"],
)

# 7일 이동 평균 에러율 계산
traces_df["date"] = pd.to_datetime(traces_df["timestamp_ms"], unit="ms").dt.date
daily = traces_df.groupby("date").agg(
    total=("status", "count"),
    errors=("status", lambda x: (x == "ERROR").sum()),
).reset_index()

daily["error_rate"] = daily["errors"] / daily["total"]
daily["rolling_7d_error_rate"] = daily["error_rate"].rolling(7).mean()

# 최근 3일 평균이 기준선의 2배 이상이면 경고
baseline = daily["rolling_7d_error_rate"].iloc[:-3].mean()
recent = daily["rolling_7d_error_rate"].iloc[-3:].mean()

if recent > baseline * 2:
    print(f"[경고] 에러율 급증 감지! 기준선: {baseline:.1%}, 최근 3일: {recent:.1%}")

7-3. 알림 설정 (Databricks Workflow 활용)

Databricks Workflow (Jobs) 를 사용하여 주기적으로 품질 지표를 체크하고 임계값 초과 시 알림을 전송합니다.

# monitoring_job.py — Databricks Job Task로 등록하여 매시간 실행
import mlflow
import requests
import os

SLACK_WEBHOOK = os.environ["SLACK_WEBHOOK_URL"]

def send_alert(message: str):
    requests.post(SLACK_WEBHOOK, json={"text": message})

# 최근 1시간 Trace 에러율 확인
import time
one_hour_ago = int((time.time() - 3600) * 1000)

recent = mlflow.search_traces(
    experiment_ids=["123456789"],
    filter_string=f"timestamp_ms >= {one_hour_ago}",
    max_results=500,
)

if len(recent) > 0:
    error_rate = (recent["status"] == "ERROR").mean()
    if error_rate > 0.05:  # 에러율 5% 초과 시 알림
        send_alert(
            f":warning: RAG 앱 에러율 급증\n"
            f"최근 1시간 에러율: {error_rate:.1%}\n"
            f"총 요청 수: {len(recent)}"
        )

8. 베스트 프랙티스

8-1. 평가 주기 가이드

상황	권장 평가 주기	비고
모델 버전 업데이트	배포 전 반드시 실행	Gold Set 전체 평가
프롬프트 변경	변경 전후 비교 평가	A/B 테스트와 병행
프로덕션 정기 평가	주 1회	샘플링된 Trace 활용
드리프트 의심 시	즉시	최근 Trace 기반

8-2. Scorer 선택 가이드

시나리오별 Scorer 조합 추천:

[RAG 챗봇]
  - Correctness: 정답 일치도
  - RetrievalGroundedness: 환각 감지
  - Guidelines: 답변 형식/언어 준수

[에이전트 (Agent)]
  - Safety: 유해 행동 방지
  - Guidelines: 도구 사용 정책 준수
  - 커스텀 Scorer: 도메인 규칙 (예: 금융 규정 준수)

[코드 생성]
  - 커스텀 Scorer: 코드 실행 가능 여부 (subprocess 실행 테스트)
  - Guidelines: 보안 취약 패턴 금지 (eval, exec 등)

8-3. 비용 vs 품질 트레이드오프

LLM-as-a-Judge 방식의 Scorer 는 판단 품질이 높지만 호출 비용이 발생합니다.

# 비용 절감 전략 1: 샘플링 평가
# 전체 Trace 의 10% 만 평가하여 비용 절감
import random

all_traces = mlflow.search_traces(experiment_ids=["123456789"], max_results=1000)
sample_size = max(50, int(len(all_traces) * 0.1))
sampled = all_traces.sample(n=sample_size, random_state=42)

# 비용 절감 전략 2: 경량 모델로 빠른 1차 필터링
# gpt-4o-mini → 문제 있는 케이스만 gpt-4o 로 심층 평가

# 비용 절감 전략 3: 규칙 기반 Scorer 우선 적용
# 명백한 오류 (에러 상태, 빈 응답) 는 LLM 호출 없이 탈락
@scorer
def basic_sanity_check(inputs, outputs) -> Assessment:
    answer = outputs.get("answer", "")
    if not answer or len(answer.strip()) < 10:
        return Assessment(name="basic_sanity_check", value=False, rationale="빈 응답 또는 너무 짧은 응답")
    return Assessment(name="basic_sanity_check", value=True, rationale="기본 형식 통과")

8-4. Flywheel (플라이휠) 효과 구축

프로덕션 데이터를 평가에 활용하는 선순환 구조를 만드는 것이 최종 목표입니다.

프로덕션 트레이스 수집
        ↓
사용자 피드백 태그 연결 (mlflow.set_trace_tag)
        ↓
고품질 Trace → Gold Set 편입 (Delta Table)
        ↓
새 모델/프롬프트 버전 평가 (mlflow.genai.evaluate)
        ↓
개선된 버전 배포 → 더 나은 트레이스 생성
        ↓ (반복)

이 사이클을 자동화하면 시간이 지날수록 GenAI 애플리케이션의 품질이 지속적으로 향상 됩니다.

정리

핵심 개념	설명
Trace 분석	블랙박스 LLM 앱의 실행 흔적을 기록하고 Latency, 토큰, 에러 패턴을 분석합니다
search_traces()	태그, 상태, 시간 조건으로 Trace 를 Python DataFrame 으로 조회합니다
genai.evaluate()	평가 데이터셋과 Scorer 를 조합하여 LLM 앱을 자동 채점합니다
기본 Scorer	Guidelines, Correctness, Safety, RetrievalGroundedness 등을 제공합니다
커스텀 Scorer	`@scorer` 데코레이터로 도메인 특화 평가 기준을 직접 정의합니다
Gold Set	검증된 평가 데이터를 Delta Table 로 버전 관리하고 CI/CD 에 통합합니다
프로덕션 모니터링	시스템 테이블 + Workflow Job 으로 드리프트를 자동 감지하고 알림을 전송합니다
Flywheel 효과	프로덕션 트레이스 → 평가 데이터 → 모델 개선의 선순환을 구축합니다

시작하기

Databricks 플랫폼

레이크하우스 아키텍처

컴퓨트

데이터 엔지니어링

데이터 웨어하우징

Lakebase

Unity Catalog

AI/BI

Genie Space 운영

머신러닝

AI 에이전트

GenAI 도구

Databricks Apps

보안과 거버넌스

모범 사례

개발 도구

플랫폼 비교

부록 — 선행 지식

부록 — 데이터 기초

부록 — 참고

​4. MLflow Evaluation — mlflow.genai.evaluate()

​4-1. 기본 개념

​4-2. 기본 Scorer 종류

​4-3. mlflow.genai.evaluate() 사용 예시

​5. 커스텀 Scorer 작성

​5-1. @scorer 데코레이터 기본 구조

​5-2. LLM-as-a-Judge 패턴 커스텀 Scorer

​5-3. 커스텀 Scorer 등록 및 실행

​6. 평가 데이터셋 구축

​6-1. Trace에서 데이터셋 추출

​6-2. 수동 레이블링 워크플로우

​6-3. Gold Set 관리

​7. 프로덕션 모니터링

​7-1. Trace 기반 품질 대시보드

​7-2. 드리프트 감지

​7-3. 알림 설정 (Databricks Workflow 활용)

​8. 베스트 프랙티스

​8-1. 평가 주기 가이드

​8-2. Scorer 선택 가이드

​8-3. 비용 vs 품질 트레이드오프

​8-4. Flywheel (플라이휠) 효과 구축

​정리

​참고 링크