모델 평가

import mlflow

results = mlflow.genai.evaluate(
    data=eval_dataset,              # 평가 데이터셋
    predict_fn=my_agent.predict,    # 평가 대상 (에이전트/모델)
    scorers=[...]                   # 평가 기준 (Scorer)
)

eval_data = [
    {
        "inputs": {"question": "반품 정책이 뭔가요?"},
        "expectations": {"expected_response": "구매 후 30일 이내 무료 반품 가능합니다."}
    },
    {
        "inputs": {"question": "배송 기간은 얼마나 걸리나요?"},
        "expectations": {"expected_response": "일반 배송 2~3 영업일, 특급 배송 당일~익일입니다."}
    },
    {
        "inputs": {"question": "회원 등급 기준이 어떻게 되나요?"},
        "expectations": {"expected_response": "연간 구매 금액 기준: Silver 50만원, Gold 200만원, Platinum 500만원 이상입니다."}
    }
]

import mlflow

results = mlflow.genai.evaluate(
    data=eval_data,
    predict_fn=my_agent.predict,
    scorers=[
        # 정확도: 기대 답변과 비교
        mlflow.genai.scorers.Correctness(),

        # 안전성: 유해 콘텐츠 검사
        mlflow.genai.scorers.Safety(),

        # 근거성: 검색 결과에 기반한 답변인지
        mlflow.genai.scorers.RetrievalGroundedness(),

        # 커스텀 가이드라인
        mlflow.genai.scorers.Guidelines(
            guidelines=[
                "답변은 반드시 한국어로 작성되어야 합니다",
                "답변에 출처를 명시해야 합니다",
                "모르는 질문에는 '확인 후 답변드리겠습니다'라고 답변해야 합니다",
                "경쟁사 제품을 추천하면 안 됩니다"
            ]
        )
    ]
)

from mlflow.genai.scorers import scorer

@scorer
def korean_check(request, response):
    """답변이 한국어인지 확인하는 커스텀 Scorer"""
    import re
    korean_ratio = len(re.findall('[가-힣]', response.content)) / max(len(response.content), 1)
    return {
        "score": 1.0 if korean_ratio > 0.5 else 0.0,
        "justification": f"한국어 비율: {korean_ratio:.1%}"
    }

@scorer
def length_check(request, response):
    """답변 길이가 적절한지 확인하는 커스텀 Scorer"""
    length = len(response.content)
    return {
        "score": 1.0 if 50 <= length <= 500 else 0.0,
        "justification": f"답변 길이: {length}자 ({'적절' if 50 <= length <= 500 else '부적절'})"
    }

# 커스텀 Scorer 사용
results = mlflow.genai.evaluate(
    data=eval_data,
    predict_fn=my_agent.predict,
    scorers=[korean_check, length_check, mlflow.genai.scorers.Safety()]
)

# 전체 메트릭 요약
print(results.metrics)
# {'correctness/mean': 0.85, 'safety/mean': 0.98, 'guidelines/mean': 0.72}

# 개별 평가 결과 테이블
display(results.tables["eval_results"])
# 각 질문별로 점수, 판단 근거(justification)가 표시됩니다

# Correctness Scorer는 의미적 유사성을 평가합니다
# "30일 이내 무료 반품" vs "한 달 내 반품 가능" → ✅ 의미적으로 일치
mlflow.genai.scorers.Correctness()

# RAG 파이프라인에서 환각을 감지합니다
mlflow.genai.scorers.RetrievalGroundedness()
# 내부적으로 각 응답 문장을 retrived context와 대조합니다

# 비즈니스 규칙을 자연어로 정의합니다
mlflow.genai.scorers.Guidelines(
    guidelines=[
        "답변은 200자를 초과하지 않아야 합니다",
        "경쟁사 제품을 비교하거나 추천하면 안 됩니다",
        "가격 정보를 언급할 때는 반드시 '부가세 별도'를 명시해야 합니다",
        "모르는 질문에는 추측하지 말고 '확인 후 답변드리겠습니다'라고 해야 합니다"
    ]
)

> 💡 **Judge 모델 커스터마이징**: 기본 Judge 모델 대신 GPT-4o나 자체 파인튜닝된 모델을 사용할 수도 있습니다. 평가의 도메인 특화 정확도를 높이려면 Judge 모델 선택이 중요합니다.

```python
# Judge 모델을 명시적으로 지정
import mlflow

mlflow.genai.scorers.Correctness(
    # 기본값: Databricks의 Foundation Model
    # 필요시 커스텀 엔드포인트를 환경 변수로 설정
)

# 환경 변수로 Judge 모델 변경
import os
os.environ["MLFLOW_DEPLOYMENTS_TARGET"] = "databricks"

# 평가 데이터셋을 Delta 테이블로 중앙 관리
eval_df = spark.createDataFrame(eval_data)
eval_df.write.mode("overwrite").saveAsTable("catalog.ml.agent_eval_dataset_v1")

# 버전 관리: 데이터셋 변경 이력 추적
spark.sql("DESCRIBE HISTORY catalog.ml.agent_eval_dataset_v1")

# 이전 버전의 데이터셋으로 평가
eval_data_v0 = spark.read.option("versionAsOf", 0) \
    .table("catalog.ml.agent_eval_dataset_v1")

# 프로덕션 트레이스에서 평가 데이터셋 추출
production_traces = spark.table("catalog.ml.agent_inference_logs")

# 최근 7일의 실제 질문에서 평가 데이터 생성
new_eval_data = production_traces.filter(
    "timestamp >= DATE_SUB(CURRENT_DATE(), 7)"
).select(
    col("request.question").alias("inputs"),
    col("response.answer").alias("actual_response")
).limit(100)

# 전문가 검토 후 expected_response 추가 → 평가 데이터셋에 병합

# ci_evaluate.py — CI/CD 파이프라인에서 실행
import mlflow
import sys

# 평가 데이터셋 로드
eval_data = spark.table("catalog.ml.agent_eval_dataset_v1").toPandas().to_dict("records")

# 에이전트 로드
agent = mlflow.pyfunc.load_model("models:/catalog.ml.customer_agent@challenger")

# 평가 실행
results = mlflow.genai.evaluate(
    data=eval_data,
    predict_fn=agent.predict,
    scorers=[
        mlflow.genai.scorers.Correctness(),
        mlflow.genai.scorers.Safety(),
        mlflow.genai.scorers.RetrievalGroundedness(),
        mlflow.genai.scorers.Guidelines(
            guidelines=["답변은 한국어로 작성되어야 합니다"]
        )
    ]
)

# 품질 게이트: 임계값 확인
thresholds = \{
    "correctness/mean": 0.80,
    "safety/mean": 0.95,
    "retrieval_groundedness/mean": 0.85,
\}

failed = []
for metric, threshold in thresholds.items():
    actual = results.metrics.get(metric, 0)
    if actual \< threshold:
        failed.append(f"\{metric\}: \{actual:.2f\} \< \{threshold:.2f\}")

if failed:
    print("❌ Quality gate FAILED:")
    for f in failed:
        print(f"  - \{f\}")
    sys.exit(1)  # CI/CD 파이프라인 실패
else:
    print("✅ Quality gate PASSED — ready for deployment")
    # champion alias 업데이트
    client = mlflow.MlflowClient()
    challenger_version = client.get_model_version_by_alias(
        "catalog.ml.customer_agent", "challenger"
    ).version
    client.set_registered_model_alias(
        "catalog.ml.customer_agent", "champion", challenger_version
    )

# databricks.yml — 평가 작업 정의
resources:
  jobs:
    agent_evaluation:
      name: "agent-quality-gate"
      tasks:
        - task_key: evaluate
          notebook_task:
            notebook_path: ./notebooks/ci_evaluate
          existing_cluster_id: $\{var.cluster_id\}

      # 트리거: 모델이 등록되면 자동 실행
      trigger:
        pause_status: UNPAUSED

# 평가 결과를 Delta 테이블에 저장 → 대시보드로 추적
import pandas as pd
from datetime import datetime

eval_record = \{
    "timestamp": datetime.now(),
    "model_version": "challenger_v5",
    "correctness_mean": results.metrics["correctness/mean"],
    "safety_mean": results.metrics["safety/mean"],
    "groundedness_mean": results.metrics["retrieval_groundedness/mean"],
    "eval_dataset_size": len(eval_data),
    "passed": len(failed) == 0
\}

spark.createDataFrame([eval_record]).write.mode("append") \
    .saveAsTable("catalog.ml.evaluation_history")