如何使用DSPy搭建RAG应用？

DSPy搭建RAG的核心是「检索+生成+优化器自动调优」，无需手动调提示词，以下是极简可落地步骤（含本地向量库适配）：

1. 前置准备：安装依赖+配置核心组件

python

# 安装依赖

pip install dspy-ai openai sentence-transformers faiss-cpu

# 配置模型与检索器（本地FAISS+Sentence-BERT，适配私有数据）

import dspy

import faiss

import numpy as np

from sentence_transformers import SentenceTransformer

# 1. 初始化大模型（支持Ollama本地模型：dspy.Ollama(model=”llama3″)）

llm = dspy.OpenAI(model='gpt-3.5-turbo-instruct', max_tokens=800)

# 2. 初始化Embedding模型（用于文档/问题向量化）

embed_model = SentenceTransformer('all-MiniLM-L6-v2')

# 3. 构建本地FAISS检索器（私有数据用这个，公开数据可换ColBERTv2）

def build_faiss_retriever(documents):

embeddings = embed_model.encode(documents) # 生成文档向量

index = faiss.IndexFlatL2(embeddings.shape[1]) # 初始化FAISS索引

index.add(embeddings) # 加入向量

return dspy.FAISSRetriever(index=index, documents=documents, embedding_model=embed_model)

# 示例私有文档（替换为你的知识库，如PDF解析后的文本）

private_docs = [

“产品A支持AI语音交互，续航24小时，售价1299元”,

“产品B主打轻便，重量200g，适配安卓/iOS系统”,

“售后政策：7天无理由退货，1年质保，全国200+服务点”

]

retriever = build_faiss_retriever(private_docs)

# 全局配置DSPy

dspy.settings.configure(lm=llm, retriever=retriever)

2. 定义RAG模块：检索+生成一体化

python

class RAGQA(dspy.Module):

def forward(self, question: str) -> dspy.Prediction:

# 步骤1：检索相关文档（k=3，返回top3相关内容）

context = dspy.Retrieve(k=3)(question).passages

# 步骤2：结合上下文生成答案（限制不编造信息）

return dspy.ChainOfThought(

signature=”question: str, context: list -> answer: str”,

instructions=”严格依据提供的上下文回答，简洁准确，不添加额外信息”

)(question=question, context=context)

3. 优化器调优（核心：自动优化提示词/检索策略）

python

from dspy.teleprompt import BootstrapFewShot

from dspy.evaluate import Evaluate

# 1. 准备训练样本（格式：[{“question”: 问题, “answer”: 正确答案}]）

train_set = [

{“question”: “产品A的续航时间是多久？”, “answer”: “24小时”},

{“question”: “产品B支持哪些系统？”, “answer”: “安卓/iOS系统”}

]

# 2. 定义评估指标（判断答案是否准确）

def rag_metric(pred, ground_truth):

return ground_truth[“answer”].lower() in pred.answer.lower()

# 3. 初始化优化器并编译

teleprompter = BootstrapFewShot(metric=rag_metric, max_bootstrapped_demos=2)

optimized_rag = teleprompter.compile(RAGQA(), trainset=train_set)

4. 调用与评估

python

# 1. 调用RAG回答问题

result = optimized_rag(question=”产品A的售价是多少？”)

print(“回答：”, result.answer) # 输出：1299元

# 2. 评估优化效果（验证集测试）

dev_set = [{“question”: “售后质保期是多久？”, “answer”: “1年”}]

evaluator = Evaluate(devset=dev_set, metric=rag_metric, display_table=True)

evaluator(optimized_rag) # 展示准确率

# 3. 保存模块（后续直接加载）

optimized_rag.save(“private_rag.json”）。