写给正在落地 AI 产品的工程师。一些代码直接可改造复用;另一些,是我踩坑后的经验之谈。
DSPy是一个2024年5月修猫曾反复推荐给大家的一个AI程序,您有兴趣的话可以翻一下之前关于DSPy的文章。毋庸置疑,DSPy.GEPA接下来会成为很多AI应用的Prompt优化器,当然最有可能成为众多AI应用程序中的一个优化模块。在国庆期间OpenAI的新产品Agentkit中就有用到DSPy.GEPA。读这篇文章不过是早了解一下,因为您迟是需要了解一下的。
关于与代理框架的协同:十一期间,OpenAI 的 AgentKit 发布,明确支持以工程化方式组织代理与工具链,开发者表示已经将 GEPA 这类提示优化器接入到 AgentKit 的开发流程中(作为反思/调优阶段)。如果您使用的类似 AgentKit 之类的流程类的无代码开发工具未内置 GEPA,也可以通过适配器把 Agent 的决策逻辑包装成一个 DSPy 模块,用 GEPA 优化其提示与子策略,二者职责清晰、边界分明。
我会在文中展示 dspy.GEPA 的两种基础用法:
不依赖 AgentKit 内置支持,也能把 GEPA 接入到代理开发中。核心是把“代理的策略/指令模板”封装为 DSPy 模块:
# 把 Agent 的一步决策包装成 DSPy.Module,便于 GEPA 优化其指令
classAgentStep(dspy.Signature):
context: str = dspy.InputField()
reply: str = dspy.OutputField()
classAgentModule(dspy.Module):
def__init__(self):
# 用 CoT 包一层,使 GEPA 能在指令层面演化
self.policy = dspy.ChainOfThought(AgentStep)
defforward(self, context: str):
# 在这里桥接到 AgentKit:把 policy 产出的回复/意图,交给 AgentKit runtime 执行
# action = agentkit.act(context, hint=self.policy_instructions) # 伪代码
out = self.policy(context=context)
return dspy.Prediction(reply=out.reply)
# metric 以任务成功率、工具调用正确率或安全规则为准则,返回分数+文字反馈
reflector = dspy.LM(model='gpt-5', temperature=1.0, max_tokens=32000)
gepa = dspy.GEPA(metric=metric, reflection_lm=reflector, auto='medium')
optimized_agent = gepa.compile(AgentModule(), trainset=train_tasks, valset=val_tasks)
实际工程里,metric 可以基于端到端任务成功(工具调用、输出格式、策略合规)产生可解释的反馈文本;AgentKit 继续负责工具与环境调用,GEPA 负责“让策略指令更聪明”。
这和“AI 控制研究”的目标意外地一致:我们需要可控、可迁移的策略,而不是一次性调好的咒语。LessWrong 上那篇关于“提示优化推动 AI 控制研究”的文章说得很实在——人工手改提示太慢、可迁移性差;把反馈变成优化面,研究速度和可重复性会不一样。
模式 A(给项目加 UI,做 Prompt 优化与对比):
import gradio as gr
# 假设已定义 program/train_set/val_set 与上文 metric/reflector
gepa = dspy.GEPA(metric=metric, reflection_lm=reflector, auto='light', track_stats=True, log_dir='./ui_runs')
defrun_once(sample_size:int=50):
# 选一个子集做快速迭代(UI 模式强调“快试快看”)
subset = train_set[:sample_size]
new_prog = gepa.compile(program, trainset=subset, valset=val_set)
# 返回关键信息用于页面展示
scores = new_prog.detailed_results.val_aggregate_scores ifhasattr(new_prog, 'detailed_results') else []
returnf"candidates={len(scores)}; best={max(scores) if scores else 'n/a'}"
demo = gr.Interface(fn=run_once, inputs=gr.Number(value=50, label='train subset size'), outputs="text")
demo.launch()
模式 B(程序内联,用 GEPA 写 AI 程序):
小提示:如果您要在 AgentKit 的代理里用 GEPA,修猫建议把“策略/指令模板”封装为一个 DSPy.Module,然后通过桥接函数把代理的调用上下文映射为该 Module 的输入;GEPA 会在这个模块上反思与演化提示,而 AgentKit 维持工具/环境交互。
适用场景(Works Well):
不太适用(Makes Less Sense):
GEPA 的“甜蜜区间”(Sweet Spots):
何时考虑替代方案:
实操建议:有好的语义指标就用起来(哪怕是 LLM‑as‑a‑judge),并把失败原因写清楚;没有就先做最小可用的“反馈器”(validator/test/logs→文字),再上 GEPA。
任务:输入是一封有关设施支持的邮件,要抽出三类信息——
先给一个基线 DSPy 程序(3 个子模块,各自 Chain‑of‑Thought),随后我们用 GEPA 自动演化提示,减少您手改的时间。
# 1) 配置推理模型(示例用小模型,可替换为您的供应商)
import dspy
lm = dspy.LM(model="openai/gpt-4.1-nano", temperature=1)
dspy.configure(lm=lm)
# 2) 定义签名与模块
from typing importList, Literal
classFacilitySupportAnalyzerUrgency(dspy.Signature):
"""Read the provided message and determine the urgency."""
message: str = dspy.InputField()
urgency: Literal['low','medium','high'] = dspy.OutputField()
classFacilitySupportAnalyzerSentiment(dspy.Signature):
"""Read the provided message and determine the sentiment."""
message: str = dspy.InputField()
sentiment: Literal['positive','neutral','negative','mixed','unclear'] = dspy.OutputField()
classFacilitySupportAnalyzerCategories(dspy.Signature):
"""Read the provided message and determine all applicable categories."""
message: str = dspy.InputField()
categories: List[Literal[
"emergency_repair_services",
"routine_maintenance_requests",
"quality_and_safety_concerns",
"specialized_cleaning_services",
"general_inquiries",
"sustainability_and_environmental_practices",
"training_and_support_requests",
"cleaning_services_scheduling",
"customer_feedback_and_complaints",
"facility_management_issues",
]] = dspy.OutputField()
classFacilitySupportAnalyzerMM(dspy.Module):
def__init__(self):
self.urgency_module = dspy.ChainOfThought(FacilitySupportAnalyzerUrgency)
self.sentiment_module = dspy.ChainOfThought(FacilitySupportAnalyzerSentiment)
self.categories_module = dspy.ChainOfThought(FacilitySupportAnalyzerCategories)
defforward(self, message: str):
u = self.urgency_module(message=message)
s = self.sentiment_module(message=message)
c = self.categories_module(message=message)
return dspy.Prediction(urgency=u.urgency, sentiment=s.sentiment, categories=c.categories)
program = FacilitySupportAnalyzerMM()
“分数 + 文本理由”让反思 LLM 抓到症结。比如:
下面是一份可直接复用的指标函数。它会为每个子模块给出定向反馈(若 GEPA 传入 pred_name),否则给出程序级反馈;并把三项任务合成一个 0–1 的平均分。
import json
from typing import Optional
ALLOWED_CATEGORIES = {
"emergency_repair_services",
"routine_maintenance_requests",
"quality_and_safety_concerns",
"specialized_cleaning_services",
"general_inquiries",
"sustainability_and_environmental_practices",
"training_and_support_requests",
"cleaning_services_scheduling",
"customer_feedback_and_complaints",
"facility_management_issues",
}
SENTIMENTS = {"positive","neutral","negative","mixed","unclear"}
URGENCIES = {"low","medium","high"}
def_safe_list(x):
return x ifisinstance(x, list) else []
defmetric(gold: dspy.Example,
pred: dspy.Prediction,
trace: Optional["DSPyTrace"] = None,
pred_name: Optional[str] = None,
pred_trace: Optional["DSPyTrace"] = None):
# gold.answer 约定为 JSON 字符串:{"urgency":..., "sentiment":..., "categories": {...或列表...}}
try:
gold_obj = json.loads(gold["answer"]) ifisinstance(gold["answer"], str) else gold["answer"]
except Exception:
# 数据脏的情况,给出温和的降级
gold_obj = {"urgency": None, "sentiment": None, "categories": []}
# 归一化 gold categories(既兼容字典,也兼容列表)
g_c = gold_obj.get("categories", [])
ifisinstance(g_c, dict):
gold_categories = sorted([k for k, v in g_c.items() if v])
else:
gold_categories = sorted([c for c in _safe_list(g_c) if c in ALLOWED_CATEGORIES])
# 取 pred 输出
p_u = getattr(pred, "urgency", None)
p_s = getattr(pred, "sentiment", None)
p_c = sorted([c for c in _safe_list(getattr(pred, "categories", [])) if c in ALLOWED_CATEGORIES])
# 逐项打分
s_u = 1.0if p_u in URGENCIES and p_u == gold_obj.get("urgency") else0.0
s_s = 1.0if p_s in SENTIMENTS and p_s == gold_obj.get("sentiment") else0.0
# 类别:F1 更稳妥,这里示例按严格一致;也可以用 Jaccard/F1
s_c = 1.0if p_c == sorted(gold_categories) else0.0
score = (s_u + s_s + s_c) / 3.0
# 文本反馈(模块定向 or 程序级)
hints = []
# 紧急程度
if s_u < 1:
hints.append(
"紧急程度:避免把礼貌性的“尽快”“感谢”当成高优先级;只有出现安全隐患、业务中断、近期限时,才提升到 high/medium。")
# 情感
if s_s < 1:
hints.append(
"情感:普通礼貌与求助语气通常是 neutral;只有出现明确赞扬/抱怨/挫败感时才判 positive/negative/mixed。")
# 类别
if s_c < 1:
if"routine_maintenance_requests"in gold_categories and"cleaning_services_scheduling"in p_c:
hints.append("类别:把 HVAC 等维护预约当成清洁安排了,应改为 routine_maintenance_requests。")
else:
hints.append("类别:仅选择文本明确支持的条目;不要外推隐含意图,也不要使用未在白名单中的标签。")
# 拼装反馈
feedback = "\n".join(hints) if hints else"各子任务判断合理。"
# 如果 GEPA 正在优化具体模块,尽量给定向建议
if pred_name:
feedback = f"[{pred_name}] 优化建议:\n" + feedback
return {"score": float(score), "feedback": feedback}
# 反思模型建议用强一些的(示例名仅占位,替换成您的可用模型)
reflector = dspy.LM(model='gpt-5', temperature=1.0, max_tokens=32000)
gepa = dspy.GEPA(
metric=metric,
reflection_lm=reflector,
auto='medium', # 或改为 max_metric_calls / max_full_evals
reflection_minibatch_size=3,
candidate_selection_strategy='pareto',
use_merge=True,
track_stats=True,
log_dir='./gepa_logs'
)
optimized = gepa.compile(program, trainset=train_set, valset=val_set)
GEPA 在每次迭代会:
您在日志里会看到类似“为 urgency_module.predict 生成了新指令”的片段;这些文本就是被优化后的“提示”,而不是手改一堆 prompt 片段。
import mlflow
mlflow.set_tracking_uri("http://localhost:5000")
mlflow.set_experiment("DSPy")
# 自动记录 GEPA 编译、评估与 Trace
mlflow.dspy.autolog(
log_compiles=True,
log_evals=True,
log_traces=True
)
我建议在团队里固定开启 autologging。谁在什么数据上优化过、提示怎样演化、哪一次回退值得保留,MLflow 都能直接回答。出了回归,也能快速 bisect。
import dspy, json, random
from typing importList, Literal
# 1) 配置推理与反思模型
base_lm = dspy.LM(model="openai/gpt-4.1-nano", temperature=1)
reflector = dspy.LM(model='gpt-5', temperature=1.0, max_tokens=32000)
dspy.configure(lm=base_lm)
# 2) 定义任务模块
classUrg(dspy.Signature):
message: str = dspy.InputField()
urgency: Literal['low','medium','high'] = dspy.OutputField()
classSent(dspy.Signature):
message: str = dspy.InputField()
sentiment: Literal['positive','neutral','negative','mixed','unclear'] = dspy.OutputField()
classCats(dspy.Signature):
message: str = dspy.InputField()
categories: List[str] = dspy.OutputField()
classApp(dspy.Module):
def__init__(self):
self.u = dspy.ChainOfThought(Urg)
self.s = dspy.ChainOfThought(Sent)
self.c = dspy.ChainOfThought(Cats)
defforward(self, message: str):
u, s, c = self.u(message=message), self.s(message=message), self.c(message=message)
return dspy.Prediction(urgency=u.urgency, sentiment=s.sentiment, categories=c.categories)
program = App()
# 3) 构造一个玩具数据集(真实项目用您自己的数据与切分)
data = [
dspy.Example({
"message": "Please adjust cleaning schedule to Tue/Fri; no urgency.",
"answer": json.dumps({
"urgency": "low",
"sentiment": "neutral",
"categories": ["cleaning_services_scheduling"]
})
}).with_inputs("message"),
]
train_set = data
val_set = data
# 4) 指标函数(可替换为上文完整版 metric)
ALLOWED_CATEGORIES = {"cleaning_services_scheduling"}
SENTIMENTS = {"positive","neutral","negative","mixed","unclear"}
URGENCIES = {"low","medium","high"}
defmetric(gold, pred, trace=None, pred_name=None, pred_trace=None):
g = json.loads(gold["answer"]) ifisinstance(gold["answer"], str) else gold["answer"]
p_u = getattr(pred, "urgency", None)
p_s = getattr(pred, "sentiment", None)
p_c = [c for c in (getattr(pred, "categories", []) or []) if c in ALLOWED_CATEGORIES]
s_u = 1.0if p_u == g.get("urgency") else0.0
s_s = 1.0if p_s == g.get("sentiment") else0.0
s_c = 1.0ifsorted(p_c) == sorted(g.get("categories", [])) else0.0
score = (s_u + s_s + s_c) / 3.0
fb = []
if s_u < 1: fb.append("紧急程度判据使用不当;仅在安全/中断/近期限时时提升等级。")
if s_s < 1: fb.append("情感标签与语气不符;礼貌语气通常为 neutral。")
if s_c < 1: fb.append("类别选择不严;只选文本明确支持的类别。")
txt = (f"[{pred_name}]\n"if pred_name else"") + "\n".join(fb or ["判断合理。"])
return {"score": float(score), "feedback": txt}
# 5) GEPA 编译
gepa = dspy.GEPA(metric=metric, reflection_lm=reflector, auto='light', track_stats=True)
optimized = gepa.compile(program, trainset=train_set, valset=val_set)
# 6) 使用优化后的程序
out = optimized(message="Need to adjust cleaning to Tue/Fri; everything is fine.")
print(out)
项目作者把这条路打通了,剩下的就是把它放进您的流水线里。
引用链接:
[1]: https://dspy.ai/tutorials/gepa_facilitysupportanalyzer/
文章来自于“Al修猫Prompt”,作者“Al修猫Prompt”。
【开源免费】AutoGPT是一个允许用户创建和运行智能体的(AI Agents)项目。用户创建的智能体能够自动执行各种任务,从而让AI有步骤的去解决实际问题。
项目地址:https://github.com/Significant-Gravitas/AutoGPT
【开源免费】MetaGPT是一个“软件开发公司”的智能体项目,只需要输入一句话的老板需求,MetaGPT即可输出用户故事 / 竞品分析 / 需求 / 数据结构 / APIs / 文件等软件开发的相关内容。MetaGPT内置了各种AI角色,包括产品经理 / 架构师 / 项目经理 / 工程师,MetaGPT提供了一个精心调配的软件公司研发全过程的SOP。
项目地址:https://github.com/geekan/MetaGPT/blob/main/docs/README_CN.md
【开源免费】FASTGPT是基于LLM的知识库开源项目,提供开箱即用的数据处理、模型调用等能力。整体功能和“Dify”“RAGFlow”项目类似。很多接入微信,飞书的AI项目都基于该项目二次开发。
项目地址:https://github.com/labring/FastGPT
【开源免费】XTuner 是一个高效、灵活、全能的轻量化大模型微调工具库。它帮助开发者提供一个简单易用的平台,可以对大语言模型(LLM)和多模态图文模型(VLM)进行预训练和轻量级微调。XTuner 支持多种微调算法,如 QLoRA、LoRA 和全量参数微调。
项目地址:https://github.com/InternLM/xtuner
【开源免费】LangGPT 是一个通过结构化和模板化的方法,编写高质量的AI提示词的开源项目。它可以让任何非专业的用户轻松创建高水平的提示词,进而高质量的帮助用户通过AI解决问题。
项目地址:https://github.com/langgptai/LangGPT/blob/main/README_zh.md
在线使用:https://kimi.moonshot.cn/kimiplus/conpg00t7lagbbsfqkq0