企业官网建设流程全解析

agno Saved Baselines：将环境回滚证据固化为 JSON 基线，实现跨进程评估对比与 CI 回归验证

【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno

基线（baseline）是评估中最容易被忽视却至关重要的资产：它把某一次特定环境与策略的回滚（rollout）结果完整保存下来，让后续任何一次运行都能与它进行同任务、同指纹的逐项对比。本文基于 agno 仓库中cookbook/environments/_13_saved_baselines/目录的完整示例，系统讲解如何使用agno.environments提供的run_rollouts/arun_rollouts、EnvironmentRunResult.save/load等 API 将评估证据持久化为纯 JSON 文件，并在同进程不可行、或 CI 需要人工审核的参考产物时，通过环境指纹（env_fingerprint）与策略指纹（policy_fingerprint）保障对比的合法性。读完本文，你将掌握基线文件的生成、重载校验、异步读写，以及将其与_14_environment_diff衔接做逐任务对比的完整实战方案。

一、什么是 Saved Baselines：把一次回滚的完整证据变成可复用的基线

在 agno 的评估体系中，environments 模块 回答了评估中最核心的两个问题：给定一个 agent 与一组任务，按 K 次重复运行得到真实的通过率（而非单次采样），以及能否把通过的尝试直接作为 SFT 训练数据。而Saved Baselines是在这一体系之上解决"跨进程、跨时间的可复现对比"问题：

• 基线文件把一次回滚的全部证据以纯 JSON形式持久化，包括任务级的历史记录与指纹；
• 由于保存的产物包含完整的提示词（prompts）与响应（responses），它应当被当作敏感评估数据妥善保管；
• 一条基线是"特定环境 + 特定策略"产生的证据，不是对未来任务或提示词修改仍可比较的承诺——一旦环境或策略改变，指纹就会变化，对比的合法性需要通过指纹来校验。

目录中的三个示例文件各司其职（对应 cookbook/environments/_13_saved_baselines/README.md）：

二、何时使用 Saved Baselines

根据官方文档，以下两类场景应当优先考虑使用保存的基线，而不是在同一个进程内直接比较：

1. 基线与候选（candidate）无法在同一进程中运行——例如候选运行在另一台机器、另一个 Python 进程，或需要长时间之后才执行；
2. CI 需要一个经过人工审核的参考产物（reviewed reference artifact）——评审通过后固化为基线，后续每次 CI 拉取该基线进行对比，而不是每次重新生成参考值。

在保存基线之后，可以继续前往 cookbook/environments/_14_environment_diff/ 对兼容的结果做逐任务（task by task）对比。那里明确规定了可比性边界：两个结果共享相同的环境指纹时，模型策略变化（如推理强度）才是可比较的；而任务、评分器、工具或提示词的改变则不可比较。

关键认知：基线是"特定环境 + 特定策略"的证据快照，不是承诺。未来任务的增删、提示词的编辑都会改变环境指纹，使旧基线失去对比合法性——这正是下文要重点讲解的指纹机制。

三、basic.py：运行环境并保存基线

basic.py演示了最基础但完整的保存流程。先看完整代码（cookbook/environments/_13_saved_baselines/basic.py）：

from pathlib import Path from agno.agent import Agent from agno.environments import Environment, Task, run_rollouts from agno.models.openai import OpenAIResponses from agno.scorer import CodeScorer from pydantic import BaseModel class Answer(BaseModel): value: int def exact_value(run, expected): return run.content.value == expected agent = Agent( model=OpenAIResponses(id="gpt-5.5", reasoning_effort="low"), output_schema=Answer, ) env = Environment( name="saved-baseline-basic", agent=agent, tasks=( Task( id="product-a", input=( "Compute 2718281828459045 times 1618033988749895. Add the " "decimal digits of that product, multiply the digit sum by " "131071, subtract the product remainder modulo 65521, and " "return the final integer." ), expected=20944939, ), Task( id="product-d", input=( "Compute 2236067977499789 times 2449489742783178. Add the " "decimal digits of that product, multiply the digit sum by " "524287, subtract the product remainder modulo 99991, and " "return the final integer." ), expected=76998482, ), ), scorer=CodeScorer(exact_value), ) baseline_path = Path(__file__).parent / "data" / "generated" / "baseline.json" if __name__ == "__main__": result = run_rollouts(env, k=6) print(result) baseline_path.parent.mkdir(parents=True, exist_ok=True) result.save(baseline_path) print(f"saved {result.n_attempts} attempts to {baseline_path}")

整个流程可以拆解为四步：

1. 定义结构化输出与评分函数

通过 Pydantic 定义Answer(value: int)作为 agent 的结构化输出 schema，随后定义评分函数exact_value(run, expected)——从run.content.value取出模型输出的整数值，与任务期望值精确比较。该函数会被CodeScorer包装为评分器。

2. 组装 Environment

Environment是run_rollouts执行的最小单元，其定义在 libs/agno/agno/environments/environment.py 中，包含四个核心字段：

• name：环境名称，会写入基线并用于标识对比；
• tasks：任务元组，每个Task由input（必填的字符串输入）、expected（期望值）、可选的id与metadata构成；未指定id时会在运行开始时按位置解析为t1..tN；
• scorer：评分器，CodeScorer会为评分函数生成digest()，参与环境指纹计算；
• agent：一个 live Agent 实例（每次尝试会做深拷贝），或一个零参工厂函数（每次尝试调用一次）——从源码注释看，Environment.agent不接受 Team，且 Agent 引用必须保持存活，因为通过Agent.from_dict重水合会丢失采样参数、base_url与凭据。

此外还有timeout_seconds参数，默认 120 秒，用于约束单次尝试的返回时限。

3. 运行回滚并保存

result = run_rollouts(env, k=6) result.save(baseline_path)

k=6表示每个任务重复运行 6 次。run_rollouts的完整签名（见 libs/agno/agno/environments/runner.py）：

def run_rollouts( env: Environment, *, k: int = 8, # 每个任务重复次数，默认 8 tasks=None, # 从 env.tasks 中筛选子集（须保持环境身份） model=None, # 模型覆盖（必须是 Model 实例，不支持字符串解析） concurrency: int = 4, # 并发度 ) -> EnvironmentRunResult:

result.save(path)会将结果序列化为纯 JSON写入磁盘。需要特别注意两点：

• save会先完成序列化再打开文件（open("w")会截断文件），因此序列化失败不会破坏已存在的基线文件——而那个文件正是diff()所依赖的；
• 保存路径使用了Path(__file__).parent / "data" / "generated" / "baseline.json"，并以mkdir(parents=True, exist_ok=True)保证目录存在。

4. 解读保存结果

保存的 JSON 是EnvironmentRunResult.save的完整序列化产物，其顶层结构包括（对应runner.py中save的实现）：

• format_version：格式版本号（当前为1），加载时会做严格校验；
• env_name、k、duration_seconds、stopped_early；
• env_fingerprint与policy_fingerprint：两个指纹字符串，是后续对比合法性的核心依据；
• task_results：每个任务的task（id/input/expected/metadata）与attempts列表，每次尝试包含完整run记录（通过run.to_dict()序列化，即完整提示词与响应）、score、stop_reason、duration_seconds、error、tool_call_limit_hit、error_type。

四、reload_baseline.py：重载并验证摘要往返

保存基线的价值在于后续能够被重新加载。reload_baseline.py（cookbook/environments/_13_saved_baselines/reload_baseline.py）演示了保存 + 重载 + 摘要比对三个动作：

from pathlib import Path from agno.agent import Agent from agno.environments import Environment, EnvironmentRunResult, Task, run_rollouts from agno.models.openai import OpenAIResponses from agno.scorer import CodeScorer from pydantic import BaseModel class Answer(BaseModel): value: int def exact_value(run, expected): return run.content.value == expected agent = Agent( model=OpenAIResponses(id="gpt-5.5", reasoning_effort="low"), output_schema=Answer, ) env = Environment( name="reload-saved-baseline", agent=agent, tasks=( Task( id="product-a", input=( "Compute 2718281828459045 times 1618033988749895. Add the " "decimal digits of that product, multiply the digit sum by " "131071, subtract the product remainder modulo 65521, and " "return the final integer." ), expected=20944939, ), Task( id="product-c", input=( "Compute 1414213562373095 times 1732050807568877. Add the " "decimal digits of that product, multiply the digit sum by " "99991, subtract the product remainder modulo 32749, and " "return the final integer." ), expected=16568751, ), ), scorer=CodeScorer(exact_value), ) baseline_path = Path(__file__).parent / "data" / "generated" / "reloaded.json" if __name__ == "__main__": result = run_rollouts(env, k=4) print(result) baseline_path.parent.mkdir(parents=True, exist_ok=True) result.save(baseline_path) loaded = EnvironmentRunResult.load(baseline_path) assert loaded.summary() == result.summary() print(f"reloaded pass rate: {loaded.pass_rate}") print(f"fingerprints preserved: {loaded.env_fingerprint == result.env_fingerprint}")

这个示例的核心验证逻辑是：

loaded = EnvironmentRunResult.load(baseline_path) assert loaded.summary() == result.summary()

summary()是冻结的 CI 契约（源码注释明确标注 "The CI contract; these keys are frozen"），返回结构化的字典，包含环境级指标与任务级指标：

• 环境级：env、k、n_tasks、n_attempts、n_scored、n_unscored、pass_rate、mean_value、env_fingerprint、policy_fingerprint、stopped_early；
• 任务级：每个任务的id、pass_rate、mean_value、n_unscored、learning_zone（该任务既有通过的尝试又有失败的尝试，处于"学习区"）。

因此loaded.summary() == result.summary()断言的是：经过 save → load 的 JSON 往返后，所有统计字段与指纹完全一致。这正是 CI 可以依赖基线的根基——摘要与指纹是确定性、可复现的。脚本还额外打印了重载后的通过率与指纹保持情况：

print(f"reloaded pass rate: {loaded.pass_rate}") print(f"fingerprints preserved: {loaded.env_fingerprint == result.env_fingerprint}")

关于load，需要注意它做了格式版本校验：当 JSON 中的format_version不等于当前构建支持的版本时，会抛出ValueError: unsupported format_version ...。这保证旧格式的基线不会在不知情的情况下被新版本错误地解读。

五、async_save_load.py：异步孪生 API 的完整往返

当外围应用已经拥有事件循环时，应使用异步孪生 API。async_save_load.py（cookbook/environments/_13_saved_baselines/async_save_load.py）展示了完整用法：

import asyncio from pathlib import Path from agno.agent import Agent from agno.environments import Environment, EnvironmentRunResult, Task, arun_rollouts from agno.models.openai import OpenAIResponses from agno.scorer import CodeScorer from pydantic import BaseModel class Answer(BaseModel): value: int def exact_value(run, expected): return run.content.value == expected agent = Agent( model=OpenAIResponses(id="gpt-5.5", reasoning_effort="low"), output_schema=Answer, ) env = Environment( name="async-saved-baseline", agent=agent, tasks=( Task( id="product-a", input=( "Compute 2718281828459045 times 1618033988749895. Add the " "decimal digits of that product, multiply the digit sum by " "131071, subtract the product remainder modulo 65521, and " "return the final integer." ), expected=20944939, ), Task( id="product-b", input=( "Compute 3141592653589793 times 2718281828459045. Add the " "decimal digits of that product, multiply the digit sum by " "104729, subtract the product remainder modulo 65537, and " "return the final integer." ), expected=16756170, ), ), scorer=CodeScorer(exact_value), ) baseline_path = Path(__file__).parent / "data" / "generated" / "async_baseline.json" async def main(): result = await arun_rollouts(env, k=4) print(result) baseline_path.parent.mkdir(parents=True, exist_ok=True) await result.asave(baseline_path) loaded = await EnvironmentRunResult.aload(baseline_path) assert loaded.summary() == result.summary() print(f"async round trip preserved {loaded.n_attempts} attempts") if __name__ == "__main__": asyncio.run(main())

这里用到的异步孪生 API 有明确的对应关系：

从源码看，asave与aload分别通过asyncio.to_thread将同步实现搬运到线程中执行，因此在已有事件循环的应用（如 FastAPI、asyncio 服务）中不会阻塞事件循环。相反，同步run_rollouts内部使用asyncio.run，如果从正在运行的事件循环中调用会抛出RuntimeError——源码明确提示"run_rollouts cannot be called from a running event loop; await arun_rollouts instead"。

示例的验证逻辑与同步版一致：await result.asave(...)保存后，await EnvironmentRunResult.aload(...)重载，并以loaded.summary() == result.summary()断言往返一致性，最后打印重载后保留的尝试数。

六、深入指纹机制：为什么基线对比是"合法"的

Saved Baselines 之所以能在跨进程场景下被信赖，关键在于运行开始前计算并刻印在结果上的两个指纹。理解它们，就理解了"什么可以对比、什么不可以对比"。

环境指纹（env_fingerprint）

定义在 libs/agno/agno/environments/environment.py 的_env_fingerprint_of中，是对"环境身份"的 sha256 摘要，参与哈希的组件包括：

• tasks：每个任务的解析后 id、input 与 expected；
• scorer：评分器的digest()——CodeScorer的摘要覆盖"评分函数的去缩进源码 + pass_threshold"，因为同一个函数配不同阈值会产生不同的评分行为；
• 声明的工具 schema 与tool_choice：后者决定了模型可以调用哪些工具；
• 提示词塑形字段：instructions、description、system_message、additional_context、expected_output、role、additional_input，以及name（仅在add_name_to_context开启时计入）；
• prompt_flags：markdown、add_name_to_context、add_location_to_context、add_datetime_to_context、add_session_state_to_context等标志值——注意这里哈希的是标志值而非渲染后的文本，因为add_datetime_to_context会注入墙钟时间，哈希渲染值会让指纹跨运行失去确定性；
• model_prompt：模型级的提示词字段；
• session_state与终止设置（timeout_seconds、tool_call_limit）。

指纹字符串带有版本前缀（当前为envfp2），因此不同格式版本写出的指纹永远不会相等。任何组件失败都会抛出FingerprintError，由运行器捕获后使指纹降级为None——而env_matches对None永远返回False，绝不会出现"两边都是 None 就视为匹配"的虚假绿灯。

策略指纹（policy_fingerprint）

_policy_fingerprint_of是对模型身份（model identity）的 sha256：包括模型类、id、provider、base_url 以及每个枚举的请求塑形参数。模型 id 明确包含在负载中——gpt-5.5与gpt-5.5-mini绝不能哈希成相同值，这正是"策略漂移"要捕获的差异。

两个指纹如何支撑对比

EnvironmentRunResult.env_matches(other)通过比较两侧的env_fingerprint判断是否来自同一环境。而后续diff()的第一步就是调用env_matches：不匹配则抛出MismatchError，明确提示"这些结果并非来自同一环境（None 永不匹配）"。_14_environment_diff的mismatch_guard.py正是演示这个异常场景。

因此一条基线只对其"同一环境"的候选运行有效：调整推理强度（reasoning_effort）只改变策略指纹，环境指纹不变，可以对比；而修改任务、评分器、工具或提示词会改变环境指纹，不可对比，此时需要改用提示词对比（_15_prompt_comparison）等专门机制。

尝试隔离保证基线质量

arun_rollouts对每次尝试执行无条件隔离：每个尝试运行在全新的内存存储与全新的用户 id上，响应缓存关闭，生产环境的解析器原样运行；随后只切断写路径（记忆捕获、知识/学习写入、会话摘要写入、save_response_to_file）。这保证了采样统计不被尝试间污染，也保证保存进基线的轨迹是干净、可用于训练的数据。需要留意三个源码明确声明的残留限制：同步评分器运行在线程中、超时无法中断其主体；tracing 是进程级全局状态，尝试的 trace 会进入调用方的 trace 存储（可用 rollout-* id 识别）；用户提供的可调用对象（hooks、fallback 回调）按引用共享，应保持幂等。

七、运行方式与前置条件

目录 cookbook/environments/_13_saved_baselines/ 下的三个示例均可直接运行：

python cookbook/environments/_13_saved_baselines/basic.py python cookbook/environments/_13_saved_baselines/reload_baseline.py python cookbook/environments/_13_saved_baselines/async_save_load.py

前置条件与限制：

• 需要配置OPENAI_API_KEY环境变量；
• 所有示例均通过OpenAIResponses使用gpt-5.5模型（basic.py与async_save_load.py使用reasoning_effort="low"）；
• 示例依赖agno.environments（导出Environment、Task、run_rollouts、arun_rollouts、EnvironmentRunResult、EnvironmentDiff等，见 libs/agno/agno/environments/init.py）；
• 运行时会调用真实 LLM 服务，产生的基线文件默认写入各脚本所在目录的data/generated/下。

八、下一步：从基线到逐任务对比

保存基线不是终点。当候选运行完成、且其环境指纹与基线一致时，即可使用EnvironmentRunResult.diff(baseline)生成EnvironmentDiff：按任务输出baseline -> current的通过率变化（improved/regressed），并标注两侧未匹配的任务（子集对比时不会静默丢弃）。完整的对比实践位于 cookbook/environments/_14_environment_diff/，其中basic.py用低/高推理强度策略在同一环境上做对比，task_subset.py演示任务子集对比，mismatch_guard.py演示环境改变时MismatchError的抛出。

由此，一条完整的评估闭环便建立起来：定义环境 → 运行回滚并保存基线 → 重载校验摘要往返 → 在任意进程/CI 中对比候选 → 依据指纹判定对比合法性。而这一切的证据基础，就是那些包含了完整提示词与响应、需要被当作敏感评估数据妥善保管的 JSON 基线文件。

【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno