HuggingFace 데이터셋 평가

!pip install datasets wandb weave

# 변수 초기화
HUGGINGFACE_DATASET = "wandb/ragbench-test-sample"
WANDB_KEY = ""
WEAVE_TEAM = ""
WEAVE_PROJECT = ""

# weave 및 필수 라이브러리 초기화
import asyncio

import nest_asyncio
import wandb
from datasets import load_dataset

import weave
from weave import Evaluation

# wandb에 로그인하고 weave 초기화
wandb.login(key=WANDB_KEY)
client = weave.init(f"{WEAVE_TEAM}/{WEAVE_PROJECT}")

# 중첩 이벤트 루프 허용을 위해 nest_asyncio 적용 (일부 노트북 환경에서 필요)
nest_asyncio.apply()

# HuggingFace 데이터셋 로드
ds = load_dataset(HUGGINGFACE_DATASET)
row_count = ds["train"].num_rows

# 데이터셋의 인덱스 매핑 생성
# HF 데이터셋 인덱스를 포함하는 딕셔너리 목록을 생성합니다
# 예시: [{"hf_id": 0}, {"hf_id": 1}, {"hf_id": 2}, ...]
hf_index = [{"hf_id": i, "hf_hub_name": HUGGINGFACE_DATASET} for i in range(row_count)]

@weave.op()
def preprocess_example(example):
    """
    Preprocesses each example before evaluation.
    Args:
        example: Dict containing hf_id
    Returns:
        Dict containing the prompt from the HF dataset
    """
    hf_row = ds["train"][example["hf_id"]]
    return {"prompt": hf_row["question"], "answer": hf_row["response"]}

@weave.op()
def hf_eval(hf_id: int, output: dict) -> dict:
    """
    Scoring function for evaluating model outputs.
    Args:
        hf_id: Index in the HF dataset
        output: The output from the model to evaluate
    Returns:
        Dict containing evaluation scores
    """
    hf_row = ds["train"][hf_id]
    return {"scorer_value": True}

@weave.op()
def function_to_evaluate(prompt: str):
    """
    The function that will be evaluated (e.g., your model or pipeline).
    Args:
        prompt: Input prompt from the dataset
    Returns:
        Dict containing model output
    """
    return {"generated_text": "testing "}

# 평가 객체 생성
evaluation = Evaluation(
    dataset=hf_index,  # 인덱스 매핑 사용
    scorers=[hf_eval],  # 점수화 함수 목록
    preprocess_model_input=preprocess_example,  # 입력 데이터 준비 함수
)

# 비동기로 평가 실행
async def main():
    await evaluation.evaluate(function_to_evaluate)

asyncio.run(main())