DSPy:声明式 LM 程序、自动优化提示词、RAG 流水线
需要以下场景时使用 DSPy:
GitHub Stars: 22,000+ | 创建方: Stanford NLP
# 稳定版
pip install dspy
# 最新开发版
pip install git+https://github.com/stanfordnlp/dspy.git
# 带特定 LM 提供商
pip install dspy[openai] # OpenAI
pip install dspy[anthropic] # Anthropic Claude
pip install dspy[all] # 所有提供商
import dspy
# 配置语言模型
lm = dspy.Claude(model="claude-sonnet-4-5-20250929")
dspy.settings.configure(lm=lm)
# 定义签名(输入 → 输出)
class QA(dspy.Signature):
"""用简短的实事答案回答问题。"""
question = dspy.InputField()
answer = dspy.OutputField(desc="通常在 1-5 个词之间")
# 创建模块
qa = dspy.Predict(QA)
# 使用
response = qa(question="法国的首都是哪里?")
print(response.answer) # "巴黎"
import dspy
lm = dspy.Claude(model="claude-sonnet-4-5-20250929")
dspy.settings.configure(lm=lm)
# 使用 ChainOfThought 以获得更好的推理
class MathProblem(dspy.Signature):
"""解数学应用题。"""
problem = dspy.InputField()
answer = dspy.OutputField(desc="数字答案")
# ChainOfThought 自动生成推理步骤
cot = dspy.ChainOfThought(MathProblem)
response = cot(problem="如果约翰有 5 个苹果,给了玛丽 2 个,他还剩多少?")
print(response.rationale) # 显示推理步骤
print(response.answer) # "3"
签名定义模块的输入和输出字段,类似于函数的类型注解:
class RAG(dspy.Signature):
"""根据给定上下文回答问题。"""
context = dspy.InputField(desc="可能包含答案的上下文")
question = dspy.InputField()
answer = dspy.OutputField(desc="1-2 句话的答案")
DSPy 提供多种内置模块:
# 选择不同的模块
cot = dspy.ChainOfThought(MathProblem)
react = dspy.ReAct(RAG)
Teleprompter 使用数据驱动方式自动优化提示词和推理策略:
from dspy.teleprompt import BootstrapFewShot
# 使用 FewShot 优化器
teleprompter = BootstrapFewShot(metric=my_metric)
compiled = teleprompter.compile(student=dspy.RAG(), trainset=trainset)
DSPy 支持组合多个模块构建复杂流水线:
class RAGModule(dspy.Module):
def __init__(self):
super().__init__()
self.retriever = dspy.Retrieve(top_k=3)
self.rag = dspy.Predict(RAG)
def forward(self, question):
context = self.retriever(query=question).passages
return self.rag(context=context, question=question)
from dspy.evaluate import Evaluate
evaluate = Evaluate(
devset=devset,
metric=my_metric,
num_threads=4,
display=True
)
results = evaluate(compiled_rag)
print(f"Accuracy: {results}")
DSPy 提示词编译器通过以下步骤工作:
对于复杂流水线,每个阶段可以使用不同签名和模块。
import dspy
# 配置
lm = dspy.Claude(model="claude-sonnet-4-5-20250929")
dspy.settings.configure(lm=lm, rm=colbertv2)
# 签名
class GenerateAnswer(dspy.Signature):
"""给定上下文,回答问题。"""
context = dspy.InputField(desc="可能包含答案的相关段落")
question = dspy.InputField()
answer = dspy.OutputField(desc="1-2 句话的答案")
# RAG 模块
class RAG(dspy.Module):
def __init__(self, num_passages=3):
super().__init__()
self.retrieve = dspy.Retrieve(top_k=num_passages)
self.generate = dspy.Predict(GenerateAnswer)
def forward(self, question):
context = self.retrieve(query=question).passages
prediction = self.generate(context=context, question=question)
return dspy.Prediction(context=context, answer=prediction.answer)
import dspy
from dspy.retrieve import ChromaRM
# 设置 Chroma 作为检索模型
rm = ChromaRM(collection_name="my_collection")
dspy.settings.configure(rm=rm)
# 定义签名
class QA(dspy.Signature):
question = dspy.InputField()
answer = dspy.OutputField()
# 创建流水线
qa_pipeline = dspy.ChainOfThought(QA)
# 运行
response = qa_pipeline(question="巴黎是哪国的首都?")
print(response.answer)
评论区