Agent Bricks 평가 파이프라인 — 심화

import mlflow
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

def predict_fn(inputs: dict) -> str:
    """Knowledge Assistant Serving Endpoint 호출"""
    response = w.serving_endpoints.query(
        name="my-ka-endpoint",
        messages=[{"role": "user", "content": inputs["query"]}],
    )
    # response 구조: OpenAI ChatCompletion 형식
    #   response.choices       → 응답 목록 (보통 1개)
    #   response.choices[0]    → 첫 번째 응답
    #   .message.content       → 실제 텍스트 답변
    return response.choices[0].message.content

from my_agent import KnowledgeAssistant

agent = KnowledgeAssistant()

def predict_fn(inputs: dict) -> str:
    result = agent.invoke(inputs["query"])
    return result["output"]

import mlflow

results = mlflow.genai.evaluate(
    data=eval_dataset,         # 평가 데이터셋
    predict_fn=predict_fn,     # 에이전트 호출 함수
    scorers=[                  # 채점 기준
        mlflow.genai.scorers.Correctness(),   # 정답 일치도
        mlflow.genai.scorers.Relevance(),     # 질문 관련성
        mlflow.genai.scorers.Safety(),        # 유해 콘텐츠 여부
    ],
)

# 요약 메트릭 출력
print(results.metrics)
# {'correctness/mean': 0.82, 'relevance/mean': 0.91, 'safety/mean': 1.0, ...}

results_df = results.tables["eval_results"]
results_df[["inputs", "outputs", "correctness/score", "relevance/score"]].head(10)

def predict_fn_with_context(inputs: dict) -> mlflow.genai.EvaluationResult:
    """RAG 파이프라인: 검색 + 생성"""
    # 1) 검색 — 질문과 관련된 문서 상위 3건 조회
    docs = retriever.search(inputs["query"], top_k=3)
    context = "\n".join([d.page_content for d in docs])

    # 2) 생성 — 검색 결과를 참고하여 답변 생성
    answer = llm.invoke(f"Context: {context}\nQuestion: {inputs['query']}")

    return mlflow.genai.EvaluationResult(
        response=answer,
        retrieved_context=[{"content": d.page_content} for d in docs],
    )

results = mlflow.genai.evaluate(
    data=eval_dataset,
    predict_fn=predict_fn_with_context,
    scorers=[
        mlflow.genai.scorers.Faithfulness(),  # 검색 결과 대비 환각 여부
        mlflow.genai.scorers.Correctness(),
        mlflow.genai.scorers.Relevance(),
    ],
)

custom_scorer = mlflow.genai.scorers.Guidelines(
    guidelines=[
        "답변은 반드시 한국어로 작성해야 한다.",
        "SQL 예시를 포함해야 한다.",
        "Databricks 공식 문서 링크를 인용해야 한다.",
    ],
    name="custom_format_check",
)

results = mlflow.genai.evaluate(
    data=eval_dataset,
    predict_fn=predict_fn,
    scorers=[custom_scorer],
)

from mlflow.genai.scorers import scorer

@scorer
def response_length_check(inputs, outputs):
    """답변 길이가 50자 이상인지 확인"""
    text = outputs or ""
    is_long_enough = len(text) >= 50
    return {
        "score": 1 if is_long_enough else 0,        # 1=통과, 0=실패
        "justification": f"응답 길이: {len(text)}자",  # 채점 근거
    }

@scorer
def contains_sql_check(inputs, outputs):
    """SQL 키워드가 포함되어 있는지 확인"""
    sql_keywords = ["SELECT", "CREATE", "INSERT", "UPDATE", "DELETE", "GRANT"]
    text = (outputs or "").upper()
    found = any(kw in text for kw in sql_keywords)
    return {
        "score": 1 if found else 0,
        "justification": f"SQL 키워드 포함: {found}",
    }

results = mlflow.genai.evaluate(
    data=eval_dataset,
    predict_fn=predict_fn,
    scorers=[
        mlflow.genai.scorers.Correctness(),  # 내장 스코어러
        response_length_check,                # 커스텀 스코어러 1
        contains_sql_check,                   # 커스텀 스코어러 2
    ],
)

# Databricks notebook source
import mlflow
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

# 1) 데이터셋 로드
df = spark.table("my_catalog.eval.ka_eval_dataset").toPandas()
eval_dataset = [
    {
        "inputs": {"query": row["question"]},
        "expectations": {"expected_response": row["expected_answer"]},
    }
    for _, row in df.iterrows()
]

# 2) predict 함수
def predict_fn(inputs: dict) -> str:
    response = w.serving_endpoints.query(
        name="my-ka-endpoint",
        messages=[{"role": "user", "content": inputs["query"]}],
    )
    return response.choices[0].message.content

# 3) 평가 실행
with mlflow.start_run(run_name="scheduled-ka-eval"):
    results = mlflow.genai.evaluate(
        data=eval_dataset,
        predict_fn=predict_fn,
        scorers=[
            mlflow.genai.scorers.Correctness(),
            mlflow.genai.scorers.Relevance(),
            mlflow.genai.scorers.Safety(),
        ],
    )

    # 4) 핵심 메트릭을 Delta 테이블에 기록 (추이 분석용)
    import pandas as pd
    from datetime import datetime

    metrics_row = {
        "eval_time": datetime.now().isoformat(),
        "correctness_mean": results.metrics.get("correctness/mean"),
        "relevance_mean": results.metrics.get("relevance/mean"),
        "safety_mean": results.metrics.get("safety/mean"),
        "num_examples": len(eval_dataset),
    }
    spark.createDataFrame(
        pd.DataFrame([metrics_row])
    ).write.mode("append").saveAsTable("my_catalog.eval.ka_eval_history")

    print(f"평가 완료: {metrics_row}")

# Lakeflow Job 생성 — 매일 오전 9시 평가 실행
databricks jobs create --json '{
  "name": "KA-Daily-Eval",
  "tasks": [{
    "task_key": "run_eval",
    "notebook_task": {
      "notebook_path": "/Workspace/Users/me/eval_notebook"
    },
    "new_cluster": {
      "spark_version": "16.2.x-scala2.12",
      "num_workers": 0,
      "node_type_id": "i3.xlarge"
    }
  }],
  "schedule": {
    "quartz_cron_expression": "0 0 9 * * ?",
    "timezone_id": "Asia/Seoul"
  }
}'