DSPy Assertions (가드레일)

DSPy Assertions (가드레일)

!!! warning "지원 중단 안내" Assertions는 지원 중단(deprecated)되었고 더는 지원되지 않아요. 대신 dspy.Refine 모듈(또는 dspy.Suggest)을 쓰세요.

아래 내용은 지원 중단된 것이며, 제거 예정입니다.

소개

언어 모델(LM)은 머신러닝과 상호작용하는 방식을 바꿔 놓았고, 자연어 이해와 생성에서 막강한 능력을 제공해요. 하지만 이 모델들이 도메인 특정 제약을 지키도록 보장하는 건 여전히 어려운 문제입니다. 파인튜닝이나 "프롬프트 엔지니어링" 같은 기법이 늘어났지만, 이런 접근법은 극도로 지루하고, LM이 특정 제약을 지키도록 유도하는 데 무겁고 수동적인 손질에 의존해요. DSPy가 프롬프팅 파이프라인을 프로그래밍하는 모듈식 방식조차도 이런 제약을 효과적이고 자동으로 강제할 메커니즘이 부족했죠.

이 문제를 해결하기 위해 DSPy Assertions를 소개합니다. LM에 계산적 제약을 자동으로 강제하도록 설계된 DSPy 프레임워크의 기능이에요. DSPy Assertions는 최소한의 수동 개입으로 개발자가 LM을 원하는 결과로 이끌 수 있게 하며, LM 출력의 신뢰성과 예측 가능성, 정확성을 높여 줍니다.

dspy.Assert와 dspy.Suggest API

DSPy Assertions에는 두 가지 주요 구성 요소가 있어요.

  • dspy.Assert:

    • 파라미터:
      • constraint (bool): Python으로 정의된 불리언 검증 검사의 결과.
      • msg (Optional[str]): 피드백이나 수정 안내를 제공하는 사용자 정의 에러 메시지.
      • backtrack (Optional[module]): 제약 실패 시 재시도할 대상 모듈. 기본 백트래킹 모듈은 assertion 직전의 마지막 모듈입니다.
    • 동작: 실패 시 재시도(retry)를 시작해 파이프라인 실행을 동적으로 조정합니다. 실패가 계속되면 실행을 중단하고 dspy.AssertionError를 던져요.

  • dspy.Suggest:

    • 파라미터: dspy.Assert와 유사.
    • 동작: 강제 중단 없이 재시도를 통한 자기 개선(self-refinement)을 유도합니다. 최대 백트래킹 시도 후 실패를 로그로 남기고 계속 실행합니다.

  • dspy.Assert vs. Python Assertions: 실패 시 프로그램을 종료하는 일반 Python assert 문과 달리, dspy.Assert는 정교한 재시도 메커니즘을 수행해 파이프라인이 스스로 조정하도록 합니다.

구체적으로, 제약이 충족되지 않으면:

  • 백트래킹 메커니즘: 내부적으로 백트래킹이 시작되어 모델에게 자기 개선 후 진행할 기회를 줍니다. 이는 시그니처 수정을 통해 이루어져요.
  • 동적 시그니처 수정: DSPy 프로그램의 시그니처를 내부적으로 수정해 다음 필드를 추가합니다.
    • Past Output: validation_fn을 통과하지 못한 모델의 과거 출력
    • Instruction: 무엇이 잘못됐고 무엇을 고치면 좋은지에 대한 사용자 정의 피드백 메시지

에러가 max_backtracking_attempts를 넘어 계속되면 dspy.Assert가 파이프라인 실행을 중단하고 dspy.AssertionError로 알려줍니다. 이렇게 하면 프로그램이 "나쁜" LM 동작으로 계속 실행되지 않게 하고, 사용자 평가를 위해 실패 샘플 출력을 즉시 강조해 주죠.

  • dspy.Suggest vs. dspy.Assert: 반면 dspy.Suggest는 더 부드러운 접근을 제공합니다. dspy.Assert와 같은 재시도 백트래킹을 유지하지만, 부드럽게 잡아당기는 역할을 합니다. 모델 출력이 max_backtracking_attempts 후에도 제약을 통과하지 못하면, dspy.Suggest는 지속된 실패를 로그로 남기고 나머지 데이터에 대해 프로그램 실행을 계속합니다. 그래서 LM 파이프라인이 실행을 중단하지 않고 "최선 노력(best-effort)" 방식으로 동작하게 해요.

  • dspy.Suggest 문은 파이프라인을 중단하지 않고 안내와 잠재적 수정을 제공하는 **평가 단계의 "헬퍼(helper)"**로 가장 잘 쓰입니다.

  • dspy.Assert 문은 **개발 단계에서 "체크커(checker)"**로 권장됩니다. LM이 기대대로 동작하는지 확인하고, 개발 주기 초기에 에러를 식별·처리하는 견고한 메커니즘을 제공하죠.

사용 사례: DSPy 프로그램에 Assertions 포함하기

소개 워크스루에서 정의한 multi-hop QA SimplifiedBaleen 파이프라인 예시로 시작할게요.

class SimplifiedBaleen(dspy.Module):
    def __init__(self, passages_per_hop=2, max_hops=2):
        super().__init__()

        self.generate_query = [dspy.ChainOfThought(GenerateSearchQuery) for _ in range(max_hops)]
        self.retrieve = dspy.Retrieve(k=passages_per_hop)
        self.generate_answer = dspy.ChainOfThought(GenerateAnswer)
        self.max_hops = max_hops

    def forward(self, question):
        context = []
        prev_queries = [question]

        for hop in range(self.max_hops):
            query = self.generate_query[hop](context=context, question=question).query
            prev_queries.append(query)
            passages = self.retrieve(query).passages
            context = deduplicate(context + passages)
        
        pred = self.generate_answer(context=context, question=question)
        pred = dspy.Prediction(context=context, answer=pred.answer)
        return pred

baleen = SimplifiedBaleen()

baleen(question = "Which award did Gary Zukav's first book receive?")

DSPy Assertions를 포함하려면, 검증 함수를 정의하고 각 모델 생성 다음에 assertion을 선언하기만 하면 됩니다.

이 사용 사례에서 다음 제약을 부과하고 싶다고 해보죠.

  1. 길이 — 각 쿼리는 100자 미만이어야 함
  2. 고유성 — 각 생성 쿼리는 이전에 생성된 쿼리와 달라야 함

이 검증 검사들을 불리언 함수로 정의할 수 있어요.

#simplistic boolean check for query length
len(query) <= 100

#Python function for validating distinct queries
def validate_query_distinction_local(previous_queries, query):
    """check if query is distinct from previous queries"""
    if previous_queries == []:
        return True
    if dspy.evaluate.answer_exact_match_str(query, previous_queries, frac=0.8):
        return False
    return True

이 검증 검사들은 dspy.Suggest 문으로 선언할 수 있습니다(최선 노력 데모로 프로그램을 테스트하고 싶으니까요). 그 위치는 쿼리 생성 query = self.generate_query[hop](context=context, question=question).query 뒤에 두면 돼요.

dspy.Suggest(
    len(query) <= 100,
    "Query should be short and less than 100 characters",
    target_module=self.generate_query
)

dspy.Suggest(
    validate_query_distinction_local(prev_queries, query),
    "Query should be distinct from: "
    + "; ".join(f"{i+1}) {q}" for i, q in enumerate(prev_queries)),
    target_module=self.generate_query
)

assertion의 효과를 비교 평가한다면, assertion이 있는 프로그램을 원래 프로그램과 별도로 정의하는 걸 권장합니다. 그게 아니면 따로 둘 필요 없이 자유롭게 써도 돼요.

Assertions가 포함된 SimplifiedBaleen 프로그램의 모습을 살펴볼게요.

class SimplifiedBaleenAssertions(dspy.Module):
    def __init__(self, passages_per_hop=2, max_hops=2):
        super().__init__()
        self.generate_query = [dspy.ChainOfThought(GenerateSearchQuery) for _ in range(max_hops)]
        self.retrieve = dspy.Retrieve(k=passages_per_hop)
        self.generate_answer = dspy.ChainOfThought(GenerateAnswer)
        self.max_hops = max_hops

    def forward(self, question):
        context = []
        prev_queries = [question]

        for hop in range(self.max_hops):
            query = self.generate_query[hop](context=context, question=question).query

            dspy.Suggest(
                len(query) <= 100,
                "Query should be short and less than 100 characters",
                target_module=self.generate_query
            )

            dspy.Suggest(
                validate_query_distinction_local(prev_queries, query),
                "Query should be distinct from: "
                + "; ".join(f"{i+1}) {q}" for i, q in enumerate(prev_queries)),
                target_module=self.generate_query
            )

            prev_queries.append(query)
            passages = self.retrieve(query).passages
            context = deduplicate(context + passages)
        
        if all_queries_distinct(prev_queries):
            self.passed_suggestions += 1

        pred = self.generate_answer(context=context, question=question)
        pred = dspy.Prediction(context=context, answer=pred.answer)
        return pred

이제 DSPy Assertions로 프로그램을 호출하려면 마지막 단계가 하나 더 필요해요. 내부 assertion 백트래킹과 Retry 로직으로 프로그램을 감싸도록 변환하는 것입니다.

from dspy.primitives.assertions import assert_transform_module, backtrack_handler

baleen_with_assertions = assert_transform_module(SimplifiedBaleenAssertions(), backtrack_handler)

# backtrack_handler is parameterized over a few settings for the backtracking mechanism
# To change the number of max retry attempts, you can do
baleen_with_assertions_retry_once = assert_transform_module(SimplifiedBaleenAssertions(), 
    functools.partial(backtrack_handler, max_backtracks=1))

대안으로, 기본 백트래킹 메커니즘(max_backtracks=2)으로 dspy.Assert/Suggest 문이 있는 프로그램에서 activate_assertions를 직접 호출할 수도 있어요.

baleen_with_assertions = SimplifiedBaleenAssertions().activate_assertions()

이제 LM 쿼리 생성의 history를 검사해 내부 LM 백트래킹을 살펴볼게요. 쿼리가 100자 미만이라는 검증을 통과하지 못하면, 백트래킹+Retry 과정에서 그 GenerateSearchQuery 시그니처가 동적으로 수정되어 과거 쿼리와 사용자 정의 지시사항: "Query should be short and less than 100 characters"를 포함하게 되는 걸 볼 수 있어요.

Write a simple search query that will help answer a complex question.

---

Follow the following format.

Context: may contain relevant facts

Question: ${question}

Reasoning: Let's think step by step in order to ${produce the query}. We ...

Query: ${query}

---

Context:
[1] «Kerry Condon | Kerry Condon (born 4 January 1983) is [...]»
[2] «Corona Riccardo | Corona Riccardo (c. 1878October 15, 1917) was [...]»

Question: Who acted in the shot film The Shore and is also the youngest actress ever to play Ophelia in a Royal Shakespeare Company production of "Hamlet." ?

Reasoning: Let's think step by step in order to find the answer to this question. First, we need to identify the actress who played Ophelia in a Royal Shakespeare Company production of "Hamlet." Then, we need to find out if this actress also acted in the short film "The Shore."

Query: "actress who played Ophelia in Royal Shakespeare Company production of Hamlet" + "actress in short film The Shore"



Write a simple search query that will help answer a complex question.

---

Follow the following format.

Context: may contain relevant facts

Question: ${question}

Past Query: past output with errors

Instructions: Some instructions you must satisfy

Query: ${query}

---

Context:
[1] «Kerry Condon | Kerry Condon (born 4 January 1983) is an Irish television and film actress, best known for her role as Octavia of the Julii in the HBO/BBC series "Rome," as Stacey Ehrmantraut in AMC's "Better Call Saul" and as the voice of F.R.I.D.A.Y. in various films in the Marvel Cinematic Universe. She is also the youngest actress ever to play Ophelia in a Royal Shakespeare Company production of "Hamlet."»
[2] «Corona Riccardo | Corona Riccardo (c. 1878October 15, 1917) was an Italian born American actress who had a brief Broadway stage career before leaving to become a wife and mother. Born in Naples she came to acting in 1894 playing a Mexican girl in a play at the Empire Theatre. Wilson Barrett engaged her for a role in his play "The Sign of the Cross" which he took on tour of the United States. Riccardo played the role of Ancaria and later played Berenice in the same play. Robert B. Mantell in 1898 who struck by her beauty also cast her in two Shakespeare plays, "Romeo and Juliet" and "Othello". Author Lewis Strang writing in 1899 said Riccardo was the most promising actress in America at the time. Towards the end of 1898 Mantell chose her for another Shakespeare part, Ophelia im Hamlet. Afterwards she was due to join Augustin Daly's Theatre Company but Daly died in 1899. In 1899 she gained her biggest fame by playing Iras in the first stage production of Ben-Hur.»

Question: Who acted in the shot film The Shore and is also the youngest actress ever to play Ophelia in a Royal Shakespeare Company production of "Hamlet." ?

Past Query: "actress who played Ophelia in Royal Shakespeare Company production of Hamlet" + "actress in short film The Shore"

Instructions: Query should be short and less than 100 characters

Query: "actress Ophelia RSC Hamlet" + "actress The Shore"

Assertion 기반 최적화

DSPy Assertions는 DSPy가 제공하는 최적화와 함께 동작합니다. 특히 BootstrapFewShotWithRandomSearch와 함께 다음 설정을 지원해요.

  • Compilation with Assertions — 컴파일 중 assertion 기반 예시 부트스트래핑과 반례(counterexample) 부트스트래핑을 포함합니다. few-shot 데모 부트스트래핑을 위한 teacher 모델이 DSPy Assertions를 활용해, 추론 중 student 모델이 배울 견고한 부트스트랩 예시를 제공할 수 있어요. 이 설정에서 student 모델은 추론 중 assertion 인지 최적화(백트래킹과 리트라이)를 수행하지 않습니다.
  • Compilation + Inference with Assertions — 컴파일과 추론 양쪽에 assertion 기반 최적화를 포함합니다. 이제 teacher 모델이 assertion 기반 예시를 제공하지만, student도 추론 중 자신만의 assertion으로 더 최적화할 수 있어요.
teleprompter = BootstrapFewShotWithRandomSearch(
    metric=validate_context_and_answer_and_hops,
    max_bootstrapped_demos=max_bootstrapped_demos,
    num_candidate_programs=6,
)

#Compilation with Assertions
compiled_with_assertions_baleen = teleprompter.compile(student = baleen, teacher = baleen_with_assertions, trainset = trainset, valset = devset)

#Compilation + Inference with Assertions
compiled_baleen_with_assertions = teleprompter.compile(student=baleen_with_assertions, teacher = baleen_with_assertions, trainset=trainset, valset=devset)