RAG 평가

검색 품질 (Retrieval)     +    생성 품질 (Generation)    =    전체 RAG 품질
- 관련 문서를 찾았는가?         - 답변이 정확한가?
- 불필요한 문서가 포함?         - 환각이 없는가?
                                - 출처와 일치하는가?

import pandas as pd

eval_dataset = pd.DataFrame([
    {
        "request": "Databricks Vector Search의 최대 벡터 차원은?",
        "expected_response": "최대 4,096차원까지 지원합니다.",
    },
    {
        "request": "Delta Sync Index와 Direct Vector Access Index의 차이는?",
        "expected_response": "Delta Sync Index는 소스 Delta Table과 자동 동기화되며, Direct Vector Access Index는 REST API로 수동 관리합니다.",
    },
    {
        "request": "Vector Search 엔드포인트당 최대 인덱스 수는?",
        "expected_response": "엔드포인트당 최대 50개 인덱스를 생성할 수 있습니다.",
    },
    {
        "request": "하이브리드 검색에서 사용하는 결과 병합 알고리즘은?",
        "expected_response": "Reciprocal Rank Fusion(RRF) 알고리즘을 사용합니다.",
    },
    {
        "request": "Foundation Model API에서 제공하는 임베딩 모델은?",
        "expected_response": "databricks-gte-large-en과 databricks-bge-large-en 모델을 제공합니다.",
    }
])

import mlflow

# 모델이 이미 로깅된 경우
model_uri = "models:/catalog.schema.rag_agent/1"

results = mlflow.evaluate(
    model=model_uri,
    data=eval_dataset,
    model_type="databricks-agent",
)

# 전체 메트릭 확인
print(results.metrics)

# 답변이 이미 포함된 데이터셋
eval_with_responses = pd.DataFrame([
    {
        "request": "Vector Search의 최대 벡터 차원은?",
        "response": "Databricks Vector Search는 최대 4,096차원의 벡터를 지원합니다.",
        "retrieved_context": [
            {"content": "임베딩 차원은 최대 4,096까지 지원됩니다."}
        ],
        "expected_response": "최대 4,096차원까지 지원합니다."
    }
])

results = mlflow.evaluate(
    data=eval_with_responses,
    model_type="databricks-agent",
)

# 요청별 상세 결과 확인
display(results.tables["eval_results"])

eval_table = results.tables["eval_results"]
low_faith = eval_table[eval_table["response/llm_judged/faithfulness/rating"] == "no"]
low_rel = eval_table[eval_table["response/llm_judged/relevance_to_query/rating"] == "no"]
print(f"환각 의심: {len(low_faith)}건 | 관련성 부족: {len(low_rel)}건")

1. 평가 데이터셋 준비 (50~100개 질문-정답 쌍)
     │
2. RAG 체인 실행 → 답변 + 검색 결과 수집
     │
3. MLflow Evaluate 실행 → 메트릭 산출
     │
4. 결과 분석 → 취약점 식별
     │
5. 개선 적용 (청킹/프롬프트/모델 변경)
     │
6. 재평가 → 개선 효과 확인
     │
7. (반복) 목표 메트릭 달성 시 배포

from langchain_databricks import ChatDatabricks

llm = ChatDatabricks(endpoint="databricks-meta-llama-3-3-70b-instruct")

def generate_eval_questions(chunk: str, num_questions: int = 3) -> list:
    """청크 텍스트에서 평가용 질문-정답 쌍을 자동 생성"""
    prompt = f"""다음 텍스트를 읽고, 이 텍스트로 답변할 수 있는 질문-정답 쌍을 {num_questions}개 생성하세요.

텍스트:
{chunk}

JSON 형식으로 출력하세요:
[{{"question": "질문", "answer": "정답"}}]"""

    response = llm.invoke(prompt)
    return response.content

import mlflow

configs = [
    {"name": "baseline", "chunk_size": 500, "model": "gte-large-en", "k": 3},
    {"name": "larger_chunks", "chunk_size": 1000, "model": "gte-large-en", "k": 3},
    {"name": "more_results", "chunk_size": 500, "model": "gte-large-en", "k": 7},
    {"name": "korean_optimized", "chunk_size": 800, "model": "multilingual-e5", "k": 5},
]

for config in configs:
    with mlflow.start_run(run_name=config["name"]):
        # 설정에 따라 RAG 체인 구성
        chain = build_rag_chain(** config)

        # 평가 실행
        results = mlflow.evaluate(
            model=chain,
            data=eval_dataset,
            model_type="databricks-agent",
        )

        # 설정 파라미터 로깅
        mlflow.log_params(config)
        print(f"{config['name']}: {results.metrics}")