1. 简介

DSPy 是一个专门用于算法优化语言模型（LM）提示和权重的框架，特别适用于在管道中多次使用 LM 的情况。没有 DSPy 时，构建复杂系统通常需要以下步骤：

1. 将问题分解为多个步骤
2. 逐步优化每个步骤的提示，直到它们单独运行良好
3. 调整这些步骤以确保它们协同工作
4. 生成合成示例来微调每个步骤
5. 使用这些示例微调较小的 LM 以降低成本

目前，这个过程既复杂又混乱：每次更改管道、LM 或数据时，所有提示（或微调步骤）可能都需要重新调整。为了使这一过程更加系统化且更强大，DSPy 提供了两大功能。

首先，它将程序的流程（模块）与每个步骤的参数（LM 提示和权重）分离开来。其次，DSPy 引入了新的优化器，这些优化器是由 LM 驱动的算法，可以根据您想要最大化的指标来调整 LM 调用的提示和/或权重。

DSPy 可以常规地教会强大的模型，如 GPT-3.5 或 GPT-4，以及本地模型，如 T5-base 或 Llama2-13b，在任务上更加可靠，即具有更高的质量和/或避免特定的失败模式。DSPy 优化器将“编译”相同的程序为不同的指令、少量提示和/或每个 LM 的权重更新（微调）。

这是一种新的范式，其中 LM 及其提示作为一个更大系统的可优化部分逐渐淡出背景，该系统可以从数据中学习。简而言之；更少的提示，更高的分数，以及一种更系统的方法来解决使用 LM 的难题。

2. Quick Start

先尝试跑一遍Quick Start，首先是安装

pip install dspy-ai

尝试最小工作脚本

import dspy
from dspy.datasets.gsm8k import GSM8K, gsm8k_metric

# Set up the LM.
turbo = dspy.OpenAI(model='gpt-3.5-turbo-instruct', max_tokens=250)
dspy.settings.configure(lm=turbo)

# Load math questions from the GSM8K dataset.
gsm8k = GSM8K()
gsm8k_trainset, gsm8k_devset = gsm8k.train[:10], gsm8k.dev[:10]

这里使用的GSM8K数据集来源于：huggingface.co/datasets/openai/gsm8k

官方介绍：

GSM8K（小学数学 8K）数据集包含了 8.5K 个高质量、语言多样的小学数学应用题，专门为支持基础数学问题解答任务而创建，这些问题解答通常需要多步骤的推理过程。

• 解决这些问题通常需要 2 至 8 个步骤。
• 解决方法主要是通过运用加减乘除等基本算术运算，进行一系列基础计算，从而得出最终答案。
• 一个聪慧的初中生理应能够应对所有题目：论文指出，“这些题目所涉及的概念均未超出初等代数的范畴，且多数题目无需明确设定变量即可解答。”
• 解决方案采用自然语言形式呈现，与纯数学表达式不同。论文中提到：“我们相信这种格式最为实用，并预计它将有助于理解大型语言模型内部思维的特性。”

说白了就是一套小学数学应用题的QA集，其中Q是题目，A则是完整详细的解题回答。

dspy在这里做的工作是预封装了数据集的下载，然后将其保存为df变量。

接下来增加以下代码：

os.environ["OPENAI_API_KEY"] = "your-openai-key"

from dspy.teleprompt import BootstrapFewShot
from dspy.evaluate import Evaluate

# Set up the optimizer: we want to "bootstrap" (i.e., self-generate) 4-shot examples of our CoT program.

config = dict(max_bootstrapped_demos=4, max_labeled_demos=4)

# Optimize! Use the `gsm8k_metric` here. In general, the metric is going to tell the optimizer how well it's doing.

teleprompter = BootstrapFewShot(metric=gsm8k_metric, **config)
optimized_cot = teleprompter.compile(CoT(), trainset=gsm8k_trainset)

# Set up the evaluator, which can be used multiple times.

evaluate = Evaluate(devset=gsm8k_devset, metric=gsm8k_metric, num_threads=4, display_progress=True, display_table=0)

# Evaluate our `optimized_cot` program.

evaluate(optimized_cot)

从输出可以看到校验的结果为：Average Metric: 7 / 10(70.0)，先理解成一个70分的水平吧。（自己验证自己分数居然这么低，但也可能是样本量过少导致）

我们可以通过查看运行记录来看DSPy刚刚实际上做了什么：

turbo.inspect_history(n=1)

至此，一个最小示例便完成。DSPy通过20行左右代码完成了：

1. 下载数据集
2. 构建训练集和测试集
3. 验证AI处理准确率
4. 封装训练好的optimized_cot作为可复用的一个处理算子

3. 使用DSPy的8个步骤

3.1 定义任务

梳理你预期的输入/输出行为：

DSPy相对于LangChain等框架而言，更重要的理念是数据驱动。因此考虑以下问题：

• 你想要基于数据构建什么？聊天机器人吗？代码助手？论文信息提取？还是一个翻译系统？
• 又或者是实现RAG并从中提取有效信息的的系统？

根据你想要构建的系统，按工作流程列出 3-4 个程序输入和输出。与此同时考虑成本，你的系统运行成本不能太贵，并且要足够快地响应用户。因此这也是LLM选型的考虑，是使用 GPT-3.5还是别的小型开源模型，比如 Mistral-7B 或 Llama2-13B-chat。

3.2 定义工作流（pipeline）

你的 DSPy 程序应该做些什么？是一个简单的思维链？还是需要RAG？或者其他工具，比如计算器或日历 API？

是否有一个典型的工作流程，通过多个明确的步骤来解决你的问题？或者你希望使用一个完全开放的大语言模型（或使用代理的开放工具）来完成任务？考虑这些可能性，从最简单的开始。

几乎每个任务都应该从一个单一的 dspy.ChainofThought 模块开始，然后逐步增加复杂性。然后编写你的（初始）DSPy 程序。

3.3 寻找例子

到这个阶段，你可能已经有了一些你想要解决的任务示例。将它们通过你的处理流程运行一遍。可以考虑使用一个大型且强大的语言模型，或者尝试几个不同的语言模型，以了解它们的潜力。

此时，你仍然在零样本的情况下使用你的处理流程，所以结果可能不太理想。DSPy 将帮助你优化指令、少样本示例，甚至是语言模型调用的权重，但了解零样本使用中的问题将非常有帮助。记录下你尝试的有趣示例（无论是简单还是复杂的）：即使没有标签，仅仅记录你尝试的输入对 DSPy 优化器来说也是有帮助的。

3.4 定义数据

现在是时候正式确定用于 DSPy 评估和优化的训练和验证数据了 - 数据指南。即使只有 10 个示例，您也可以有效地使用 DSPy 优化器，但如果有 50-100 个示例（甚至更好，300-500 个示例），效果会更好。那如何获取这些示例呢？

如果您的任务非常特殊，请花时间手工准备大约 10 个示例。通常情况下，根据您的指标，您只需要输入数据而不需要标签，所以这并不难。然而，您的任务很可能并不那么独特。

您几乎总是可以在 HuggingFace 数据集或其他形式的数据中找到一些相近的数据集来利用。如果数据的许可足够宽松，我们建议您使用这些数据。否则，您也可以开始使用、部署或演示您的系统，并通过这种方式收集一些初始数据。

3.5 定义评价方式（metric）

是什么决定了你的系统输出结果的好坏？你需要投入时间和资源来定义评估指标，并逐步改进它们。如果你无法定义某个东西，就很难持续改进它。一个评估指标其实就是一个函数，它会从你的数据中获取示例，并结合系统的输出，返回一个量化输出质量的分数 - 这就是指标指南。

对于简单的任务，这个指标可能只是“准确率”或“精确匹配”或“F1分数”。这些指标适用于简单的分类任务或短问答任务。然而，对于大多数应用场景，你的系统会输出长篇的结果。

在这种情况下，你的评估指标可能需要是一个较小的 DSPy 程序，它会检查输出的多个属性（很可能会使用来自大语言模型的 AI 反馈）。第一次就能做到完美是不太可能的，但你应该从简单的指标开始，并逐步迭代改进。（如果你的评估指标本身是一个 DSPy 程序，请注意，迭代的最强大方法之一是编译（优化）你的评估指标本身。

这通常很容易，因为评估指标的输出通常是一个简单的值（例如，5分中的一个分数），因此通过收集一些示例，定义和优化评估指标的指标是很容易的。）

3.6 收集 zero-shot 评估集

现在你已经有了一些数据和一个评估指标，在运行任何优化器之前，先对你的流程进行评估。查看输出结果和指标分数。这可能会帮助你发现一些主要问题，并为你的下一步设定一个基准。

3.7 用DSPy优化器编译

给定一些数据和一个度量标准，我们现在可以优化你构建的程序 - 优化器指南。DSPy 提供了许多不同功能的优化器。请记住：DSPy 优化器会为每一步创建示例，编写指令，和/或更新 LM 权重。

通常，你不需要为管道步骤提供标签，但数据示例需要包含输入值和度量标准所需的标签（例如，如果度量标准是无参考的，则不需要标签，但大多数情况下需要最终输出标签）。

以下是入门的一般指导：如果你有非常少的数据，例如 10 个任务示例，使用 BootstrapFewShot。如果你有稍多的数据，例如 50 个任务示例，使用 BootstrapFewShotWithRandomSearch。如果你有更多的数据，例如 300 个或更多示例，使用 MIPRO。如果你能够使用其中一个大 LM（例如，7B 参数或以上）并且需要一个非常高效的程序，将其编译为一个小 LM 使用 BootstrapFinetune。

3.8 迭代

此时，你可能对一切都非常满意（我们见过不少人在第一次尝试 DSPy 时就成功了），但更有可能的是，你已经取得了很大进展，但对最终的程序或指标不太满意。

此时，请回到第 1 步，重新审视主要问题。你是否清晰地定义了你的任务？你是否需要为你的问题收集（或在线查找）更多数据？你是否想更新你的指标？你是否想使用更复杂的优化器？你是否需要考虑像 DSPy Assertions 这样的高级功能？或者，可能最重要的是，你是否想在你的 DSPy 程序中添加更多复杂性或步骤？你是否想在序列中使用多个优化器？

迭代开发是关键。DSPy 为你提供了逐步实现这一目标的工具：迭代你的数据、程序结构、断言、指标和优化步骤。在撰写本文时，优化复杂的 LM 程序是一个全新的范式，只存在于 DSPy 中，因此关于该做什么的规范仍在不断涌现。

4. 使用实例

4.1 训练过程说明

数据来自于readhub_news_qa_records.log，是原有的一套抓取readhub新闻并撰写为一篇推文的程序，原程序调用OpenAI模型gpt-3.5-turbo-1106。

以下程序尝试使用该数据（包含27个QA对），利用DSPy框架训练一个基于qwen:14b的模型。

第一步：初始化

import os
import dspy
from dspy.teleprompt import BootstrapFewShot
from dspy.evaluate import Evaluate
import re

ollama_qwen = dspy.OllamaLocal(model='qwen:14b')
dspy.settings.configure(lm=ollama_qwen)

第二步：读取数据文件，将70%数据划分为训练集，30%数据划分为测试集

# get current data into train-set
def parse_data_file(file_path):
    with open(file_path, 'r', encoding='utf-8') as file:
        content = file.read()
    
    # 使用正则表达式匹配问题和答案对
    pattern = r"'question':'(.*?)'\s*'answer':'(.*?)'"
    matches = re.findall(pattern, content, re.DOTALL)
    
    # 创建列表并添加每个问答对
    readhub_data_openai = [
        {'question': question.strip(), 'answer': answer.strip()}
        for question, answer in matches
    ]
    
    return readhub_data_openai

file_path = 'your-data-file-path'  # 替换为您的数据文件路径
readhub_data_openai = parse_data_file(file_path)
train_dev_seq = int(len(readhub_data_openai) * 0.7)

定义Writer_CoT思维链，article_quality_metric为度量函数，通过Assess函数反复调用大模型对结果进行评分

teleprompter.compile即为训练过程

class Writer_CoT(dspy.Module):
    def __init__(self):
        super().__init__()
        self.prog = dspy.ChainOfThoughtWithHint("question -> answer")
    
    def forward(self, question):
        prompt = """
        你是一个专业自媒体编辑，请按以下要求加工一段内容，要求：
        {- 站在中立角度对新闻进行重新陈述
         - 使用简明扼要的叙述描述新闻事件，设计独特、新颖的内容
         - 减少元话语/废话/引导词的输出
         - 确保文章逻辑清晰，段落结构流畅，没有小标题
         - 输出一整段可直接发布的文字}
        请开始用中文扩写"""
        return self.prog(question=question, hint=prompt)

class Assess(dspy.Signature):
    """Assess the quality of a tweet along the specified dimension."""

    assessed_text = dspy.InputField()
    assessment_question = dspy.InputField()
    assessment_answer = dspy.OutputField(desc="only Yes or No, no only output")

def article_quality_metric(gold, pred, trace=None):
    question, answer, tweet = gold.question, gold.answer, pred.answer
    
    engaging = "Does the assessed text make for a long and detailed part of an article?"
    correct = f"The text should expand the text `{question}` with `{answer}`. Does the assessed text expended properly and correctly?"
    
    with dspy.context(lm=ollama_qwen):
        correct =  dspy.Predict(Assess)(assessed_text=tweet, assessment_question=correct)
        engaging = dspy.Predict(Assess)(assessed_text=tweet, assessment_question=engaging)

    correct, engaging = [('yes' in m.assessment_answer.lower() or '是的' in m.assessment_answer.lower()) for m in [correct, engaging]]
    score = (correct + engaging) if correct else 0

    if trace is not None: return score >= 2
    return score / 2.0

# 转换数据集
readhub_data_openai_traintset = [dspy.Example(item).with_inputs("question") for item in readhub_data_openai[0:train_dev_seq]]

config = dict(max_bootstrapped_demos=4, max_labeled_demos=4)

teleprompter = BootstrapFewShot(metric=article_quality_metric, **config)
optimized_cot = teleprompter.compile(Writer_CoT(), trainset=readhub_data_openai_traintset)

评估过程使用Evaluate函数完成，最后输出评分结果

readhub_data_openai_devset = [dspy.Example(item).with_inputs("question") for item in readhub_data_openai[train_dev_seq:]]

evaluate = Evaluate(devset=readhub_data_openai_devset, metric=article_quality_metric, num_threads=4, display_progress=True, display_table=0)

# Evaluate our `optimized_cot` program.
evaluate(optimized_cot)

Average Metric: 9.0 / 9(100.0): 100%|
100.0

保存模型到文件

optimized_cot.save("your-model-path")

4.2 个人使用感受

建议仔细阅读 dspy-docs.vercel.app/docs/building-blocks/solving_your_task 读完比较明白DSPy在做的是一件什么事情。与LangChain野蛮生长有点不同的是，DSPy一开始就是按着Pytorch机器学习的那套思路来展开的。几乎大部分机器学习过程都可以抽象成那个处理模式，塞数据，拆训练集测试集，跑算法，再Evaluate。

初步感觉相对于LangChain来说其实是拆得更细了，不过感觉这样可控的和不可控的部分会分得更清晰一些，LangChain最大的问题是有时候一个东西封装得太死你都不知道怎么改，然后它又极不稳定，不停在更新，你pip upgrade了一个什么东西就崩了。动不动就要去改LangChain源码，维护自己本地的一个分支版本，到后面想用新特性就挺麻烦了。

以上面案例来说，当你有一个原本用OpenAI跑的工作流，把输入输出积累下来当成Few Shots，那后续可以用来优化成某个本地大模型（如qwen:14b）的表现，官方文档有写训练原理，也提供了许多不同模式的优化器：dspy-docs.vercel.app/docs/building-blocks/optimizers

后续再研究下结合RAG的部分，应该可以提升RAG的质量。

总的来说，一个AI应用在初上线可以采用最好的模型，积累了数据之后切换到DSPy训练好的优化器，进一步降低成本，做到完全不需要调用在线模型，也能达到同样输出效果。

因此DSPy相当于是提供了一套无需大规模算力的fine-tune机制，但其局限性估计在这里：

其机制调这三类东西：权重，提示，输入输出行为。上述应用案例其实还没体现DSPy的真正价值，正确的玩法应该是尽可能减少人工提示词，而是利用数据来优化结果。具体可参考更复杂的应用案例：github.com/stanford-oval/storm。