Verdict - Weights & Biases Documentation

Weave는 Verdict Python library를 통해 이루어지는 모든 Call을 자동으로 추적하고 로깅하도록 설계되었습니다. AI 평가 파이프라인을 다룰 때 디버깅은 중요합니다. 파이프라인의 step이 실패하거나, 출력이 예상과 다르거나, 중첩된 오퍼레이션으로 인해 혼란이 생기면 문제 지점을 정확히 파악하기 어려울 수 있습니다. Verdict 애플리케이션은 여러 파이프라인 step, judge, transformation으로 구성되는 경우가 많으므로, 평가 워크플로의 내부 동작을 이해하면 도움이 됩니다. Weave는 Verdict 애플리케이션의 트레이스를 자동으로 수집해 이 과정을 간소화합니다. 이를 통해 파이프라인 성능을 모니터링하고 분석하여 AI 평가 워크플로를 디버깅하고 최적화할 수 있습니다.

시작하기

Verdict 파이프라인에 Weave 트레이싱을 사용 설정하려면 스크립트 시작 부분에서 weave.init(project=...)를 호출하세요. project 인수를 사용해 team-name/project-name 형식으로 특정 W&B Teams 이름에 로깅하거나, project-name을 전달해 기본 팀 또는 entity에 로깅할 수 있습니다.

import weave
from verdict import Pipeline
from verdict.common.judge import JudgeUnit
from verdict.schema import Schema

# Initialize Weave with your project name
weave.init("verdict_demo")

# Create a simple evaluation pipeline
pipeline = Pipeline()
pipeline = pipeline >> JudgeUnit().prompt("Rate the quality of this text: {source.text}")

# Create sample data
data = Schema.of(text="This is a sample text for evaluation.")

# 파이프라인 실행 - Weave가 자동으로 트레이스합니다
output = pipeline.run(data)

print(output)

call 메타데이터 추적

Verdict 파이프라인 call에 맞춤형 메타데이터를 첨부하려면 weave.attributes 컨텍스트 관리자를 사용하세요. 이 컨텍스트 관리자를 사용하면 파이프라인 run이나 평가 배치와 같은 특정 코드 블록에 태그를 지정해, 나중에 Weave UI에서 관련 트레이스를 필터링하고 그룹화할 수 있습니다.

import weave
from verdict import Pipeline
from verdict.common.judge import JudgeUnit
from verdict.schema import Schema

# 프로젝트 이름으로 Weave 초기화
weave.init("verdict_demo")

pipeline = Pipeline()
pipeline = pipeline >> JudgeUnit().prompt("Evaluate sentiment: {source.text}")

data = Schema.of(text="I love this product!")

with weave.attributes({"evaluation_type": "sentiment", "batch_id": "batch_001"}):
    output = pipeline.run(data)

print(output)

Weave는 Verdict 파이프라인 call의 트레이스에 메타데이터를 자동으로 기록합니다. Weave 웹 인터페이스에서 이 메타데이터를 볼 수 있습니다.

트레이스

AI 평가 파이프라인의 트레이스를 중앙 데이터베이스에 저장하면 개발 환경과 프로덕션 환경 모두에서 유용합니다. 이러한 트레이스는 평가 워크플로를 디버깅하고 개선하는 데 도움이 되며, 유용한 데이터셋도 제공합니다. Weave는 Verdict 애플리케이션의 트레이스를 자동으로 캡처합니다. Verdict 라이브러리를 통해 발생하는 모든 Call을 추적하고 로깅하며, 여기에는 다음이 포함됩니다:

Pipeline 실행 step
JudgeUnit 평가
Layer 변환
풀링 오퍼레이션
맞춤형 유닛 및 변환

트레이스는 파이프라인 실행의 계층 구조를 보여주는 Weave 웹 인터페이스에서 확인할 수 있습니다.

파이프라인 트레이싱 예시

다음 예시는 Weave가 중첩된 파이프라인 오퍼레이션을 트레이스하는 방식을 보여주며, 이를 통해 다단계 Verdict 파이프라인에서 각 step이 어떻게 캡처되는지 확인할 수 있습니다:

import weave
from verdict import Pipeline, Layer
from verdict.common.judge import JudgeUnit
from verdict.transform import MeanPoolUnit
from verdict.schema import Schema

# 프로젝트 이름으로 Weave 초기화
weave.init("verdict_demo")

# 여러 step로 구성된 파이프라인 생성
pipeline = Pipeline()
pipeline = pipeline >> Layer([
    JudgeUnit().prompt("Rate coherence: {source.text}"),
    JudgeUnit().prompt("Rate relevance: {source.text}"),
    JudgeUnit().prompt("Rate accuracy: {source.text}")
], 3)
pipeline = pipeline >> MeanPoolUnit()

# 샘플 데이터
data = Schema.of(text="This is an evaluation of text quality across multiple dimensions.")

# 파이프라인 실행 - Weave가 모든 오퍼레이션을 트레이스합니다
result = pipeline.run(data)

print(f"Average score: {result}")

이렇게 하면 다음 내용을 보여주는 상세한 트레이스가 생성됩니다:

메인 Pipeline 실행.
Layer 내 각 JudgeUnit 평가.
MeanPoolUnit 집계 step.
각 오퍼레이션의 타이밍 정보.

설정

weave.init()를 호출하면 Verdict 파이프라인에서 트레이싱이 자동으로 활성화됩니다. 이 인테그레이션은 Pipeline.__init__() 메서드를 패치해 모든 트레이스 데이터를 Weave로 전달하는 VerdictTracer를 주입하는 방식으로 작동합니다. 추가 설정은 필요하지 않습니다. Weave가 자동으로 다음을 수행합니다:

모든 파이프라인 오퍼레이션을 캡처합니다.
실행 시간을 추적합니다.
입력과 출력을 로깅합니다.
트레이스 계층 구조를 유지합니다.
동시 파이프라인 실행을 처리합니다.

맞춤형 트레이서와 Weave

애플리케이션에서 이미 맞춤형 Verdict 트레이서를 사용 중이라면, Weave의 VerdictTracer도 함께 실행할 수 있으므로 인테그레이션 중 하나를 선택하지 않고 둘 다 사용할 수 있습니다:

import weave
from verdict import Pipeline
from verdict.common.judge import JudgeUnit
from verdict.util.tracing import ConsoleTracer
from verdict.schema import Schema

# Initialize Weave with your project name
weave.init("verdict_demo")

# You can still use Verdict's built-in tracers
console_tracer = ConsoleTracer()

# Create pipeline with both Weave (automatic) and Console tracing
pipeline = Pipeline(tracer=[console_tracer])  # Weave tracer is added automatically
pipeline = pipeline >> JudgeUnit().prompt("Evaluate: {source.text}")

data = Schema.of(text="Sample evaluation text")

# Weave와 콘솔 모두에 트레이싱됩니다
result = pipeline.run(data)

Models 및 평가

여러 파이프라인 컴포넌트로 구성된 AI 시스템을 정리하고 평가하는 일은 까다로울 수 있습니다. weave.Model을 사용하면 프롬프트, 파이프라인 설정, 평가 매개변수와 같은 실험 세부 정보를 담아 체계적으로 정리할 수 있어, 서로 다른 반복 버전을 더 쉽게 비교할 수 있습니다. 다음 예제에서는 Verdict 파이프라인을 weave.Model로 래핑하는 방법을 보여줍니다:

import asyncio
import weave
from verdict import Pipeline
from verdict.common.judge import JudgeUnit
from verdict.schema import Schema

# 프로젝트 이름으로 Weave를 초기화합니다
weave.init("verdict_demo")

class TextQualityEvaluator(weave.Model):
    judge_prompt: str
    pipeline_name: str

    @weave.op()
    async def predict(self, text: str) -> dict:
        pipeline = Pipeline(name=self.pipeline_name)
        pipeline = pipeline >> JudgeUnit().prompt(self.judge_prompt)
        
        data = Schema.of(text=text)
        result = pipeline.run(data)
        
        return {
            "text": text,
            "quality_score": result.score if hasattr(result, 'score') else result,
            "evaluation_prompt": self.judge_prompt
        }

model = TextQualityEvaluator(
    judge_prompt="Rate the quality of this text on a scale of 1-10: {source.text}",
    pipeline_name="text_quality_evaluator"
)

text = "This is a well-written and informative piece of content that provides clear value to readers."

prediction = asyncio.run(model.predict(text))

# Jupyter Notebook을 사용 중이라면 다음을 실행하세요:
# prediction = await model.predict(text)

print(prediction)

이 코드는 Weave UI에서 시각화해 파이프라인 구조와 평가 결과를 모두 확인할 수 있는 모델을 생성합니다.

평가

평가는 평가 파이프라인 자체의 성능을 측정하는 데 도움이 됩니다. weave.Evaluation 클래스를 사용하면 특정 작업이나 데이터셋에서 Verdict 파이프라인이 얼마나 잘 동작하는지 파악할 수 있습니다:

import asyncio
import weave
from verdict import Pipeline
from verdict.common.judge import JudgeUnit
from verdict.schema import Schema

# Weave 초기화
weave.init("verdict_demo")

# 평가 모델 생성
class SentimentEvaluator(weave.Model):
    @weave.op()
    async def predict(self, text: str) -> dict:
        pipeline = Pipeline()
        pipeline = pipeline >> JudgeUnit().prompt(
            "Classify sentiment as positive, negative, or neutral: {source.text}"
        )

        data = Schema.of(text=text)
        result = pipeline.run(data)

        return {"sentiment": result}

# 테스트 데이터
texts = [
    "I love this product, it's amazing!",
    "This is terrible, worst purchase ever.",
    "The weather is okay today."
]
labels = ["positive", "negative", "neutral"]

examples = [
    {"id": str(i), "text": texts[i], "target": labels[i]}
    for i in range(len(texts))
]

# 점수 산출 함수
@weave.op()
def sentiment_accuracy(target: str, output: dict) -> dict:
    predicted = output.get("sentiment", "").lower()
    return {"correct": target.lower() in predicted}

model = SentimentEvaluator()

evaluation = weave.Evaluation(
    dataset=examples,
    scorers=[sentiment_accuracy],
)

scores = asyncio.run(evaluation.evaluate(model))
# Jupyter Notebook에서 실행하는 경우:
# scores = await evaluation.evaluate(model)

print(scores)

이렇게 하면 다양한 테스트 케이스에서 Verdict 파이프라인의 수행 방식을 보여주는 평가 트레이스가 생성됩니다.

모범 사례

다음 섹션에서는 Verdict 파이프라인과 함께 Weave를 사용할 때의 성능 모니터링 및 오류 처리 모범 사례를 설명합니다.

성능 모니터링

Weave는 모든 파이프라인 오퍼레이션의 실행 시간 정보를 자동으로 캡처하며, 이 정보를 바탕으로 Runs 전반의 성능 병목 지점을 파악할 수 있습니다:

import weave
from verdict import Pipeline, Layer
from verdict.common.judge import JudgeUnit
from verdict.schema import Schema

weave.init("verdict_demo")

# 성능 변동이 있을 수 있는 파이프라인 생성
pipeline = Pipeline()
pipeline = pipeline >> Layer([
    JudgeUnit().prompt("Quick evaluation: {source.text}"),
    JudgeUnit().prompt("Detailed analysis: {source.text}"),  # 더 느릴 수 있음
], 2)

data = Schema.of(text="Sample text for performance testing")

# 타이밍 패턴 확인을 위해 여러 번 실행
for i in range(3):
    with weave.attributes({"run_number": i}):
        result = pipeline.run(data)

오류 처리

Weave는 파이프라인 실행 중 발생하는 예외를 자동으로 캡처하므로 애플리케이션에서 예외를 처리하더라도 Weave 트레이스에 실패가 로깅됩니다:

import weave
from verdict import Pipeline
from verdict.common.judge import JudgeUnit
from verdict.schema import Schema

weave.init("verdict_demo")

pipeline = Pipeline()
pipeline = pipeline >> JudgeUnit().prompt("Process: {source.invalid_field}")  # 이 코드는 오류를 발생시킵니다

data = Schema.of(text="Sample text")

try:
    result = pipeline.run(data)
except Exception as e:
    print(f"Pipeline failed: {e}")
    # 오류 세부 정보는 Weave 트레이스에 캡처됩니다

Weave를 Verdict와 통합하면 AI 평가 파이프라인 전반을 관측할 수 있어 평가 워크플로를 디버그하고 최적화하며 이해하기가 더 쉬워집니다.

​시작하기

​call 메타데이터 추적

​트레이스

​파이프라인 트레이싱 예시

​설정

​맞춤형 트레이서와 Weave

​Models 및 평가

​평가

​모범 사례

​성능 모니터링

​오류 처리