RAG(검색 증강 생성) 튜토리얼

이 튜토리얼에서는 DSPy로 기본 질문 답변과 **검색 증강 생성(Retrieval-Augmented Generation, RAG)**을 각각 만들어 비교해 볼게요. 구체적으로는 Linux나 iPhone 앱 같은 Tech 질문에 답하는 시스템을 만들어 볼 거예요.

설치는 pip install -U dspy로 최신 DSPy를 받아서 따라오시면 되고, 개념적 개요가 더 필요하다면 최근 강의를 추천해요. 추가로 pip install datasets도 필요해요. 이 튜토리얼은 dspy.Embedder와 dspy.retrievers.Embeddings를 쓰는데, 이건 numpy가 필요해서 pip install dspy[numpy]로 설치해야 해요.

출처: Retrieval-Augmented Generation (RAG)

DSPy 환경 설정

DSPy에게 모듈에서 OpenAI의 gpt-4o-mini를 쓰겠다고 알려줄게요. 인증은 OPENAI_API_KEY에서 읽어요. 다른 프로바이더나 로컬 모델로도 쉽게 바꿀 수 있어요.

권장사항: 내부에서 무슨 일이 벌어지는지 이해하려면 MLflow Tracing을 설정해 두세요.

MLflow DSPy 통합

MLflow는 DSPy와 기본적으로 통합되는 LLMOps 도구예요. 설정은 네 단계로 끝나요. MLflow를 설치하고(%pip install mlflow>=2.20), mlflow ui --port 5000으로 UI를 띄우고, mlflow.set_tracking_uri("http://localhost:5000")과 mlflow.dspy.autolog()로 노트북을 연결하면 각 프로그램 실행의 trace를 볼 수 있어요. 프로그램 실행마다 trace가 남아 모델 동작을 잘 드러내 주고, DSPy 개념을 이해하는 데 큰 도움이 돼요. 자세한 내용은 MLflow DSPy 문서를 참고하세요.

import dspy

lm = dspy.LM('openai/gpt-4o-mini')
dspy.configure(lm=lm)

기본 DSPy 모듈 살펴보기

언제든 LM에 직접 lm(prompt="prompt")나 lm(messages=[...])로 물어볼 수 있어요. 하지만 DSPy는 Module이라는 더 나은 방식으로 LM 함수를 정의하게 해 줘요.

가장 단순한 모듈은 dspy.Predict예요. DSPy Signature, 즉 구조화된 입출력 스키마를 받아 지정한 동작을 하는 호출 가능한 함수를 돌려줘요. 인라인 표기법으로 question(str)을 입력받아 response를 출력하는 모듈을 선언해 볼게요.

qa = dspy.Predict('question: str -> response: str')
response = qa(question="what are high memory and low memory on linux?")

print(response.response)

이 질문에 대한 답은 "리눅스에서 high memory와 low memory는 시스템 메모리 주소 공간의 서로 다른 영역, 특히 32비트 아키텍처 맥락에서..." 같은 내용이에요. signature에 지정한 변수 이름이 입출력 인자 이름과 역할을 결정한다는 걸 알 수 있어요.

DSPy가 이 qa 모듈을 만들 때 뭘 했을까요? 이 예시에서는 아직 특별한 건 없어요. 모듈이 signature, LM, 입력을 Adapter에 넘기고, Adapter가 입출력 구조화와 구조화된 출력 파싱을 처리해요. dspy.inspect_history(n=1)로 DSPy가 보낸 최근 프롬프트를 직접 확인할 수 있어요. 그 프롬프트는 question을 입력으로 받아 response를 만들라는 구조예요.

DSPy에는 dspy.ChainOfThought, dspy.ProgramOfThought, dspy.ReAct 같은 다양한 내장 모듈이 있어요. 이것들은 기본 dspy.Predict와 서로 바꿔 쓸 수 있어요. 당신의 작업에 특화된 signature를 받아서, 거기에 범용 프롬프트 기법과 추론 시점 전략을 적용해요. 예를 들어 dspy.ChainOfThought는 LM이 signature에서 요청한 출력을 내기 전에 reasoning을 끌어내기 쉬운 방법이에요.

cot = dspy.ChainOfThought('question -> response')
cot(question="should curly braces appear on their own line?")

여기서는 str 타입을 생략했어요(기본 타입이 문자열이라서). 다른 필드와 타입(예: topics: list[str]나 is_realistic: bool)도 실험해 보면 좋아요. 흥미롭게도 reasoning을 요청하면 이 경우 출력 response가 더 짧아질 수 있어요. 이것이 좋은지 나쁜지는 필요에 따라 달라요. 공짜 점심은 없지만, DSPy는 다양한 전략을 매우 빠르게 실험할 수 있는 도구를 줘요. 참고로 dspy.ChainOfThought는 DSPy 내부에서 dspy.Predict로 구현돼 있어요.

평가와 반복 개발

이쯤이면 DSPy에 대해 꽤 많이 알게 됐어요. 빠른 스크립팅만 원한다면 이 정도로도 많은 걸 할 수 있어요. Python 제어 흐름에 DSPy signature와 모듈을 섞는 건 LM으로 일을 처리하기에 꽤 편리한 방식이에요.

하지만 아마 여러분은 고품질 시스템을 만들고 시간이 지나며 개선하고 싶어서 여기에 왔을 거예요. DSPy에서 그렇게 하려면 시스템 품질을 평가해 빠르게 반복하고, Optimizer 같은 DSPy의 강력한 도구를 쓰면 돼요.

DSPy에서 Example 다루기

DSPy 시스템의 품질을 측정하려면 1) question 같은 입력 값들이 잔뜩 필요하고, 2) 시스템 출력을 채점할 metric이 필요해요. metric은 천차만별이에요. 일부는 분류·질문답변처럼 이상적 출력의 정답 라벨이 필요하고, 다른 일부는 충실성(faithfulness)이나 환각(hallucination) 여부를 확인하는 자기 지도 방식이에요. 때로는 DSPy 프로그램을 판사로 써서 이런 특성을 채점하기도 해요.

질문과 (꽤 긴) 정답을 담은 데이터셋을 로드할게요. 이 노트북은 Tech 질문에 답하는 시스템을 만드는 게 목표니까, RAG-QA Arena 데이터셋에서 StackExchange 기반 질문과 정답을 가져왔어요.

import orjson
from dspy.utils import download

# Download question--answer pairs from the RAG-QA Arena "Tech" dataset.
download("https://huggingface.co/dspy/cache/resolve/main/ragqa_arena_tech_examples.jsonl")

with open("ragqa_arena_tech_examples.jsonl") as f:
    data = [orjson.loads(line) for line in f]

데이터 포인트 하나는 이런 모양이에요.

{'question': 'why igp is used in mpls?',
 'response': "An IGP exchanges routing prefixes between gateways/routers.  \nWithout a routing protocol, you'd have to configure each route on every router ...",
 'gold_doc_ids': [2822, 2823]}

이런 단순 dict를 dspy.Example 목록으로 만들게요. dspy.Example은 DSPy에서 훈련(또는 테스트) 데이터 포인트를 담는 데이터 타입이에요. dspy.Example을 만들 때 보통 .with_inputs("field1", "field2", ...)로 어떤 필드가 입력인지 지정해요. 나머지 필드는 라벨이나 메타데이터로 취급돼요.

data = [dspy.Example(**d).with_inputs('question') for d in data]

# Let's pick an `example` here from the data.
example = data[2]
example

이제 데이터를 나눌게요.

Train/Validation 세트: 보통 DSPy 옵티마이저에 주는 분할이에요. 옵티마이저는 훈련 예시에서 직접 학습하고 검증 예시로 진행을 확인해요. 훈련·검증 각각 30~300개 정도가 좋아요. 특히 프롬프트 옵티마이저는 훈련보다 검증을 더 많이 넘기는 게 좋을 때가 많아요. 여기서는 총 200개를 쓸 건데, MIPROv2가 valset을 안 넘기면 20% 훈련 / 80% 검증으로 나눠요.
Dev/Test 세트: 나머지(보통 30~1000개)로 개발(시스템을 반복하며 살펴보기)과 테스트(최종 홀드아웃 평가)에 써요.

import random

random.Random(0).shuffle(data)
trainset, devset, testset = data[:200], data[200:500], data[500:1000]

len(trainset), len(devset), len(testset)

결과는 (200, 300, 500)이에요.

DSPy에서 평가하기

질문 답변 작업에 어울리는 metric은 뭘까요? 정답이 길기 때문에 두 가지를 물어볼 수 있어요. 시스템 응답이 정답(gold response)의 핵심 사실을 얼마나 잘 커버하는지, 그리고 반대로 시스템 응답이 정답에 없는 내용을 얼마나 말하지 않는지요. 이 metric은 본질적으로 의미론적 F1(semantic F1) 이에요. DSPy에서 SemanticF1 metric을 불러올게요. 이 metric은 아주 단순한 DSPy 모듈로 구현돼 있고, 작업 중인 LM을 사용해요.

from dspy.evaluate import SemanticF1

# Instantiate the metric.
metric = SemanticF1(decompositional=True)

# Produce a prediction from our `cot` module, using the `example` above as input.
pred = cot(**example.inputs())

# Compute the metric score for the prediction.
score = metric(example, pred)

print(f"Question: \t {example.question}\n")
print(f"Gold Response: \t {example.response}\n")
print(f"Predicted Response: \t {pred.response}\n")
print(f"Semantic F1 Score: {score:.2f}")

예를 들어 "왜 내 문자 메시지가 maybe로 뜨지?"라는 질문에서 정답은 iOS 9의 Proactivity 기능과 iOS 11.2 버그를 언급하지만, 예측 응답은 그 구체적 내용을 빼고 일반적인 설명만 해서 Semantic F1 점수가 0.33이 나와요.

실제로 metric 내부에서도 DSPy 모듈 호출이 일어나요. dspy.inspect_history(n=1)로 살펴보면 metric이 question, ground_truth, system_response를 입력으로 받아 정답의 핵심 아이디어와 응답의 핵심 아이디어를 나열하고, 겹침을 논의한 뒤 recall과 precision을 산출하는 구조예요.

평가에는 위 metric을 단순 루프로 돌려 평균을 내도 되지만, 병렬 처리와 유틸리티를 위해 dspy.Evaluate를 쓰는 게 좋아요.

# Define an evaluator that we can re-use.
evaluate = dspy.Evaluate(devset=devset, metric=metric, num_threads=24,
                         display_progress=True, display_table=2)

# Evaluate the Chain-of-Thought program.
evaluate(cot)

결과는 Average Metric: 125.68 / 300 (41.9%) 예요. 순수 chain-of-thought 모듈은 dev set에서 약 42%의 semantic F1을 내요.

MLflow에 평가 결과 기록하기

import mlflow

with mlflow.start_run(run_name="rag_evaluation"):
    evaluate = dspy.Evaluate(
        devset=devset,
        metric=metric,
        num_threads=24,
        display_progress=True,
    )

    # Evaluate the program as usual
    result = evaluate(cot)


    # Log the aggregated score
    mlflow.log_metric("semantic_f1_score", result.score)
    # Log the detailed evaluation results as a table
    mlflow.log_table(
        {
            "Question": [example.question for example in eval_set],
            "Gold Response": [example.response for example in eval_set],
            "Predicted Response": [output[1] for output in result.results],
            "Semantic F1 Score": [output[2] for output in result.results],
        },
        artifact_file="eval_results.json",
    )

지금까지 질문 답변용 아주 단순한 chain-of-thought 모듈을 만들고 작은 데이터셋에서 평가했어요. 더 잘할 수 있을까요? 나머지에서는 같은 작업에 RAG 프로그램을 만들어 어떻게 점수를 크게 끌어올리는지 보고, DSPy 옵티마이저 중 하나로 RAG 프로그램을 더 높은 품질의 프롬프트로 컴파일해 점수를 더 올릴게요.

기본 RAG

먼저 RAG 검색에 쓸 코퍼스 데이터를 내려받을게요. 예전 버전 튜토리얼은 전체(65만 문서) 코퍼스를 썼는데, 빠르고 저렴하게 돌리려고 28,000 문서로 다운샘플링했어요.

download("https://huggingface.co/dspy/cache/resolve/main/ragqa_arena_tech_corpus.jsonl")

리트리버 설정

DSPy 입장에서는 도구나 리트리버를 호출하는 Python 코드를 아무거나 끼워 넣을 수 있어요. 여기서는 편의를 위해 OpenAI Embeddings를 쓰고 top-K 검색을 로컬에서 수행할게요.

참고: 아래 단계는 pip install -U faiss-cpu를 하거나 dspy.retrievers.Embeddings에 brute_force_threshold=30_000을 넘겨 faiss를 피해야 해요.

max_characters = 6000  # for truncating >99th percentile of documents
topk_docs_to_retrieve = 5  # number of documents to retrieve per search query

with open("ragqa_arena_tech_corpus.jsonl") as f:
    corpus = [orjson.loads(line)['text'][:max_characters] for line in f]
    print(f"Loaded {len(corpus)} documents. Will encode them below.")

embedder = dspy.Embedder('openai/text-embedding-3-small', dimensions=512)
search = dspy.retrievers.Embeddings(embedder=embedder, corpus=corpus, k=topk_docs_to_retrieve)

28,436개 문서를 로드해 512차원 임베딩으로 FAISS 인덱스를 만들어요.

첫 RAG 모듈 만들기

앞서 dspy.Predict("question -> answer")처럼 개별 모듈만 봤는데, 여러 단계를 가진 DSPy 프로그램을 만들고 싶다면 어떻게 할까요? dspy.Module 문법으로 리트리버와 생성 모듈 같은 조각들을 연결하면 전체 시스템을 최적화할 수 있어요.

__init__ 메서드에서 필요한 하위 모듈을 선언하고(여기서는 검색된 context와 질문을 받아 응답을 만드는 dspy.ChainOfThought('context, question -> response') 모듈), forward 메서드에서 원하는 Python 제어 흐름을 표현하면 돼요. 여기서는 앞서 정의한 search 함수를 먼저 호출한 다음 self.respond ChainOfThought 모듈을 호출해요.

class RAG(dspy.Module):
    def __init__(self):
        self.respond = dspy.ChainOfThought('context, question -> response')

    def forward(self, question):
        context = search(question).passages
        return self.respond(context=context, question=question)

RAG 모듈을 사용해 볼게요.

rag = RAG()
rag(question="what are high memory and low memory on linux?")

이제 검색된 문서 5개가 context에 실려 프롬프트로 들어가요. LM이 "High Memory는 커널 주소 공간에 영구 매핑되지 않은 메모리로 사용자 공간 앱과 임시 버퍼에 쓰이고, Low Memory는 커널이 직접 접근·부팅 시 정적 매핑된 메모리"라는, 검색 문서에 근거한 훨씬 정확한 답을 만들어요.

앞서 CoT 모듈은 dev set에서 semantic F1 기준 약 40%였어요. 이 RAG 모듈은 더 잘할까요?

evaluate(RAG())

결과는 Average Metric: 166.54 / 300 (55.5%) 예요. RAG만으로 42%에서 55%로 눈에 띄게 올랐어요.

DSPy Optimizer로 RAG 프롬프트 개선하기

기본 RAG 모듈이 55%를 기록했어요. 더 강하게 만들 옵션은 뭐가 있을까요? DSPy가 제공하는 여러 선택지 중 하나는 파이프라인의 프롬프트를 최적화하는 거예요.

프로그램에 하위 모듈이 많으면 모두 함께 최적화돼요. 여기서는 self.respond = dspy.ChainOfThought('context, question -> response') 하나뿐이에요.

DSPy의 MIPRO (v2) 옵티마이저를 설정하고 쓸게요. 아래 실행 비용은 약 $1.5(medium auto 설정 기준)이고, 스레드 수에 따라 20~30분 정도 걸릴 수 있어요.

tp = dspy.MIPROv2(metric=metric, auto="medium", num_threads=24)  # use fewer threads if your rate limit is small

optimized_rag = tp.compile(RAG(), trainset=trainset,
                           max_bootstrapped_demos=2, max_labeled_demos=2)

프롬프트 최적화 과정은 꽤 체계적이에요. 이 논문에서 자세히 배울 수 있어요. 중요한 건 이게 마법의 버튼이 아니라는 거예요. 훈련 세트에 과적합돼 홀드아웃 세트에 잘 일반화되지 않을 수도 있으니, 프로그램을 반복적으로 검증하는 게 필수적이에요.

같은 질문을 최적화되지 않은 기준선 rag = RAG()과, 프롬프트 최적화 후의 optimized_rag에 각각 물어보며 비교해 볼게요.

baseline = rag(question="cmd+tab does not work on hidden or minimized windows")
print(baseline.response)

pred = optimized_rag(question="cmd+tab does not work on hidden or minimized windows")
print(pred.response)

기준선은 "맞아요, cmd+tab은 숨겨진·최소화된 창에는 동작하지 않아요..."라고 짧게 답하고, 최적화 버전은 macOS의 Command+Tab 동작, 최소화 창은 활성으로 안 세는 점, Command + Option + H + M 같은 대안 단축키, Dock에서 창을 복구하는 방법까지 꽤 상세하게 답해요.

dspy.inspect_history(n=2)로 최적화 전 RAG 프롬프트와 최적화 후를 볼 수 있어요. 이 노트북의 실행 중 하나에서 최적화된 프롬프트는 이런 일을 했어요.

다음 지시문을 만들었어요.

Using the provided `context` and `question`, analyze the information step by step to generate a comprehensive and informative `response`. Ensure that the response clearly explains the concepts involved, highlights key distinctions, and addresses any complexities noted in the context.

"how to transfer whatsapp voice message to computer?" 같은 합성 reasoning과 답이 달린, 완전히 풀린 RAG 예시 두 개를 포함했어요.

전체 dev set에서 평가해 볼게요.

evaluate(optimized_rag)

결과는 Average Metric: 183.32 / 300 (61.1%) 예요. 프롬프트 최적화로 55.5%에서 61.1%로 또 올랐어요.

비용 관리하기

DSPy는 프로그램의 비용을 추적할 수 있어요.

cost = sum([x['cost'] for x in lm.history if x['cost'] is not None])  # in USD, as calculated by LiteLLM for certain providers

저장과 로딩

최적화된 프로그램은 내부 구조가 꽤 단순해요. optimized_rag를 저장해서 나중에 처음부터 다시 최적화하지 않고 로드할 수 있어요.

optimized_rag.save("optimized_rag.json")

loaded_rag = RAG()
loaded_rag.load("optimized_rag.json")

loaded_rag(question="cmd+tab does not work on hidden or minimized windows")

로드된 RAG도 cmd+tab 질문에 대해 앞서 최적화 버전과 같은 상세한 답을 내놓아요.

MLflow에 프로그램 저장

로컬 파일 대신 MLflow에 저장하면 재현성과 협업에 유리해요. 1) MLflow가 고정된 환경 메타데이터를 프로그램과 함께 자동 저장하고, 2) 프로그램의 성능·비용을 함께 추적하며, 3) 실험 공유로 팀과 결과를 나눌 수 있어요.

import mlflow

# Start an MLflow Run and save the program
with mlflow.start_run(run_name="optimized_rag"):
    model_info = mlflow.dspy.log_model(
        optimized_rag,
        artifact_path="model", # Any name to save the program in MLflow
    )

# Load the program back from MLflow
loaded = mlflow.dspy.load_model(model_info.model_uri)

다음은?

이 작업에서 SemanticF1 기준 약 42%에서 약 61%까지 올리기는 꽤 쉬웠어요. 하지만 DSPy는 시스템 품질을 이어서 개선할 길을 계속 열어 주는데, 우리는 아직 표면만 살짝 건드렸어요.

일반적으로 이런 도구가 있어요.

프로그램에 더 나은 시스템 아키텍처 탐색. 예를 들어 LM에게 리트리버용 검색 쿼리를 생성하게 하면 어떨까요? DSPy로 만든 STORM 파이프라인을 참고하세요.
다른 프롬프트 옵티마이저나 가중치 옵티마이저 탐색. Optimizers 문서를 보세요.
DSPy Optimizer로 추론 시점 연산 확장. 예를 들어 최적화 후 여러 프로그램을 앙상블하는 방법.
프롬프트·가중치 최적화로 더 작은 LM에 증류해 비용 절감.

어떤 것부터 시작할지 어떻게 정할까요? 첫 단계는 시스템 출력을 살펴보는 거예요. 그러면 성능 저하 원인(있다면)을 찾을 수 있어요. 이러면서 metric도 계속 다듬고(예: 자신의 판단을 기준으로 최적화), 관련 도메인에서든 사용자 앞에 데모를 내놓아서든 더 많고 더 현실적인 데이터를 모으는 걸 잊지 마세요.

더 알아보기 (Learn more)

MIPROv2 옵티마이저 — 프롬프트 최적화
SemanticF1 metric — 의미론적 F1 평가
Multi-Hop RAG 튜토리얼 — 검색 쿼리를 생성하는 고급 RAG