APO¶

快捷方式

您可以使用快捷方式 agl.APO(...) 创建 APO 实例。

import agentlightning as agl

agl.APO(...)

安装¶

pip install agentlightning[apo]

当前实现范围¶

APO 当前的范围是优化单个提示模板。尚未支持优化多个提示模板。

但是，提示模板中变量占位符的数量没有限制（范围可以从零到很多）。在优化过程中可能会创建无效的提示。确保提示模板对代理的任务有效是代理开发人员的责任。

初始提示¶

APO 期望在 initial_resources 字典中提供初始提示。这可以通过两种方法完成

传递给 Trainer 构造函数

trainer = agl.Trainer(
    algorithm=agl.APO(...),
    initial_resources={"main_prompt": agl.PromptTemplate(template="You are a helpful assistant.", engine="f-string")},
)

传递给 [APO][agentlightning.algorithm.apo.APO].set_initial_resources() 方法

algo = agl.APO(...)
algo.set_initial_resources(
    {"this_is_also_valid_key": agl.PromptTemplate(template="You are a helpful assistant.", engine="f-string")}
)

资源键可以是任意的，用于在基于类的实现中，当您有多个资源时，标识提示模板。当键更改时，代理开发人员需要在 rollout 方法中更新键。

使用APO的教程¶

使用APO训练第一个代理 - 一步一步地指导您使用APO训练您的第一个代理。

参考¶

`agentlightning.algorithm.apo` ¶

`APO` ¶

基类：Algorithm, Generic[T_task]

使用文本梯度和波束搜索的自动提示优化 (APO) 算法。

APO 是一种迭代式提示优化算法，它使用 LLM 生成的文本梯度来通过束搜索过程改进提示。它在 rollout 上评估提示，根据结果计算评论，并应用编辑以生成改进的提示。

该算法以轮次为单位运行，每个轮次

从当前束中采样父提示
通过计算文本梯度和应用编辑来生成新提示
在验证集上评估所有候选者
为下一轮选择前 k 个提示

基于以下思想

`init(async_openai_client, *, gradient_model='gpt-5-mini', apply_edit_model='gpt-4.1-mini', diversity_temperature=1.0, gradient_batch_size=4, val_batch_size=16, beam_width=4, branch_factor=4, beam_rounds=3, rollout_batch_timeout=3600.0, run_initial_validation=True, gradient_prompt_files=None, apply_edit_prompt_files=None, _poml_trace=False)` ¶

使用配置参数初始化 APO 算法。

参数

async_openai_client (AsyncOpenAI) –

用于进行 LLM API 调用的 AsyncOpenAI 客户端。
gradient_model (str, 默认值: 'gpt-5-mini' ) –

用于计算文本梯度（评论）的模型名称。
apply_edit_model (str, 默认值: 'gpt-4.1-mini' ) –

用于基于评论应用编辑的模型名称。
diversity_temperature (float, 默认值: 1.0 ) –

用于控制多样性的 LLM 调用的温度参数。
gradient_batch_size (int, 默认值: 4 ) –

用于梯度计算的 rollout 结果数量。
val_batch_size (int, 默认值: 16 ) –

用于评估的验证示例数量。
beam_width (int, 默认值: 4 ) –

在每个轮次在束中保留的得分最高的提示数量。
branch_factor (int, 默认值: 4 ) –

从每个父提示生成的新提示候选者数量，通过应用文本梯度编辑。这控制着搜索树的扩展。
beam_rounds (int, 默认值: 3 ) –

要执行的束搜索轮次数量。
rollout_batch_timeout (float, 默认值: 3600.0 ) –

等待 rollout 批次完成的最大时间（秒）。
run_initial_validation (bool, 默认值: True ) –

如果为 True，则在开始优化之前对种子提示进行验证，以建立基线分数。默认值为 True。
gradient_prompt_files (Optional[List[Path]], 默认值: None ) –

用于计算文本梯度（评论）的提示模板。
apply_edit_prompt_files (Optional[List[Path]], 默认值: None ) –

用于基于评论应用编辑的提示模板。

`compute_textual_gradient(current_prompt, rollout_results, *, prefix=None)` `async` ¶

基于 rollout 结果为当前提示计算文本梯度（评论）。

此方法对 rollout 结果进行采样，将其与当前提示一起发送到 LLM，并生成描述如何改进提示的评论。

参数

current_prompt (VersionedPromptTemplate) –

要评论的提示模板。
rollout_results (List[RolloutResultForAPO]) –

包含 spans、messages 和 rewards 的 rollout 结果列表。

返回

Optional[str] –

由 LLM 生成的文本评论，如果生成失败则为 None。

`evaluate_prompt_on_batch(prompt, resource_name, dataset, mode, *, prefix=None)` `async` ¶

通过运行 rollout 并计算平均奖励来评估一批任务上的提示。

此方法

将提示作为命名资源添加到存储中
为数据集中的每个任务排队进行 rollout
等待 rollout 完成（带超时）
计算并返回平均奖励

参数

prompt (VersionedPromptTemplate) –

要评估的提示模板字符串。
resource_name (str) –

在存储中注册提示的名称。
dataset (Sequence[T_task]) –

评估提示的任务序列。
mode (RolloutMode) –

用于日志记录/跟踪的 rollout 模式（“train”或“val”）。

返回

List[RolloutResultForAPO] –

一个元组 (rollout_results, average_reward)，其中 rollout_results 包含
float –

每个 rollout 的详细信息，average_reward 是平均最终奖励。

`get_adapter()` ¶

获取用于将 spans 转换为消息的适配器。

返回

TraceToMessages –

此算法的 TraceToMessages 实例。

引发

ValueError –

如果适配器不是 TraceToMessages。

`get_best_prompt()` ¶

检索在优化期间发现的最佳提示。

返回

PromptTemplate –

到目前为止，发现的具有最高验证分数的提示模板。

引发

ValueError –

如果尚未找到最佳提示（未调用 run()）。

`get_rollout_results(store, rollout, *, prefix=None)` `async` ¶

将完成的 rollout 转换为与 APO 兼容的结果格式。

获取每个 rollout 的 spans，将其适配为消息，并将其与奖励和状态信息打包在一起，以便进行梯度计算。

参数

rollout (List[Rollout]) –

完成的 rollout 元数据列表。

返回

List[RolloutResultForAPO] –

格式化为 APO 处理的 rollout 结果列表。

`get_seed_prompt_template()` ¶

从算法的资源中提取初始提示模板。

返回

Tuple[str, PromptTemplate] –

一个元组 (resource_name, prompt_template)，表示种子提示。

引发

ValueError –

如果未设置 initial_resources 或未找到 PromptTemplate。

`run(store, llm_proxy, train_dataset=None, val_dataset=None)` `async` ¶

执行 APO 算法，通过带有文本梯度的束搜索来优化提示。

该算法在多个轮次中执行迭代提示优化

每个轮次：采样父提示，通过文本梯度生成新的候选提示，在验证数据上评估所有候选提示，并保留表现最佳的提示
跟踪所有轮次中历史最佳提示
为每次梯度计算使用不同的训练数据集样本，以确保多样性

参数

train_dataset (Optional[Dataset[T_task]], default: None ) –

用于计算文本梯度的任务数据集。必需。
val_dataset (Optional[Dataset[T_task]], default: None ) –

用于评估和选择提示的任务数据集。必需。

引发

ValueError –

如果 train_dataset 或 val_dataset 为 None，或者资源未设置。

`textual_gradient_and_apply_edit(current_prompt, rollout, *, prefix=None)` `async` ¶

通过计算文本梯度并应用编辑来生成改进的提示。

这是主要的优化步骤，它

基于 rollout 性能计算一个 critique（文本梯度）
使用另一个 LLM 应用 critique 并生成改进的提示

参数

current_prompt (VersionedPromptTemplate) –

要改进的当前提示模板。
rollout (List[RolloutResultForAPO]) –

基于该 critique 的 rollout 结果列表。

返回

Optional[str] –

改进的提示文本，或者如果梯度计算失败，则为原始提示。

APO¶

安装¶

当前实现范围¶

初始提示¶

使用APO的教程¶

参考¶

agentlightning.algorithm.apo ¶

APO ¶

compute_textual_gradient(current_prompt, rollout_results, *, prefix=None) async ¶

evaluate_prompt_on_batch(prompt, resource_name, dataset, mode, *, prefix=None) async ¶

get_adapter() ¶

get_best_prompt() ¶

get_rollout_results(store, rollout, *, prefix=None) async ¶

get_seed_prompt_template() ¶

run(store, llm_proxy, train_dataset=None, val_dataset=None) async ¶

textual_gradient_and_apply_edit(current_prompt, rollout, *, prefix=None) async ¶