手工写 Prompt → 反复调参 → 测试 → 改 Prompt → 测试 → 部署 → 发现效果差 → 再改 Prompt...

定义模块/签名 → 编写 Pipeline 逻辑 → 准备数据 → 选择编译器 → 自动优化 → 部署

import dspy

# 定义签名 —— 取代写 Prompt
class GenerateAnswer(dspy.Signature):
    """基于上下文信息回答问题。"""
    context = dspy.InputField(desc="相关的参考文档片段")
    question = dspy.InputField(desc="用户提出的问题")
    answer = dspy.OutputField(desc="准确、简洁的答案")

class RAG(dspy.Module):
    def __init__(self, num_passages=3):
        self.retrieve = dspy.Retrieve(k=num_passages)
        self.generate_answer = dspy.ChainOfThought(GenerateAnswer)

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

from dspy.teleprompt import BootstrapFewShot

optimizer = BootstrapFewShot(
    metric=validate_answer,  # 评估函数
    max_bootstrapped_demos=8,  # 最大引导示例数
    max_labeled_demos=16,     # 最大标注示例数
)

# 编译优化 —— 自动生成 Few-Shot 示例并优化 Prompt
optimized_rag = optimizer.compile(RAG(), trainset=trainset)

trainset (训练数据)
    │
    ▼
┌─────────────────────┐
│   BootstrapFewShot   │
│   (引导学习)         │
└─────────┬───────────┘
          │ 生成高质量示例
          ▼
┌─────────────────────┐
│   Pipeline 编译      │
│   (Prompt 自动化化)   │
└─────────┬───────────┘
          │ 优化后的 Pipeline
          ▼
┌─────────────────────┐
│   Evaluator (评估)   │
│   + Metric           │
└─────────┬───────────┘
          │ 验证通过
          ▼
    生产部署

class SentimentAnalysis(dspy.Signature):
    """分析文本的情感倾向。"""
    text = dspy.InputField()
    sentiment = dspy.OutputField(desc="positive, negative, or neutral")
    confidence = dspy.OutputField(desc="confidence score 0.0-1.0")

predictor = dspy.Predict(SentimentAnalysis)
result = predictor(text="这个产品用起来非常顺手！")
# result.sentiment = "positive", result.confidence = 0.95

class MathReasoning(dspy.Signature):
    problem = dspy.InputField()
    answer = dspy.OutputField(desc="最终答案")

cot = dspy.ChainOfThought(MathReasoning)
result = cot(problem="小明有5个苹果，吃了2个后妈妈又给了3个，现在有几个？")
# result.reasoning = "小明开始有5个苹果，吃了2个还剩5-2=3个..."
# result.answer = "6"

cot_hint = dspy.ChainOfThoughtWithHint(MathReasoning)
result = cot_hint(
    problem="...",
    hint="先用减法计算剩余数量，再用加法计算最终数量"
)

class DataAnalysis(dspy.Signature):
    data = dspy.InputField(desc="CSV 格式的数据")
    analysis = dspy.OutputField(desc="数据分析结果")

pot = dspy.ProgramOfThought(DataAnalysis)
result = pot(data="日期,销售额\n2024-01,100\n2024-02,150\n2024-03,120")

# 配置检索模型
dspy.settings.configure(
    rm=dspy.ColBERTv2(url="http://localhost:8080/api/search"),
    lm=dspy.OpenAI(model="gpt-4o")  # 或本地模型
)

retriever = dspy.Retrieve(k=5)
result = retriever("什么是思维链推理？")
# result.passages = ["思维链推理是...", ...]

class CustomRetriever(dspy.Retrieve):
    def __init__(self, k=3):
        super().__init__(k=k)
        self.vector_store = MyVectorStore()

    def forward(self, query_or_queries, **kwargs):
        queries = [query_or_queries] if isinstance(query_or_queries, str) else query_or_queries
        all_passages = []
        for query in queries:
            results = self.vector_store.search(query, k=self.k)
            all_passages.extend(results)
        return dspy.Prediction(passages=all_passages)

class CalculatorTool(dspy.Tool):
    name = "calculator"
    desc = "执行数学计算"

    def __call__(self, expression: str) -> str:
        try:
            return str(eval(expression))
        except Exception as e:
            return f"Error: {e}"

class SearchTool(dspy.Tool):
    name = "web_search"
    desc = "搜索网络信息"

    def __call__(self, query: str) -> str:
        # 调用搜索 API
        return search_web(query)

agent = dspy.ReAct(
    signature="question -> answer",
    tools=[CalculatorTool(), SearchTool()],
    max_iters=10
)

result = agent(question="2024年诺贝尔物理学奖得主是谁？他是哪一年出生的？")

1. Thought: 我需要搜索诺贝尔物理学奖得主
2. Action: web_search(query="2024 诺贝尔物理学奖")
3. Observation: John Hopfield 和 Geoffrey Hinton...
4. Thought: 现在需要查 Hinton 的出生年份
5. Action: web_search(query="Geoffrey Hinton 出生年份")
6. Observation: 1947年...
7. Final Answer: Geoffrey Hinton，1947年出生...

from dspy.teleprompt import BootstrapFewShot

def validate_context_and_answer(example, pred, trace=None):
    """验证函数：检查答案是否准确且基于上下文"""
    # 1. 答案不能为空
    if not pred.answer:
        return False
    # 2. 答案相关度检查（简化版）
    if example.answer not in pred.answer:
        return False
    return True

optimizer = BootstrapFewShot(
    metric=validate_context_and_answer,
    max_bootstrapped_demos=4,
    max_labeled_demos=8,
    max_rounds=1,
    max_errors=5
)

compiled_rag = optimizer.compile(RAG(), trainset=trainset)

from dspy.teleprompt import BootstrapFewShotWithRandomSearch

optimizer = BootstrapFewShotWithRandomSearch(
    metric=validate_answer,
    num_candidate_programs=8,  # 候选程序数量
    num_threads=4,             # 并行线程
)

compiled_rag = optimizer.compile(RAG(), 
    trainset=trainset,
    valset=valset,  # 验证集
)

from dspy.teleprompt import MIPROv2

optimizer = MIPROv2(
    metric=validate_answer,
    num_candidates=10,      # 候选指令数
    init_temperature=0.8,   # 指令生成温度
    verbose=True,
)

optimized_rag = optimizer.compile(
    RAG(),
    trainset=trainset,
    max_bootstrapped_demos=4,
    max_labeled_demos=8,
    num_trials=30,          # 搜索试验次数
    seed=42,
)

1. 训练数据 → 生成候选示例
2. 对每个模块 → 用 LLM 生成候选指令（描述更好格式/约束）
3. 组合候选指令+示例 = 候选程序
4. 使用 Bayesian Optimization 优化组合
5. 返回验证集上得分最高的程序

from dspy.teleprompt import COPRO

optimizer = COPRO(
    metric=validate_answer,
    breadth=10,      # 每轮生成的候选数
    depth=3,         # 优化深度（轮数）
    init_temperature=0.8,
)

compiled = optimizer.compile(RAG(), trainset=trainset, 
    eval_kwargs={"num_threads": 4})

from dspy.teleprompt import BootstrapFewShot, MIPROv2, Ensemble

# 训练多个优化器
optimizer1 = BootstrapFewShot(metric=validate_answer)
optimizer2 = MIPROv2(metric=validate_answer, num_candidates=5)

compiled1 = optimizer1.compile(RAG(), trainset=trainset[:50])
compiled2 = optimizer2.compile(RAG(), trainset=trainset[:50])

# 集成
ensemble = Ensemble()
ensemble.add_program(compiled1, weight=0.4)
ensemble.add_program(compiled2, weight=0.6)

# 运行时自动加权投票
result = ensemble(question="...")

class MultiHopRAG(dspy.Module):
    def __init__(self, num_hops=2):
        self.num_hops = num_hops
        self.retrieve = dspy.Retrieve(k=5)

        # 每跳的查询生成器
        self.generate_query = dspy.ChainOfThought(
            "context, question -> search_query"
        )
        # 最终答案生成器
        self.generate_answer = dspy.ChainOfThought(
            "context, question -> answer"
        )

    def forward(self, question):
        context = []
        current_query = question

        for hop in range(self.num_hops):
            # 生成子查询
            query_result = self.generate_query(
                context="\n".join(context[-3:]) if context else "",
                question=current_query
            )

            # 检索
            retrieved = self.retrieve(query_result.search_query)
            new_passages = retrieved.passages

            # 累积上下文
            context.extend(new_passages)
            current_query = query_result.search_query

        # 最终回答
        return self.generate_answer(
            context="\n".join(context[-10:]),
            question=question
        )

class DatabaseQueryTool(dspy.Tool):
    name = "query_database"
    desc = "用 SQL 查询数据库获取信息"

    def __call__(self, sql_query: str) -> str:
        import sqlite3
        try:
            conn = sqlite3.connect(":memory:")
            cursor = conn.execute(sql_query)
            return str(cursor.fetchall())
        except Exception as e:
            return f"Query error: {e}"

class EmailTool(dspy.Tool):
    name = "send_email"
    desc = "发送邮件到指定地址"

    def __call__(self, to: str, subject: str, body: str) -> str:
        # 实际发送邮件
        return f"Email sent to {to}"

class CustomerServiceAgent(dspy.Module):
    def __init__(self, lm):
        self.lm = lm
        self.react = dspy.ReAct(
            signature="customer_query, user_info -> response, action_taken",
            tools=[DatabaseQueryTool(), EmailTool()],
            max_iters=15,
        )

    def forward(self, customer_query, user_info):
        result = self.react(
            customer_query=customer_query,
            user_info=user_info
        )
        return dspy.Prediction(
            response=result.response,
            action_taken=result.action_taken,
        )

class IntentClassifier(dspy.Signature):
    """对用户意图进行分类。"""
    query = dspy.InputField()
    intent = dspy.OutputField(desc="refund, order_status, complaint, or general")

class RefundHandler(dspy.Module):
    def __init__(self):
        self.check = dspy.ChainOfThought(
            "order_id, reason -> eligibility, refund_amount, explanation"
        )
    def forward(self, **kwargs):
        return self.check(**kwargs)

class OrderStatusHandler(dspy.Module):
    def __init__(self):
        self.track = dspy.ChainOfThought(
            "order_id -> status, estimated_delivery, tracking_link"
        )
    def forward(self, **kwargs):
        return self.track(**kwargs)

class IntelligentRouter(dspy.Module):
    def __init__(self):
        self.classifier = dspy.Predict(IntentClassifier)
        self.handlers = {
            "refund": RefundHandler(),
            "order_status": OrderStatusHandler(),
            "complaint": RefundHandler(),  # 投诉也转退款
            "general": dspy.Predict(
                "query -> response"
            ),
        }

    def forward(self, query):
        intent = self.classifier(query=query).intent
        handler = self.handlers.get(intent, self.handlers["general"])
        return handler(query=query)

class SelfRefinePipeline(dspy.Module):
    def __init__(self, max_refine_rounds=3):
        self.max_refine_rounds = max_refine_rounds
        self.generator = dspy.ChainOfThought("task -> initial_output")
        self.critic = dspy.ChainOfThought(
            "task, generated_output -> issues, improvement_suggestions"
        )
        self.refiner = dspy.ChainOfThought(
            "task, previous_output, critic_feedback -> refined_output"
        )

    def forward(self, task):
        output = self.generator(task=task)

        for round_idx in range(self.max_refine_rounds):
            # 批评阶段
            critique = self.critic(
                task=task,
                generated_output=output.initial_output
            )

            # 检查是否还需改进
            if "无问题" in critique.issues or "no issues" in critique.issues.lower():
                break

            # 优化阶段
            output = self.refiner(
                task=task,
                previous_output=output.initial_output,
                critic_feedback=critique.improvement_suggestions
            )

        return dspy.Prediction(final_output=output.refined_output)

import dspy
from dspy.teleprompt import MIPROv2
from dspy.datasets import HotPotQA
from dspy.evaluate import Evaluate

# ============ 1. 配置 LM 与 RM ============
# OpenAI
# dspy.settings.configure(lm=dspy.OpenAI(model="gpt-4o"))

# 或使用本地模型（llama.cpp 等）
local_lm = dspy.HFModel(model="meta-llama/Meta-Llama-3-70B-Instruct")
colbert_rm = dspy.ColBERTv2(url="http://localhost:8080/api/search")

dspy.settings.configure(lm=local_lm, rm=colbert_rm)

# ============ 2. 定义签名 ============
class GenerateAnswer(dspy.Signature):
    """根据检索到的上下文提供准确、客观的答案。
    如果上下文中没有足够信息直接回答，请明确说明。
    使用中文回答用户问题。"""
    context = dspy.InputField(desc="与问题相关的文档片段")
    question = dspy.InputField(desc="用户原始问题")
    answer = dspy.OutputField(desc="准确的答案（中文）")
    confidence = dspy.OutputField(desc="可信度评分 0.0-1.0")
    sources = dspy.OutputField(desc="引用来源")

# ============ 3. 定义模块 ============
class ProductionRAG(dspy.Module):
    def __init__(self, num_passages=3):
        self.num_passages = num_passages
        self.retrieve = dspy.Retrieve(k=num_passages)
        self.analyze_query = dspy.ChainOfThought(
            "question -> query_type, keywords"
        )
        self.generate_answer = dspy.ChainOfThought(GenerateAnswer)

    def forward(self, question):
        # 查询分析
        analysis = self.analyze_query(question=question)

        # 多关键词检索
        contexts = []
        keywords = [question, analysis.keywords] if hasattr(analysis, 'keywords') else [question]
        for q in keywords[:3]:
            retrieved = self.retrieve(q)
            contexts.extend(retrieved.passages)

        # 去重
        seen = set()
        unique_contexts = []
        for ctx in contexts:
            if ctx not in seen:
                seen.add(ctx)
                unique_contexts.append(ctx)

        context_str = "\n---\n".join(unique_contexts[:self.num_passages])

        # 生成答案
        return self.generate_answer(
            context=context_str,
            question=question
        )

# ============ 4. 加载数据 ============
# HotPotQA 数据集示例
dataset = HotPotQA()
trainset = [dataset[i] for i in range(50)]
valset = [dataset[i] for i in range(50, 100)]

# ============ 5. 定义评估 metric ============
def validate_answer(example, pred, trace=None):
    # 检查是否有答案
    if not pred.answer or len(pred.answer) < 3:
        return False
    # 检查是否包含正确答案关键词
    correct = example.answer.lower()
    if correct not in pred.answer.lower():
        return False
    return True

# ============ 6. 编译优化 ============
optimizer = MIPROv2(
    metric=validate_answer,
    num_candidates=8,
    init_temperature=0.8,
)

compiled_rag = optimizer.compile(
    ProductionRAG(),
    trainset=trainset,
    max_bootstrapped_demos=4,
    max_labeled_demos=8,
    num_trials=20,
)

# ============ 7. 评估 ============
evaluator = Evaluate(
    devset=valset,
    metric=validate_answer,
    num_threads=4,
    display_progress=True
)

eval_score = evaluator(compiled_rag)
print(f"评估得分: {eval_score}")

# ============ 8. 保存 & 加载 ============
compiled_rag.save("compiled_rag.dspy")
# loaded_rag = ProductionRAG()
# loaded_rag.load("compiled_rag.dspy.compiled")

# ============ 9. 运行 ============
result = compiled_rag(question="什么是思维链提示？")
print(f"答案: {result.answer}")
print(f"可信度: {result.confidence}")
print(f"来源: {result.sources}")

# 在 LangChain Pipeline 中使用 DSPy 优化模块
from langchain.chains import RetrievalQA
from langchain.vectorstores import Chroma

class LangChainDSPyRAG(dspy.Module):
    def __init__(self, vector_store):
        self.retriever = vector_store.as_retriever(search_kwargs={"k": 5})
        self.generate = dspy.ChainOfThought(
            "context, question -> answer, confidence"
        )

    def forward(self, question):
        docs = self.retriever.get_relevant_documents(question)
        context = "\n".join([d.page_content for d in docs])
        return self.generate(context=context, question=question)

# 用 DSPy 优化 LangChain 的检索器
rag = LangChainDSPyRAG(chroma_store)
optimizer = BootstrapFewShot(metric=my_metric)
optimized_rag = optimizer.compile(rag, trainset=trainset)

from dspy.evaluate import Evaluate

# 定义评估
evaluator = Evaluate(
    devset=devset,
    metric=validate_answer,
    num_threads=8,
    display_progress=True,
    display_table=5,  # 显示前5个示例的详细结果
)

# 评估原始 vs 优化后
print("原始 RAG 得分:", evaluator(RAG()))
print("优化后 RAG 得分:", evaluator(compiled_rag))

# 追踪每次调用
dspy.inspect_history(n=3)  # 查看最近3次 LLM 调用

# 输出示例：
# 1. 调用 ChainOfThought (GenerateAnswer) #1
# Prompt:
# 根据上下文提供答案...
# 上下文: [检索到的文档...]
# 问题: ...
#
# 思考: 根据上下文...
# 答案: ...

import hashlib
import json
from typing import Dict, Any

class SemanticCache:
    def __init__(self, cache_size=100):
        self.cache: Dict[str, Any] = {}
        self.cache_size = cache_size

    def _make_key(self, question: str, module_name: str) -> str:
        content = f"{module_name}:{question.lower().strip()}"
        return hashlib.md5(content.encode()).hexdigest()

    def get(self, question: str, module_name: str):
        key = self._make_key(question, module_name)
        return self.cache.get(key)

    def set(self, question: str, module_name: str, result):
        key = self._make_key(question, module_name)
        if len(self.cache) >= self.cache_size:
            # LRU 淘汰
            oldest = min(self.cache.keys(), 
                key=lambda k: self.cache[k].get("timestamp", 0))
            del self.cache[oldest]
        self.cache[key] = {"result": result, "timestamp": time.time()}

class CachedRAG(dspy.Module):
    def __init__(self, base_rag):
        self.base_rag = base_rag
        self.cache = SemanticCache()

    def forward(self, question):
        cached = self.cache.get(question, "rag")
        if cached:
            return cached["result"]
        result = self.base_rag(question=question)
        self.cache.set(question, "rag", result)
        return result

# OpenAI
dspy.settings.configure(lm=dspy.OpenAI(model="gpt-4o"))

# Anthropic
dspy.settings.configure(lm=dspy.Anthropic(model="claude-3-opus-20240229"))

# 本地模型（通过 vLLM 或 llama.cpp）
dspy.settings.configure(lm=dspy.HFModel(model="path/to/model"))

# Google
dspy.settings.configure(lm=dspy.Google(model="gemini-1.5-pro"))

# 多模型配置
gpt4 = dspy.OpenAI(model="gpt-4o", max_tokens=4000)
claude = dspy.Anthropic(model="claude-3-opus", max_tokens=4000)

class EnsembleGenerator(dspy.Module):
    def __init__(self):
        self.gpt4_generator = dspy.ChainOfThought(GenerateAnswer, lm=gpt4)
        self.claude_generator = dspy.ChainOfThought(GenerateAnswer, lm=claude)
        self.selector = dspy.Predict("q1, q2, question -> best")

    def forward(self, question):
        gpt4_result = self.gpt4_generator(question=question, context="")
        claude_result = self.claude_generator(question=question, context="")
        selection = self.selector(
            q1=gpt4_result.answer, 
            q2=claude_result.answer, 
            question=question
        )
        return selection.best

class VagueSignature(dspy.Signature):
    input_text = dspy.InputField()
    output = dspy.OutputField()

class PreciseSignature(dspy.Signature):
    """将给定的技术摘要转换为适合非技术读者的通俗解释。"""
    technical_summary = dspy.InputField(
        desc="技术文章或论文摘要"
    )
    plain_explanation = dspy.OutputField(
        desc="通俗易懂的解释，避免专业术语"
    )
    analogy = dspy.OutputField(
        desc="一个帮助理解的类比"
    )

def bad_metric(example, pred, trace=None):
    return "正确" in pred.answer

def good_metric(example, pred, trace=None):
    score = 0.0

    # 答案非空
    if not pred.answer or len(pred.answer) < 5:
        return 0.0
    score += 0.3

    # 答案相关（包含正确答案关键词）
    if example.answer.lower() in pred.answer.lower():
        score += 0.4

    # 答案长度合理
    if 10 < len(pred.answer) < 1000:
        score += 0.15

    # 可信度合理
    if hasattr(pred, 'confidence') and 0 <= pred.confidence <= 1:
        score += 0.15

    return score

from sklearn.model_selection import train_test_split

# 划分数据集
train_data, test_data = train_test_split(full_dataset, test_size=0.2)
train_data, val_data = train_test_split(train_data, test_size=0.2)

# 编译用训练集
compiled_program = optimizer.compile(program, trainset=train_data)

# 评估用测试集
evaluator = Evaluate(devset=test_data, metric=validate_answer)
score = evaluator(compiled_program)

1. 定义清晰的 Signature（带 desc 字段）
2. 编写最小可行 Pipeline
3. 构建评估 Metric（多维 + 量化）
4. 准备 50-100 条训练数据
5. 运行 BootstrapFewShot 基准
6. 切换到 MIPROv2 精细优化
7. 用测试集验证泛化能力
8. 保存优化后的 Pipeline
9. 集成到生产环境

# ✅ 好的命名
class FinancialReportAnalyzer(dspy.Module):
    """分析财务报表并生成摘要。"""

# ✅ 合理的 desc
profit = dspy.OutputField(desc="净利润（单位：万元），取整")

# ✅ 模块化设计
class Step1Extraction(dspy.Module): ...
class Step2Analysis(dspy.Module): ...
class Step3Report(dspy.Module): ...

# ✅ 缓存策略
class CachedPipeline(dspy.Module):
    def __init__(self, pipeline):
        self.pipeline = pipeline
        self.cache = {}