DSPy：让 Prompt 自动优化而非手调

# 第 1 版
prompt_v1 = "从以下文本中提取人名：{text}"
# 准确率 60%

# 第 2 版
prompt_v2 = """你是一个专业的文本分析助手。
请仔细阅读以下文本，识别其中提到的人名。
人名通常以大写字母开头。
文本：{text}
请输出人名列表："""
# 准确率 65%

# 第 3 版
prompt_v3 = """从文本中提取所有人名。要求：
1. 输出为 JSON 数组
2. 忽略职位（Dr., Mr.）
3. 包含中英文名
文本：{text}"""
# 准确率 72%

# 第 4 版...
# ...永远调不到 100%

import dspy

# 定义一个 signature
class ExtractNames(dspy.Signature):
    """从文本中提取所有人名。"""
    
    text = dspy.InputField(desc="包含人名的原始文本")
    names = dspy.OutputField(desc="提取的人名列表，JSON 数组格式")

class NameExtractor(dspy.Module):
    def __init__(self):
        super().__init__()
        self.extract = dspy.ChainOfThought(ExtractNames)
    
    def forward(self, text):
        return self.extract(text=text)

from dspy.teleprompt import MIPROv2

# 准备训练数据
trainset = [
    dspy.Example(
        text="Steve Jobs 和 Bill Gates 创立了科技公司。",
        names=["Steve Jobs", "Bill Gates"]
    ).with_inputs("text"),
    # ... 50-200 条
]

# 配置优化器
optimizer = MIPROv2(
    metric=lambda ex, pred: 1.0 if set(ex.names) == set(pred.names) else 0.0,
    num_candidates=10,
    init_temperature=1.0
)

# 编译
optimized = optimizer.compile(
    NameExtractor(),
    trainset=trainset,
    num_trials=50  # 试 50 种 prompt 组合
)

# 配置
optimizer = MIPROv2(
    metric=accuracy_metric,
    num_candidates=10,        # 10 个候选指令
    num_bootstrap=4,          # 4 个 few-shot 候选
    num_trials=30,            # 总共试 30 种组合
    max_bootstrapped_demos=2,
    max_labeled_demos=2,
)

# 编译结果
optimized_program = optimizer.compile(
    program,
    trainset=trainset,
)

# 看看编译器找到了什么
for predictor in optimized_program.predictors():
    print(f"Signature: {predictor.signature}")
    print(f"Instructions: {predictor.extended_signature.instructions}")
    print(f"Demos: {len(predictor.demos)}")

from dspy.teleprompt import GEPA

optimizer = GEPA(
    metric=accuracy_metric,
    reflection_model=dspy.LM("gpt-4o"),  # 用 LLM 反思
    num_generations=10,
    population_size=20,
)

optimized = optimizer.compile(
    program,
    trainset=trainset,
)

import dspy

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

# 2. 自定义 retriever（用你自己的向量库）
def search_wikipedia(question: str, k: int = 3) -> list[str]:
    results = chroma_db.search(question, n_results=k)
    return [r["content"] for r in results]

# 3. 定义 Signature
class GenerateAnswer(dspy.Signature):
    """用给定的上下文回答用户问题。"""
    context = dspy.InputField(desc="检索到的相关文档")
    question = dspy.InputField()
    answer = dspy.OutputField(desc="2-3 句话的精炼答案")

# 4. 定义 RAG Module
class RAG(dspy.Module):
    def __init__(self, num_passages=3):
        super().__init__()
        self.num_passages = num_passages
        self.generate_answer = dspy.ChainOfThought(GenerateAnswer)
    
    def forward(self, question):
        context = search_wikipedia(question, k=self.num_passages)
        prediction = self.generate_answer(context=context, question=question)
        return prediction

# 5. 准备训练集
trainset = [
    dspy.Example(
        question="苹果公司的创始人是谁？",
        context="苹果公司由 Steve Jobs、Steve Wozniak 和 Ronald Wayne 于 1976 年创立。",
        answer="苹果公司由 Steve Jobs、Steve Wozniak 和 Ronald Wayne 于 1976 年创立。"
    ).with_inputs("question"),
    # ... 100 条
]

# 6. 编译
from dspy.teleprompt import MIPROv2

optimizer = MIPROv2(metric=answer_accuracy, num_trials=20)
compiled_rag = optimizer.compile(RAG(), trainset=trainset)

# 7. 部署
compiled_rag.save("./compiled_rag.json")

# 加载
loaded = RAG()
loaded.load("./compiled_rag.json")

from pydantic import BaseModel
import dspy

# Pydantic 保证结构
class PersonOutput(BaseModel):
    name: str
    age: int
    role: str

# DSPy Signature 用于优化
class ExtractPerson(dspy.Signature):
    """从文本中提取人物信息。"""
    text = dspy.InputField()
    person_json = dspy.OutputField(desc="JSON 格式，符合 PersonOutput schema")

class PersonExtractor(dspy.Module):
    def __init__(self):
        super().__init__()
        self.extract = dspy.ChainOfThought(ExtractPerson)
    
    def forward(self, text):
        result = self.extract(text=text)
        # Pydantic 二次校验
        person = PersonOutput.model_validate_json(result.person_json)
        return person

# 编译产物
compiled_rag.save("./rag_v1.json")

# 产物内容（简化）
{
    "generate_answer.predict.instructions": "用以下上下文回答问题。如果上下文没有相关信息，明确说'我不知道'...",
    "generate_answer.predict.demos": [
        {
            "context": "苹果公司由 Steve Jobs...",
            "question": "苹果创始人？",
            "answer": "Steve Jobs, Wozniak, Wayne"
        },
        # ...
    ]
}

# 方式 1：直接加载 JSON
loaded_rag = RAG()
loaded_rag.load("./rag_v1.json")

# 方式 2：导出 prompt 到 LangChain
from dspy.utils import export_prompt

prompt_text = export_prompt(loaded_rag.predictors()[0])
# 把 prompt_text 写到 LangChain 的 PromptTemplate