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

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

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, ...}

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%}",
    )

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"],
    )

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

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)}건")

# 검토가 필요한 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"], "낮은 사용자 평점 - 전문가 검토 필요")

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()],
)

-- 일별 핵심 품질 지표 집계
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;

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%}")

# 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)}"
        )

시나리오별 Scorer 조합 추천:

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

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

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

# 비용 절감 전략 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="기본 형식 통과")

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