DSPy: Declarative Self-Improving LLM Pipelines

In frameworks like LangChain that are commonly used for prompt engineering, the prompts written for specific tasks tend to be very large.

In particular, because these prompts are manually crafted upfront through trial and error, they are widespread but fall short in terms of performance and efficiency, and have clear scalability limitations.

DSPy was developed to address these prompt engineering shortcomings. It is a declarative, self-improving natural language processing framework written in a Pythonic style. You declare what an LLM pipeline should do, and DSPy internally learns and optimizes itself to improve performance.

DSPy provides a simple API to get started quickly, but building a real AI system requires iterative refinement. To support this, DSPy recommends three broad stages: programming (defining the task and designing the pipeline), evaluation (collecting datasets and defining metrics), and optimization (tuning prompts and weights). Working through these stages in order is the efficient path; attempting optimization on top of a weak initial design or poorly defined metrics wastes effort.

1. Programming

In DSPy, the core work is designing code-based control flow. Start by clearly defining the system's inputs and outputs, sketch a simple initial pipeline, and then introduce retrieval or additional APIs as needed. Experiment with diverse input examples along the way, and use powerful language models to explore possible solution strategies. The examples you build up here feed directly into the evaluation and optimization stages.

(1) LLM Model

The code below summarizes the key capabilities: setting environment variables, configuring the default LM, calling the LM directly, integrating with DSPy modules, switching between LMs, setting generation parameters, and inspecting call history.

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())  # 마지막 호출 관련 메타데이터 확인

(2) Signatures

A DSPy Signature declaratively defines the inputs and outputs of a module, giving the LM a clear specification of what it needs to do. Unlike a traditional function signature that simply describes input/output types, a DSPy Signature uses meaningful field names (e.g., question, answer) to explicitly communicate the role the model should play. Using signatures, you can define LM behavior in a modular and reproducible way without hand-writing long prompts, and then use DSPy's compiler and optimizers to automatically derive high-quality prompts or fine-tuning.

Signatures can be defined concisely as strings, can include multiple input/output fields, and can alternatively be written as classes with docstrings, desc keywords, and type constraints (e.g., Literal) to express the nature of the task more precisely. This makes them applicable to a wide range of tasks — sentiment analysis, question answering, summarization, citation verification, and more — and you can compose and compile DSPy modules to build increasingly optimized LM pipelines.

Here are some example signatures:

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

DSPy modules are the building blocks that abstract LM calls.

• dspy.Predict: The basic predictor. Calls the LM according to the signature.
• dspy.ChainOfThought: Guides the LM to derive an answer step by step through a chain of thought.
• dspy.ProgramOfThought: Outputs code and uses the result of executing that code as the answer.
• dspy.ReAct: Enables tool use via an agent-style approach.
• dspy.MultiChainComparison: Compares multiple candidate answers to produce a final answer.

All of these declaratively define inputs and outputs via a signature and return results when called. Outputs are accessed as fields; modules like ChainOfThought additionally expose a reasoning field. When calling a module, you can control generation with parameters such as n, temperature, and max_len, and you can try different modules against the same signature.

Composing multiple modules into a larger program is straightforward — just call modules and pass their outputs to other modules, exactly like calling Python functions.

The examples below show how to declare DSPy modules, assign signatures, pass input arguments when calling a module, and access output fields. Composing modules into a larger pipeline is handled just like regular Python function calls.

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)

1. Evaluation

• Notes from within DSPy

Once you have an initial system, gather a small set of development examples to enable systematic improvement — as few as 20 input examples is enough to get started, and around 200 is considerably more helpful. Depending on the task, you may need only inputs, or you may need final outputs as well. DSPy rarely requires labels for intermediate steps.

Try existing data sources such as HuggingFace Datasets or StackExchange, or use any data whose license permits it. If none is available, manually label a few examples yourself or deploy a system demo to collect data organically.

Next, define a DSPy metric. Decide how you will judge whether system outputs are good or bad, and plan to refine that judgment iteratively. Simple tasks can start with a straightforward metric like accuracy, but for complex tasks, build a small DSPy program that checks multiple attributes and use that as your metric.

Finally, evaluate your pipeline using the data and metric you have collected. Examine the outputs and scores to identify the main failure modes, and use that understanding to plan the next round of improvements.

(1) Data Handling

This section covers how to construct and use datasets with DSPy's Example object, organized around code examples for clarity.

An Example object is essentially similar to a Python dict, but provides the following conveniences:

• Value access via the dot (.) operator
• The ability to designate input keys
• A unified data type for both training examples and model predictions

For instance, you can create an Example object as follows. (An Example consists of key-value pairs and can hold values of various types, most commonly strings.)

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.

Training Set Construction

Multiple Example objects can be grouped into a list to form a training dataset.

trainset = [
    dspy.Example(report="LONG REPORT 1", summary="short summary 1"),
    dspy.Example(report="LONG REPORT 2", summary="short summary 2"),
    # ...
]

Specifying Input Keys

Just as traditional machine learning separates input data from labels, DSPy lets you designate specific fields of an Example object as inputs using the with_inputs() method.

# 단일 필드 입력 지정
qa_input = qa_pair.with_inputs("question")
print(qa_input)

# 다중 필드 입력 지정 (필요에 따라 레이블도 입력으로 지정 가능)
qa_input_output = qa_pair.with_inputs("question", "answer")
print(qa_input_output)

Separating Inputs and Labels

You can also split an Example object into its input keys and non-input keys. The inputs() method returns only the input keys, while labels() returns everything else (e.g., 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: a function that compares model output against a ground-truth answer and returns a score.
• Simple metric example: returns a bool indicating an exact match.
• Complex metric example: evaluates multiple dimensions such as answer matching and context matching, returning a float score.
• LLM feedback: use an LLM to assess the quality of long-form outputs (correctness, engagement, etc.).
• trace usage: during optimization, the execution trace of each program step can be captured and used to evaluate intermediate behavior, not just final outputs.

# 간단한 지표: 예측 답변이 정답과 정확히 일치하는지 확인하는 함수
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

1. Optimization

Optimizing a program with DSPy is an iterative, incremental process. Here are the key points.

1. Build your datasets:

• Set up your system and evaluation methodology first, then construct train, validation, and test sets.
• Aim for at least 30 examples, preferably 300 or more. Some optimizers only need a training set; others require both training and validation sets.
• For prompt optimization in particular, it is recommended to start with roughly 20% training data and 80% validation data — the inverse of typical deep learning practice.

1. Review results after optimization:

• If the initial optimization results are unsatisfactory, revisit your problem definition, data collection, evaluation metrics, and choice of optimizer.
• Consider using DSPy Assertions, or try restructuring your program to be more complex — experiment with different approaches to drive improvement.

1. Iterative development process:

• DSPy provides a framework for sequentially refining data, program structure, Assertions, evaluation metrics, and optimization procedures.
• This enables iterative development that gradually improves performance and output quality.

1. Leverage community resources:

• Best practices for DSPy are still evolving. The recently launched community Discord server is a good place to get help.

In summary, the DSPy optimization process emphasizes acquiring sufficient data, constructing well-balanced dataset splits, iterating continuously, and leveraging community support.

(1) Optimizer
What is a DSPy Optimizer?

• A DSPy optimizer is an algorithm that tunes the parameters of a DSPy program — such as prompts and LM weights — to maximize a given metric (e.g., accuracy).
• It can be applied to DSPy programs ranging from simple single-module setups to complex multi-module pipelines, and can get started with just a handful of examples. More data generally yields better results.

What DSPy Optimizers tune:

• Demo generation: optimizers like dspy.BootstrapRS automatically generate better few-shot examples for each module.
• Prompt improvement: dspy.MIPROv2 refines the text instructions and searches for better demonstration examples.
• LM fine-tuning: dspy.BootstrapFinetune takes a prompt-based program and fine-tunes the model weights directly.

Representative DSPy Optimizers:

• LabeledFewShot: generates simple few-shot examples from provided labeled data points.
• BootstrapFewShot: uses the same program (as a teacher) to generate additional demos, then filters them with the metric and keeps only the best ones.
• BootstrapFewShotWithRandomSearch: runs BootstrapFewShot multiple times with random search to explore a larger candidate pool and select the best program.
• KNNFewShot: selects optimal few-shot examples using k-nearest neighbors over input examples, then runs BootstrapFewShot optimization.
• COPRO / MIPROv2: refines per-step instructions or uses Bayesian optimization to automatically search for the best instructions and demos.
• BootstrapFinetune: leverages a prompt-based program to update actual model weights (fine-tuning) and integrates the fine-tuned model back into the DSPy program.

Usage guide:

• With very few examples, try BootstrapFewShot; with more examples, try BootstrapFewShotWithRandomSearch; for instruction-only optimization or when you have more data and compute, try MIPROv2.
• If you want a more efficient, faster model via fine-tuning, consider BootstrapFinetune.

Optimization cost:

• Simple optimization runs can cost just a few dollars, but using large models or large datasets can push costs to tens of dollars or more — plan accordingly.

Saving and reloading optimized programs:

• Save optimization results to a file with the .save() method.
• Reload a saved program later with the .load() method.

# 필요한 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")