GEPA实践指南：从入门到精通

快速入门：5分钟上手GEPA

最简单的GEPA使用示例

import dspy

# 1. 定义你的DSPy程序
class SimpleQA(dspy.Module):
    def __init__(self):
        super().__init__()
        self.answer = dspy.Predict("question -> answer")
    
    def forward(self, question):
        return self.answer(question=question)

# 2. 准备数据
trainset = [
    dspy.Example(question="什么是人工智能？", answer="人工智能是模拟人类智能的计算机系统"),
    dspy.Example(question="Python是什么语言？", answer="Python是一种高级编程语言"),
    # ... 更多训练数据
]

# 3. 定义评估指标
def simple_metric(gold, pred, trace=None, pred_name=None, pred_trace=None):
    """简单的精确匹配指标"""
    score = 1.0 if gold.answer.lower() == pred.answer.lower() else 0.0
    feedback = f"正确答案是: {gold.answer}, 你的回答是: {pred.answer}"
    return {"score": score, "feedback": feedback}

# 4. 运行GEPA优化
gepa = dspy.GEPA(
    metric=simple_metric,
    auto="light",  # 快速实验模式
    reflection_lm=dspy.LM(model='gpt-4'),  # 使用GPT-4进行反思
    track_stats=True
)

optimized_program = gepa.compile(
    student=SimpleQA(),
    trainset=trainset
)

# 5. 使用优化后的程序
result = optimized_program(question="机器学习是什么？")
print(f"优化后的回答: {result.answer}")

# 查看优化详情
print(f"最佳分数: {optimized_program.detailed_results.val_aggregate_scores[optimized_program.detailed_results.best_idx]}")

中级应用：构建复杂系统的优化

多步骤推理任务的优化

class MultiStepReasoning(dspy.Module):
    def __init__(self):
        super().__init__()
        self.generate_thought = dspy.ChainOfThought("question -> reasoning")
        self.generate_answer = dspy.Predict("question, reasoning -> final_answer")
    
    def forward(self, question):
        reasoning = self.generate_thought(question=question)
        return self.generate_answer(question=question, reasoning=reasoning.reasoning)

# 高级反馈函数
def reasoning_metric(gold, pred, trace, pred_name, pred_trace):
    """针对推理任务的细粒度反馈"""
    
    if pred_name == "generate_thought":
        # 对推理步骤的专门反馈
        reasoning_quality = assess_reasoning_quality(gold.reasoning, pred.reasoning)
        feedback = f"推理步骤分析: {reasoning_feedback}"
        return {"score": reasoning_quality, "feedback": feedback}
    
    elif pred_name == "generate_answer":
        # 对最终答案的反馈
        answer_score = 1.0 if gold.final_answer == pred.final_answer else 0.0
        reasoning_context = "基于之前的推理步骤" if trace else ""
        feedback = f"最终答案评估: {answer_score}, {reasoning_context}"
        return {"score": answer_score, "feedback": feedback}
    
    else:
        # 系统级别的整体评估
        overall_score = calculate_overall_score(gold, pred)
        return overall_score

# 配置GEPA进行深度优化
gepa = dspy.GEPA(
    metric=reasoning_metric,
    auto="medium",
    reflection_lm=dspy.LM(model='gpt-4', temperature=0.7, max_tokens=4000),
    reflection_minibatch_size=4,
    candidate_selection_strategy="pareto",
    use_merge=True,
    track_stats=True
)

工具调用任务的优化

class ToolUsingAgent(dspy.Module):
    def __init__(self):
        super().__init__()
        self.plan = dspy.ChainOfThought("task -> steps")
        self.execute = dspy.ReAct("task, steps -> result", tools=[calculator, web_search])
    
    def forward(self, task):
        plan = self.plan(task=task)
        return self.execute(task=task, steps=plan.steps)

# 工具使用评估指标
def tool_metric(gold, pred, trace, pred_name, pred_trace):
    """评估工具使用效果"""
    
    if pred_name == "execute" and pred_trace:
        # 分析工具调用轨迹
        tool_calls = extract_tool_calls(pred_trace)
        tool_feedback = analyze_tool_usage(tool_calls, gold.result)
        
        score = tool_feedback["score"]
        feedback = f"工具使用分析: {tool_feedback['details']}"
        return {"score": score, "feedback": feedback}
    
    return calculate_task_completion_score(gold, pred)

高级技巧：最大化GEPA效能

1. 智能预算分配

# 根据任务复杂度动态调整预算
def adaptive_budget_planning(num_predictors, dataset_size, task_complexity):
    """
    自适应预算规划
    - 简单任务：使用light模式
    - 中等任务：使用medium模式  
    - 复杂任务：手动配置精细预算
    """
    if task_complexity == "simple":
        return {"auto": "light"}
    elif task_complexity == "medium":
        return {"auto": "medium"}
    else:
        # 复杂任务的精细配置
        estimated_calls = dataset_size * num_predictors * 10
        return {
            "max_metric_calls": estimated_calls,
            "reflection_minibatch_size": min(5, dataset_size // 10),
            "use_merge": True,
            "max_merge_invocations": 3
        }

config = adaptive_budget_planning(
    num_predictors=len(program.predictors()),
    dataset_size=len(trainset),
    task_complexity="complex"
)

gepa = dspy.GEPA(metric=your_metric, **config)

2. 反馈函数的最佳实践

def advanced_feedback_metric(gold, pred, trace, pred_name, pred_trace):
    """
    高级反馈函数设计原则：
    1. 分层评估：系统级 + 预测器级
    2. 语义丰富：提供具体的改进建议
    3. 上下文感知：利用轨迹信息
    """
    
    # 基础分数计算
    base_score = calculate_base_score(gold, pred)
    
    if pred_name and pred_trace:
        # 预测器级别的细粒度反馈
        predictor_analysis = analyze_predictor_performance(
            pred_name, pred_trace, gold, pred
        )
        
        return {
            "score": predictor_analysis["score"],
            "feedback": predictor_analysis["detailed_feedback"]
        }
    
    elif trace:
        # 利用完整轨迹的系统级反馈
        system_analysis = analyze_system_trace(trace, gold, pred)
        
        feedback_parts = [
            f"整体表现: {base_score}",
            f"关键发现: {system_analysis['key_insights']}",
            f"改进建议: {system_analysis['suggestions']}"
        ]
        
        return {
            "score": base_score,
            "feedback": "\n".join(feedback_parts)
        }
    
    else:
        # 简单的系统级评估
        return base_score

3. 多模态任务优化

from dspy.teleprompt.gepa.instruction_proposal import MultiModalInstructionProposer

class VisualQA(dspy.Module):
    def __init__(self):
        super().__init__()
        self.analyze_image = dspy.Predict("image, question -> description")
        self.answer_question = dspy.Predict("description, question -> answer")
    
    def forward(self, image, question):
        description = self.analyze_image(image=image, question=question)
        return self.answer_question(description=description.description, question=question)

# 多模态GEPA配置
gepa = dspy.GEPA(
    metric=multimodal_metric,
    auto="medium",
    reflection_lm=dspy.LM(model='gpt-4-vision-preview'),  # 视觉模型
    instruction_proposer=MultiModalInstructionProposer(),  # 多模态提案器
    track_stats=True
)

性能调优和问题排查

1. 性能瓶颈分析

# 启用详细日志分析性能
gepa = dspy.GEPA(
    metric=your_metric,
    log_dir="./gepa_logs",  # 保存详细日志
    track_stats=True,
    use_wandb=True  # 使用wandb进行可视化
)

# 分析优化过程
def analyze_optimization_performance(detailed_results):
    """分析GEPA优化效果"""
    print(f"总评估次数: {detailed_results.total_metric_calls}")
    print(f"最佳分数: {detailed_results.val_aggregate_scores[detailed_results.best_idx]}")
    print(f"发现的候选数量: {len(detailed_results.candidates)}")
    
    # 分析收敛曲线
    plot_convergence_curve(detailed_results.discovery_eval_counts, 
                          detailed_results.val_aggregate_scores)

2. 常见问题解决方案

#### 问题1：优化过程太慢 解决方案：

# 减少预算或使用更小的验证集
gepa = dspy.GEPA(
    metric=metric,
    auto="light",  # 使用轻量模式
    reflection_minibatch_size=2,  # 减小反思批量
    num_threads=2  # 限制并行线程
)

#### 问题2：优化效果不明显 解决方案：

# 增强反馈质量和反思模型
gepa = dspy.GEPA(
    metric=more_detailed_metric,  # 使用更详细的反馈
    reflection_lm=dspy.LM(model='gpt-4', temperature=1.0),  # 更强的反思模型
    candidate_selection_strategy="pareto",  # 使用帕累托选择
    use_merge=True  # 启用合并优化
)

#### 问题3：内存消耗过大 解决方案：

# 优化内存使用
gepa = dspy.GEPA(
    metric=metric,
    max_metric_calls=500,  # 限制总评估次数
    track_stats=False,  # 不跟踪详细统计（节省内存）
    reflection_minibatch_size=2  # 减小批量大小
)

生产环境部署

1. 检查点和恢复

# 设置检查点目录
gepa = dspy.GEPA(
    metric=production_metric,
    log_dir="./checkpoints/run_001",  # 检查点目录
    auto="heavy"
)

# 如果运行中断，可以从检查点恢复
# 使用相同的log_dir重新运行即可自动恢复

2. 监控和告警

import time
from datetime import datetime

def monitored_gepa_optimization(program, trainset, valset, config):
    """带监控的GEPA优化"""
    start_time = time.time()
    
    gepa = dspy.GEPA(**config)
    
    try:
        optimized_program = gepa.compile(
            student=program,
            trainset=trainset,
            valset=valset
        )
        
        duration = time.time() - start_time
        log_optimization_success(duration, optimized_program.detailed_results)
        
        return optimized_program
        
    except Exception as e:
        log_optimization_failure(e, duration=time.time()-start_time)
        raise

3. A/B测试框架

def ab_test_gepa_variants(base_program, trainset, valset, test_cases):
    """
    对比不同GEPA配置的效果
    """
    results = {}
    
    for config_name, gepa_config in test_cases.items():
        print(f"测试配置: {config_name}")
        
        gepa = dspy.GEPA(**gepa_config)
        optimized = gepa.compile(student=base_program, trainset=trainset, valset=valset)
        
        # 在独立测试集上评估
        test_score = evaluate_on_test_set(optimized, test_set)
        results[config_name] = {
            'program': optimized,
            'test_score': test_score,
            'optimization_stats': optimized.detailed_results
        }
    
    return results

总结：GEPA最佳实践清单

✅ 必做事项

• [ ] 使用强力的反思语言模型（如GPT-4）
• [ ] 设计详细的反馈函数，提供具体改进建议
• [ ] 设置合适的预算（从light开始，根据需要调整）
• [ ] 启用track_stats以获取优化详情
• [ ] 使用验证集避免过拟合

⚠️ 注意事项

• [ ] 避免反馈函数中的非确定性评分
• [ ] 监控内存使用，特别是大型数据集
• [ ] 测试不同组件选择策略的效果
• [ ] 验证优化后的程序在未见数据上的表现

🚀 进阶技巧

• [ ] 使用自定义指令提案器处理特殊输入类型
• [ ] 实现预测器级别的细粒度反馈
• [ ] 利用帕累托前沿进行多目标优化
• [ ] 设置检查点支持长时间运行的优化