如何用DSPy生成prompt示例-编程阁

之前探索了使用DSPy的简单示例

https://blog.csdn.net/liliang199/article/details/155614507

这里进一步探索DSPy的符合应用，包括问答、情感分类、RAG系统等。

所用示例参考和修改自网络资料。

1 DSPy基础应用

1.1 DSPy配置

首先是LLM模型设置，这里配置LLM模型ollama/gemma3n:e2b，示例代码如下。

import dspy # 1. 配置语言模型 (这里以OpenAI为例，需提前设置API密钥) lm = dspy.LM(model="ollama/gemma3n:e2b", api_base="http://localhost:11434") dspy.configure(lm=lm)

另外，还可以配置兼容openai接口的LLM模型，过程参考如下链接。

https://blog.csdn.net/liliang199/article/details/155614507

1.2 DSPy测试

这是一个最简流程，展示了定义签名、选择模块、调用的核心步骤，验证DSPy的有效性。

# 2. 定义签名：指定输入和输出 class BasicQA(dspy.Signature): """回答简短的事实性问题。""" question: str = dspy.InputField() answer: str = dspy.OutputField(desc="通常为1到5个词") # 3. 创建预测模块 qa_module = dspy.Predict(BasicQA) # 4. 调用模块 result = qa_module(question="量子计算的创始人是谁？") print(f"答案: {result.answer}")

输出如下，这里通过BasicQA限定输入、输出的约束。

要求回答简单的事实性问题，答案要求1到5个词。

显然，DSPy生成的回答简明扼要，就Peter Shor2两个词，遵循了BasicQA的要求。

答案: Peter Shor

2 DSPy深度应用

这里考虑DSPy的进一步应用，比如应用到情感分类、构建RAG系统、使用优化器提升性能。

2.1 加入推理链 - 情感分类

使用ChainOfThought模块，让模型展示推理过程，通常能得到更准确的结果。

这里没有使用提示词，而是通过SentimentSignature给定任务定义，由DSPy生成提示词。

相比直接使用提示词的方式，在保持效果的前提下，这种实现方式更简洁明了。

import dspy from typing import Literal # 1. 定义更详细的签名 class SentimentSignature(dspy.Signature): """对文本进行情感分类。""" sentence: str = dspy.InputField(desc="需要分类的文本") sentiment: Literal['positive', 'negative', 'neutral'] = dspy.OutputField(desc="情感类别") # 2. 使用思维链模块 classifier = dspy.ChainOfThought(SentimentSignature) # 3. 调用并查看推理过程 result = classifier(sentence="这部电影的视觉效果令人惊叹，但剧情略显拖沓。") print(f"情感: {result.sentiment}") print(f"推理: {result.reasoning}") # ChainOfThought会额外提供推理字段

输出如下所示，这里通过SentimentSignature定义输入和输出需要分类的类别。

然后，使用思维链模块适配SentimentSignature。

可以看出，尽管没有进行模型训练，DSPy对输入进行了准确分类。

情感: neutral
推理: 这句话表达了对电影视觉效果的积极评价，但同时也指出了剧情的不足。整体而言，这句话的情感倾向是复杂的，既有积极的，也有消极的。考虑到视觉效果的突出，以及剧情的不足，可以认为整体情感偏向积极。

2.2 构建RAG问答系统

这是DSPy的典型应用。这里提供一个RAG问答系统示例，整合检索和生成。

这个例子使用本地数据模拟检索，能直接运行并理解DAG的核心流程。

以下是RAG逻辑核心逻辑签名模块

# ===== 3. 定义DSPy签名 ===== class GenerateAnswer(dspy.Signature): """基于给定上下文回答问题。""" context = dspy.InputField(desc="相关背景信息") question = dspy.InputField() answer = dspy.OutputField(desc="简洁、准确的答案，基于上下文")

相比直接采用提示词实现RAG检索和生成，这种方式逻辑清晰，简洁有效，更方便编程实现。

RAG整体代码示例如下。

import dspy import json # ===== 2. 构建模拟知识库（实际项目中替换为真实向量数据库） ===== class SimpleRetriever: """一个简单的内存检索器，模拟向量数据库功能""" def __init__(self, documents): # documents格式: [{"text": "...", "id": 1}, ...] self.documents = documents def retrieve(self, query, k=3): """简单关键词匹配检索（实际应用应使用向量检索）""" query_lower = query.lower() scored_docs = [] for doc in self.documents: text = doc["text"].lower() # 简单评分：计算查询词在文档中出现的次数 score = sum(1 for word in query_lower.split() if word in text) if score > 0: scored_docs.append((score, doc["text"])) # 按分数排序并返回前k个 scored_docs.sort(reverse=True, key=lambda x: x[0]) return [text for _, text in scored_docs[:k]] # 创建示例知识库（你的实际文档数据） knowledge_base = [ {"id": 1, "text": "爱因斯坦在1921年因对理论物理的贡献，特别是发现光电效应定律而获得诺贝尔物理学奖。"}, {"id": 2, "text": "光电效应是指当光照射到金属表面时，会从金属中发射出电子的现象。这一发现对量子力学的发展至关重要。"}, {"id": 3, "text": "阿尔伯特·爱因斯坦（1879-1955）是德裔理论物理学家，相对论的创始人，也是量子力学的重要奠基人之一。"}, {"id": 4, "text": "诺贝尔物理学奖是根据阿尔弗雷德·诺贝尔的遗嘱设立的，旨在表彰在物理学领域做出杰出贡献的科学家。"}, {"id": 5, "text": "1921年的诺贝尔物理学奖颁奖典礼于1922年举行，因为1921年没有候选人被认为符合获奖标准。"}, ] # 初始化检索器 retriever = SimpleRetriever(knowledge_base) # ===== 3. 定义DSPy签名 ===== class GenerateAnswer(dspy.Signature): """基于给定上下文回答问题。""" context = dspy.InputField(desc="相关背景信息") question = dspy.InputField() answer = dspy.OutputField(desc="简洁、准确的答案，基于上下文") # ===== 4. 构建RAG模块 ===== class RAG(dspy.Module): def __init__(self, retriever, num_passages=3): super().__init__() self.retriever = retriever self.num_passages = num_passages # 使用ChainOfThought让模型先推理再回答 self.generate_answer = dspy.ChainOfThought(GenerateAnswer) def forward(self, question): # 1. 检索阶段：获取相关文档 contexts = self.retriever.retrieve(question, k=self.num_passages) context_str = "\n---\n".join(contexts) # 2. 生成阶段：基于检索到的上下文生成答案 prediction = self.generate_answer( context=context_str, question=question ) # 返回完整结果 return dspy.Prediction( contexts=contexts, answer=prediction.answer, reasoning=prediction.reasoning # ChainOfThought提供的推理过程 ) # ===== 5. 初始化RAG系统 ===== rag_system = RAG(retriever, num_passages=2) # ===== 6. 测试RAG系统 ===== def test_rag_system(): """测试RAG系统的示例问题""" test_questions = [ "爱因斯坦因什么获得诺贝尔奖？", "什么是光电效应？", "谁创立了相对论？" ] for question in test_questions: print(f"\n{'='*60}") print(f"问题: {question}") print(f"{'='*60}") # 获取答案 result = rag_system(question) # 打印检索到的上下文 print("检索到的上下文:") for i, ctx in enumerate(result.contexts, 1): print(f"{i}. {ctx[:100]}...") # 只显示前100字符 # 打印推理过程（如果有） if hasattr(result, 'reasoning') and result.reasoning: print(f"\n模型推理: {result.reasoning}") # 打印最终答案 print(f"\n最终答案: {result.answer}") # ===== 7. 优化RAG系统（可选：使用BootstrapFewShot） ===== def optimize_rag_system(): """使用少量示例优化RAG提示""" from dspy.teleprompt import BootstrapFewShot # 准备训练示例 trainset = [ dspy.Example( question="爱因斯坦的诺贝尔奖贡献是什么？", contexts=[ "爱因斯坦在1921年因对理论物理的贡献，特别是发现光电效应定律而获得诺贝尔物理学奖。", "光电效应是指当光照射到金属表面时，会从金属中发射出电子的现象。" ], answer="发现光电效应定律" ).with_inputs('question'), dspy.Example( question="谁创立了相对论？", contexts=[ "阿尔伯特·爱因斯坦（1879-1955）是德裔理论物理学家，相对论的创始人。", "爱因斯坦是相对论的创始人，也是量子力学的重要奠基人之一。" ], answer="阿尔伯特·爱因斯坦" ).with_inputs('question'), ] # 定义评估指标 def validate_answer(example, prediction, trace=None): # 简单验证：预测答案是否包含关键词 correct_keywords = { "爱因斯坦的诺贝尔奖贡献是什么？": ["光电效应"], "谁创立了相对论？": ["爱因斯坦"] } question = example.question if question in correct_keywords: return any(keyword in prediction.answer for keyword in correct_keywords[question]) return True # 创建优化器 teleprompter = BootstrapFewShot( metric=validate_answer, max_bootstrapped_demos=2, max_labeled_demos=2 ) # 优化RAG系统 print("正在优化RAG系统...") optimized_rag = teleprompter.compile(RAG(retriever), trainset=trainset) return optimized_rag # ===== 8. 主程序入口 ===== if __name__ == "__main__": print("RAG问答系统启动...") # 测试基础版本 test_rag_system() # 询问是否要优化（实际使用时可注释掉这部分） print(f"\n{'='*60}") user_question = input("\n请输入你的问题（或直接回车使用默认问题）: ").strip() if not user_question: user_question = "爱因斯坦在哪年获得诺贝尔奖？" # 获取答案 result = rag_system(user_question) print(f"\n问题: {user_question}") print(f"答案: {result.answer}") if hasattr(result, 'reasoning') and result.reasoning: print(f"推理过程: {result.reasoning}") # 显示来源 print(f"\n信息来源:") for i, ctx in enumerate(result.contexts, 1): print(f"[{i}] {ctx}")

输出如下所示

RAG问答系统启动...
============================================================
问题: 爱因斯坦因什么获得诺贝尔奖？
============================================================
检索到的上下文:
模型推理: 爱因斯坦在1921年获得诺贝尔物理学奖，获奖的原因是由于他对光电效应的解释。光电效应是光与物质相互作用时，电子被激发的过程。爱因斯坦提出了光量子（光子）的概念来解释光电效应，这在当时是一个革命性的想法。他的解释为量子力学的发展奠定了基础。
最终答案: 爱因斯坦因他对光电效应的解释而获得诺贝尔奖。
============================================================
问题: 什么是光电效应？
============================================================
检索到的上下文:
模型推理: 光电效应是指光子与物质相互作用时，电子被发射出来的现象。它揭示了光具有粒子性，而不是仅仅是波动。当光子具有足够的能量时，它可以激发电子从原子中脱离。
最终答案: 光电效应是指光线照射到物质表面时，电子被发射出来的现象。它表明光具有粒子性，光子撞击原子并激发电子从原子中脱离。
============================================================
问题: 谁创立了相对论？
============================================================
检索到的上下文:
模型推理: 相对论是由阿尔伯特·爱因斯坦创立的。狭义相对论于1905年发表，广义相对论于1915年发表。
最终答案: 阿尔伯特·爱因斯坦
============================================================
请输入你的问题（或直接回车使用默认问题）:
问题: 爱因斯坦在哪年获得诺贝尔奖？
答案: 1905
推理过程: 爱因斯坦于1905年获得诺贝尔物理学奖。
信息来源:

3 DSPy进阶应用

3.1 使用优化器提升性能

对一些进阶应用，DSPy需要用户给出少量标注数据，辅助DSPy优化器优化prompt。

DSPy优化器可以自动利用这些少量标注数据来优化提示。

以下是使用BootstrapFewShot的简化流程，包括

1）准备少量训练数据

2）定义评估指标，比如准确率

然后通过bootstrapfewshot模块变异优化给定任务的提示词。

类似于模型训练，分类器优化后，DSPy就可以作预测了，推理延迟会更小，性能会更好。

import dspy from dspy.teleprompt import BootstrapFewShot # 1. 定义签名和初始模块（同示例二） classifier = dspy.ChainOfThought(SentimentSignature) # 2. 准备少量训练数据 trainset = [ dspy.Example(sentence="产品非常好用，强烈推荐！", sentiment='positive').with_inputs('sentence'), dspy.Example(sentence="服务糟糕，体验极差。", sentiment='negative').with_inputs('sentence'), # ... 更多示例 ] # 3. 定义评估指标（例如：准确率） def sentiment_metric(example, prediction): # 简单比较预测情感和真实情感是否一致 return example.sentiment == prediction.sentiment # 4. 创建并运行优化器 optimizer = BootstrapFewShot(metric=sentiment_metric, max_bootstrapped_demos=2) optimized_classifier = optimizer.compile(classifier, trainset=trainset) # 5. 使用优化后的模块 result = optimized_classifier(sentence="这本书内容一般，没什么亮点。") print(f"优化后的预测: {result.sentiment}")

输出如下

100%|██████████| 2/2 [00:38<00:00, 19.25s/it]
Bootstrapped 0 full traces after 1 examples for up to 1 rounds, amounting to 2 attempts.
优化后的预测: neutral