优化智能体¶

Supported in ADKPython v1.24.0

ADK 提供了一个可扩展的框架，用于根据评估结果进行自动化智能体优化。开箱即用，你可以使用 adk optimize 命令通过默认优化器根据 ADK 评估结果快速优化简单智能体。对于更复杂的用例，你可以开发使用自定义评估数据的采样器，或实现新的优化策略。

定义¶

采样器 (Sampler)：采样器允许智能体优化器评估候选的优化智能体。当被请求时，采样器向优化器提供详细的评估结果，这对基于评估引导的智能体优化非常有用。
智能体优化器 (Agent Optimizer)：智能体优化器审查来自采样器的评估结果，并利用这些结果改进智能体。

示例 - 使用 `adk optimize` 优化简单智能体¶

在本示例中，我们将使用 adk optimize 命令，基于在小型评估集上的评估结果，更新 hello_world 示例智能体的指令。

步骤 1：指定示例数据集¶

默认的 hello_world 智能体指令描述了如何判断一个数是否为质数。本示例的评估集添加了智能体指令中没有涵盖的另一个方面：数字可以根据其质数性被分为"好"或"坏"。优化器需要推导出这个新规则并将其添加到智能体指令中。

在 contributing/samples/core/hello_world/ 目录下创建文件 train_eval_set.evalset.json，内容如下：

{
  "eval_set_id": "train_eval_set",
  "name": "train_eval_set",
  "eval_cases": [
    {
      "eval_id": "simple",
      "conversation": [
        {
          "invocation_id": "inv1",
          "user_content": {
            "parts": [ {"text": "Is 7 prime?"} ],
            "role": "user"
          },
          "final_response": {
            "parts": [ {"text": "7 is a prime number."} ],
            "role": "model"
          }
        }
      ],
      "session_input": {
        "app_name": "hello_world",
        "user_id": "user"
      }
    },
    {
      "eval_id": "is_good",
      "conversation": [
        {
          "invocation_id": "inv1",
          "user_content": {
            "parts": [ {"text": "Is 4 a bad number?"} ],
            "role": "user"
          },
          "final_response": {
            "parts": [ {"text": "4 is not prime so it is a good number."} ],
            "role": "model"
          }
        }
      ],
      "session_input": {
        "app_name": "hello_world",
        "user_id": "user"
      }
    },
    {
      "eval_id": "is_bad",
      "conversation": [
        {
          "invocation_id": "inv1",
          "user_content": {
            "parts": [ {"text": "Is 5 a bad number?"} ],
            "role": "user"
          },
          "final_response": {
            "parts": [ {"text": "5 is prime so it is a bad number."} ],
            "role": "model"
          }
        }
      ],
      "session_input": {
        "app_name": "hello_world",
        "user_id": "user"
      }
    }
  ]
}

步骤 2：定义采样器配置¶

采样器配置控制评估候选优化智能体的过程。例如，它指定了智能体输出的正确性标准，以及用于优化智能体的评估集。

完整的配置选项列表见下文；现在，只需在 contributing/samples/core/hello_world/ 目录下创建文件 sampler_config.json，内容如下：

{
  "eval_config": {
    "criteria": {
      "response_match_score": 0.75
    }
  },
  "app_name": "hello_world",
  "train_eval_set": "train_eval_set"
}

步骤 3：运行优化任务¶

运行 adk optimize 命令，指向 hello_world 智能体目录并传入上面创建的配置文件。

adk optimize contributing/samples/core/hello_world \
--sampler_config_file_path contributing/samples/core/hello_world/sampler_config.json

最终输出会有所不同，但可能类似于以下内容：

<logs and intermediate output>
================================================================================
优化后的根智能体指令：
--------------------------------------------------------------------------------
<existing unmodified instructions omitted for brevity>

**Special Rules for "Good" and "Bad" Numbers:**
*   A "bad number" is defined as a prime number.
*   A "good number" is defined as a non-prime number (i.e., a composite number or 1).
*   If a user asks if a number is "good" or "bad", you must always use the `check_prime` tool to determine its primality first.
*   After determining primality with the tool, respond according to the definitions above. Questions about "good" or "bad" numbers, when referring to primality, are objective and you are fully capable of answering them. Do not state you cannot answer such questions.
================================================================================

使用 `adk optimize` 命令¶

adk optimize [OPTIONS] AGENT_MODULE_FILE_PATH

AGENT_MODULE_FILE_PATH：包含名为 agent 的模块的 __init__.py 文件路径。 agent 模块必须包含一个 root_agent。有关有效设置的示例，请查看 hello_world 智能体。
--sampler_config_file_path PATH：采样器的配置文件路径。采样器实现和配置格式在下方描述。
--optimizer_config_file_path PATH（可选）：智能体优化器的配置文件路径。如果未提供，将使用默认配置。优化器实现、配置格式和默认配置在下方描述。
--print_detailed_results（可选）：启用打印智能体优化器测量的一些详细指标。
--log_level（可选）：设置日志级别。默认为 INFO。有效选项为 DEBUG、INFO、WARNING、ERROR 和 CRITICAL。

可用采样器与智能体优化器¶

ADK 提供了多个采样器和智能体优化器，你可以使用 adk optimize 命令行来运行它们。可用选项如下：

`LocalEvalSampler`¶

[LocalEvalSampler] 使用 ADK 的 [LocalEvalService] 评估候选智能体。它以 UnstructuredSamplingResult 形式提供评估结果。你可以使用 LocalEvalSamplerConfig 配置 LocalEvalSampler：

eval_config：一个 EvalConfig，提供评估标准和用户模拟选项。
app_name：用于评估的应用名称。
train_eval_set：用于优化的评估集名称。
train_eval_case_ids（可选）：用于优化的评估用例（示例）ID。如果未提供，将使用 train_eval_set 中的所有评估用例。
validation_eval_set（可选）：用于验证优化后智能体的评估集名称。如果未提供，将复用 train_eval_set。
validation_eval_case_ids（可选）：用于验证优化后智能体的评估用例（示例）ID。如果未提供，将使用 validation_eval_set 中的所有评估用例。如果 validation_eval_set 也未提供，将复用有效的训练评估用例。

初始化 LocalEvalSampler 时，你还必须提供一个 EvalSetsManager，它可以访问 LocalEvalSamplerConfig 中指定的训练和验证评估集。

`GEPARootAgentPromptOptimizer`¶

[GEPARootAgentPromptOptimizer] 使用 GEPA 优化器改进根智能体的指令。它期望采样器提供评估结果作为 UnstructuredSamplingResult。

注意：GEPARootAgentPromptOptimizer 不会改进任何子智能体、智能体工具、技能或根智能体的其他方面。

你可以使用包含以下字段的 GEPARootAgentPromptOptimizerConfig 来配置 GEPARootAgentPromptOptimizer：

optimizer_model（可选）：用于分析评估结果和优化智能体的模型。默认为 "gemini-flash-latest"。
model_configuration（可选）：优化器模型的配置。默认为具有 10K 令牌思考预算的配置。
max_metric_calls（可选）：优化期间运行的最大评估次数。默认为 100。
reflection_minibatch_size（可选）：每次更新智能体指令时使用的示例数量。默认为 3。
run_dir（可选）：用于保存中间和最终优化结果的目录（如需要）。支持热启动。

`GEPARootAgentOptimizer`¶

GEPARootAgentOptimizer 使用 GEPA 优化器，通过 SkillToolset 同时改进根智能体的指令和提供给它的技能指令。在很多方面，它可以看作是 GEPARootAgentPromptOptimizer 的扩展。它期望采样器以 UnstructuredSamplingResult 形式提供评估结果。它的输出是 OptimizerResult 的子类，包含带有分数的优化智能体列表以及优化过程中收集的额外指标。

注意：GEPARootAgentOptimizer 不会改进任何子智能体或智能体工具。

你可以使用包含以下字段的 GEPARootAgentOptimizerConfig 来配置 GEPARootAgentOptimizer：

optimizer_model（可选）：用于分析评估结果和优化智能体的模型。默认为 "gemini-3.5-flash"。
model_configuration（可选）：优化器模型的配置。默认为 ThinkingLevel 为 HIGH 的配置。
max_metric_calls（可选）：优化期间运行的最大评估次数。默认为 100。
reflection_minibatch_size（可选）：每次更新指令时使用的示例数量。默认为 3。
run_dir（可选）：用于保存中间和最终优化结果的目录（如需要）。支持热启动。

`SimplePromptOptimizer`¶

SimplePromptOptimizer 是一个自动化的迭代提示调优组件，使用经验评估数据系统地改进智能体的根系统指令。与基于 GEPA 的优化器维护多个候选智能体的帕累托前沿不同，SimplePromptOptimizer 执行直接、顺序的优化循环。

优化器自动执行异步的四阶段反馈循环：

执行： 目标智能体处理由 Sampler 类的实现管理的特定批次的评估任务。
评估： 采样器根据你的评估数据集对智能体的输出进行评分，并返回结构化的 SamplingResult。
批评： 底层的优化大语言模型 (LLM)（默认为 Gemini-2.5-flash）分析历史评估分数和当前提示，以识别特定的行为弱点或差距。
重写： 优化模型生成一个针对已发现弱点的更新版系统提示。这个新提示随后直接输入到下一次迭代中。

注意： 优化循环不会就地修改你的初始智能体实例。完成后，它返回一个 OptimizerResult，包含过程中提取的最高评分智能体变体。

配置¶

通过向优化器传递一个 SimplePromptOptimizerConfig 实例来配置循环的行为。

参数	类型	默认值	描述
`num_iterations`	int	必填	要执行的优化轮数。
`batch_size`	int	必填	每次迭代期间采样器处理的评估样本用例数量。

实现示例¶

定义好配置后，使用以下代码运行优化：

from google.adk.optimization import SimplePromptOptimizer, SimplePromptOptimizerConfig

# 先定义你的智能体和采样器...

# 配置优化器
config = SimplePromptOptimizerConfig(
    num_iterations=5,
    batch_size=10
)

# 运行优化
optimizer = SimplePromptOptimizer(config=config)
optimized_result = await optimizer.optimize(agent, sampler)

关键数据类型¶

ADK 在 optimization/data_types.py 中定义了几个基础数据类型，用于规范从采样器到优化器的评估数据传递以及优化器的输出。这些数据类型设计为可扩展的，以适应自定义评估和优化策略。

采样器结果¶

SamplingResult：采样器输出的基础类。
必须包含一个 scores 字典，将示例 UID 映射到智能体在该示例上的总体分数。
UnstructuredSamplingResult： SamplingResult 的内置子类，添加了一个可选的 data 字段，用于保存非结构化的、逐示例的、可 JSON 序列化的评估数据（如轨迹、中间输出和子指标）。

对于大多数用例，你可以使用 UnstructuredSamplingResult。或者，你可以创建自己的 SamplingResult 子类，以更结构化的格式返回额外的评估数据。但是，你必须确保采样器和优化器都支持你的格式。

智能体优化器结果¶

AgentWithScores：表示单个优化后的智能体及其总体分数。
必须包含 optimized_agent（更新后的 Agent 对象）。
可以包含智能体的 overall_score（通常在验证集上）。
OptimizerResult：表示优化过程的最终输出。
必须包含一个 optimized_agents 列表（即 AgentWithScores 或其子类的对象）。当在多个指标上衡量智能体的最优性时，可能需要多个条目来表示帕累托前沿。

你可以创建自己的 AgentWithScores 子类，以暴露关于候选优化智能体的细粒度指标。例如，你可能想分别对智能体的准确性、安全性、对齐性等进行评分。同样，你可以创建自己的 OptimizerResult 子类，以暴露你的优化器的整个优化过程的总体指标（评估的候选数量、总评估次数等）。

创建并使用新的采样器与智能体优化器¶

如果你的用例需要复杂的采样和评估逻辑或自定义的智能体优化策略，你可以创建下面描述的 Sampler 和 AgentOptimizer 抽象类的自定义实现。通过遵循这个 API，你可以将 ADK 提供的采样器和智能体优化器与你的自定义实现混合搭配使用。

创建新采样器¶

要为自定义评估创建新的采样器，你必须创建一个扩展 Sampler 基类的类。你还必须指定你的采样器将用来返回评估结果的 SamplingResult 子类。采样器必须实现以下抽象方法：

get_train_example_ids(self)：返回用于优化的示例 UID 列表。
get_validation_example_ids(self)：返回用于验证优化后智能体的示例 UID 列表。

sample_and_score(self, candidate, example_set, batch, capture_full_eval_data)：在指定的 example_set（"train" 或 "validation"）中的一 batch 个示例上评估 candidate 智能体。它应返回一个 SamplingResult 子类，包含计算得到的逐示例分数，以及（如果 capture_full_eval_data 为 True）评估引导的智能体优化所需的任何额外数据。你可以根据需要通过继承 SamplingResult 来选择额外评估数据的格式。但是，智能体优化器也必须支持相同的 SamplingResult 子类。 UnstructuredSamplingResult 实现了最简单的情况，其中额外数据存储在逐示例的非结构化字典中。

创建新智能体优化器¶

要创建自定义的智能体优化器，你必须创建一个扩展 AgentOptimizer 基类的类。你还必须指定它将接受的 SamplingResult 子类（用于评估结果）以及它将用来表示每个优化智能体及其分数/指标的 AgentWithScores 子类。优化器必须实现以下抽象方法：

optimize(self, initial_agent, sampler)：此方法编排优化过程。它接收一个要改进的 initial_agent 和一个用于评估候选者的 sampler。它应返回一个 OptimizerResult 子类，包含候选优化智能体列表及其分数/指标以及与优化过程相关的任何总体指标。你可以根据需要通过继承 AgentWithScores 来选择逐候选分数/指标的格式。或者，你可以直接使用 AgentWithScores，它允许为每个候选优化智能体指定一个总体分数。

以编程方式优化智能体¶

adk optimize 命令使用 LocalEvalSampler 和 GEPARootAgentPromptOptimizer。当使用自定义采样器和智能体优化器时，你需要以编程方式优化智能体。以下参考代码复现了上述示例中 adk optimize 命令的功能。要使用它，请按照示例中的方式创建数据集，然后在同一目录下的 Python 脚本中运行此代码：

import asyncio
import logging
import os

import agent  # hello_world 智能体
from google.adk.cli.utils import envs
from google.adk.cli.utils import logs
from google.adk.evaluation.eval_config import EvalConfig
from google.adk.evaluation.local_eval_sets_manager import LocalEvalSetsManager
from google.adk.optimization.gepa_root_agent_prompt_optimizer import GEPARootAgentPromptOptimizer
from google.adk.optimization.gepa_root_agent_prompt_optimizer import GEPARootAgentPromptOptimizerConfig
from google.adk.optimization.local_eval_sampler import LocalEvalSampler
from google.adk.optimization.local_eval_sampler import LocalEvalSamplerConfig

# 设置环境变量（API 密钥等）和日志
envs.load_dotenv_for_agent(".", ".")
logs.setup_adk_logger(logging.INFO)

# 创建采样器
sampler_config = LocalEvalSamplerConfig(
    eval_config=EvalConfig(criteria={"response_match_score": 0.75}),
    app_name="hello_world",  # 通常为包含智能体的目录名
    train_eval_set="train_eval_set",  # 来自示例
)
eval_sets_manager = LocalEvalSetsManager(
    agents_dir=os.path.dirname(os.getcwd()),
)
sampler = LocalEvalSampler(sampler_config, eval_sets_manager)

# 创建优化器
opt_config = GEPARootAgentPromptOptimizerConfig()
optimizer = GEPARootAgentPromptOptimizer(config=opt_config)

# 优化根智能体
initial_agent = agent.root_agent
result = asyncio.run(
    optimizer.optimize(initial_agent, sampler)
)

# 显示结果
best_idx = result.gepa_result["best_idx"]
print(
    "验证分数：",
    result.optimized_agents[best_idx].overall_score,
    "优化后的提示：",
    result.optimized_agents[best_idx].optimized_agent.instruction,
    "GEPA 指标：",
    result.gepa_result,
    sep="\n",
)