코드에서 평가 데이터 기록하기

import weave
from openai import OpenAI
from weave import EvaluationLogger

weave.init('[YOUR-TEAM]/[YOUR-PROJECT]')

# 토큰 추적을 위해 모델 호출 전에 EvaluationLogger를 초기화하세요
eval_logger = EvaluationLogger(
    model="my_model",
    dataset="my_dataset"
)

# 예제 입력 데이터 (원하는 데이터 구조를 사용할 수 있습니다)
eval_samples = [
    {'inputs': {'a': 1, 'b': 2}, 'expected': 3},
    {'inputs': {'a': 2, 'b': 3}, 'expected': 5},
    {'inputs': {'a': 3, 'b': 4}, 'expected': 7},
]

# OpenAI를 사용한 예제 모델 로직
@weave.op
def user_model(a: int, b: int) -> int:
    oai = OpenAI()
    response = oai.chat.completions.create(
        messages=[{"role": "user", "content": f"What is {a}+{b}?"}],
        model="gpt-4o-mini"
    )
    # 응답을 어떤 방식으로든 활용합니다 (여기서는 단순하게 a + b를 반환합니다)
    return a + b

# 예제를 순회하며 예측하고 로깅합니다
for sample in eval_samples:
    inputs = sample["inputs"]
    model_output = user_model(**inputs) # 입력을 kwargs로 전달합니다

    # 예측 입력과 출력을 로깅합니다
    prediction = eval_logger.log_prediction(
        inputs=inputs,
        output=model_output
    )

    # 이 예측에 대한 점수를 계산하고 로깅합니다
    expected = sample["expected"]
    correctness_score = model_output == expected
    prediction.log_score(
        scorer="correctness", # 스코어러의 단순 문자열 이름
        score=correctness_score
    )

    # 이 특정 예측에 대한 로깅을 완료합니다
    prediction.finish()

# 전체 평가에 대한 최종 요약을 로깅합니다.
# Weave는 위에서 로깅된 'correctness' 점수를 자동으로 집계합니다.
summary_stats = {"subjective_overall_score": 0.8}
eval_logger.log_summary(summary_stats)

print("Evaluation logging complete. View results in the Weave UI.")

span.set_attributes(
    {
        "weave.eval.run_id": "[EVALUATION-RUN-CALL-ID]",
        "weave.eval.predict_and_score_call_id": "[PREDICT-AND-SCORE-CALL-ID]",
        "weave.eval.kind": "agent",
        "weave.eval.row_digest": "[ROW-DIGEST]",
        "weave.eval.example_id": "[EXAMPLE-ID]",
        "weave.eval.trial_index": 0,
        "weave.eval.evaluation_name": "[EVALUATION-NAME]",
    }
)

import weave
from weave import EvaluationLogger

weave.init("[YOUR-TEAM]/[YOUR-PROJECT]")

# 데이터셋 게시(한 번만 수행하면 됨)
dataset = weave.Dataset(
    name="my_eval_dataset",
    rows=[
      {"question": "What is the capital of France?", "expected": "Paris"},
      {"question": "What U.S. state is Seattle in?", "expected": "Washington"},
      {"question": "In which country is Mount Fuji?", "expected": "Japan"},
    ],
)
weave.publish(dataset)

# 게시된 데이터셋 조회
dataset = weave.ref("my_eval_dataset").get()

# 토큰 추적을 위해 모델을 호출하기 전에 EvaluationLogger를 초기화합니다
ev = EvaluationLogger(
    model="example_model",
    dataset="example_dataset"
)

# 토큰 추적을 위해 모델 출력(예: OpenAI Call)은 로거 초기화 후에 발생해야 합니다
outputs = [your_output_generator(**inputs) for inputs in your_dataset]
predictions = [ev.log_prediction(inputs, output) for inputs, output in zip(your_dataset, outputs)]
for prediction, output in zip(predictions, outputs):
    prediction.log_score(scorer="greater_than_5_scorer", score=output > 5)
    prediction.log_score(scorer="greater_than_7_scorer", score=output > 7)
    prediction.finish()

ev.log_summary()

import io
import wave
import struct
from PIL import Image
import random
from typing import Any
import weave

def generate_random_audio_wave_read(duration=2, sample_rate=44100):
    n_samples = duration * sample_rate
    amplitude = 32767  # 16비트 최대 진폭

    buffer = io.BytesIO()

    # 버퍼에 wave 데이터를 씁니다
    with wave.open(buffer, 'wb') as wf:
        wf.setnchannels(1)
        wf.setsampwidth(2)  # 16비트
        wf.setframerate(sample_rate)

        for _ in range(n_samples):
            sample = random.randint(-amplitude, amplitude)
            wf.writeframes(struct.pack('<h', sample))

    # 읽을 수 있도록 버퍼를 처음 위치로 되돌립니다
    buffer.seek(0)

    # Wave_read 객체를 반환합니다
    return wave.open(buffer, 'rb')

rich_media_dataset = [
    {
        'image': Image.new(
            "RGB",
            (100, 100),
            color=(
                random.randint(0, 255),
                random.randint(0, 255),
                random.randint(0, 255),
            ),
        ),
        "audio": generate_random_audio_wave_read(),
    }
    for _ in range(5)
]

@weave.op
def your_output_generator(image: Image.Image, audio) -> dict[str, Any]:
    return {
        "result": random.randint(0, 10),
        "image": image,
        "audio": audio,
    }

ev = EvaluationLogger(model="example_model", dataset="example_dataset")

for inputs in rich_media_dataset:
    output = your_output_generator(**inputs)
    prediction = ev.log_prediction(inputs, output)
    prediction.log_score(scorer="greater_than_5_scorer", score=output["result"] > 5)
    prediction.log_score(scorer="greater_than_7_scorer", score=output["result"] > 7)

ev.log_summary()

import weave

models = [
    "model1",
    "model2",
     {"name": "model3", "metadata": {"coolness": 9001}}
]

for model in models:
    # 토큰을 캡처하려면 모델 Call 전에 EvalLogger를 초기화해야 합니다
    ev = EvaluationLogger(
        name="comparison-eval",
        model=model, 
        dataset="example_dataset",
        scorers=["greater_than_3_scorer", "greater_than_5_scorer", "greater_than_7_scorer"],
        eval_attributes={"experiment_id": "exp_123"}
    )
    for inputs in your_dataset:
        output = your_output_generator(**inputs)
        prediction = ev.log_prediction(inputs=inputs, output=output)
        prediction.log_score(scorer="greater_than_3_scorer", score=output > 3)
        prediction.log_score(scorer="greater_than_5_scorer", score=output > 5)
        prediction.log_score(scorer="greater_than_7_scorer", score=output > 7)
        prediction.finish()

    ev.log_summary()