평가 데이터셋 설계와 CI/CD 통합

# 평가 데이터셋을 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")