Verdict

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

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

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

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

​

시작하기

시작하려면 스크립트의 시작 부분에서 weave.init(project=...)을 호출하면 됩니다. project 인자를 사용해 team-name/project-name 형식으로 특정 W&B 팀 이름에 기록하거나, 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 컨텍스트 매니저를 사용할 수 있습니다. 이 컨텍스트 매니저를 사용하면 파이프라인 실행이나 평가 배치와 같은 특정 코드 블록에 대해 커스텀 메타데이터를 설정할 수 있습니다.

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(자동) 및 콘솔 트레이싱을 모두 사용하는 파이프라인 생성
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을(를) 사용하면 프롬프트, 파이프라인 구성, 평가 매개변수와 같은 실험 세부 정보를 기록하고 정리할 수 있어, 서로 다른 iteration을 더 쉽게 비교할 수 있습니다.

다음 예시는 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 파이프라인이 어떻게 동작하는지 보여주는 evaluation trace가 생성됩니다.

​

모범 사례

​

성능 모니터링

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 평가 파이프라인에 대한 종합적인 가시성을 확보할 수 있어 평가 워크플로우를 디버깅하고 최적화하며 더 잘 이해할 수 있습니다.