DSPy(Declarative Self-improving Language Programs, pythonically)
프롬프트 엔지니어링을 위해 흔히 사용하는 LangChain 등의 프레임워크 에서 특정 작업을 위해 사용하는 프롬프트의 크기가 굉장히 크다.
특히, trial and error 방식으로 사전에 수작업으로 작성된 프롬프트를 사용하기 때문에 보편적이지만, 성능이나 효율성 측면에서는 최선이 아닐뿐더러, 확장성 측면에서는 한계가 있다.
이러한 프롬프트 엔지니어링을 극복하기 위해서 파이썬 스타일로 작성된 선언적이고 스스로 개선되는 기능을 갖춘 자연어 처리 프로그램인 DSPy가 나왔다, 이 프레임워크에서는 LLM 파이프라인이 무엇을 할 것인지를 명확히 선언하면, 내부적으로 스스로 학습하고 최적화하여 성능을 향상시키는 기능이 있다.
DSPy는 간단한 API를 통해 빠르게 시작할 수 있지만, AI 시스템을 구현하는 과정은 반복적인 개선이 필요하다. 이를 위해 DSPy는 크게 프로그래밍(작업 정의 및 파이프라인 설계), 평가(데이터셋 수집 및 메트릭 정의), 최적화(프롬프트 및 가중치 조정)라는 3단계를 권장하며, 각 단계를 차례로 밟아 나가는 것이 효율적이며, 부실한 초기 설계나 지표로 최적화를 시도하는 것은 비효율적입니다.
1) 프로그래밍
DSPy에서는 코드 기반의 제어 흐름 설계가 핵심이다. 먼저 시스템의 입력과 출력 목표를 명확히 하고, 이를 위한 초기 파이프라인을 단순하게 잡은 뒤, 필요에 따라 검색 기능이나 추가 API를 도입한다. 작업 과정에서 다양한 입력 예시를 실험하며, 강력한 언어 모델도 활용해 가능한 해결 방식을 탐색한다. 이를 통해 구축한 예시들은 이후 평가 및 최적화 단계에 도움이 된다.
(1) LLM Model
아래의 코드는 환경 변수 설정, 기본 LM 구성, LM 직접 호출, DSPy 모듈과의 연계, LM 전환, 파라미터 설정, 호출 기록 확인 등 주요 기능을 코드로 요약한 것이다.
import dspy
import os
# 1. 환경 변수로 인증 (OPENAI_API_KEY)
os.environ['OPENAI_API_KEY'] = 'YOUR_OPENAI_API_KEY'
# 2. 언어 모델 설정(예: OpenAI GPT-4o-mini)
lm = dspy.LM('openai/gpt-4o-mini')
dspy.configure(lm=lm)
# 3. LM 직접 호출
result = lm("Say this is a test!", temperature=0.7)
print(result) # => ['This is a test!']
# 4. Message 형식으로 호출
result = lm(messages=[{"role": "user", "content": "Say this is a test!"}])
print(result) # => ['This is a test!']
# 5. DSPy 모듈 사용 (예: ChainOfThought)
qa = dspy.ChainOfThought('question -> answer')
response = qa(question="How many floors are in the castle David Gregory inherited?")
print(response.answer)
# 6. dspy.configure 또는 dspy.context를 사용한 LM 교체
with dspy.context(lm=dspy.LM('openai/gpt-3.5-turbo')):
response = qa(question="How many floors are in the castle David Gregory inherited?")
print(response.answer)
# 7. LM 생성 파라미터 설정
gpt_4o_mini = dspy.LM('openai/gpt-4o-mini', temperature=0.9, max_tokens=3000, stop=None, cache=False)
# 8. LM 호출 기록 조회
print(len(lm.history)) # LM 호출 횟수
print(lm.history[-1].keys()) # 마지막 호출 관련 메타데이터 확인DSPy의 시그니처(Signature)는 모듈의 입력과 출력을 선언적으로 정의하여 LM에게 해야 할 일을 명확하게 전달하는 기능이다. 이는 단순히 함수의 입출력을 기술하는 전통적 시그니처와 달리, 의미 있는 필드명(예: question, answer)을 통해 모델이 수행할 역할을 명확히 한다. 시그니처를 사용하면 긴 프롬프트를 직접 작성하지 않고도 모듈화·재현 가능한 형태로 LM 동작을 정의하고, DSPy 컴파일러와 최적화기를 활용해 고품질의 프롬프트나 파인튜닝을 자동으로 얻을 수 있다.
시그니처는 간단히 문자열로 정의할 수 있으며, 입력/출력 필드를 여러 개 둘 수도 있고, 필요하다면 클래스 기반으로 작성하여 docstring, desc 키워드, 타입 제약(Literal 등)으로 태스크 성격을 더욱 명확히 표현할 수 있다. 이로써 감정 분석, 질의응답, 요약, 인용 검증 등 다양한 태스크에 적용할 수 있고, DSPy 모듈들을 조합·컴파일하여 더욱 최적화된 LM 파이프라인을 구축할 수 있다.
import dspy
from typing import Literal
# 1. 인라인 시그니처 예시 (질문-응답)
qa = dspy.ChainOfThought('question -> answer')
response = qa(question="How many floors are in the castle David Gregory inherited?")
print(response.answer)
# 2. 인라인 시그니처 예시 (감정 분석)
sentence = "it's a charming and often affecting journey."
classify = dspy.Predict('sentence -> sentiment: bool')
print(classify(sentence=sentence).sentiment) # True/False
# 3. 인라인 시그니처 예시 (문서 요약)
document = """The 21-year-old made seven appearances for the Hammers ..."""
summarize = dspy.ChainOfThought('document -> summary')
response = summarize(document=document)
print(response.summary)
print(response.reasoning) # ChainOfThought는 reasoning 필드를 자동으로 추가
# 4. 클래스 기반 시그니처 예시 (감정 분류)
class Emotion(dspy.Signature):
"""Classify emotion."""
sentence: str = dspy.InputField()
sentiment: Literal['sadness', 'joy', 'love', 'anger', 'fear', 'surprise'] = dspy.OutputField()
emotion_classify = dspy.Predict(Emotion)
print(emotion_classify(sentence="i started feeling a little vulnerable...").sentiment)
# 5. 클래스 기반 시그니처 예시 (인용 검증)
class CheckCitationFaithfulness(dspy.Signature):
"""Verify that the text is based on the provided context."""
context: str = dspy.InputField(desc="facts here are assumed to be true")
text: str = dspy.InputField()
faithfulness: bool = dspy.OutputField()
evidence: dict[str, list[str]] = dspy.OutputField(desc="Supporting evidence for claims")
context = "The 21-year-old made seven appearances for the Hammers..."
text = "Lee scored 3 goals for Colchester United."
faithfulness = dspy.ChainOfThought(CheckCitationFaithfulness)
result = faithfulness(context=context, text=text)
print(result.reasoning)
print(result.faithfulness)
print(result.evidence)
(3) Module
DSPy 모듈은 LM 호출을 추상화하는 빌딩 블록이다.
- dspy.Predict: 기본 예측자. 시그니처에 맞춰 LM 호출.
- dspy.ChainOfThought: "사고의 흐름"을 통해 단계적으로 답을 도출하도록 LM 유도.
- dspy.ProgramOfThought: 코드를 출력하고 이를 실행한 결과로 답을 얻는 모듈.
- dspy.ReAct: 에이전트 방식을 이용해 도구를 사용 가능.
- dspy.MultiChainComparison: 여러 답변을 비교하여 최종 답변을 산출.
이들은 모두 시그니처를 통해 입력과 출력을 선언적으로 정의하고, 모듈을 호출해 결과를 얻는다. 출력은 필드 형태로 접근하며, ChainOfThought 모듈 등은 reasoning 필드를 추가 제공한다. 모듈을 호출할 때 n, temperature, max_len 등의 파라미터로 결과를 제어할 수 있고, 동일한 서명(signature)을 사용하여 다양한 모듈을 시도해볼 수 있다.
여러 모듈을 조합해 더 큰 프로그램을 만들 때에도 그냥 Python 함수를 호출하듯 모듈들을 호출하고 결과를 전달하면 된다.
아래 예제들을 통해 DSPy 모듈을 선언하고, 시그니처를 지정하며, 모듈 호출 시 입력 인자를 전달하고 출력 필드를 접근하는 방법을 확인할 수 있다. 또한 모듈을 조합하여 더 큰 파이프라인을 구축하는 것도 단순히 Python 함수 호출처럼 처리할 수 있다.
import dspy
# 1. 기본 예측 (Predict) 예제
sentence = "it's a charming and often affecting journey."
classify = dspy.Predict('sentence -> sentiment: bool')
response = classify(sentence=sentence)
print(response.sentiment) # True/False
# 2. ChainOfThought 예제: 질문-답변
question = "What's something great about the ColBERT retrieval model?"
qa = dspy.ChainOfThought('question -> answer', n=5) # n=5: 5개 결과 요청
response = qa(question=question)
print("Answer:", response.answer) # 첫 번째 답변
print("Reasoning:", response.reasoning) # 사고 과정(ChainOfThought에서 자동 추가)
print("All answers:", response.completions.answer) # 모든 답변 리스트
# 3. 다양한 모듈 예시 (수학 문제)
math = dspy.ChainOfThought("question -> answer: float")
response = math(question="Two dice are tossed. What is the probability that the sum equals two?")
print("Reasoning:", response.reasoning)
print("Answer:", response.answer)
# 4. 복잡한 모듈 조합
# DSPy는 PyTorch처럼 동적 계산 그래프 스타일로 모듈을 사용할 수 있다.
# 즉, 그냥 Python 함수 호출하듯 모듈 호출 결과를 다른 모듈에 전달 가능.
# 아래는 예시적인 구조(실제 동작 예시는 상황에 따라 변경 가능):
# 가상의 시나리오: 먼저 어떤 문서에서 문맥(context)을 추출한 뒤,
# 해당 문맥을 바탕으로 질의응답을 수행.
# context 추출 모듈(예: 단순 Predict 기반, 실제 구현은 임의)
extract_context = dspy.Predict("document -> context: str")
# 질의응답 모듈(ChainOfThought 이용)
qa_context = dspy.ChainOfThought("context: str, question: str -> answer")
# 문서에서 context 추출
doc = "Some long document text..."
context_result = extract_context(document=doc)
# 추출한 context를 이용해 질의응답
final_answer = qa_context(context=context_result.context, question="What is the main topic?")
print(final_answer.answer)
2) Evaluation
- DSPy 내의 설명
초기 시스템을 갖춘 뒤에는, 체계적으로 개선할 수 있도록 초기 개발용 예시 데이터를 모아 20개 정도의 입력 예시만 있어도 시작할 수 있으며, 200개 정도면 훨씬 도움이 됩니다. 작업에 따라 입력만 필요하거나, 최종 출력까지 필요할 수 있습니다. DSPy에서는 중간 단계 라벨링은 거의 필요 없습니다.
HuggingFace Datasets나 StackExchange처럼 이미 존재하는 데이터 소스를 활용해보거나, 라이선스가 허용되는 데이터를 활용할 수 있습니다. 그렇지 않다면 직접 몇 개의 예시를 라벨링하거나, 시스템 데모를 배포하여 데이터를 모을 수도 있습니다.
다음으로는 DSPy 평가지표(metric)를 정의하세요. 시스템 출력을 어떻게 좋고 나쁨으로 판단할지 정하고, 점진적으로 개선해나가는 것이 중요합니다. 단순한 작업은 정확도 같은 간단한 지표로 시작할 수 있지만, 복잡한 작업이라면 다양한 속성을 체크하는 작은 DSPy 프로그램을 만들어 지표로 삼으세요.
마지막으로, 수집한 데이터와 평가지표로 파이프라인을 평가해보세요. 결과 출력과 점수를 살펴보면서 주요 문제를 파악하고, 이를 바탕으로 다음 개선 단계를 계획할 수 있습니다.
(1) Data Handling
DSPy의 Example 객체를 이용해 데이터셋을 구성하고 활용하는 방법에 대한 정리이다. 코드를 중심으로 쉽게 이해할 수 있도록 구성하였다.
Example 객체는 기본적으로 Python의 dict와 유사하지만, 다음과 같은 편의 기능을 제공한다.
- .(dot) 연산자를 통한 값 접근
- 입력 키(Input Keys) 지정 기능
- 훈련 예제(Example)와 모델 예측(Prediction) 데이터 타입을 동일하게 다룰 수 있는 구조
예를 들어, 다음과 같이 Example 객체를 만들 수 있다. (Example 객체는 key-value 쌍으로 이루어져 있으며, 다양한 타입의 값을 가질 수 있다(주로 문자열).)
import dspy
qa_pair = dspy.Example(question="This is a question?", answer="This is an answer.")
print(qa_pair)
print(qa_pair.question)
print(qa_pair.answer)
# 출력 예시
Example({'question': 'This is a question?', 'answer': 'This is an answer.'}) (input_keys=None)
This is a question?
This is an answer.
훈련 세트 구성
여러 개의 Example 객체를 리스트로 묶어 학습용 데이터셋(Training set)을 구성할 수 있다.
trainset = [
dspy.Example(report="LONG REPORT 1", summary="short summary 1"),
dspy.Example(report="LONG REPORT 2", summary="short summary 2"),
# ...
]
Input Keys 지정하기
전통적인 머신 러닝에서 입력 데이터와 레이블을 구분하듯, DSPy에서도 Example 객체의 특정 필드를 입력으로 지정할 수 있다. 이를 위해 with_inputs() 메서드를 사용한다.
# 단일 필드 입력 지정
qa_input = qa_pair.with_inputs("question")
print(qa_input)
# 다중 필드 입력 지정 (필요에 따라 레이블도 입력으로 지정 가능)
qa_input_output = qa_pair.with_inputs("question", "answer")
print(qa_input_output)
Inputs와 Labels 분리
Example 객체에서 입력키와 비입력키를 분리하여 확인할 수도 있습니다. inputs() 메서드는 입력 키만 반환하며, labels() 메서드는 그 외(레이블 등) 키만 반환합니다.
article_summary = dspy.Example(
article="This is an article.",
summary="This is a summary."
).with_inputs("article")
# 입력 필드만 추출
input_key_only = article_summary.inputs()
# 비입력 필드만 추출
non_input_key_only = article_summary.labels()
print("Input fields only:", input_key_only)
print("Non-Input fields only:", non_input_key_only)
#출력 예시
Input fields only: Example({'article': 'This is an article.'}) (input_keys=None)
Non-Input fields only: Example({'summary': 'This is a summary.'}) (input_keys=None)
(2) Metrics
- 지표(metric): 모델 출력과 정답을 비교하여 점수를 반환하는 함수.
- 간단한 지표 예: 정확 일치 여부를 bool로 반환.
- 복잡한 지표 예: 정답 매칭, 문맥 매칭 등 다차원 평가 후 float 점수 반환.
- LLM 피드백 활용: LLM을 통해 장문 출력 품질(정확성, 참여도 등)을 평가.
- trace 활용: 최적화 시점에 프로그램 실행 단계를 추적하고, 이를 활용해 내부 과정까지 평가 가능.
# 간단한 지표: 예측 답변이 정답과 정확히 일치하는지 확인하는 함수
def validate_answer(example, pred, trace=None):
return example.answer.lower() == pred.answer.lower()
# 좀 더 복잡한 지표: 답변 정확도와 문맥 출처 여부를 동시에 평가
def validate_context_and_answer(example, pred, trace=None):
answer_match = example.answer.lower() == pred.answer.lower()
context_match = any((pred.answer.lower() in c) for c in pred.context)
if trace is None:
# 평가 및 최적화 시에는 float 점수 반환
return (answer_match + context_match) / 2.0
else:
# 부트스트래핑 시에는 bool 반환
return answer_match and context_match
# 기본적인 평가 루프 예시 (실제 devset, program, metric 대입 필요)
'''
scores = []
for x in devset:
pred = program(**x.inputs())
score = metric(x, pred)
scores.append(score)
'''
# dspy.evaluate.Evaluate 유틸리티 사용 예시
from dspy.evaluate import Evaluate
# evaluator = Evaluate(devset=YOUR_DEVSET, num_threads=1, display_progress=True, display_table=5)
# evaluator(YOUR_PROGRAM, metric=YOUR_METRIC)
# LLM 피드백을 위한 시그니처 정의
import dspy
class Assess(dspy.Signature):
"""Assess 시그니처: LLM을 통해 텍스트가 기준에 부합하는지 boolean으로 평가"""
assessed_text = dspy.InputField()
assessment_question = dspy.InputField()
assessment_answer: bool = dspy.OutputField()
# LLM 피드백을 활용한 복잡한 지표 예시
def metric(gold, pred, trace=None):
question, answer, tweet = gold.question, gold.answer, pred.output
# 평가 기준 문구 정의
engaging_query = "Does the assessed text make for a self-contained, engaging tweet?"
correct_query = f"The text should answer `{question}` with `{answer}`. Does the assessed text contain this answer?"
# LLM 호출을 통한 평가
correct = dspy.Predict(Assess)(assessed_text=tweet, assessment_question=correct_query)
engaging = dspy.Predict(Assess)(assessed_text=tweet, assessment_question=engaging_query)
correct, engaging = [m.assessment_answer for m in [correct, engaging]]
score = (correct + engaging) if correct and (len(tweet) <= 280) else 0
# trace 있을 때(부트스트래핑)에는 bool, 없을 때는 float 점수 반환
if trace is not None:
return score >= 2
return score / 2.0
# trace를 활용한 평가 예시
def validate_hops(example, pred, trace=None):
# trace에서 쿼리 히스토리(hops) 추출: 첫 질문 + 이후 단계별 query
hops = [example.question] + [outputs.query for *_, outputs in trace if 'query' in outputs]
# 조건 1: hop 중 하나라도 너무 길면 False
if max(len(h) for h in hops) > 100:
return False
# 조건 2: 이전 hop과 80% 이상 동일한 쿼리가 재사용되는 경우 False
# (dspy.evaluate.answer_exact_match_str 사용)
if any(dspy.evaluate.answer_exact_match_str(hops[idx], hops[:idx], frac=0.8) for idx in range(2, len(hops))):
return False
return True
3) Optimization
DSPy를 활용한 프로그램 최적화 과정은 반복적이고 단계적으로 진행된다. 다음은 핵심 포인트의 요약이다.
- 데이터 세트 구성:
- 시스템과 평가 방법을 마련한 뒤, 학습(train), 검증(validation), 테스트(test) 세트를 구축한다.
- 최소 30개, 가능하면 300개 이상의 학습 예시를 마련하고, 일부 최적화 기법은 학습 세트만 필요하지만, 다른 경우 학습 및 검증 세트를 모두 요구한다.
- 특히 프롬프트 최적화 시, 일반적인 딥러닝과 반대로 학습 데이터 20%, 검증 데이터 80%로 시작하는 것을 권장한다.
- 최적화 후 재검토:
- 초기 최적화 결과가 만족스럽지 않다면, 문제 정의, 데이터 수집, 평가 지표, 사용한 최적화 방법 등을 재검토한다.
- DSPy Assertions를 고려하거나, 프로그램 구조를 복잡하게 바꾸어 보는 등 다양한 시도를 통해 성능 개선을 모색한다.
- 반복적 개발 프로세스:
- DSPy는 데이터, 프로그램 구조, 가설(Assertions), 평가 지표, 최적화 절차 등을 순차적으로 개선할 수 있는 틀을 제공한다.
- 이를 통해 점진적으로 성능과 품질을 끌어올리는 반복적(Iterative) 개발이 가능하다.
- 커뮤니티 리소스 활용:
- DSPy 관련 노하우는 아직 형성 중이며, 최근 개설된 커뮤니티 디스코드 서버를 통해 도움을 받을 수 있다.
요약하자면, DSPy 최적화 프로세스는 충분한 데이터 확보, 적절한 세트 구성, 반복적 개선, 그리고 커뮤니티 지원 활용을 통한 지속적 발전을 강조한다.
(1) Optimizer
DSPy Optimizer란?
- DSPy 최적화기는 DSPy 프로그램의 파라미터(프롬프트, LM 가중치 등)를 최적화하여 주어진 평가 지표(예: 정확도)를 최대화하는 알고리즘이다.
- 단순한 예시부터 복잡한 멀티 모듈 프로그램까지 다양한 형태의 DSPy 프로그램에 적용할 수 있으며, 몇 개의 예시 데이터만으로도 시작 가능하다. 많은 데이터가 있으면 더욱 효과적이다.
DSPy Optimizer가 조정하는 것:
- 예시(데모) 제작: dspy.BootstrapRS 등은 모듈별로 더 나은 few-shot 예시를 자동 생성.
- 프롬프트 개선: dspy.MIPROv2는 주어진 텍스트 명령어(Instructions)를 개선하고, 더 나은 설명 예시를 탐색.
- LM 파라미터 미세조정(Finetuning): dspy.BootstrapFinetune는 만든 프롬프트 기반 프로그램을 파인튜닝하여 모델 가중치 자체를 수정.
대표적인 DSPy Optimizer 예시:
- LabeledFewShot: 제공된 라벨 데이터 포인트를 사용해 간단한 few-shot 예시 생성.
- BootstrapFewShot: 동일 프로그램(teacher)을 사용해 추가 데모를 생성하고, metric을 통해 검증해 최적의 데모만 채택.
- BootstrapFewShotWithRandomSearch: BootstrapFewShot을 반복 실행하며 랜덤 탐색으로 더 많은 후보 프로그램 중 최고의 프로그램을 선택.
- KNNFewShot: 입력 예제에 대한 k-최근접 이웃(k-NN) 기반으로 최적의 few-shot 예시를 선택하여 BootstrapFewShot 최적화를 수행.
- COPRO/MIPROv2: 주어진 단계별 인스트럭션을 개선하거나, 베이지안 최적화를 통해 최적 지시문과 데모를 자동 탐색.
- BootstrapFinetune: 프롬프트 기반 프로그램을 활용해 실제 파라미터(가중치)를 업데이트하고, 이를 파인튜닝한 모델을 DSPy 프로그램에 통합.
사용 가이드:
- 예시 수가 매우 적다면(BootstrapFewShot), 좀 더 많다면(BootstrapFewShotWithRandomSearch), 인스트럭션 최적화만 하려면(MIPROv2), 더 길고 많은 데이터와 자원을 사용하려면(MIPROv2) 등 상황에 따라 다른 Optimizer를 시도해볼 수 있다.
- 파인튜닝으로 효율적이고 빠른 모델을 원한다면(BootstrapFinetune)을 고려.
최적화 비용:
- 간단한 최적화는 수 달러 내외로 해결 가능하지만, 대형 모델이나 큰 데이터세트 사용 시 수십 달러 이상이 들 수 있으니 주의.
프로그램 결과 저장 및 재사용:
- 최적화 결과를 .save() 메서드로 파일 형태로 저장 가능.
- 이후 .load() 메서드를 통해 저장된 프로그램을 불러와 재사용할 수 있음.
# 필요한 Optimizer 임포트
from dspy.teleprompt import BootstrapFewShotWithRandomSearch, MIPROv2
import dspy
from dspy.datasets import HotPotQA
# LM 설정: 대규모 언어모델(LLM) 설정
dspy.configure(lm=dspy.LM('openai/gpt-4o-mini'))
# 검색 함수 예시 - ReAct 에이전트에서 사용할 툴
def search(query: str) -> list[str]:
"""
검색 함수를 정의합니다.
query에 대해 Wikipedia 추출을 수행하고
결과 리스트를 반환합니다.
"""
results = dspy.ColBERTv2(url='http://20.102.90.50:2017/wiki17_abstracts')(query, k=3)
return [x['text'] for x in results]
# HotPotQA에서 500개의 QA 예시 로드
trainset = [x.with_inputs('question') for x in HotPotQA(train_seed=2024, train_size=500).train]
# ReAct 에이전트 정의: 도구로 위의 search 함수를 사용
react = dspy.ReAct("question -> answer", tools=[search])
# MIPROv2 Optimizer 설정:
# metric: 정답 정확도 평가 함수 사용
# auto="light": 적은 비용 모드
# num_threads=24: 병렬 처리
tp = MIPROv2(metric=dspy.evaluate.answer_exact_match, auto="light", num_threads=24)
# Optimizer로 ReAct 프로그램 최적화 수행
optimized_react = tp.compile(react, trainset=trainset)
# 최적화 결과 저장
optimized_react.save("optimized_react_program.json")
# 이후 사용할 때 로드하기
loaded_program = type(react)() # react와 동일한 타입의 객체 생성
loaded_program.load("optimized_react_program.json")