W&B Weave와 W&B Tables를 사용하여 모델 평가하기

import weave
import wandb
from typing import Any

# Weave 초기화
weave.init("your-entity/your-project")

# 레지스트리에서 로드하는 ChatModel 정의
class ChatModel(weave.Model):
    model_name: str
    
    def model_post_init(self, __context):
        # W&B Models 레지스트리에서 모델 다운로드
        with wandb.init(project="your-project", job_type="model_download") as run:
            artifact = run.use_artifact(self.model_name)
            self.model_path = artifact.download()
            # 여기서 모델 초기화
    
    @weave.op()
    async def predict(self, query: str) -> str:
        # 모델 추론 로직
        return self.model.generate(query)

# 평가 데이터셋 생성
dataset = weave.Dataset(name="eval_dataset", rows=[
    {"input": "What is the capital of France?", "expected": "Paris"},
    {"input": "What is 2+2?", "expected": "4"},
])

# Scorer 정의
@weave.op()
def exact_match_scorer(expected: str, output: str) -> dict:
    return {"correct": expected.lower() == output.lower()}

# 평가 실행
model = ChatModel(model_name="wandb-entity/registry-name/model:version")
evaluation = weave.Evaluation(
    dataset=dataset,
    scorers=[exact_match_scorer]
)
results = await evaluation.evaluate(model)

# W&B 추적 기능으로 평가 실행
with weave.attributes({"wandb-run-id": wandb.run.id}):
    summary, call = await evaluation.evaluate.call(evaluation, model)

# W&B Models에 메트릭 로깅
wandb.run.log(summary)
wandb.run.config.update({
    "weave_eval_url": f"https://wandb.ai/{entity}/{project}/r/call/{call.id}"
})

@weave.op()
def llm_judge_scorer(expected: str, output: str, judge_model) -> dict:
    prompt = f"Is this answer correct? Expected: {expected}, Got: {output}"
    judgment = await judge_model.predict(prompt)
    return {"judge_score": judgment}

models = [
    ChatModel(model_name="model:v1"),
    ChatModel(model_name="model:v2"),
]

for model in models:
    results = await evaluation.evaluate(model)
    print(f"{model.model_name}: {results}")

import wandb

# run을 초기화합니다
run = wandb.init(project="model-evaluation")

# 평가 결과를 담은 table을 만듭니다
columns = ["id", "input", "ground_truth", "prediction", "confidence", "correct"]
eval_table = wandb.Table(columns=columns)

# 평가 데이터를 추가합니다
for idx, (input_data, label) in enumerate(test_dataset):
    prediction = model(input_data)
    confidence = prediction.max()
    predicted_class = prediction.argmax()
    
    eval_table.add_data(
        idx,
        wandb.Image(input_data),  # 이미지 또는 기타 미디어를 로깅합니다
        label,
        predicted_class,
        confidence,
        label == predicted_class
    )

# table을 로깅합니다
run.log({"evaluation_results": eval_table})

# 모델 A 평가
with wandb.init(project="model-comparison", name="model_a") as run:
    eval_table_a = create_eval_table(model_a, test_data)
    run.log({"test_predictions": eval_table_a})

# 모델 B 평가  
with wandb.init(project="model-comparison", name="model_b") as run:
    eval_table_b = create_eval_table(model_b, test_data)
    run.log({"test_predictions": eval_table_b})

for epoch in range(num_epochs):
    train_model(model, train_data)
    
    # 이 에포크의 예측을 평가하고 로깅하세요
    eval_table = wandb.Table(columns=["image", "truth", "prediction"])
    for image, label in test_subset:
        pred = model(image)
        eval_table.add_data(wandb.Image(image), label, pred.argmax())
    
    wandb.log({f"predictions_epoch_{epoch}": eval_table})

# 분석 열을 추가할 수 있는 변경 가능한 table을 생성합니다
eval_table = wandb.Table(
    columns=["id", "image", "label", "prediction"],
    log_mode="MUTABLE"  # 나중에 열을 추가할 수 있습니다
)

# 초기 예측
for idx, (img, label) in enumerate(test_data):
    pred = model(img)
    eval_table.add_data(idx, wandb.Image(img), label, pred.argmax())

run.log({"eval_analysis": eval_table})

# 오류 분석을 위한 신뢰도 점수를 추가합니다
confidences = [model(img).max() for img, _ in test_data]
eval_table.add_column("confidence", confidences)

# 오류 유형을 추가합니다
error_types = classify_errors(eval_table.get_column("label"), 
                            eval_table.get_column("prediction"))
eval_table.add_column("error_type", error_types)

run.log({"eval_analysis": eval_table})