DSPy：大模型应用编程框架入门与实践

加载测试数据

多跳问题可以使用 HotPotQA 进行测试。

from dspy.datasets import HotPotQA      

dataset = HotPotQA(train_seed=1, train_size=20, eval_seed=2023, dev_size=50, test_size=0)      
trainset = [x.with_inputs('question') for x in dataset.train]   
devset = [x.with_inputs('question') for x in dataset.dev]      
print(len(trainset), len(devset))   
# Output: (20, 50)

设置 Signatures

签名是最小形式的输入和输出字段的元组。与手写提示相比，签名可以通过引导示例编译成自完善和自适应管道的 prompt 或 fine-tune。

class GenerateAnswer(dspy.Signature):       
    context = dspy.InputField()       
    question = dspy.InputField()       
    answer = dspy.OutputField()          

class GenerateSearchQuery(dspy.Signature):          
    context = dspy.InputField()       
    question = dspy.InputField()       
    query = dspy.OutputField()    

构建 DSPy Pipeline

DSPy 实现了多种 Module，包括 dspy.Predict、dspy.ChainOfThought、dspy.Retrieve 等。我们将签名传递给 ChainOfThought 模块，并使用输入字段的值进行调用。

以下是一个简化版的多跳检索生成器 SimplifiedBaleen 的实现：

from dsp.utils import deduplicate      

class SimplifiedBaleen(dspy.Module):       
    def __init__(self, passages_per_hop=3, max_hops=2):           
        super().__init__()              
        self.generate_query = [dspy.ChainOfThought(GenerateSearchQuery) for _ in range(max_hops)]  
        self.retrieve = dspy.Retrieve(k=passages_per_hop)           
        self.generate_answer = dspy.ChainOfThought(GenerateAnswer)           
        self.max_hops = max_hops           

    def forward(self, question):           
        context = []                      
        for hop in range(self.max_hops):               
            query = self.generate_query[hop](context=context, question=question).query       
            passages = self.retrieve(query).passages               
            context = deduplicate(context + passages)              
            pred = self.generate_answer(context=context, question=question)           
        return dspy.Prediction(context=context, answer=pred.answer)    

测试

my_question = "How many storeys are in the castle that David Gregory inherited?"  
uncompiled_baleen = SimplifiedBaleen()     
pred = uncompiled_baleen(my_question)      
print(f"Question: {my_question}")   
print(f"Predicted Answer: {pred.answer}")   
print(f"Retrieved Contexts (truncated): {[c[:200] + '...' for c in pred.context]}")    

优化

可以使用打分来优化 DSPy 的结果。定义评估函数，检查预测答案是否相符、检索到的上下文是否包含真实答案、生成的搜索查询长度及重复率等。

def validate_context_and_answer_and_hops(example, pred, trace=None):       
    if not dspy.evaluate.answer_exact_match(example, pred): return False       
    if not dspy.evaluate.answer_passage_match(example, pred): return False          
    hops = [example.question] + [outputs.query for *_, outputs in trace if 'query' in outputs]          
    if max([len(h) for h in hops]) > 100: return False       
    if any(dspy.evaluate.answer_exact_match_str(hops[idx], hops[:idx], frac=0.8) for idx in range(2, len(hops))): return False          
    return True    

使用 DSPy 中的 BootstrapFewShot 优化流程。DSPy 实现了多个提词器，如 LabeledFewShot、BootstrapFewShot、Ensemble 等，它们在优化成本和质量等方面提供不同的权衡。

from dspy.teleprompt import BootstrapFewShot      
teleprompter = BootstrapFewShot(metric=validate_context_and_answer_and_hops)   
compiled_baleen = teleprompter.compile(SimplifiedBaleen(), teacher=SimplifiedBaleen(passages_per_hop=2), trainset=trainset)    

评估

from dspy.evaluate.evaluate import Evaluate         
def gold_passages_retrieved(example, pred, trace=None):       
    gold_titles = set(map(dspy.evaluate.normalize_text, example["gold_titles"]))       
    found_titles = set(           
    map(dspy.evaluate.normalize_text, [c.split(" | ")[0] for c in pred.context])       
    )       
    return gold_titles.issubset(found_titles)      
evaluate_on_hotpotqa = Evaluate(devset=devset, num_threads=1, display_progress=True, display_table=5)      
uncompiled_baleen_retrieval_score = evaluate_on_hotpotqa(uncompiled_baleen, metric=gold_passages_retrieved, display=False)      
compiled_baleen_retrieval_score = evaluate_on_hotpotqa(compiled_baleen, metric=gold_passages_retrieved)      
print(f"## Retrieval Score for uncompiled Baleen: {uncompiled_baleen_retrieval_score}")   
print(f"## Retrieval Score for compiled Baleen: {compiled_baleen_retrieval_score}")         
# Output: 
# ## Retrieval Score for uncompiled Baleen: 36.0   
# ## Retrieval Score for compiled Baleen: 60.0    

DSPy 与 LangChain 或 LlamaIndex 有何不同？

LangChain、LlamaIndex 和 DSPy 都是帮助开发人员轻松构建基于 LM 应用的框架。使用 LangChain 和 LlamaIndex 的典型 pipeline 通常使用 prompt template 来实现，这使得整个 pipeline 对组件更改非常敏感。

相比之下，DSPy 将构建基于 LM 的管道从操作 prompt 转移到更贴近编程。DSPy 中新引入的编译器在更改基于 LM 的应用程序（如 LM 或数据）中的部件时，消除了任何额外的 Prompt 工程或微调工作。相反，开发人员可以简单地重新编译程序，以优化 pipeline 适应新添加的更改。因此，与 LangChain 或 LlamaIndex 相比，DSPy 可以帮助开发人员以更少的努力获得 pipeline 的性能。

总结

本文回顾了 Prompt 使用过程中的痛点和问题，介绍了 DSPy 这一基于编程的大模型应用新范式。通过多跳问答任务的例子，展示了 DSPy 如何通过签名、模块和优化器自动构建和增强应用。目前生成式人工智能社区对该框架非常感兴趣，因为它将构建基于 LM 的应用程序从手动提示工程转向了可编程的自动化流程。

DSPy 的核心价值在于其声明式编程能力，使得开发者能够专注于任务逻辑而非底层提示细节，从而显著提升系统的稳定性和可维护性。