DSPy 声明式Prompt编程

import dspy

# 方式一：使用 OpenAI
lm = dspy.LM('openai/gpt-4o', api_key='your-api-key')
dspy.configure(lm=lm)

# 方式二：使用本地模型（通过 Ollama）
lm = dspy.LM('ollama_chat/llama3', api_base='http://localhost:11434', api_key='')
dspy.configure(lm=lm)

# 方式三：使用其他提供商
lm = dspy.LM('anthropic/claude-3-sonnet', api_key='your-api-key')
dspy.configure(lm=lm)

# 方式一：字符串语法定义（最常用）
class QASignature(dspy.Signature):
    """回答给定问题。"""  # 文档字符串作为任务描述
    question: str = dspy.InputField(desc="需要回答的问题")
    answer: str = dspy.OutputField(desc="问题的答案")

# 方式二：简写字符串语法（适用于简单场景）
# "输入字段1, 输入字段2 -> 输出字段1, 输出字段2"
signature = "question -> answer"
signature = "question, context -> answer, confidence"

# 方式三：完整类定义（推荐，可加描述）
class SummarizeSignature(dspy.Signature):
    """将长文本总结为简洁的摘要。"""
    text: str = dspy.InputField(desc="需要总结的原文")
    length: int = dspy.InputField(desc="摘要的目标字数")
    summary: str = dspy.OutputField(desc="生成的摘要")
    key_points: list = dspy.OutputField(desc="提取的关键要点")

import dspy

# 使用 Signature 类
predictor = dspy.Predict(QASignature)

# 使用字符串 Signature
predictor = dspy.Predict("question -> answer")

# 调用
result = predictor(question="法国的首都是哪里？")
print(result.answer)  # "巴黎"

# 访问完整输出
print(result.toDict())  # {'question': '...', 'answer': '...'}

cot = dspy.ChainOfThought("question -> answer")

result = cot(question="如果一辆火车每小时行驶60公里，3小时行驶多少公里？")
print(result.answer)      # "180公里"
print(result.rationale)   # 推理过程："我需要计算 60 × 3 = 180"

# 定义工具函数
def search_wikipedia(query: str) -> str:
    """搜索维基百科"""
    # 实际实现中调用搜索 API
    return f"关于 {query} 的搜索结果..."

def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        return str(eval(expression))
    except:
        return "计算错误"

# 创建 ReAct 模块
react = dspy.ReAct(
    signature="question -> answer",
    tools=[search_wikipedia, calculate],
    max_iters=5  # 最大推理-行动迭代次数
)

result = react(question="爱因斯坦的出生年份的平方是多少？")
print(result.answer)

pot = dspy.ProgramOfThought("question -> answer")

result = pot(question="计算1到100的和")
# LLM 会生成类似 sum(range(1, 101)) 的代码并执行
print(result.answer)  # "5050"

class RAGModule(dspy.Module):
    """检索增强生成模块"""

    def __init__(self, num_passages=3):
        super().__init__()
        self.retrieve = dspy.Retrieve(k=num_passages)
        self.generate_answer = dspy.ChainOfThought("context, question -> answer")

    def forward(self, question: str):
        # 检索相关文档
        context = self.retrieve(question).passages
        # 生成答案
        prediction = self.generate_answer(context=context, question=question)
        return prediction

# 使用
rag = RAGModule(num_passages=5)
result = rag(question="什么是量子计算？")
print(result.answer)

from dspy.teleprompt import BootstrapFewShot

# 定义评估指标
def answer_exact_match(example, prediction, trace=None):
    """检查预测答案是否与真实答案完全匹配"""
    return example.answer.lower().strip() == prediction.answer.lower().strip()

# 创建优化器
optimizer = BootstrapFewShot(
    metric=answer_exact_match,    # 评估指标函数
    max_bootstrapped_demos=4,     # 最大引导示例数
    max_labeled_demos=4,          # 最大标注示例数
    max_rounds=1,                 # 优化轮数
    max_errors=0                  # 允许的最大错误数
)

# 准备训练数据
trainset = [
    dspy.Example(question="1+1=?", answer="2").with_inputs("question"),
    dspy.Example(question="2+2=?", answer="4").with_inputs("question"),
    dspy.Example(question="3*3=?", answer="9").with_inputs("question"),
    # ... 更多示例
]

# 编译模块
compiled_module = optimizer.compile(
    student=RAGModule(),          # 待优化的模块
    trainset=trainset,            # 训练集
    # teacher=RAGModule(),        # 可选：教师模块（默认使用 student 自身）
)

from dspy.teleprompt import MIPROv2

optimizer = MIPROv2(
    metric=answer_exact_match,
    num_threads=4,                # 并行线程数
    num_candidates=10,            # 候选 prompt 数量
    max_bootstrapped_demos=4,
    max_labeled_demos=4,
    num_trials=20,                # 优化试验次数
    minibatch_size=25,            # 小批量评估大小
    minibatch_full_eval_steps=5,  # 全量评估间隔
    autobase="auto",              # 自动选择基础配置
)

# 编译
compiled_module = optimizer.compile(
    student=RAGModule(),
    trainset=trainset,
    eval_kwargs={"display": True, "display_progress": True},
)

from dspy.teleprompt import BootstrapFewShotWithRandomSearch

optimizer = BootstrapFewShotWithRandomSearch(
    metric=answer_exact_match,
    max_bootstrapped_demos=4,
    max_labeled_demos=4,
    num_candidate_programs=10,    # 候选程序数
    num_threads=4,                # 并行线程数
)

compiled_module = optimizer.compile(
    student=RAGModule(),
    trainset=trainset,
)

# 创建 Example
example = dspy.Example(
    question="法国的首都是哪里？",
    answer="巴黎",
    context="法国是欧洲国家..."
).with_inputs("question")  # 指定哪些字段是输入

# 访问字段
print(example.question)  # "法国的首都是哪里？"
print(example.answer)    # "巴黎"

# with_inputs() 指定输入字段
# 未在 with_inputs 中的字段被视为输出/标签
example = dspy.Example(
    question="1+1=?",
    answer="2"
).with_inputs("question")

# 创建数据集
class MyDataset(dspy.Dataset):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._train = [
            dspy.Example(question="Q1", answer="A1").with_inputs("question"),
            dspy.Example(question="Q2", answer="A2").with_inputs("question"),
        ]
        self._dev = [
            dspy.Example(question="Q3", answer="A3").with_inputs("question"),
        ]
        self._test = [
            dspy.Example(question="Q4", answer="A4").with_inputs("question"),
        ]

dataset = MyDataset()
trainset = dataset.train  # 训练集
devset = dataset.dev      # 验证集
testset = dataset.test    # 测试集

# 内置数据集
from dspy.datasets import HotPotQA
hotpot = HotPotQA(train_seed=1, train_size=20, eval_seed=2023, dev_size=50, test_size=0)
trainset = hotpot.train
devset = hotpot.dev

import dspy

# 定义评估指标
def answer_match(example, prediction, trace=None):
    """答案匹配指标"""
    return example.answer.lower().strip() == prediction.answer.lower().strip()

def answer_f1(example, prediction, trace=None):
    """基于 F1 分数的指标"""
    pred_tokens = set(prediction.answer.lower().split())
    gold_tokens = set(example.answer.lower().split())
    if not pred_tokens or not gold_tokens:
        return 0.0
    precision = len(pred_tokens & gold_tokens) / len(pred_tokens)
    recall = len(pred_tokens & gold_tokens) / len(gold_tokens)
    if precision + recall == 0:
        return 0.0
    return 2 * precision * recall / (precision + recall)

# 使用 Evaluate 类
evaluator = dspy.Evaluate(
    devset=devset,               # 评估数据集
    metric=answer_match,          # 评估指标
    num_threads=4,                # 并行线程数
    display_progress=True,        # 显示进度条
    display_table=5,              # 显示前5个样本的详细结果
)

# 评估模块
score = evaluator(module)
print(f"准确率: {score}")

# 评估编译前后的对比
uncompiled_score = evaluator(rag_module)
compiled_score = evaluator(compiled_module)
print(f"优化前: {uncompiled_score}, 优化后: {compiled_score}")

def metric(example: dspy.Example, prediction: dspy.Prediction, trace=None) -> bool | float:
    """
    参数:
        example: 标准答案示例
        prediction: 模块预测结果
        trace: 可选的执行追踪（用于高级评估）
    返回:
        bool 或 float 评分
    """
    pass

import dspy

# 全局配置 LLM
lm = dspy.LM(
    model='openai/gpt-4o',       # 模型标识: 提供商/模型名
    api_key='sk-...',             # API 密钥
    api_base='https://api.openai.com/v1',  # 可选：自定义 API 端点
    temperature=0.0,              # 生成温度
    max_tokens=2048,              # 最大生成 token 数
    num_retries=3,                # API 调用重试次数
    cache=True,                   # 是否缓存响应
)
dspy.configure(lm=lm)

# 查看当前配置
print(dspy.settings.lm)  # 当前 LM 对象

# 配置检索器（用于 RAG）
from dspy.retrieve import ColBERTv2
rm = ColBERTv2(url='http://20.102.90.50:2017/wiki17_abstracts')
dspy.configure(lm=lm, rm=rm)

# 多 LM 配置（不同模块使用不同模型）
class MyModule(dspy.Module):
    def __init__(self):
        super().__init__()
        self.cheap_lm = dspy.LM('openai/gpt-4o-mini', temperature=0.0)
        self.expert_lm = dspy.LM('openai/gpt-4o', temperature=0.3)

    def forward(self, question):
        # 使用 dspy.context 临时切换 LM
        with dspy.context(lm=self.cheap_lm):
            initial = dspy.Predict("question -> draft")(question=question)

        with dspy.context(lm=self.expert_lm):
            final = dspy.Predict("draft, question -> answer")(
                draft=initial.draft, question=question
            )
        return final

import dspy

class ValidatedQA(dspy.Module):
    def __init__(self):
        super().__init__()
        self.generate = dspy.ChainOfThought("question -> answer")

    def forward(self, question):
        prediction = self.generate(question=question)

        # dspy.Assert: 硬约束，条件不满足时抛出断言错误并重试
        dspy.Assert(
            len(prediction.answer) > 0,
            "答案不能为空",
        )

        # dspy.Suggest: 软约束，条件不满足时建议修改并重试
        dspy.Suggest(
            len(prediction.answer) < 500,
            "答案应该简洁，不超过500字",
        )

        return prediction

# 使用断言模块时需要包装
from dspy.primitives.assertions import assert_transform_module
validated_qa = assert_transform_module(ValidatedQA(), backtrack_handler)

# 或者直接在 forward 中使用
qa = ValidatedQA()
result = qa(question="什么是Python？")

# 配置最大回溯次数
from dspy.primitives.assertions import backtrack_handler

handler = backtrack_handler(max_backtracks=3)  # 最多重试3次
validated = assert_transform_module(ValidatedQA(), handler)

import dspy
from dspy.teleprompt import BootstrapFewShot
from dspy.retrieve import ColBERTv2

# 配置
lm = dspy.LM('openai/gpt-4o', api_key='your-key')
rm = ColBERTv2(url='http://20.102.90.50:2017/wiki17_abstracts')
dspy.configure(lm=lm, rm=rm)

# 定义 Signature
class GenerateAnswer(dspy.Signature):
    """根据检索到的上下文回答问题。"""
    context: str = dspy.InputField(desc="检索到的相关文档")
    question: str = dspy.InputField(desc="用户的问题")
    answer: str = dspy.OutputField(desc="基于上下文的答案")

# 定义模块
class RAG(dspy.Module):
    def __init__(self, num_passages=3):
        super().__init__()
        self.retrieve = dspy.Retrieve(k=num_passages)
        self.generate = dspy.ChainOfThought(GenerateAnswer)

    def forward(self, question):
        context = self.retrieve(question).passages
        prediction = self.generate(context=context, question=question)
        return prediction

# 评估指标
def gold_answer_match(example, prediction, trace=None):
    return example.answer.lower().strip() == prediction.answer.lower().strip()

# 准备数据
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]

# 优化前评估
uncompiled_rag = RAG()
evaluator = dspy.Evaluate(devset=devset, metric=gold_answer_match, num_threads=4)
before_score = evaluator(uncompiled_rag)
print(f"优化前准确率: {before_score:.2f}")

# 编译优化
optimizer = BootstrapFewShot(
    metric=gold_answer_match,
    max_bootstrapped_demos=4,
    max_labeled_demos=4,
)
compiled_rag = optimizer.compile(uncompiled_rag, trainset=trainset)

# 优化后评估
after_score = evaluator(compiled_rag)
print(f"优化后准确率: {after_score:.2f}")

# 使用优化后的模块
result = compiled_rag(question="阿尔伯特·爱因斯坦在哪出生？")
print(result.answer)

import dspy

# 配置
lm = dspy.LM('openai/gpt-4o', api_key='your-key')
dspy.configure(lm=lm)

# 数学推理：使用 ProgramOfThought
math_solver = dspy.ProgramOfThought("problem -> solution")
result = math_solver(problem="计算斐波那契数列第10项")
print(result.solution)

# 逻辑推理：使用 ChainOfThought
logic_solver = dspy.ChainOfThought("premise, hypothesis -> entailment, explanation")
result = logic_solver(
    premise="所有猫都是动物",
    hypothesis="小花是猫，所以小花是动物"
)
print(f"蕴含关系: {result.entailment}")
print(f"解释: {result.explanation}")

# 多模块组合
class MathReasoner(dspy.Module):
    def __init__(self):
        super().__init__()
        self.analyze = dspy.ChainOfThought("problem -> analysis, approach")
        self.solve = dspy.ProgramOfThought("analysis, approach -> solution")
        self.verify = dspy.ChainOfThought("problem, solution -> is_correct, correction")

    def forward(self, problem):
        # 步骤1：分析问题
        analysis = self.analyze(problem=problem)

        # 步骤2：求解
        solution = self.solve(
            analysis=analysis.analysis,
            approach=analysis.approach
        )

        # 步骤3：验证
        verification = self.verify(
            problem=problem,
            solution=solution.solution
        )

        # 如果不正确，返回修正
        if verification.is_correct.lower() == "no":
            return dspy.Prediction(
                solution=verification.correction,
                verified=False
            )

        return dspy.Prediction(
            solution=solution.solution,
            verified=True
        )

reasoner = MathReasoner()
result = reasoner(problem="一个圆的半径是5，求面积")
print(f"答案: {result.solution}, 已验证: {result.verified}")

import dspy
from dspy.primitives.assertions import assert_transform_module, backtrack_handler

lm = dspy.LM('openai/gpt-4o', api_key='your-key')
dspy.configure(lm=lm)

class QualityControlledSummary(dspy.Module):
    def __init__(self):
        super().__init__()
        self.summarize = dspy.ChainOfThought("document -> summary")
        self.extract = dspy.ChainOfThought("summary -> key_points")

    def forward(self, document):
        # 生成摘要
        pred = self.summarize(document=document)

        # 断言：摘要长度在合理范围内
        dspy.Assert(
            50 <= len(pred.summary) <= 500,
            f"摘要长度 {len(pred.summary)} 不在50-500范围内，请调整"
        )

        # 建议：摘要应包含原文关键词
        dspy.Suggest(
            any(kw in pred.summary for kw in document.split()[:5]),
            "摘要应包含原文的关键词"
        )

        # 提取要点
        points = self.extract(summary=pred.summary)

        # 断言：至少提取一个要点
        dspy.Assert(
            len(points.key_points) > 0,
            "必须提取至少一个关键要点"
        )

        return dspy.Prediction(
            summary=pred.summary,
            key_points=points.key_points
        )

# 包装断言模块
handler = backtrack_handler(max_backtracks=3)
module = assert_transform_module(QualityControlledSummary(), handler)

result = module(document="人工智能是计算机科学的一个分支...")
print(result.summary)
print(result.key_points)

# 1. 使用 TypedDict 风格的 Signature 并添加描述
class GoodSignature(dspy.Signature):
    """清晰的任务描述。"""
    question: str = dspy.InputField(desc="需要回答的问题，用中文表述")
    answer: str = dspy.OutputField(desc="简洁准确的答案，不超过100字")

# 避免过于简单的 Signature
# BAD: "q -> a"  (字段名和描述都不清晰)

# 2. 从小规模优化开始，逐步增加复杂度
# 第一步：使用 BootstrapFewShot 快速验证
optimizer_small = BootstrapFewShot(metric=my_metric, max_bootstrapped_demos=2)
compiled_v1 = optimizer_small.compile(module, trainset=trainset[:10])

# 第二步：使用更多数据和更高级优化器
optimizer_full = MIPROv2(metric=my_metric, num_trials=20)
compiled_v2 = optimizer_full.compile(module, trainset=trainset)

# 3. 保存和加载编译后的模块
compiled_module.save("optimized_module.json")
loaded_module = RAGModule()
loaded_module.load("optimized_module.json")

# 4. 使用 dspy.inspect 检查生成的 prompt
import dspy
# 查看最近的 LLM 调用
for entry in dspy.settings.lm.history:
    print(f"Prompt: {entry['prompt']}")
    print(f"Response: {entry['response']}")
    print("---")

# 5. 断言中使用具体的错误信息
dspy.Assert(
    len(prediction.answer) > 0,
    "答案不能为空，请重新生成一个非空的答案"  # 具体的修正建议
)

# 6. 合理划分训练/验证/测试集
# 训练集：用于优化器搜索 few-shot 示例
# 验证集：用于优化器选择最优配置
# 测试集：最终评估，不参与优化过程

# 7. 使用 with_inputs 明确输入字段
example = dspy.Example(
    question="Q", context="C", answer="A"
).with_inputs("question", "context")  # 明确指定输入字段
# answer 不在 with_inputs 中，被视为标签/输出

# 8. 评估时使用多线程加速
evaluator = dspy.Evaluate(
    devset=testset,
    metric=my_metric,
    num_threads=8,           # 根据API限制调整
    display_progress=True,
    display_table=0,         # 生产环境设为0，不显示详情
)

# 查看编译后的模块配置
print(compiled_module)  # 显示模块结构和优化后的配置

# 追踪单次执行
with dspy.context(trace=True):
    result = compiled_module(question="测试问题")
    for step in dspy.settings.trace:
        print(f"模块: {step[0].__class__.__name__}")
        print(f"输入: {step[1]}")
        print(f"输出: {step[2]}")

# 对比优化前后的 prompt
# 优化前
result_before = uncompiled_module(question="测试")
print("优化前最后一个 prompt:", dspy.settings.lm.history[-1])

# 优化后
result_after = compiled_module(question="测试")
print("优化后最后一个 prompt:", dspy.settings.lm.history[-1])