用DSPy替代手写Prompt：编程化优化LLM应用的工程实践

import dspy

# 1. 配置LLM（DSPy的统一接口，底层支持OpenAI/Anthropic/本地模型）
lm = dspy.LM("openai/gpt-4o-mini", api_key="sk-xxx")
dspy.configure(lm=lm)

# 2. Signature（签名）：声明输入输出，而不是写Prompt
class Sentiment(dspy.Signature):
    """判断文本的情感倾向"""
    text: str = dspy.InputField()
    sentiment: str = dspy.OutputField(desc="positive, negative, or neutral")

# 3. Module（模块）：可复用的推理单元，替代硬编码的Prompt Chain
classify = dspy.Predict(Sentiment)

# 使用
result = classify(text="这个产品太好用了！")
print(result.sentiment)  # positive

# ❌ 传统方式：Prompt硬编码，换模型就废了
prompt = f"""请判断以下文本的情感倾向，只返回positive、negative或neutral三个词之一。
文本：{text}
情感："""
response = openai.chat(prompt)

# ✅ DSPy方式：声明式签名，框架自动编排
classify = dspy.Predict(Sentiment)
result = classify(text=text)

import dspy

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

# 假设有一个向量检索器（DSPy支持接入任何检索后端）
from dspy.retrieve.chromadb_rm import ChromadbRM

retriever = ChromadbRM(
    collection_name="docs",
    persist_path="./chroma_db",
    embedding_function=dspy.SentenceTransformersEmbeddings("BAAI/bge-base-zh"),
)

# 定义RAG签名
class RAG(dspy.Signature):
    """根据检索到的上下文回答问题"""
    context: list[str] = dspy.InputField(desc="检索到的相关文档片段")
    question: str = dspy.InputField()
    answer: str = dspy.OutputField()

# 定义RAG模块
class RAGModule(dspy.Module):
    def __init__(self):
        self.retrieve = dspy.Retrieve(k=5)
        self.generate = dspy.ChainOfThought(RAG)

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

rag = RAGModule()
result = rag(question="什么是向量数据库？")
print(result.answer)

from dspy.teleprompt import BootstrapFewShot

# 定义评估指标
def validate_answer(example, prediction, trace=None):
    """简单的评估函数：答案不为空且包含关键词"""
    has_answer = len(prediction.answer.strip()) > 10
    return has_answer

# 准备训练数据（只需要问题+可选的答案）
trainset = [
    dspy.Example(question="什么是向量数据库？", answer="向量数据库...").with_inputs("question"),
    dspy.Example(question="Redis怎么用？", answer="Redis是一个...").with_inputs("question"),
    dspy.Example(question="Python异步编程是什么？", answer="async/await...").with_inputs("question"),
    # 10-20个例子就够
]

# 编译：自动优化few-shot示例的选择和排列
compiler = BootstrapFewShot(metric=validate_answer, max_bootstrapped_demos=4)
optimized_rag = compiler.compile(RAGModule(), trainset=trainset)

# 优化后的模型直接使用，Prompt已经被框架自动调整
result = optimized_rag(question="什么是Kubernetes？")

from dspy.teleprompt import MIPROv2

# 准备验证集
devset = [
    dspy.Example(question="什么是Docker？", answer="容器化技术...").with_inputs("question"),
    # 更多验证数据...
]

# MIPROv2：自动优化指令+示例
optimizer = MIPROv2(
    metric=validate_answer,
    num_candidates=10,       # 生成10个候选指令
    init_temperature=1.0,
    verbose=True,
)

optimized = optimizer.compile(
    RAGModule(),
    trainset=trainset,
    num_trials=20,           # 20轮优化实验
    max_bootstrapped_demos=4,
    max_labeled_demos=2,
)

# 保存优化后的模型
optimized.save("optimized_rag.json")

# 加载
loaded = RAGModule()
loaded.load("optimized_rag.json")

# Chain of Thought：自动加"Let's think step by step"
class MathQA(dspy.Signature):
    """解决数学应用题"""
    question: str = dspy.InputField()
    reasoning: str = dspy.OutputField(desc="推理过程")
    answer: str = dspy.OutputField(desc="最终数值答案")

solver = dspy.ChainOfThought(MathQA)
result = solver(question="小明有5个苹果，给了小红2个，又买了3个，现在有几个？")
print(result.reasoning)  # 小明开始有5个，给小红2个后剩3个，又买了3个，共6个
print(result.answer)     # 6

# ReAct：带工具调用的推理
class SearchAgent(dspy.Signature):
    """搜索并回答问题"""
    question: str = dspy.InputField()
    answer: str = dspy.OutputField()

def search_wikipedia(query: str) -> list[str]:
    """搜索维基百科"""
    # 实际实现接入Wikipedia API
    return ["相关文档片段..."]

agent = dspy.ReAct(SearchAgent, tools=[search_wikipedia])
result = agent(question="图灵测试是谁提出的？")

import dspy
from pydantic import BaseModel
from enum import Enum
from typing import Optional

# 定义意图类别
class Intent(str, Enum):
    ORDER_STATUS = "订单查询"
    REFUND = "退款申请"
    TECH_SUPPORT = "技术支持"
    COMPLAINT = "投诉建议"
    CHITCHAT = "闲聊"

class IntentClassification(dspy.Signature):
    """客服意图分类器：根据用户消息判断意图类别"""
    message: str = dspy.InputField(desc="用户发送的消息")
    intent: str = dspy.OutputField(desc="意图类别，从以下选择：订单查询、退款申请、技术支持、投诉建议、闲聊")
    confidence: float = dspy.OutputField(desc="置信度，0-1之间")

class CustomerServiceClassifier(dspy.Module):
    def __init__(self):
        self.classify = dspy.ChainOfThought(IntentClassification)

    def forward(self, message):
        result = self.classify(message=message)
        return result

# 准备训练数据
trainset = [
    dspy.Example(message="我的订单到哪了", intent="订单查询").with_inputs("message"),
    dspy.Example(message="买了三天还没发货", intent="订单查询").with_inputs("message"),
    dspy.Example(message="我要退货", intent="退款申请").with_inputs("message"),
    dspy.Example(message="东西坏了能退吗", intent="退款申请").with_inputs("message"),
    dspy.Example(message="App打不开闪退", intent="技术支持").with_inputs("message"),
    dspy.Example(message="你们服务太差了", intent="投诉建议").with_inputs("message"),
    dspy.Example(message="今天天气不错", intent="闲聊").with_inputs("message"),
    # ... 更多训练数据
]

# 评估指标
def evaluate(example, prediction, trace=None):
    return example.intent == prediction.intent

# 编译优化
from dspy.teleprompt import BootstrapFewShot
optimizer = BootstrapFewShot(metric=evaluate, max_bootstrapped_demos=3)
classifier = optimizer.compile(CustomerServiceClassifier(), trainset=trainset)

# 测试
test_messages = [
    "快递怎么还没到",
    "我要退款退货",
    "系统登不上去了",
    "你们公司真垃圾",
    "哈哈今天心情好",
]

for msg in test_messages:
    result = classifier(message=msg)
    print(f"消息: {msg}")
    print(f"意图: {result.intent} (置信度: {result.confidence})")
    print("---")