Verdict

Weave와 Verdict 평가 프레임워크를 함께 사용하여 LLM 평가 파이프라인을 추적하고 모니터링합니다

Weave는 Verdict Python 라이브러리를 통해 이루어지는 모든 호출을 손쉽게 추적하고 로깅할 수 있도록 설계되었습니다.

AI 평가 파이프라인을 다룰 때는 디버깅이 매우 중요합니다. 파이프라인 단계가 실패하거나, 출력이 예상과 다르거나, 중첩된 연산으로 인해 혼란이 생기는 경우 문제의 정확한 지점을 찾기 어렵습니다. Verdict 애플리케이션은 보통 여러 파이프라인 단계, judge(평가자), 변환 단계로 구성되므로, 평가 워크플로의 내부 동작을 이해하는 것이 필수적입니다.

Weave는 Verdict 애플리케이션에 대한 트레이스를 자동으로 캡처하여 이 과정을 단순화합니다. 이를 통해 파이프라인의 성능을 모니터링하고 분석할 수 있으며, AI 평가 워크플로를 더 쉽게 디버깅하고 최적화할 수 있습니다.

​

시작하기

시작하려면 스크립트의 시작 부분에서 weave.init(project=...)를 호출하면 됩니다. project 인수를 사용해 team-name/project-name 형식으로 특정 W&B Team 이름으로 로그를 남기거나, project-name만 지정해 기본 팀/엔티티에 로그를 남길 수 있습니다.

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("Rate the quality of this text: {source.text}")

# 샘플 데이터 생성
data = Schema.of(text="This is a sample text for evaluation.")

# 파이프라인 실행 - 자동으로 추적됩니다
output = pipeline.run(data)

print(output)

​

호출 메타데이터 추적

Verdict 파이프라인 호출의 메타데이터를 추적하려면 weave.attributes 컨텍스트 매니저를 사용할 수 있습니다. 이 컨텍스트 매니저를 사용하면 파이프라인 run 또는 평가 배치와 같은 특정 코드 블록에 대해 사용자 지정 메타데이터를 설정할 수 있습니다.

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 파이프라인 호출의 트레이스와 함께 메타데이터를 자동으로 추적합니다. Weave 웹 인터페이스에서 해당 메타데이터를 확인할 수 있습니다.

​

트레이스

AI 평가 파이프라인의 트레이스를 중앙 데이터베이스에 저장하는 것은 개발 및 프로덕션 환경 모두에서 매우 중요합니다. 이러한 트레이스는 유용한 데이터셋을 제공하여 평가 워크플로우를 디버깅하고 개선하는 데 필수적입니다.

Weave는 Verdict 애플리케이션에 대한 트레이스를 자동으로 캡처합니다. Verdict 라이브러리를 통해 수행되는 모든 호출을 추적하고 로깅하며, 여기에는 다음이 포함됩니다:

• 파이프라인 실행 단계
• Judge 유닛의 평가
• 레이어 변환 작업
• 풀링 연산
• 사용자 정의 유닛 및 변환

Weave 웹 인터페이스에서 트레이스를 확인할 수 있으며, 이를 통해 파이프라인 실행의 계층적 구조를 파악할 수 있습니다.

​

파이프라인 트레이싱 예시

다음은 Weave가 중첩된 파이프라인 연산을 추적하는 방식을 보여 주는 좀 더 복잡한 예시입니다.

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")

# 여러 단계로 구성된 복잡한 파이프라인 생성
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 a comprehensive evaluation of text quality across multiple dimensions.")

# 파이프라인 실행 - 모든 작업이 추적됩니다
result = pipeline.run(data)

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

이 설정은 다음과 같은 상세 추적을 생성합니다:

• 주요 Pipeline 실행
• Layer 내 각 JudgeUnit 평가
• MeanPoolUnit의 집계 단계
• 각 연산의 타이밍 정보

​

설정

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

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

# Verdict의 내장 트레이서를 계속 사용할 수 있습니다
console_tracer = ConsoleTracer()

# Weave(자동) 및 Console 트레이싱을 모두 사용하는 파이프라인 생성
pipeline = Pipeline(tracer=[console_tracer])  # Weave 트레이서가 자동으로 추가됩니다
pipeline = pipeline >> JudgeUnit().prompt("Evaluate: {source.text}")

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

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

​

모델 및 평가

여러 파이프라인 구성 요소로 이루어진 AI 시스템을 체계적으로 구성하고 평가하는 일은 쉽지 않을 수 있습니다. weave.Model을 사용하면 프롬프트, 파이프라인 설정, 평가 파라미터와 같은 실험 세부 정보를 기록하고 정리할 수 있어 서로 다른 반복 실험을 더 쉽게 비교할 수 있습니다.

다음 예시는 Verdict 파이프라인을 WeaveModel로 래핑하는 방법을 보여줍니다.

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 파이프라인이 어떻게 동작하는지를 보여주는 평가 트레이스를 생성합니다.

​

모범 사례

​

성능 모니터링

Weave는 모든 파이프라인 연산에 대한 타이밍 정보를 자동으로 수집합니다. 이를 활용해 성능 병목 구간을 식별할 수 있습니다.

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는 파이프라인 실행 중에 발생하는 예외를 자동으로 포착합니다:

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 평가 파이프라인에 대한 포괄적인 옵저버빌리티를 확보할 수 있어, 평가 워크플로우를 디버깅하고 최적화하며 이해하기가 한결 쉬워집니다.