Agent Optimizer 与 Sampler:用 ADK 的 google.adk.optimization 自动优化 Agent 指令
2026/9/14 0:25:57 网站建设 项目流程

Agent Optimizer 与 Sampler:用 ADK 的 google.adk.optimization 自动优化 Agent 指令

【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python

google.adk.optimization是 ADK 内置的离线指令优化子系统:它自动改写 Agent 的 instruction,用评估集(eval set)为候选提示词打分,并保留分数更高的一方。你只需要提供一个知道"如何评估你的 Agent"的Sampler,ADK 负责驱动搜索的优化器(SimplePromptOptimizerGEPARootAgentOptimizer)。读完本文,你将掌握这套机制的接口契约、两种优化器的内部原理、配置参数的含义,以及如何基于已有评估集或自定义评分体系写出可落地的优化流水线。

引言:从"手改提示词"到"自动搜索提示词"

传统做法是:手动改一句 system prompt → 在几个用例上跑一遍 Agent → 凭感觉判断"看起来更好了"。google.adk.optimization把这条流程自动化:优化器提出一条新指令,评估器在一组固定的样例上给它打分,分数高者胜出。整套过程替换了人工试错,且不在请求时(request time)运行——优化是一次离线的批处理作业,产出是一个携带更好指令的Agent对象,你需要再把它复制回源码。

从源码结构看,这个包被清晰地切成两半,中间只通过一个接口对接:

  • AgentOptimizer负责搜索:决定尝试哪些提示词、以什么顺序、何时停止。包内自带两个实现——SimplePromptOptimizer(对单一提示词做爬山式迭代)和GEPARootAgentOptimizer(封装 GEPA 算法,返回一个 Pareto 前沿的多条提示词而非单一赢家)。基类定义在 agent_optimizer.py,核心只有一个抽象方法optimize(initial_agent, sampler)
  • Sampler负责评分:回答"存在哪些样例"以及"候选 Agent 在样例上表现如何"。LocalEvalSampler基于 ADK 自带的 eval sets 实现它,你也可以针对任何自己信任的评分体系实现Sampler

注意google/adk/optimization/__init__.py不导出任何符号,因此所有导入都必须走完整模块路径:

from google.adk.optimization.agent_optimizer import AgentOptimizer from google.adk.optimization.data_types import AgentWithScores from google.adk.optimization.data_types import OptimizerResult from google.adk.optimization.data_types import UnstructuredSamplingResult from google.adk.optimization.sampler import Sampler from google.adk.optimization.simple_prompt_optimizer import SimplePromptOptimizer from google.adk.optimization.simple_prompt_optimizer import SimplePromptOptimizerConfig

快速上手:实现 Sampler,交给优化器

先实现一个Sampler,覆盖你已有的评分逻辑,然后交给优化器。下面的示例 Sampler 是"假装打分"(真实场景里它应该真正运行 Agent):

from google.adk.agents import Agent from google.adk.optimization.sampler import Sampler from google.adk.optimization.data_types import UnstructuredSamplingResult class MySampler(Sampler[UnstructuredSamplingResult]): def get_train_example_ids(self) -> list[str]: return ["case-1", "case-2", "case-3"] def get_validation_example_ids(self) -> list[str]: return ["holdout-1", "holdout-2"] async def sample_and_score( self, candidate, example_set=Sampler.VALIDATION_SET, batch=None, capture_full_eval_data=False, ) -> UnstructuredSamplingResult: if batch is None: batch = ( self.get_train_example_ids() if example_set == Sampler.TRAIN_SET else self.get_validation_example_ids() ) scores = {example_id: await my_score(candidate, example_id) for example_id in batch} return UnstructuredSamplingResult(scores=scores) agent = Agent( name="support_agent", instruction="Help the user with their order.", tools=[check_order_status, issue_refund], ) optimizer = SimplePromptOptimizer( SimplePromptOptimizerConfig(num_iterations=5, batch_size=3) ) result = await optimizer.optimize(agent, MySampler()) best = result.optimized_agents[0] print(best.overall_score) print(best.optimized_agent.instruction)

三个关键约定:

  1. optimize绝不修改你传入的 Agent。每一个候选都是通过clone(update={"instruction": ...})构建的(见 simple_prompt_optimizer.py 的_run_optimization_iterations),运行结束时你构造的原始对象仍然带着最初的指令。
  2. 分数是浮点数,越高越好。这是唯一的契约——取值范围由你决定,优化器只会在你自己的两个数值之间做比较。
  3. example_set只有两个取值Sampler.TRAIN_SETSampler.VALIDATION_SET是类常量,分别持有字符串"train""validation",定义见 sampler.py。

工作原理:优化器与 Sampler 之间的契约

优化器只调用 Sampler,不碰其他东西:先要一次样例 ID 列表,然后反复调用sample_and_score(candidate, example_set, batch)并读取result.scores——一个以样例 ID 为键的dict

当优化器需要知道候选为什么失败时,会传入capture_full_eval_data=True,此时 Sampler 需要填充UnstructuredSamplingResult.data。这是第二个 dict,同样以样例 ID 为键,存放任何有助于模型反思失败原因的、可 JSON 序列化的材料:输入、响应、工具调用、指标判定等。从 data_types.py 可以看到,UnstructuredSamplingResultSamplingResult基础上增加了可选的data字段。

两种优化器对这个接口的索取程度不同:SimplePromptOptimizer只要分数GEPARootAgentOptimizer依赖data,并会仔细反思你放进去的内容。

SimplePromptOptimizer 做了什么

其核心是一个"无记忆的爬山"循环(见 simple_prompt_optimizer.py 的_run_optimization_iterations):

  1. 在一批随机采样的训练样例上给初始 Agent 打分,作为待超越的基线分数;
  2. 每一轮迭代,让优化器模型仅依据"当前最佳指令 + 它的分数"重写指令,克隆出候选 Agent,在新的一批随机训练样例上打分,只有分数更高才保留;
  3. 最后一轮结束后,把幸存者放到完整验证集上打一次分,这个数字就是overall_score

因此,num_iterations=n的一次运行会产生n + 2sample_and_score调用:1 次基线 + n 次候选 + 1 次最终验证。选择完全发生在训练分数上;验证只在最后测一次,永不参与选择

这个设计带来两个你需要理解的后果:

  • 每次比较用的都是不同的、大小为batch_size的随机批次,所以一个候选可能因为批次噪声而非真实实力获胜;
  • 优化器模型只看到上一条提示词和一个数字,永远看不到失败的案例,它是在猜测该改进什么。

另外,SimplePromptOptimizer__init__里就解析优化器模型(LLMRegistry().new_llm(...)),而不是在optimize里——所以模型名写错会在构造时立刻失败,而不是跑了一半才报错。

GEPARootAgentOptimizer 做了什么

GEPA 是"反思失败"而不是"盲猜":它运行候选 Agent,把低分运行捕获的轨迹喂给反思模型,用反思模型的诊断来提出下一条提示词。它维护一个Pareto 前沿而非单一冠军,所以optimized_agents会返回多个各有所长的 Agent,同时GEPARootAgentOptimizerResult.gepa_result携带原始算法输出的 dict(见 gepa_root_agent_optimizer.py)。

它比简单优化器多优化两类东西:

  1. 根 Agent 的指令
  2. Agent 工具中所有通过SkillToolset可达的Skillinstructions文本

每一类都成为独立进化的组件(component)。从源码看,seed_candidate会把每个 skill 的instructionsskill_instructions:{skill_name}为键、把根指令以agent_prompt为键放入候选字典(gepa_root_agent_optimizer.py),反思模型据此分别生成新的根指令与各 skill 指令(_AGENT_PROMPT_UPDATOR_INST_TEMPLATE_SKILL_INST_UPDATOR_INST_TEMPLATE)。

需要注意:子 Agent 的提示词不会被优化。如果initial_agent.sub_agents非空,优化器会记录一条 warning 并只对根 Agent 进行。

此外,模块列表里还有GEPARootAgentPromptOptimizer——它是同类实现的更早、更窄的版本:只进化根指令,完全忽略 skill。它的配置与 GEPA 版几乎一致,仅有两处默认值差异(gemini-2.5-flash,以及用 thinking预算而非 thinking级别)。优先使用GEPARootAgentOptimizer,它包含了前者的全部能力。

两个需要知道的工程细节:

  • GEPARootAgentOptimizer带有@experimental装饰器,构造时会发出UserWarning,内容形如[EXPERIMENTAL] GEPARootAgentOptimizer: ...
  • 算法本体位于第三方gepa包中,ADK 在optimize内部惰性导入。未安装时会抛出ImportError: Eval module is not installed, please install via pip install "google-adk[eval]"
  • 它还要求initial_agent.instruction纯字符串。指令是 callable provider 的 Agent 会在任何评估运行之前抛出ValueError,因为请求作用域的 provider 无法在没有真实调用上下文的情况下被解析(见 _gepa_utils.py 的require_static_instruction)。该检查位于gepa导入之后,所以请先安装 extra,否则你看到的会是ImportError而不是这个ValueError

LocalEvalSampler:连接优化器与 ADK 评估体系的现成桥梁

LocalEvalSampler是优化器到 ADK 评估服务 的即用桥梁(源码见 local_eval_sampler.py)。给它一个EvalConfig、一个 app 名和 eval set 的 ID,它就会对每个你点名的 eval case,把候选 Agent 送入LocalEvalService走一遍"先推理、后指标"的完整评估。

一个关键行为:它按状态而非指标值打分。通过的 case 给优化器1.0,其他任何结果都给0.0(local_eval_sampler.py)。所以如果某个 case 以 0.94 的分数卡在 0.95 的阈值下、另一个 case 直接报错,优化器无法区分这两者。

另外,导入local_eval_sampler会连带拉入整个评估栈(依赖evalextra)。未安装时,导入失败的表现是ModuleNotFoundError: No module named 'vertexai',而不是 ADK 自己的安装提示——这个报错信息容易误导,需提前知晓。

配置项详解

每个优化器接受各自的配置对象,两者界定一次运行的方式不同:简单优化器按迭代次数计数,GEPA 按计分评估次数计数。

SimplePromptOptimizerConfig

四个设置分别覆盖:由哪个模型重写提示词、该模型如何生成、尝试多少次重写、每次重写由多少个样例评判(见 simple_prompt_optimizer.py)。

选项类型默认值说明
optimizer_modelstr"gemini-2.5-flash"负责重写提示词的模型。不是被优化 Agent 自身的模型。
model_configurationGenerateContentConfig开启 thinking,预算 10240优化器模型的生成配置。
num_iterationsint10尝试的候选提示词数量。
batch_sizeint5每个候选被评判的训练样例数。
  • optimizer_model指定做重写的模型,与被优化 Agent 跑在什么模型上完全独立。默认配置开启 thinking、预算 10240 token,因为重写本身是一项推理任务。
  • batch_size是成本与信号的权衡:每个候选要花batch_size次 Agent 运行,因此一次运行大约需要(num_iterations + 1) * batch_size次训练运行,外加一次完整的验证遍历。批次太小会让比较充满噪声——这种噪声正是"运行结束得到的提示词其实并没有更好"的主要原因。
  • 注意:当batch_size超过训练样例总数时,optimize会把它钳制到训练样例数,而且是通过直接写入你传入的配置对象完成的(simple_prompt_optimizer.py)。运行后读config.batch_size,你可能会发现它和你设置的值不一样。

GEPARootAgentOptimizerConfig

GEPA 用总评估预算而非迭代次数界定运行,所以这些设置关心的是"你愿意花多少"以及"反思模型每次能看到多少"(见 gepa_root_agent_optimizer.py)。

选项类型默认值说明
optimizer_modelstr"gemini-3.5-flash"用于反思和提出新提示词的模型。
model_configurationGenerateContentConfigthinking 级别HIGH反思模型的生成配置。
max_metric_callsint100整次运行的总评估预算。
reflection_minibatch_sizeint3每一步展示给反思模型的样例数。
run_dirstr \| NoneNone检查点目录。设置后运行可断点续跑。
  • max_metric_calls是决定一次运行成本的唯一旋钮:它限制整个搜索过程中计分评估的次数,调高它会线性地买到更多探索。GEPA 示例建议从默认的 100 起步,认真跑一次可以提到 500 以上。
  • run_dir值得在任何"长得足够被中断"的运行上设置:设置后 GEPA 会写入中间状态并从最后一个检查点恢复;不设置的话,中断的运行只能从头再来。

进阶应用

优化工作几乎全部发生在 Sampler 里,所以真正重要的问题是:你的评分从哪来、报告的数字能不能被信任。

针对已有 eval set 优化

你已经拥有的 eval cases 和指标就是现成的评分代码,不必写第二遍。直接用LocalEvalSampler替代自研Sampler,分别指向一个训练集和一个验证集:

sampler = LocalEvalSampler( LocalEvalSamplerConfig( eval_config=eval_config, app_name="my_app", train_eval_set="train_set", validation_eval_set="holdout_set", ), eval_sets_manager=eval_sets_manager, ) result = await GEPARootAgentOptimizer( GEPARootAgentOptimizerConfig(max_metric_calls=200, run_dir="/tmp/gepa_run") ).optimize(agent, sampler)

这里的EvalConfig与 eval 配置文件产生的对象是同一个(相关配置见 EvalConfig 指南),所以你已经在配置文件里调好的指标直接决定这里的"更好"是什么含义。

一个容易踩的坑:省略validation_eval_set会让验证复用训练集,此时报告的分数是优化器已经拟合过的分数,对新案例没有任何信息量。

保持训练集与验证集真正分离

在相同样例上做选择又报告分数的优化器看起来总是成功——因为它报告的分数正是它调优时用的分数。务必让get_train_example_idsget_validation_example_ids返回互不相交的 ID 列表。

GEPARootAgentOptimizer会替你检查:当两个集合有交集时记录 warning(gepa_root_agent_optimizer.py),因为一个共享 UID 在两个集合中如果代表不同样例,会静默地造成别名问题。其他优化器不做这项检查,在这些优化器上,分离只能靠你自己保证。

基于自己的评分体系写 Sampler

人工评分、生产指标或基于 rubric 的评判器都不一定适配 ADK 的 eval set——它们也不需要适配。实现Sampler的三个方法,把数值放进scores即可。如果打算用 GEPA,还需要在收到capture_full_eval_data=True时用反思模型应读取的轨迹材料填充data——因为 GEPA 在轨迹缺失时会抛异常而不是继续执行。除此之外的一切(批处理、缓存、并发)都由你决定,因为优化器只是 await 那个协程。

相关示例:GEPA 集成样例

仓库中的 GEPA 集成示例 在 Tau-bench 零售基准上用 GEPA 优化 ADK Agent 的提示词,包括在没有奖励信号时使用的 LLM 评判器。它早于google.adk.optimization出现,是直接对着gepa包手写自己的 GEPA adapter 的——把它当作"一个Sampler封装的评估工作长什么样"的参考,而不是 API 用法示例。该样例还给出了超参数选择经验:max_metric_calls从 100 起步可上调到 500+;reflection_minibatch_size(即其--train_batch_size)论文默认 3;num_eval_trials建议 4–8 以获得更稳定的指标。

已知限制

  • 没有任何内容被写回。optimize返回的是内存中的Agent。把获胜指令复制进你的源码、并针对它重新跑你的常规测试,这一步是手动的。
  • SimplePromptOptimizer内置了一段客服提示词。它的优化器模板写着 "the agent needs to solve customer support tasks by using tools correctly and following policies",而且没有提供替换选项。对任何其他领域,这句话都会在每一轮迭代中成为发给重写模型的误导信息。
  • SimplePromptOptimizer只返回一个 Agent,尽管OptimizerResult.optimized_agents被文档描述为 Pareto 前沿的列表。只有 GEPA 会往里面放多个。
  • 选择只基于训练分数。SimplePromptOptimizer运行会报告一个它从未针对其优化的验证分数,甚至可能报告比初始 Agent更差的验证分数,同时仍然返回重写后的提示词。
  • 成本真实存在且不受包约束。每一次评分都是一次针对真实模型的完整 Agent 运行。max_metric_calls能约束 GEPA;自定义 Sampler 则没有任何约束。
  • LocalEvalSampler丢弃了指标分辨率。只有通过与失败,所以优化器看不到一个从 0.4 提升到 0.9 但没有越过阈值的候选。
  • 仓库中的示例没有使用这些类。上面链接的 GEPA 示例是直接调用第三方gepa包。
  • 导入data_types会发出 Pydantic 弃用警告。其中的模型给Field传了一个 Pydantic v2 已不认识的required=True关键字。字段仍然是必需的,这个警告只是噪声。

相关指南

  • EvalConfig 与 eval 配置文件:讲解决定LocalEvalSampler判定"通过"的指标。
  • Evaluator 指南:讲解编写 Sampler 所针对评分的指标。
  • SkillToolset:讲解GEPARootAgentOptimizer与根提示词一并进化的 skill 指令。
  • 评估服务指南:LocalEvalSampler底层调用的评估机制。

【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询