当前，DSPy 在全球引起了轰动。它是斯坦福最受欢迎的开源项目，并且正在改变人们在大语言模型上构建应用的方式。然而，它的入门以及理解都存在一定的困难性，尤其是对于小白用户。本文旨在提供一个 DSPy 的易于理解的入门介绍。本文参考 YouTube 视频系列 Learn By Building AI[1]，以及翻译自文章《A gentle introduction to DSPy 》。
本文将从解释 DSPy 是什么以及它提供的主要优势开始。然后，我们会通过一个简单的例子，展示如何使用 DSPy 将文本翻译成“Grug 语言”。在这个过程中，我们会强调一些重要的概念，如构建数据集、定义提示（prompts）以及零样本提示（zero-shot prompting）。最后，我们将展示如何自动测量和优化提示。
本文将展示 DSPy 如何使这一过程更加数据驱动和系统化，从而让我们能够微调模型以获得更好的结果。
在本文结束时，你将能够对 DSPy 的工作原理以及如何构建基本的 DSPy 程序有一个扎实的理解。
背景我们都经历过这样的情况——你有一个很棒的项目想法，能够真正得益于大型语言模型（LMs）的强大功能，但实际操作它们的过程却像是穿越雷区。从确定合适的提示到获得一致可靠的结果，很容易在细节上陷入困境，而无法专注于核心目标。
需要很多手动调整。提示工程。微调。调试。简直是一团糟。DSPy 旨在改变这一点，提供一个更加系统化的方法。DSPy 是一个新的框架，它使得与 LMs 合作更加系统化且强大。
通过将程序的流程（模块）与参数（提示和权重）分离，DSPy 让用户能够专注于定义高层逻辑，而将 LM 的微调留给专门的优化器处理。
在本文中，将简要介绍 DSPy，并通过一个简单的示例来演示如何使用它将文本翻译成“Grug 语言”。在这个过程中，将探讨 DSPy 的方法和优化能力如何帮助你在大型语言模型之上构建更可靠且更强大的应用程序。
什么是 DSPyDSPy，即“演示-搜索-预测”框架[2]。在较高层面上，DSPy 将程序的流程（模块与逻辑）与控制 LLM 行为的参数（例如，提示）分离。这使得本文可以轻松地尝试不同的提示，调整它们，并在不重写核心应用程序逻辑的情况下衡量性能。DSPy 甚至可以替本文编写提示！这意味着，理想情况下，我们进需要专注于定义想要模型执行的任务，让 DSPy 处理使其可靠工作的细节。通过将程序流程与模型参数解耦，并提供优化工具，DSPy 试图减少对特定 LLMs 的依赖。
案例研究: Grug 语言翻译▶数据集 Datasets
DSPy 最重要的方面之一可能是你需要某种类型的数据集。当本文帮助初学者开始使用 DSPy 时，这通常是人们最大的知识空白或困惑点。在这个例子中，我们将使用/参考一篇精彩的文章：
The Grug Brained Developer A layman's guide to thinking like the self-aware smol brained
首先，我们需要构建一个包含 Grug 语音示例的数据集。为此，我们将抓取原始的 Grug 大脑网站，该网站包含用 Grug 语言编写的文本：
import requestsfrom bs4 import BeautifulSoupimport dspyres = requests.get("https://grugbrain.dev/")soup = BeautifulSoup(res.text, 'html.parser')raw_text = [p.text for p in soup.find_all('p') if p.text]raw_text[:10]
现在我们有了 一些 Grug 文本示例，我们需要一种方法将其翻译成普通英语。以下是本文编写的一个简单函数：
from openai import OpenAIclient = OpenAI()openai_model_name= "gpt-3.5-turbo"class BuildMessages:def __init__(self, system_prompt, user_prompt):self.system_prompt = system_promptself.user_prompt = user_promptdef render(self, **kwargs):sys = self.system_prompt.format(**kwargs)user = self.user_prompt.format(**kwargs)return [{"role":"system", "content":sys},{"role":"user", "content":user},]from functools import cache@cachedef translate_grug(grug_text):prompt = BuildMessages("You are an expert in deciphering strange text. The user will provide text written by someone named Grug and you will provide the translation.","""Translate the following text into plain english: '{text}'.Do not respond with any other text. Only provide that text. Now take a deep breath and begin.""")result = client.chat.completions.create(messages=prompt.render(text=grug_text), model=openai_model_name)return result.choices[0].message.content
现在我们可以通过翻译前 10 个 Grug 文本样本来构建我们的数据集。
dataset = []for grug_text in raw_text[:10]:translated = translate_grug(grug_text)dataset.append({"grug_text":grug_text, "plain_english":translated})这是我们基础的数据集，包含 Grug 文本及其“翻译”。为了在 DSPy 中使用这个数据集，我们需要从数据集中创建一系列 dspy.Example 对象：
examples = []for row in dataset:examples.append(dspy.Example(grug_text=row["grug_text"], plain_english=row["plain_english"]).with_inputs("plain_english"))这里的关键洞察是，我们创建了一组示例，DSPy 将代表我们使用这些示例来进行未来的翻译。我们指定了一个“普通英语”的输入，并获得 Grug 文本。通过调用 with_inputs("plain_english")，我们告诉 DSPy 我们示例的输入。然后我们可以将示例分为训练集和测试集：
import numpy as npfrom random import shuffledef split_for_train_test(values, test_size = 1/3.0):shuffle(values)train = int(len(values)-test_size*len(values))print(train)return values[:train], values[train:]train, test = split_for_train_test(examples)
▶定义 Signatures (or prompts)
现在从概念上来说，签名有点像提示。想象一下签名就像是任务规范。你有输入，你有输出，还有一个简单的提示（就像文档字符串）来描述任务。让我们为我们的 Grug 文本创建一个。
import dspyclass GrugTranslation(dspy.Signature):"Translate plain english to Grug text."plain_english = dspy.InputField()grug_text = dspy.OutputField()
在使用 DSPy 时，有一件事让本文感到困惑，那就是如何调试这个问题。本文知道，在某个点上，这个会被传递到一个语言模型中。在深入代码研究后，本文发现你可以打印出签名的字符串表示：
turbo = dspy.OpenAI(model='gpt-3.5-turbo', max_tokens=1000)dspy.settings.configure(lm=turbo)from dspy.signatures.signature import signature_to_templategrug_translation_as_template = signature_to_template(GrugTranslation)print(str(grug_translation_as_template))print(grug_translation_as_template.query(examples[0]))GrugTranslation.signatureGrugTranslation.with_instructions
这并不是对提示的完美表示，但这是一个不错的开始。
▶定义 Module
DSPy 使用模块来封装特定任务的逻辑。这个模块将接受纯英文文本作为输入，并在调用 forward 方法时返回相应的 Grug 文本。
class CoT(dspy.Module):def __init__(self):super().__init__()self.prog = dspy.ChainOfThought(GrugTranslation)def forward(self, plain_english):return self.prog(plain_english=plain_english)c = CoT()
DSPy 模块其实就是 Python。本文可以添加新的方法，自己的代码，随便什么。没有复杂抽象。这里强大的是 ChainOfThought 类。不必手动指定模型按照 chain of thought 进行操作，只需调用这个更高级别的抽象，DSPy 就会为本文处理一切。
到目前为止，本文已经介绍了基础知识。本文从收集数据开始，然后定义了翻译任务作为一个签名，并最终将其封装在一个可重用的模块中。这种系统化的方法使得本文可以轻松地进行实验，测量性能并优化翻译模型。在优化之前，本文可以进行一次零样本的前向传播：
c.forward("You should not construct complex systems.")这将给出模型对于 Grug 文本应该是什么的最佳猜测。
评估与优化到目前为止，本文已经进行了基本的提示工程，并没有特别花哨的内容。DSPy 能力的一个核心支柱就是优化。本文可以定义自定义指标，并使用它们来微调模型，这使得方法更加系统和数据驱动。接下来，探讨如何定义自定义指标以及如何使用它们来微调模型。现在对于 Grug 文本，本文需要一些指标来优化对抗。
▶Metric 1: Readability
最常见的可读性度量之一是自动可读性指数（ARI），它是一个公式，可以生成一个分数，近似于理解文本所需的年级水平。这个方法并不完美，但作为一个开始是很好的。
# https://apps.dtic.mil/sti/tr/pdf/AD0667273.pdfdef automated_readability_index(text):import recharacters = len(re.sub(r'\s+', '', text)) # Count characters (ignoring whitespace)words = len(text.split()) # Count words by splitting the text# Count sentences by finding period, exclamation, or question marksentences = len(re.findall(r'[.!?\n]', text))# small change is to add a new line character as grug doesn't seem to use punctuation.if words == 0 or sentences == 0:# Prevent division by zeroreturn 0# Calculate the Automated Readability Index (ARI)ari = (4.71 * (characters / words)) + (0.5 * (words / sentences)) - 21.43return round(ari, 2)
然后，本文可以使用这个函数来比较原始文本和"Grug 说法"翻译的可读性。
for ex in examples:source_ari = automated_readability_index(ex.plain_english)grug_ari = automated_readability_index(ex.grug_text)print(f"ARI {source_ari} => {grug_ari}")你会在这里看到一堆度量值，但基本上 Grug 说话的方式并不是最复杂的（谁会想到呢）。
▶Metric 2: AI 反馈
我们要考虑的另一个重要指标是语义相似度，它衡量的是翻译后的文本意义与原文的匹配程度。现在就来使用人工智能来评估一下输出结果。
# https://dspy-docs.vercel.app/docs/building-blocks/metrics#intermediate-using-ai-feedback-for-your-metricclass AssessBasedOnQuestion(dspy.Signature):"""Given the assessed text provide a yes or no to the assessment question."""assessed_text = dspy.InputField(format=str)assessment_question = dspy.InputField(format=str)assessment_answer = dspy.OutputField(desc="Yes or No")example_question_assessment = dspy.Example(assessed_text="This is a test.", assessment_question="Is this a test?", assessment_answer="Yes").with_inputs("assessed_text", "assessment_question")print(signature_to_template(AssessBasedOnQuestion).query(example_question_assessment))
注：example_question_assessment对象在技术上是一个`Prediction`对象[3]，但它具有Example的功能。
现在，本文可以实际定义一个相似度指标。这个指标接收一个真实值和一个预测值，并使用人工智能反馈来评估两个文本之间的语义相似度。这里使用的是 GPT4-Turbo，但您也可以使用任何您有权访问的模型。
gpt4T = dspy.OpenAI(model='gpt-4-turbo', max_tokens=500)def similarity_metric(truth, pred, trace=None):truth_grug_text = truth.grug_textproposed_grug_text = pred.grug_textsimilarity_question = f"""Does the assessed text have the same meaning as the gold_standard text provided?Gold Standard: "{truth_grug_text}"Provide only a yes or no answer."""with dspy.context(lm=gpt4T):assessor = dspy.Predict(AssessBasedOnQuestion)raw_similarity_result = assessor(assessed_text=proposed_grug_text, assessment_question=similarity_question)print(raw_similarity_result) # for debuggingraw_similarity = raw_similarity_result.assessment_answer.lower().strip()same_meaning = raw_similarity == 'yes'return same_meaning
您会注意到，需要指定truth和pred参数。这是 DSPy 用于优化时任何指标的标准接口。
def ari_metric(truth, pred, trace=None):truth_grug_text = truth.grug_textproposed_grug_text = pred.grug_textgold_ari = automated_readability_index(truth_grug_text)pred_ari = automated_readability_index(proposed_grug_text)print(f"ARI {gold_ari} => {pred_ari}")ari_result = pred_ari <= 7.01return ari_result
现在，已经定义了这两个指标。可以将它们结合成一个整体性能指标，该指标同时评估可读性和语义相似度。
def overall_metric(provided_example, predicted, trace=None):similarity = similarity_metric(provided_example, predicted, trace)ari = ari_metric(provided_example, predicted, trace)if similarity and ari:return Truereturn False
从流程上可以这样理解：
现在，可以使用优化技术，比如少样本学习，来微调模型并提高其性能。这使得本文能够采取一种更系统化、数据驱动的方法来处理语言模型。
▶运行优化流程
现在，本文将运行优化过程。本文指定了优化器，并使用我们的模块和数据对其进行编译。
from dspy.teleprompt import BootstrapFewShotconfig = dict(max_bootstrapped_demos=4, max_labeled_demos=4)optimizer = BootstrapFewShot(metric=overall_metric, **config)optimizer.max_errors = 1 # helpful to debug errors fasteroptimized_cot = optimizer.compile(CoT(), trainset=train, valset=test)
如果一直使用的是样本数据集，这个过程不会花费太长时间。需要注意的是，DSPy 将会自动缓存模型 API 调用。
▶评估输出
现在需要评估模型了！本文可以使用Evaluate类在优化后的模型上运行各项指标。在这种情况下，将打印出每个单独指标的结果。这样做是为了帮助识别可能出现问题的位置。
from dspy.evaluate import Evaluateindividual_metrics = [similarity_metric, ari_metric]for metric in individual_metrics:evaluate = Evaluate(metric=metric, devset=train, num_threads=1, display_progress=True, display_table=5)evaluate(optimized_cot)
显然，也可以手动检查输出结果。
optimized_cot.forward("You should not construct complex systems.")最后，为了以后的使用，可以保存模型：
optimized_cot.save(path="/tmp/model.json")
本文为您提供了一个关于 DSPy 的简单介绍，这是一个强大的框架系统，它对处理大型语言模型的过程进行了系统化。本文展示了 DSPy 如何允许我们将程序的流程与控制模型行为的参数分离，以及这种分离如何实现更数据驱动和可优化的方法。
通过将文本翻译成“Grug 语言”的例子，本文演示了 DSPy 如何轻松实现以下功能：
构建数据集 定义提示 使用自定义指标测量和优化性能
这种系统化的方法使我们能够创建更可靠、更有效的语言模型，这些模型可以退居幕后，使我们能够专注于核心任务和目标。后续，我们将更深入地探讨诸如提示优化、微调、建议、断言等高级用例。
