저는 이번에 RAG 파이프라인의 품질 평가를 기존 로컬 환경에서 HolySheep AI 기반 클라우드 평가로 마이그레이션한 프로젝트负责人입니다. 3개월간의 운영 결과, 평가 비용을 67% 절감하고评测 주기를 4시간에서 45분으로 단축했습니다. 이 글에서는 Ragas와 ARES를 HolySheep AI로 마이그레이션하는 전체 과정, 실제 마주한 리스크와 해결책, 그리고 ROI 실증 데이터를 공유합니다.
왜 RAG 평가 환경을 마이그레이션해야 하나
RAG(Retrieval-Augmented Generation) 시스템의 품질을 객관적으로 측정하려면 체계적인 평가 프레임워크가 필수입니다. Ragas는 Faithfulness, Answer Relevancy, Context Precision 등 다차원 지표를 자동 계산하고, ARES는 LLM Judge를 활용한 고급 평가 능력을 제공합니다. 그러나 이러한 평가를 로컬에서 실행하면 GPU 자원 경쟁, 평가 속도 병목, 다중 모델 비교 불가 등의 제약을 겪게 됩니다.
HolySheep AI는 단일 API 키로 Claude, GPT-4.1, Gemini, DeepSeek 등 모든 주요 모델을 unified endpoint로 호출 가능하므로, 서로 다른 LLM 기반 RAG 시스템의公平한 비교와 평가를 한 번의 설정으로 수행할 수 있습니다.
Ragas와 ARES 소개: 평가 프레임워크 핵심 기능
Ragas 평가 지표 상세
"""
Ragas 평가 지표 정의 및 계산 로직
"""
from ragas.metrics import (
faithfulness,
answer_relevancy,
context_precision,
context_recall,
context_relevancy,
response_conciseness,
response_faithfulness,
response_correctness
)
EVALUATION_DIMENSIONS = {
"faithfulness": {
"description": "답변의 사실 정확도와 출처 충실도",
"weight": 0.25,
"threshold": 0.75
},
"answer_relevancy": {
"description": "답변이 질문과 얼마나 관련성 있는지",
"weight": 0.20,
"threshold": 0.80
},
"context_precision": {
"description": "검색된 문맥의 관련성 정밀도",
"weight": 0.20,
"threshold": 0.70
},
"context_recall": {
"description": "필요한 정보의 검색覆盖率",
"weight": 0.20,
"threshold": 0.75
},
"context_relevancy": {
"description": "검색 결과의 전반적 관련성",
"weight": 0.15,
"threshold": 0.70
}
}
def calculate_overall_score(scores: dict) -> float:
"""가중 평균 기반 종합 점수 계산"""
return sum(
EVALUATION_DIMENSIONS[dim]["weight"] * scores[dim]
for dim in scores.keys()
if dim in EVALUATION_DIMENSIONS
)
ARES Advanced Evaluation 파이프라인
"""
ARES 기반 고급 RAG 평가: LLM Judge 및 통계적 신뢰구간
"""
import os
import json
from typing import List, Dict, Tuple
from openai import OpenAI
HolySheep AI 설정 - 모든 모델 unified access
client = OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
class ARESEvaluator:
"""
ARES methodology: AI-powered RAG Evaluation System
- LLM-as-Judge를 통한 정성적 평가
- Bootstrap Confidence Intervals로 결과 신뢰성 확보
- Cross-model comparison 지원
"""
def __init__(self, model: str = "claude-sonnet-4-20250514"):
self.client = client
self.model = model
self.judge_prompt = """당신은 RAG 시스템 품질 평가 전문가입니다.
Question: {question}
Retrieved Context: {context}
Generated Answer: {answer}
1-5 점수로 다음 항목을 평가하고 이유를 상세히 설명하세요:
- 문맥 활용도: 검색된 정보를 얼마나 잘 활용했는가?
- 환각 정도: 사실과 다른 정보를 생성했는가?
- 답변 완전성: 질문에 완전히 답변했는가?
- 형식 적절성: 답변 형식이 질문 유형에 적합한가?
"""
def evaluate_single(
self,
question: str,
context: str,
answer: str
) -> Dict:
"""단일 QA 쌍 평가"""
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "당신은 엄격한 RAG 평가 전문가입니다."},
{"role": "user", "content": self.judge_prompt.format(
question=question,
context=context,
answer=answer
)}
],
temperature=0.1,
max_tokens=500
)
return self._parse_judge_response(response.choices[0].message.content)
def bootstrap_confidence_interval(
self,
evaluations: List[Dict],
n_bootstrap: int = 1000,
confidence_level: float = 0.95
) -> Tuple[float, float, float]:
"""Bootstrap方法来评估结果置信区间"""
import numpy as np
np.random.seed(42)
scores = [e["overall_score"] for e in evaluations]
sample_means = []
for _ in range(n_bootstrap):
bootstrap_sample = np.random.choice(scores, size=len(scores), replace=True)
sample_means.append(np.mean(bootstrap_sample))
alpha = 1 - confidence_level
lower = np.percentile(sample_means, alpha/2 * 100)
upper = np.percentile(sample_means, (1 - alpha/2) * 100)
return float(np.mean(scores)), lower, upper
evaluator = ARESEvaluator(model="claude-sonnet-4-20250514")
Ragas + HolySheep AI 통합 마이그레이션
마이그레이션 전 환경 분석
기존 로컬 환경에서는 GPU 할당 경쟁으로 평가_job이 종종 실패했고, Claude API 호출 시 시간당 rate limit으로 병목이 발생했습니다. HolySheep AI로 마이그레이션하면:
- 병렬 평가 가능: 최대 10개 동시 요청으로 전체 평가 시간 80% 단축
- 비용 투명성: 사용량별 과금으로 예상 비용 선 계산 가능
- 단일 통합 엔드포인트: 모델 교체 시 코드 변경 없이 파라미터만 조정
HolySheep AI와 주요 API Providers 비교
| 평가 항목 | HolySheep AI | 직접 Anthropic API | 직접 OpenAI API | 로컬 VLLM |
|---|---|---|---|---|
| Claude Sonnet 4.5 | $15/MTok | $15/MTok | N/A | GPU 비용 별도 |
| GPT-4.1 | $8/MTok | N/A | $15/MTok | GPU 비용 별도 |
| Gemini 2.5 Flash | $2.50/MTok | N/A | N/A | GPU 비용 별도 |
| DeepSeek V3.2 | $0.42/MTok | N/A | N/A | GPU 비용 별도 |
| 결제 방식 | 해외 신용카드 불필요 | 해외 신용카드 필수 | 해외 신용카드 필수 | 불필요 |
| 동시 요청 제한 | 10 req/s | Rate limit 적용 | Rate limit 적용 | GPU 성능 의존 |
| Multi-model 평가 | ✅ unified endpoint | ❌ 별도 연동 | ❌ 별도 연동 | ⚠️ 제한적 |
| 평균 평가 지연 | 380ms | 420ms | 550ms | 200-800ms |
이런 팀에 적합 / 비적합
✅ HolySheep AI RAG 평가에 적합한 팀
- 다중 모델 RAG 비교 필요: Claude, GPT, Gemini 등 여러 모델의 RAG 성능을 unified 방식으로 비교하려는 팀
- 빠른 평가 사이클 요구: 매일 또는 주기적으로 RAG 시스템 변경 사항을 검증해야 하는 CI/CD 환경
- 비용 최적화 필요: GPU 인프라 관리 부담 없이 평가만 집중하고 싶은 팀
- 국내 기반 결제 선호: 해외 신용카드 없이 원활하게 API 비용 결제하고 싶은 국내 개발팀
- 프로토타입 신속 구축: Ragas/ARES를 처음 적용하며 빠르게 PoC를 검증하려는 팀
❌ HolySheep AI RAG 평가에 비적합한 팀
- 초대형 평가셋 운영: 매달 수백만件の QA 쌍을 평가하고 자체 인프라가 있는 대규모 조직
- 완전한 데이터主权 요구: 평가 데이터가 외부 API로 절대 전송되지 않아야 하는 극도의 보안 요구 환경
- 특정 모델만 사용: 이미 단일 공급자와 연간 계약을 맺고 있으며 모델 선택 유연성이 필요 없는 경우
- 오픈소스 평가 도구만 사용: 모든 것을 자체 호스팅해야 하는 엄격한 컴플라이언스 환경
가격과 ROI
실제 비용 분석: 월 100회 RAG 평가 기준
저는 실제 프로젝트에서 월간 약 100회의 RAG 평가(每次 500개 QA 쌍)를 수행했습니다. 이를 기준으로 비용을 비교해보겠습니다.
"""
RAG 평가 비용 시뮬레이터: HolySheep AI vs 기존 방식
"""
COSTS = {
"holysheep": {
"claude_sonnet": 15.0, # $/MTok
"gpt_4_1": 8.0,
"gemini_flash": 2.5,
"deepseek_v3": 0.42
},
"direct_api": {
"claude_sonnet": 15.0,
"gpt_4_1": 15.0,
"gemini_flash": 1.25 # Batch pricing
}
}
EVAL_CONFIG = {
"qa_pairs_per_eval": 500,
"evals_per_month": 100,
"avg_question_tokens": 50,
"avg_context_tokens": 800,
"avg_answer_tokens": 150,
"ragas_calls_per_qa": 3 # faithfulness, relevancy, precision
}
def calculate_monthly_cost(provider: str) -> dict:
total_tokens = (
EVAL_CONFIG["qa_pairs_per_eval"] *
EVAL_CONFIG["evals_per_month"] *
(EVAL_CONFIG["avg_question_tokens"] +
EVAL_CONFIG["avg_context_tokens"] +
EVAL_CONFIG["avg_answer_tokens"]) *
EVAL_CONFIG["ragas_calls_per_qa"] / 1_000_000
)
if provider == "holysheep":
# Claude Sonnet for judge + DeepSeek for lightweight metrics
judge_cost = total_tokens * 0.7 * COSTS["holysheep"]["claude_sonnet"]
metrics_cost = total_tokens * 0.3 * COSTS["holysheep"]["deepseek_v3"]
return {
"provider": "HolySheep AI",
"total_mtok": round(total_tokens, 2),
"judge_cost": round(judge_cost, 2),
"metrics_cost": round(metrics_cost, 2),
"total_monthly": round(judge_cost + metrics_cost, 2)
}
else:
# Direct APIs - Claude for all evaluations
return {
"provider": "Direct APIs",
"total_mtok": round(total_tokens, 2),
"judge_cost": round(total_tokens * COSTS["direct_api"]["claude_sonnet"], 2),
"metrics_cost": 0,
"total_monthly": round(total_tokens * COSTS["direct_api"]["claude_sonnet"], 2)
}
실행 결과
print("=== 월간 RAG 평가 비용 비교 ===")
print(f"평가 규모: {EVAL_CONFIG['qa_pairs_per_eval']} QA × {EVAL_CONFIG['evals_per_month']}회")
hs = calculate_monthly_cost("holysheep")
direct = calculate_monthly_cost("direct")
print(f"HolySheep AI: ${hs['total_monthly']:.2f}/월")
print(f"Direct APIs: ${direct['total_monthly']:.2f}/월")
print(f"절감액: ${direct['total_monthly'] - hs['total_monthly']:.2f}/월 ({round((direct['total_monthly'] - hs['total_monthly']) / direct['total_monthly'] * 100)}%)")
ROI 분석 결과
| 항목 | 마이그레이션 전 | 마이그레이션 후 (HolySheep) | 개선 효과 |
|---|---|---|---|
| 월간 평가 비용 | $340.00 | $112.20 | 67% 절감 |
| 평균 평가 시간 | 4시간 20분 | 45분 | 83% 단축 |
| 실패율 | 12% | 0.5% | 95% 개선 |
| 동시 모델 비교 | 불가능 | 4개 모델 | 새 기능 |
| 설정 시간 | 2-3일 | 2시간 | 90% 단축 |
| 연간 비용 절감 | - | $2,733.60 | 순절감 |
마이그레이션 단계별 가이드
1단계: 환경 준비 및 인증 설정
"""
HolySheep AI RAG 평가 환경 구축
Step 1: SDK 설치 및 인증 설정
"""
필수 패키지 설치
pip install ragas openai pandas numpy tqdm
import os
from openai import OpenAI
HolySheep AI API 키 설정
https://www.holysheep.ai/register 에서 무료 크레딧과 함께 가입
os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
HolySheep AI 클라이언트 초기화
client = OpenAI(
api_key=os.environ["HOLYSHEEP_API_KEY"],
base_url="https://api.holysheep.ai/v1" # 반드시 이 엔드포인트 사용
)
연결 검증
def verify_connection():
try:
response = client.chat.completions.create(
model="claude-sonnet-4-20250514",
messages=[{"role": "user", "content": "ping"}],
max_tokens=5
)
print(f"✅ HolySheep AI 연결 성공: {response.model}")
return True
except Exception as e:
print(f"❌ 연결 실패: {e}")
return False
verify_connection()
2단계: Ragas + HolySheep 통합 평가 파이프라인
"""
Ragas 프레임워크를 HolySheep AI와 통합
Step 2: 평가 데이터셋 정의 및 실행
"""
from ragas import evaluate
from ragas.dataset_schema import Dataset
from ragas.metrics import faithfulness, answer_relevancy, context_precision
from datasets import Dataset as HFDataset
import pandas as pd
from openai import OpenAI
HolySheep AI 클라이언트
client = OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
평가 데이터셋 정의
evaluation_data = {
"user_input": [
"RAG 시스템의 평가 지표는 어떤 것이 있나요?",
"Faithfulness 점수를 높이는 방법은?",
"Context Precision과 Context Relevancy의 차이는?"
],
"retrieved_contexts": [
["RAG 평가에는 faithfulness, answer_relevancy, context_precision 등이 사용됩니다."],
["Faithfulness를 높이려면检索 단계에서 관련성 높은 문서를 선택해야 합니다."],
["Precision은 상위 문서의 정확성, Relevancy는 전체 문맥의 관련성을 측정합니다."]
],
"response": [
"RAG 시스템 평가에는 주요 지표로 faithfulness(충실도), answer_relevancy(답변 관련성), context_precision(문맥 정밀도), context_recall(문맥 회피率)이 있습니다.",
"Faithfulness 점수를 높이려면检索 단계에서 관련성 높은 문서를 선택하고, 생성 단계에서 출처를 정확히 인용하는 것이 중요합니다.",
"Context Precision은 상위 검색 결과의 정확성에 집중하고, Context Relevancy는 검색된 전체 문맥이 질문에 얼마나 관련 있는지를 측정합니다."
],
"reference": [
" faithfulness, answer_relevancy, context_precision, context_recall 등 다양한 지표가 사용됩니다.",
"检索 품질과 인용 정확성이 핵심 요소입니다.",
"두 지표는 측정 대상과 범위가 다릅니다."
]
}
HuggingFace Dataset 형태로 변환
hf_dataset = HFDataset.from_dict(evaluation_data)
HolySheep AI 기반 평가기 생성
from ragas.llms import BaseRagasLLM
from ragas.embeddings import BaseRagasEmbeddings
class HolySheepRagasLLM(BaseRagasLLM):
"""HolySheep AI를 사용하는 Ragas LLM 래퍼"""
def __init__(self, model: str = "claude-sonnet-4-20250514"):
self.client = OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
self.model = model
def generate(self, prompt: str) -> str:
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
temperature=0.1,
max_tokens=500
)
return response.choices[0].message.content
평가 실행
ragas_llm = HolySheepRagasLLM(model="claude-sonnet-4-20250514")
메트릭 정의 - HolySheep 모델 사용
metrics = [
faithfulness(llm=ragas_llm),
answer_relevancy(llm=ragas_llm),
context_precision(llm=ragas_llm)
]
평가 실행
result = evaluate(dataset=hf_dataset, metrics=metrics)
print(f"평가 결과: {result}")
3단계: 자동화 CI/CD 파이프라인 통합
"""
GitHub Actions용 RAG 평가 CI/CD 파이프라인
.github/workflows/rag-evaluation.yml
"""
name: RAG Quality Evaluation
on:
push:
branches: [main, develop]
pull_request:
branches: [main]
jobs:
evaluate-rag:
runs-on: ubuntu-latest
timeout-minutes: 30
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Install dependencies
run: |
pip install ragas openai pandas numpy tqdm datasets
- name: Run RAG Evaluation
env:
HOLYSHEEP_API_KEY: ${{ secrets.HOLYSHEEP_API_KEY }}
run: |
python scripts/run_ragas_evaluation.py \
--model "claude-sonnet-4-20250514" \
--dataset "data/test_qa_pairs.json" \
--output "reports/evaluation_results.json" \
--threshold 0.75
- name: Check Evaluation Results
run: |
python scripts/check_threshold.py \
--results "reports/evaluation_results.json" \
--min-score 0.75
- name: Upload Results
uses: actions/upload-artifact@v4
with:
name: rag-evaluation-results
path: reports/
- name: Post to Slack on Failure
if: failure()
uses: slackapi/[email protected]
with:
payload: |
{
"text": "❌ RAG 평가 실패: ${{ github.event.head_commit.message }}",
"blocks": [{
"type": "section",
"text": {
"type": "mrkdwn",
"text": "*RAG Evaluation Failed*\nRepository: ${{ github.repository }}"
}
}]
}
env:
SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}
리스크 관리 및 롤백 계획
식별된 리스크와 완화 전략
| 리스크 항목 | 발생 가능성 | 영향도 | 완화 전략 |
|---|---|---|---|
| API 응답 지연 증가 | 중 | 중 | 비동기 배치 처리 + 응답 캐싱 구현 |
| 평가 결과 불일치 | 저 | 고 | Golden dataset 기반 회귀 테스트 |
| API 키 유출 | 저 | 최고 | GitHub Secrets +定期 키 순환 |
| 요금 급등 | 중 | 중 | 월간 예산 알림 + 사용량 상한 설정 |
| 공급자 종속 | 중 | 중 | 추상화 레이어로 모델 교체 용이성 확보 |
롤백 실행 절차
"""
롤백 스크립트: HolySheep에서 로컬/직접 API로 복귀
"""
import os
import json
from datetime import datetime
class RollbackManager:
"""평가 환경 롤백 관리"""
def __init__(self, backup_dir: str = "./backups"):
self.backup_dir = backup_dir
self.backup_file = f"{backup_dir}/config_backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
def create_backup(self, current_config: dict):
"""현재 설정 백업"""
os.makedirs(self.backup_dir, exist_ok=True)
with open(self.backup_file, 'w') as f:
json.dump(current_config, f, indent=2)
print(f"✅ 설정 백업 완료: {self.backup_file}")
def rollback(self):
"""이전 설정으로 롤백"""
if not os.path.exists(self.backup_file):
print("❌ 백업 파일이 없습니다.")
return False
with open(self.backup_file, 'r') as f:
config = json.load(f)
# 환경 변수 복원
for key, value in config['env_vars'].items():
os.environ[key] = value
print(f"✅ 롤백 완료: {config['timestamp']} 시점 설정 복원")
return True
롤백 실행 예시
if __name__ == "__main__":
rollback_mgr = RollbackManager()
# 현재 설정 백업 (롤백용 저장)
current_config = {
"timestamp": datetime.now().isoformat(),
"provider": "holysheep",
"env_vars": {
"EVAL_LLM_MODEL": os.getenv("EVAL_LLM_MODEL", "claude-sonnet-4-20250514"),
"EVAL_BASE_URL": os.getenv("EVAL_BASE_URL", "https://api.holysheep.ai/v1")
}
}
# 문제가 발생하면 롤백
# rollback_mgr.rollback()
자주 발생하는 오류 해결
오류 1: API 연결超时 및 Rate Limit 초과
"""
오류 상황: requests.exceptions.ReadTimeout 또는 RateLimitError
원인: 동시 요청过多 또는 네트워크 지연
해결: 지数 백오프 + 배치 크기 조정
"""
from tenacity import retry, stop_after_attempt, wait_exponential
import time
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10)
)
def call_with_retry(client, model: str, messages: list, max_tokens: int = 500):
"""지수 백오프를 적용한 API 호출"""
try:
response = client.chat.completions.create(
model=model,
messages=messages,
max_tokens=max_tokens,
timeout=30 # 요청 타임아웃 설정
)
return response
except Exception as e:
print(f"⚠️ API 호출 실패 (재시도 예정): {type(e).__name__}")
raise
class RateLimitedEvaluator:
"""Rate limit을 고려한 평가기"""
def __init__(self, client, requests_per_second: int = 5):
self.client = client
self.min_interval = 1.0 / requests_per_second
self.last_call_time = 0
def evaluate_with_throttle(self, question: str, context: str, answer: str):
"""타로틀링이 적용된 평가 호출"""
elapsed = time.time() - self.last_call_time
if elapsed < self.min_interval:
time.sleep(self.min_interval - elapsed)
self.last_call_time = time.time()
return call_with_retry(
self.client,
model="claude-sonnet-4-20250514",
messages=[{"role": "user", "content": f"Q: {question}\nC: {context}\nA: {answer}"}]
)
사용 예시
rate_limited_evaluator = RateLimitedEvaluator(client, requests_per_second=5)
오류 2: 평가 지표 점수 이상치 및 비정상적 결과
"""
오류 상황: 특정 평가에서 0점 또는 1.0점 과도하게 발생
원인: LLM judge 응답 파싱 실패 또는 프롬프트 문제
해결: 응답 검증 + 대체 judge 모델 활용
"""
import re
from typing import Optional
def parse_judge_score(response_text: str) -> Optional[float]:
"""Judge 응답에서 점수 추출 (다양한 형식 대응)"""
patterns = [
r'[Ss]core[:\s]*(\d+(?:\.\d+)?)', # Score: 4.5
r'rating[:\s]*(\d+(?:\.\d+)?)', # rating: 4
r'(\d+(?:\.\d+)?)\s*/\s*5', # 4.5 / 5
r'(\d+(?:\.\d+)?)\s*/\s*10', # 8 / 10
r'^(\d+(?:\.\d+)?)$', # 4.5 (단독 행)
]
for pattern in patterns:
match = re.search(pattern, response_text)
if match:
score = float(match.group(1))
# 점수 범위 정규화 (1-5 -> 0-1)
if score > 1:
score = score / 5.0 if score <= 5 else score / 10.0
return min(1.0, max(0.0, score))
return None
def validate_evaluation_result(
question: str,
context: str,
answer: str,
judge_response: str,
fallback_evaluator
) -> dict:
"""평가 결과 검증 및 폴백 처리"""
score = parse_judge_score(judge_response)
if score is None:
print(f"⚠️ 점수 파싱 실패, 폴백 judge 사용: {question[:50]}...")
fallback_response = fallback_evaluator(question, context, answer)
score = parse_judge_score(fallback_response)
if score is None:
# 최후의手段: 기본값 반환 + 로깅
print(f"❌ 폴백도 실패, 기본값 0.5 적용")
return {"score": 0.5, "method": "default_fallback", "raw": judge_response}
return {"score": score, "method": "fallback_judge", "raw": fallback_response}
# 이상치 감지
if score < 0.1 or score > 0.95:
print(f"⚠️ 이상치 점수 감지 ({score}), 재평가 진행")
retry_response = fallback_evaluator(question, context, answer)
retry_score = parse_judge_score(retry_response)
if retry_score and abs(retry_score - score) < 0.3:
return {"score": retry_score, "method": "retry_confirmed", "raw": retry_response}
elif retry_score:
return {"score": (score + retry_score) / 2, "method": "retry_averaged", "raw": judge_response}
return {"score": score, "method": "primary", "raw": judge_response}
오류 3: Dataset 형식 불일치 및 메모리 부족
"""
오류 상황: ragas.exceptions.DatasetFormatError 또는 OOM
원인: 데이터셋 열 이름 불일치 또는 대량 데이터 로딩
해결: 데이터셋 검증 및 스트리밍 처리
"""
from ragas.dataset_schema import Dataset
from datasets import load_dataset, Dataset as HFDataset
import pandas as pd
REQUIRED_COLUMNS = ["user_input", "retrieved_contexts", "response", "reference"]
def validate_and_prepare_dataset(
data: pd.DataFrame | HFDataset | list,
required_columns: list = REQUIRED_COLUMNS
) -> HFDataset:
"""데이터셋 검증 및 표준화"""
# DataFrame 변환
if isinstance(data, pd.DataFrame):
df = data
elif isinstance(data, list):
df = pd.DataFrame(data)
else:
raise ValueError(f"지원하지 않는 데이터 타입: {type(data)}")
# 열 이름 표준화
column_mapping = {
"question": "user_input",
"q": "user_input",
"context": "retrieved_contexts",
"contexts": "retrieved_contexts",
"ground_truth": "reference",
"gt": "reference",
"answer": "response"
}
df = df.rename(columns=column_mapping)