Prompt Engineering Dspy 使用示例

# DSPy 实际案例 使用 DSPy 构建生产系统的实用示例。 ## 目录 - RAG 系统 - 智能体系统 - 分类 - 数据处理 - 多阶段流水线 ## RAG 系统 ### 基础 RAG python import dspy class BasicRAG(dspy.Module): def __init__(self, num_passages=3): super().__init__() self.retrieve = dspy.Retrieve(k=num_passages) self.generate = dspy.ChainOfThought("context, question -> answer") def forward(self, question): passages = self.retrieve(question).passages context = "nn".join(passages) return self.generate(context=context, question=question) # 配置检索器 (以 Chroma 为例) from dspy.retrieve.chromadb_rm import ChromadbRM retriever = ChromadbRM( collection_name="my_docs", persist_directory="./chroma_db", k=3 ) dspy.settings.configure(rm=retriever) # 使用 RAG rag = BasicRAG() result = rag(question="什么是 DSPy？") print(result.answer) ### 优化后的 RAG python from dspy.teleprompt import BootstrapFewShot # 包含问答对的训练数据 trainset = [ dspy.Example( question="什么是检索增强生成？", answer="RAG 将相关文档的检索与生成相结合..." ).with_inputs("question"), # ... 更多示例 ] # 定义指标 def answer_correctness(example, pred, trace=None): # 检查答案是否包含关键信息 return example.answer.lower() in pred.answer.lower() # 优化 RAG optimizer = BootstrapFewShot(metric=answer_correctness) optimized_rag = optimizer.compile(rag, trainset=trainset) # 优化后的 RAG 在类似问题上表现更好 result = optimized_rag(question="解释 RAG 系统") print(result.answer) ### 多跳 (Multi-Hop) RAG python class MultiHopRAG(dspy.Module): """跨文档推理链的 RAG。""" def __init__(self): super().__init__() self.retrieve = dspy.Retrieve(k=3) self.generate_query = dspy.ChainOfThought("question -> search_query") self.generate_answer = dspy.ChainOfThought("context, question -> answer") def forward(self, question): # 第一次检索 query1 = self.generate_query(question=question).search_query passages1 = self.retrieve(query1).passages # 基于第一次结果生成后续查询 context1 = "n".join(passages1) query2 = self.generate_query( question=f"基于: {contex