# 背景 

学习下自动优化prompts方案，减少写提示词的工作量

# 用途

可以参考官网的例子，dspy可以用于：

Math

RAG

Classification

Information Extraction

Agents

Multi-Stage Pipelines

# 资源

https://dspy.ai/

# 启动

关注网页中以下几种不同调用方式的启动

OpenAI
Anthropic
Databricks
Gemini
Local LMs on your laptop
Local LMs on a GPU server
Other providers

官网提供的示例代码较简单，但实际会有更多入参，如下进行了一些介绍

import dspy
lm = dspy.LM("ollama_chat/llama3.2:1b", api_base="http://localhost:11434", api_key="",stream=True,complete_response=True,cache=True)
dspy.configure(lm=lm)
"""
model：ollama_chat/llama3.2:1b
api_base：本地接口
api_key:秘钥
stream：是否流式
complete_response：是否等待流式处理结束后返回所有内容
cache：是否基于缓存内容，如为True，且为之前一样的问题，会返回上次的回答。不再通过大模型即时生成
"""

# 模块：Predict、ChainofThought、ReAct

- predict

直接输出结果

- ChainofThought

CoT，返回思考过程，返回结果中会有reasoning

可以通过参数要求输出多个答案，如这里是5：

classify = dspy.ChainOfThought('question -> answer', n=5)

- ReAct

Reasoning+Action，调用工具（如查询某个网页的信息），进行推理及动作。官网上给了一个agents的示例，question是xxx除以某某的出生年份是多少。dspy根据question思考下一步act，并调用tools，再基于tools返回的结果继续reasoning和act，直到完成任务。

- ChainOfThoughtWithHint 

带提示的CoT

适合分类、意图识别、多标签任务

# label 更新时从classes直接拍选择
signature = dspy.Signature("text -> label").with_updated_fields('label', type_=Literal[tuple(CLASSES)])
# 训练时使用提示hint
classify = dspy.ChainOfThoughtWithHint(signature)

ChainOfThoughtWithHint和ChainOfThought的区别：ChainOfThoughtWithHint用到了一些提示，引导模型思考过程，hint可以是类别、关键词、分类逻辑等。

hint在训练和推理时都能用，训练时可以在trainset中显式地带上hint（text、label、hint）。推理阶段也可以传入hint提升准确率，推理时可以result = optimized(text=xxx,hint=xxx)

- ProgramOfThought

教导语言模型输出代码，代码的执行结果将决定响应。

- MultiChainComparison

可以比较来自 ChainOfThought 的多个输出，以产生最终预测。

- majority

可以进行基本投票，从一组预测中返回最常见的响应。

- Module 自定义模块

可以组合多个模块，形成一个更大的程序

# signature 签名

用于定义输入dspy.InputField、输出dspy.OutputField、相关的数据类型（str、list、dict）、字段描述desc

可以用dspy.Signature搭建一个类，用于定义复杂的数据流，再控制，如下例

from typing import Literal

class Classify(dspy.Signature):
    """Classify sentiment of a given sentence."""

    sentence: str = dspy.InputField()
    sentiment: Literal['positive', 'negative', 'neutral'] = dspy.OutputField()
    confidence: float = dspy.OutputField()

classify = dspy.Predict(Classify)
classify(sentence="This book was super fun to read, though not the last chapter.")

还可以指定数据类型，默认的是str："question -> answer"，这等同于 "question: str -> answer: str"

布尔型"sentence -> sentiment: bool"

复合型"question, choices: list[str] -> reasoning: str, selection: int"

可以结合指令instruct

toxicity = dspy.Predict(
    dspy.Signature(
        "comment -> toxic: bool",
        instructions="Mark as 'toxic' if the comment includes insults, harassment, or sarcastic derogatory remarks.",
    )
)

# Agent & tools & React 自我调用工具

可以定义一些tools，由DSPy搭建的智能体自己决定如何调用


def evaluate_math(expression: str):
    return dspy.PythonInterpreter({}).execute(expression)

def search_wikipedia(query: str):
    results = dspy.ColBERTv2(url='http://20.102.90.50:2017/wiki17_abstracts')(query, k=3)
    return [x['text'] for x in results]

react = dspy.ReAct("question -> answer: float", tools=[evaluate_math, search_wikipedia])

pred = react(question="What is 9362158 divided by the year of birth of David Gregory of Kinnairdy castle?")
print(pred.answer)

# dspy.Module & 多阶段流水线 & desc 自我扩展提示

官网给了个例子，根据主题先写大纲然后每个模块润色，可以看到这个例子里，最后写出一篇1500文字的文章，可以看到这里先用signature定义了2个class，每个class是一个步骤，然后再用Module定义一个class即DraftArticle用来串联。

这里的desc可以理解为简单写了点提示词？然后让depy来进行扩展？

class Outline(dspy.Signature):
    """Outline a thorough overview of a topic."""

    topic: str = dspy.InputField()
    title: str = dspy.OutputField()
    sections: list[str] = dspy.OutputField()
    section_subheadings: dict[str, list[str]] = dspy.OutputField(desc="mapping from section headings to subheadings")

class DraftSection(dspy.Signature):
    """Draft a top-level section of an article."""

    topic: str = dspy.InputField()
    section_heading: str = dspy.InputField()
    section_subheadings: list[str] = dspy.InputField()
    content: str = dspy.OutputField(desc="markdown-formatted section")

class DraftArticle(dspy.Module):
    def __init__(self):
        self.build_outline = dspy.ChainOfThought(Outline)
        self.draft_section = dspy.ChainOfThought(DraftSection)

    def forward(self, topic):
        outline = self.build_outline(topic=topic)
        sections = []
        for heading, subheadings in outline.section_subheadings.items():
            section, subheadings = f"## {heading}", [f"### {subheading}" for subheading in subheadings]
            section = self.draft_section(topic=outline.title, section_heading=section, section_subheadings=subheadings)
            sections.append(section.content)
        return dspy.Prediction(title=outline.title, sections=sections)

draft_article = DraftArticle()
article = draft_article(topic="World Cup 2002")

# 推理（非训练）

“将 AI 行为描述为代码，而非字符串。”，“DSPy 将您的签名扩展为提示并解析您的类型化输出”

例如上面多阶段流水线的例子，是简化了提示词,让depy来根据inputfield名称、outputfield名称、desc来自动扩展.

只需要告诉语言模型 需要做什么，而不是指定 如何 让语言模型去做。因此英文名称非常重要

# 优化器 & 编译

优化器可以参考pytorch中的概念

这里的编译指的是生成优化后的提示

优化器可以分为自动少样本学习、自动指令优化、自动微调、程序转换多种，具体看官网（https://dspy.org.cn/learn/optimization/optimizers/   https://dspy.org.cn/api/optimizers/MIPROv2/）吧，很复杂，官网还讲了些选择方法，下面介绍一些常见的。

- dspy.MIPROv2

“为每个提示提出并智能探索更好的自然语言指令”


import dspy
from dspy.datasets import HotPotQA

dspy.configure(lm=dspy.LM('openai/gpt-4o-mini'))

def search_wikipedia(query: str) -> list[str]:
    results = dspy.ColBERTv2(url='http://20.102.90.50:2017/wiki17_abstracts')(query, k=3)
    return [x['text'] for x in results]

trainset = [x.with_inputs('question') for x in HotPotQA(train_seed=2024, train_size=500).train]
react = dspy.ReAct("question -> answer", tools=[search_wikipedia])

tp = dspy.MIPROv2(metric=dspy.evaluate.answer_exact_match, auto="light", num_threads=24)
# 优化器.编译（任务，训练集）
optimized_react = tp.compile(react, trainset=trainset)


class RAG(dspy.Module):
    def __init__(self, num_docs=5):
        self.num_docs = num_docs
        self.respond = dspy.ChainOfThought('context, question -> response')

    def forward(self, question):
        context = search(question, k=self.num_docs)   # defined in tutorial linked below
        return self.respond(context=context, question=question)

tp = dspy.MIPROv2(metric=dspy.evaluate.SemanticF1(decompositional=True), auto="medium", num_threads=24)
# max_bootstrapped_demos 提示词中模型自己编的示例数量
# max_labeled_demos 提示词中标注的示例数量
optimized_rag = tp.compile(RAG(), trainset=trainset, max_bootstrapped_demos=2, max_labeled_demos=2)

- dspy.BootstrapFinetune 自举微调

“为您的模块构建数据集并使用它们微调系统中语言模型的权重”

- dspy.BootstrapRS

“为每个模块合成良好的 Few-shot 示例”

# 优化（之前称为 teleprompters）

需要少量训练输入。数量可以非常少（例如，只有 5 或 10 个样本），并且可能不完整（只有程序的输入，没有标签）。也可以是很多数据

# 评估

可以是自定义的，也可以是dspy内置的方法

metric=(lambda x, y, trace=None: x.label == y.label)

metric=dspy.evaluate.answer_exact_match

metric=dspy.evaluate.SemanticF1(decompositional=True)

dspy.evaluate.metrics.answer_passage_match

除了内置方法，还可以写一些复杂函数作为度量

def metric(gold, pred, trace=None):
    question, answer, tweet = gold.question, gold.answer, pred.output

    engaging = "Does the assessed text make for a self-contained, engaging tweet?"
    correct = f"The text should answer `{question}` with `{answer}`. Does the assessed text contain this answer?"

    correct =  dspy.Predict(Assess)(assessed_text=tweet, assessment_question=correct)
    engaging = dspy.Predict(Assess)(assessed_text=tweet, assessment_question=engaging)

    correct, engaging = [m.assessment_answer for m in [correct, engaging]]
    score = (correct + engaging) if correct and (len(tweet) <= 280) else 0

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

# 使用自定义数据集

网上给的示例是使用from dspy.datasets import HotPotQA加载数据，也可以用hf下载后放到本地，再从本地加载。但是我想用自己标注的数据，就需要处理为DSPy支持的格式。

d = {"1":"a","2":"b","3":"c","4":"d"}
train_set = [dspy.Example(input_key=k,answer=v).with_inputs('input_key') for k,v in d.items()]

generate_answer = dspy.Predict("input_key -> answer")
generate_answer(input_key=xxx)

 with_inputs() 方法，可以将特定字段标记为输入。（其余的只是元数据或标签。）

可以使用 .（点）运算符访问值，train_set[0].answer

带hint的数据集

train_set = [
    dspy.Example(
        input_key=k,
        label=v,
        hint = f"This question belongs to the category:{v}"
    ).with_inputs('input_key') for k,v in d.items()
]

要访问或排除某些键，请使用 inputs() 和 labels() 方法分别返回仅包含输入键或非输入键的新 Example 对象。

article_summary = dspy.Example(article= "This is an article.", summary= "This is a summary.").with_inputs("article")

input_key_only = article_summary.inputs()
non_input_key_only = article_summary.labels()

print("Example object with Input fields only:", input_key_only)
print("Example object with Non-Input fields only:", non_input_key_only)

# 保存与加载

官方提供了save()方法，对优化器编译后的模块保存为json文件

def AAA(dspy.Module):
    def __init__(self):
        ...
    def forward():
        ...
tp = dspy.MIPROv2(...) # 优化器
model = AAA() # 初始化模型
trainset = ... # 数据
res = tp.compile(model, trainset=trainset )

# 将模型结构、优化后的提示、配置保存到文件中
res.save('xxx.json')

# 实例化并加载权重
model2 = AAA()
model2.load('xxx.json')

# 官方教程

https://dspy.org.cn/tutorials/

结合自己的需要，选择对应的教程

# windows
Logo
2048 AI社区

有“AI”的1024 = 2048，欢迎大家加入2048 AI社区

更多推荐

cover

【TextIn大模型加速器 + 火山引擎】跨国药企23语言产品手册同步实战

avatar 2048 AI社区
cover

Transformer架构及其源码实现

avatar 2048 AI社区
cover

2025必备！8个AI论文平台，助你轻松搞定本科毕业论文！

avatar 2048 AI社区