출력 개선하기: BestOfN과 Refine

출력 개선하기: BestOfN과 Refine

BestOfNRefine은 모두 캐싱을 우회하기 위해 서로 다른 rollout ID로 LM 호출을 여러 번 만들어 예측의 신뢰도와 품질을 높이도록 설계된 DSPy 모듈이에요. 두 모듈 모두 N번 시도를 마치거나, reward_fnthreshold를 넘는 보상을 돌려주면 멈춰요.

출처: Output Refinement: BestOfN and Refine

BestOfN

BestOfN은 제공된 모듈을 서로 다른 rollout ID로 (최대 N번) 여러 번 실행하는 모듈이에요. 지정한 threshold를 통과한 첫 번째 예측을 돌려주거나, 그런 게 없으면 보상이 가장 높은 예측을 돌려줘요.

기본 사용법

모델에게서 한 단어로 된 답을 받을 확률을 최대한 높이고 싶다고 해볼게요. 여러 rollout ID로 시도한 뒤 최상의 결과를 돌려주도록 BestOfN을 쓰면 돼요.

import dspy

def one_word_answer(args, pred: dspy.Prediction) -> float:
    return 1.0 if len(pred.answer.split()) == 1 else 0.0

best_of_3 = dspy.BestOfN(
    module=dspy.ChainOfThought("question -> answer"), 
    N=3, 
    reward_fn=one_word_answer, 
    threshold=1.0
)

result = best_of_3(question="What is the capital of Belgium?")
print(result.answer)  # Brussels

오류 처리

기본적으로 모듈이 시도 중 오류를 만나면 N번까지 계속 시도해요. fail_count 파라미터로 이 동작을 조정할 수 있어요.

best_of_3 = dspy.BestOfN(
    module=qa, 
    N=3, 
    reward_fn=one_word_answer, 
    threshold=1.0,
    fail_count=1
)

best_of_3(question="What is the capital of Belgium?")
# 첫 실패 후 오류를 던진다

Refine

Refine은 자동 피드백 루프를 더해 BestOfN의 기능을 확장한 모듈이에요. 실패한 각 시도(마지막 시도 제외) 뒤에 모듈 성능에 대한 상세 피드백을 자동으로 생성하고, 그 피드백을 이후 실행의 힌트로 써요.

기본 사용법

import dspy

def one_word_answer(args, pred: dspy.Prediction) -> float:
    return 1.0 if len(pred.answer.split()) == 1 else 0.0

refine = dspy.Refine(
    module=dspy.ChainOfThought("question -> answer"), 
    N=3, 
    reward_fn=one_word_answer, 
    threshold=1.0
)

result = refine(question="What is the capital of Belgium?")
print(result.answer)  # Brussels

오류 처리

BestOfN과 마찬가지로 Refine도 기본적으로 오류가 나도 최대 N번까지 시도해요. fail_count 파라미터로 조절할 수 있어요.

# 첫 오류 후 멈추기
refine = dspy.Refine(
    module=qa, 
    N=3, 
    reward_fn=one_word_answer, 
    threshold=1.0,
    fail_count=1
)

BestOfN vs. Refine 비교

두 모듈은 비슷한 목적을 가지지만 접근 방식이 달라요.

  • BestOfN은 서로 다른 rollout ID를 시도해 보고, reward_fn이 정의한 대로 최상의 예측을 고르기만 해요.
  • Refine은 피드백 루프를 더해요. 이전 예측과 reward_fn 속 코드를 바탕으로 LM이 모듈 자신의 성능에 대한 상세 피드백을 만들고, 그 피드백을 이후 실행의 힌트로 활용해요.

실전 예제

사실적 정확성 보장하기

import dspy

class FactualityJudge(dspy.Signature):
    """Determine if a statement is factually accurate."""
    statement: str = dspy.InputField()
    is_factual: bool = dspy.OutputField()

factuality_judge = dspy.ChainOfThought(FactualityJudge)

def factuality_reward(args, pred: dspy.Prediction) -> float:
    statement = pred.answer    
    result = factuality_judge(statement)    
    return 1.0 if result.is_factual else 0.0

refined_qa = dspy.Refine(
    module=dspy.ChainOfThought("question -> answer"),
    N=3,
    reward_fn=factuality_reward,
    threshold=1.0
)

result = refined_qa(question="Tell me about Belgium's capital city.")
print(result.answer)

요약 - 응답 길이 조절하기

import dspy

def ideal_length_reward(args, pred: dspy.Prediction) -> float:
    """
    Reward the summary for being close to 75 words with a tapering off for longer summaries.
    """
    word_count = len(pred.summary.split())
    distance = abs(word_count - 75)
    return max(0.0, 1.0 - (distance / 125))

optimized_summarizer = dspy.BestOfN(
    module=dspy.ChainOfThought("text -> summary"),
    N=50,
    reward_fn=ideal_length_reward,
    threshold=0.9
)

result = optimized_summarizer(
    text="[Long text to summarize...]"
)
print(result.summary)

dspy.Suggestdspy.Assert에서의 마이그레이션

BestOfNRefine은 DSPy 2.6부터 dspy.Suggestdspy.Assert를 대체하는 모듈이에요.

더 알아보기 (Learn more)