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

3. 에이전트 호출 함수 정의

mlflow.genai.evaluate()에 전달할 predict 함수를 정의합니다. 이 함수는 inputs dict를 받아서 에이전트의 응답 문자열을 반환해야 합니다.

Serving Endpoint 호출 방식

배포된 엔드포인트를 호출하는 방식입니다. 프로덕션 환경의 실제 성능을 측정할 때 사용합니다.

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

`response.choices[0].message.content` 구조 설명

Serving Endpoint는 OpenAI ChatCompletion API와 동일한 응답 구조를 따릅니다:

필드	설명
`response.choices`	생성된 응답의 리스트. 보통 1개 (`n=1`이 기본)
`response.choices[0].message`	첫 번째 응답의 메시지 객체
`.message.role`	`"assistant"` (항상 동일)
`.message.content`	에이전트가 생성한 실제 답변 텍스트

Agent 직접 호출 방식 (개발 중 테스트)

배포 전에 로컬에서 에이전트 코드를 직접 호출하여 빠르게 테스트할 때 사용합니다.

from my_agent import KnowledgeAssistant

agent = KnowledgeAssistant()

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

4. 평가 실행

기본 실행 (내장 스코어러 3종)

아래 코드는 평가 데이터셋의 각 질문에 대해 predict_fn을 호출하고, 3가지 내장 스코어러로 응답 품질을 채점합니다. 결과는 자동으로 현재 MLflow Experiment에 기록됩니다.

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

결과 테이블 확인

개별 질문별 점수를 DataFrame으로 확인할 수 있습니다. 낮은 점수를 받은 항목을 집중적으로 분석하세요.

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

5. 내장 스코어러 상세

스코어러	측정 대상	점수 범위	설명
`Correctness()`	정답 일치도	1~5	expected_response 대비 정확성
`Relevance()`	질문 관련성	1~5	답변이 질문에 적절한지
`Safety()`	안전성	0 또는 1	유해 콘텐츠 여부
`Faithfulness()`	근거 충실도	1~5	retrieved_context 대비 환각 여부
`Guidelines()`	커스텀 기준	1~5	사용자 정의 가이드라인 준수

Faithfulness 사용 (RAG 평가)

retrieved_context가 있어야 동작합니다. RAG 파이프라인에서 검색된 문서를 EvaluationResult에 함께 반환해야 합니다.

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

커스텀 Guidelines 스코어러

조직별 답변 규칙(한국어 작성, SQL 포함 등)을 Guidelines로 정의하면 자동으로 채점됩니다.

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

6. 커스텀 스코어러 작성

내장 스코어러로 부족할 때 직접 만들 수 있습니다.

`@scorer` 데코레이터 동작 원리

@scorer 데코레이터를 붙인 함수는 MLflow 평가 프레임워크가 자동으로 호출합니다. 함수의 파라미터 이름에 따라 자동으로 값이 주입됩니다:

inputs: 평가 데이터셋의 입력 값 (eval_dataset[i]["inputs"])
outputs: predict_fn이 반환한 에이전트 응답 문자열
expectations: 평가 데이터셋의 기대 값 (eval_dataset[i]["expectations"])

반환값은 {"score": 숫자, "justification": "이유 문자열"} 형식의 dict입니다.

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

7. Lakeflow Job으로 자동화

평가를 주기적으로 실행하려면 노트북을 만들어 Lakeflow Job에 등록합니다. 이렇게 하면 모델 변경이나 데이터 업데이트 후에도 품질을 지속적으로 모니터링할 수 있습니다.

평가 노트북 (eval_notebook.py)

아래 노트북은 평가 데이터셋 로드 -> 에이전트 호출 -> 평가 실행 -> 결과를 Delta 테이블에 기록하는 전 과정을 자동화합니다.

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

Job 등록 (Databricks CLI)

아래 명령은 매일 오전 9시(한국 시간)에 평가 노트북을 자동 실행하는 Lakeflow Job을 생성합니다.

Cron 표현식 해석: `0 0 9 * * ?`

필드	값	의미
초	`0`	0초
분	`0`	0분
시	`9`	오전 9시
일	`*`	매일
월	`*`	매월
요일	`?`	요일 무관

즉, **매일 오전 9시 0분 0초 (Asia/Seoul 시간대)**에 실행됩니다.

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

트러블슈팅

인증 실패: `PERMISSION_DENIED` 또는 `401 Unauthorized`

증상	원인	해결 방법
`PERMISSION_DENIED` on endpoint query	Serving Endpoint에 대한 호출 권한 없음	Endpoint 설정 > Permissions에서 사용자/SP에 `CAN_QUERY` 부여
`401 Unauthorized`	`WorkspaceClient()` 인증 실패	노트북에서는 자동 인증. 로컬에서는 `databricks configure --profile` 실행 후 `DATABRICKS_CONFIG_PROFILE` 환경변수 설정
`mlflow.genai` 모듈 없음	MLflow 2.x 사용 중	`pip install --upgrade mlflow>=3.0`

타임아웃: `TimeoutError` 또는 응답이 매우 느림

증상	원인	해결 방법
평가 실행이 30분 이상 소요	평가 건수가 너무 많거나 Endpoint가 느림	데이터셋을 50건 이하로 줄이고 테스트. Endpoint 스케일 아웃 확인
개별 호출에서 `TimeoutError`	Endpoint cold start 또는 모델 로딩 지연	평가 전에 warm-up 호출 1~2건 실행. Endpoint의 `min_provisioned_throughput` 설정 확인
`Rate limit exceeded`	단시간에 너무 많은 요청	`predict_fn`에 `time.sleep(0.5)` 추가하여 호출 간격 조절

점수가 비정상적으로 낮을 때

correctness/mean이 0.3 이하로 나온다면 에이전트 문제가 아니라 평가 데이터셋 문제일 수 있습니다:

expected_response가 너무 구체적이면 의미는 맞지만 표현이 다른 답변에 낮은 점수를 줌
핵심 키워드 중심으로 기대 답변을 작성하고, 세부 표현은 유연하게 허용하세요

평가 품질 개선 팁

팁	설명
평가 셋 다양성	쉬운 질문·어려운 질문·엣지 케이스를 골고루 포함
정기 업데이트	새 문서가 추가되면 평가 셋도 함께 갱신
A/B 비교	프롬프트 변경 전후로 동일 데이터셋에 평가 실행
임계값 알림	correctness_mean < 0.7이면 Slack 알림 전송
Trace 분석	낮은 점수 건은 MLflow Trace에서 검색→생성 과정 추적

개요

Platform Setup

Data Engineering

한국어 RAG & Vector Search

Genie Space 운영

Apps & Lakebase

AI Agent 평가

Delta Lake 성능

부록 — Claude Code 사용 가이드

모범 사례

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

3. 에이전트 호출 함수 정의

Serving Endpoint 호출 방식

`response.choices[0].message.content` 구조 설명

Agent 직접 호출 방식 (개발 중 테스트)

4. 평가 실행

기본 실행 (내장 스코어러 3종)

결과 테이블 확인

5. 내장 스코어러 상세

Faithfulness 사용 (RAG 평가)

커스텀 Guidelines 스코어러

6. 커스텀 스코어러 작성

`@scorer` 데코레이터 동작 원리

7. Lakeflow Job으로 자동화

평가 노트북 (eval_notebook.py)

Job 등록 (Databricks CLI)

Cron 표현식 해석: `0 0 9 * * ?`

트러블슈팅

인증 실패: `PERMISSION_DENIED` 또는 `401 Unauthorized`

타임아웃: `TimeoutError` 또는 응답이 매우 느림

점수가 비정상적으로 낮을 때

평가 품질 개선 팁

​3. 에이전트 호출 함수 정의

​Serving Endpoint 호출 방식

​response.choices[0].message.content 구조 설명

​Agent 직접 호출 방식 (개발 중 테스트)

​4. 평가 실행

​기본 실행 (내장 스코어러 3종)

​결과 테이블 확인

​5. 내장 스코어러 상세

​Faithfulness 사용 (RAG 평가)

​커스텀 Guidelines 스코어러

​6. 커스텀 스코어러 작성

​@scorer 데코레이터 동작 원리

​7. Lakeflow Job으로 자동화

​평가 노트북 (eval_notebook.py)

​Job 등록 (Databricks CLI)

​Cron 표현식 해석: 0 0 9 * * ?

​트러블슈팅

​인증 실패: PERMISSION_DENIED 또는 401 Unauthorized

​타임아웃: TimeoutError 또는 응답이 매우 느림

​점수가 비정상적으로 낮을 때

​평가 품질 개선 팁

3. 에이전트 호출 함수 정의

Serving Endpoint 호출 방식

`response.choices[0].message.content` 구조 설명

Agent 직접 호출 방식 (개발 중 테스트)

4. 평가 실행

기본 실행 (내장 스코어러 3종)

결과 테이블 확인

5. 내장 스코어러 상세

Faithfulness 사용 (RAG 평가)

커스텀 Guidelines 스코어러

6. 커스텀 스코어러 작성

`@scorer` 데코레이터 동작 원리

7. Lakeflow Job으로 자동화

평가 노트북 (eval_notebook.py)

Job 등록 (Databricks CLI)

Cron 표현식 해석: `0 0 9 * * ?`

트러블슈팅

인증 실패: `PERMISSION_DENIED` 또는 `401 Unauthorized`

타임아웃: `TimeoutError` 또는 응답이 매우 느림

점수가 비정상적으로 낮을 때

평가 품질 개선 팁