说白了就是一套小学数学应用题的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行左右代码完成了：

下载数据集构建训练集和测试集验证AI处理准确率封装训练好的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')code>

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:code>

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")code>

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 个人使用感受