Agno Environments 任务集（Task Sets）实战：用稳定 ID、JSONL 与元数据组织可复现评测-平芜编程栈

Agno Environments 任务集（Task Sets）实战：用稳定 ID、JSONL 与元数据组织可复现评测

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

导读

本文围绕 Agno 开源仓库中 Environments 模块的Task Sets（任务集）展开，讲解如何把可重复的输入、期望值、ID 与元数据组织成一个环境（Environment）所验证的任务集合，让任务 ID 标注评测网格、对齐后续的差异对比（diff）。你将掌握三种任务组织方式——内联Task对象、JSONL 数据文件加载、按元数据筛选校准切片——并理解其底层run_rollouts执行与评分机制，最终能依据测试日志中的通过率分布定位模型的"学习区"。

一、Task Sets 是什么：为什么任务需要"集合化"

单次成功的 Agent 运行不足以证明可靠性。Agno 的 Environments 模块把"一次运行"升级为"对一个任务集合的 K 次重复采样"，每次尝试独立评分，最终形成一个通过/失败网格——全满的行表示已被掌握的任务，全空的行表示需要换一种干预手段，而部分通过的行就是值得深入研究的学习区（learning zone）。

任务集正是这个网格的"输入骨架"：它把四类信息收集在一起，交给Environment统一验证：

信息	字段	作用
输入	`input`	提交给 Agent 的原始提示词，必填且必须是字符串
期望值	`expected`	Agent 应该产出的结果，可选
稳定 ID	`id`	标注网格行、作为后续 diff 的 join 键，可选
元数据	`metadata`	划分 split / difficulty 等切片，可选，不会自动进入提示词

对应地，cookbook/environments/_02_task_sets/ 目录提供了三个从易到难的示例，以及一份本地可审查的 JSONL 任务夹具：

basic.py—— 以内联元组声明少量带稳定 ID 的任务；
from_jsonl.py—— 从版本控制的 JSONL 文件加载同构任务集；
with_metadata.py—— 通过任务元数据挑选"校准"行后再运行；
data/chained_arithmetic.jsonl—— 本地、可审查的任务数据夹具。

二、三种任务组织方式实战

2.1 内联 Task：紧凑示例的首选

basic.py 展示了最小可运行的任务集。核心是Task数据类，每个任务三要素齐全——输入、期望值、稳定 ID：

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 answer_matches(run, expected): return run.content.value == expected TASKS = ( Task(input="What is 43 multiplied by 47?", expected=2021, id="easy-product"), Task( input=( "Compute 2718281828459045 multiplied by 1618033988749895. Add " "the product's decimal digits, multiply the sum by 131071, " "subtract the product remainder modulo 65521, and return the result." ), expected=20944939, id="chained-product-a", ), Task( input=( "Compute two chained values, then add them. First: multiply " "2718281828459045 by 1618033988749895, ... Second: multiply " "3141592653589793 by 1414213562373095, ... Return the sum of the two chained values." ), expected=37676112, id="dual-chain-sum", ), )

任务之后是 Agent、Environment 与运行入口：

agent = Agent( model=OpenAIResponses(id="gpt-5.5", reasoning_effort="low"), output_schema=Answer, instructions="Return only the requested final integer in the typed field.", ) environment = Environment( name="task-set-basics", agent=agent, tasks=TASKS, scorer=CodeScorer(answer_matches), ) if __name__ == "__main__": results = run_rollouts(environment, k=4) print(results) print() for task_result in results.task_results: print(f"{task_result.task.id}: {task_result.n_passed}/{task_result.n_scored}")

从 environment.py 的源码可以看到Task的完整字段定义：input（必填字符串）、expected（可选任意值）、id（可选字符串）、metadata（默认空映射）。其中id用于展示与选择，未声明时会在运行开始时按位置自动解析为t1、t2……tN。

2.2 JSONL 加载：任务集作为"数据"来治理

当任务集合由数据团队拥有、需要评审与版本控制时，内联代码就不再合适。from_jsonl.py 展示了Task.from_jsonl的用法：

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 TASKS_PATH = Path(__file__).parent / "data" / "chained_arithmetic.jsonl" agent = Agent( model=OpenAIResponses(id="gpt-5.5", reasoning_effort="low"), output_schema=Answer, instructions="Return only the requested final integer in the typed field.", ) environment = Environment( name="jsonl-task-set", agent=agent, tasks=Task.from_jsonl(TASKS_PATH), scorer=CodeScorer(answer_matches), ) if __name__ == "__main__": results = run_rollouts(environment, k=4) print(results) print(f"loaded {len(environment.tasks)} tasks from {TASKS_PATH}")

对应数据文件 chained_arithmetic.jsonl 每一行是一个 JSON 对象，同时携带id、input、expected与metadata：

{"id":"easy-product","input":"What is 43 multiplied by 47?","expected":2021,"metadata":{"split":"smoke","difficulty":"anchor"}} {"id":"chained-product-a","input":"Compute 2718281828459045 multiplied by 1618033988749895. Add the product's decimal digits, multiply the sum by 131071, subtract the product remainder modulo 65521, and return the result.","expected":20944939,"metadata":{"split":"validation","difficulty":"calibration"}} {"id":"dual-chain-sum","input":"Compute two chained values, then add them. First: multiply 2718281828459045 by 1618033988749895, ... Return the sum of the two chained values.","expected":37676112,"metadata":{"split":"validation","difficulty":"calibration"}}

需要特别说明的是Task.from_jsonl的严格校验语义（见 environment.py）：每行必须是一个对象，顶层只允许input、expected、id、metadata四个键，任何未知键都会抛出ValueError并指出具体行号与键名。这样设计是为了防止"拼错列名"（例如把expected_output误写成expected）导致每个任务都拿到expected=None，从而在宽松评分器下"全绿"的静默事故。此外input必须为字符串、metadata必须为对象、id若存在必须是字符串，行内 JSON 语法错误也会被精确到行号地拒绝。

2.3 按元数据挑选任务：校准切片的正确姿势

with_metadata.py 演示了元数据的核心价值——为任务打上split与difficulty标签，然后在运行前从原任务对象中挑选子集：

TASKS = ( Task( input=("Compute 2718281828459045 multiplied by 1618033988749895. ..."), expected=20944939, id="chained-product-a", metadata={"split": "validation", "difficulty": "calibration"}, ), Task( input=("Compute 3141592653589793 multiplied by 1414213562373095. ..."), expected=16731173, id="chained-product-b", metadata={"split": "validation", "difficulty": "calibration"}, ), Task( input="What is 43 multiplied by 47?", expected=2021, id="easy-product", metadata={"split": "smoke", "difficulty": "anchor"}, ), ) environment = Environment( name="metadata-task-set", agent=agent, tasks=TASKS, scorer=CodeScorer(answer_matches), ) if __name__ == "__main__": calibration_tasks = tuple( task for task in environment.tasks if task.metadata["difficulty"] == "calibration" ) results = run_rollouts(environment, tasks=calibration_tasks, k=4) print(results) print() for task_result in results.task_results: split = task_result.task.metadata["split"] difficulty = task_result.task.metadata["difficulty"] print( f"{task_result.task.id}: split={split}, difficulty={difficulty}, " f"pass_rate={task_result.pass_rate}" )

这里有三个容易踩坑的细节值得展开：

元数据只用于组织，不污染提示词：文档明确说明"它不自动加入 Agent 提示词"，评分器也不会因此改变——split/difficulty 纯粹是数据集治理层面的切片标签。
子集选择必须按身份（identity）进行：从 runner.py 的实现看，run_rollouts的tasks=参数要求从env.tasks中选出的原始对象（例如[t for t in env.tasks if ...]），重建出的新Task会因身份不匹配而抛出ValueError——因为选择必须保持环境身份，才能保证指纹与 diff 语义成立。
重复 ID 会被拦截：Environment.__post_init__会对显式声明的 ID 做唯一性校验（见 environment.py），因为重复 ID 会让diff()静默地把行配错到错误的基线任务上。

三、运行方式与环境要求

三个示例的运行命令（来自 README.md）：

python cookbook/environments/_02_task_sets/basic.py python cookbook/environments/_02_task_sets/from_jsonl.py python cookbook/environments/_02_task_sets/with_metadata.py

运行前提：

需要OPENAI_API_KEY环境变量；
所有模型调用都通过OpenAIResponses，模型为gpt-5.5（示例中以reasoning_effort="low"降低成本）；
三个脚本内部都以k=4运行（每个任务重复采样 4 次）。

学习路径上，建议先阅读 _01_first_environment/ 的 README 掌握最小运行器，再继续到 _03_code_scorer/ 的 README 了解期望值如何转化为分数。

四、源码级原理：run_rollouts 如何执行与评分

任务集只是输入，真正干活的是run_rollouts/arun_rollouts（见 runner.py）。其完整签名与默认值：

async def arun_rollouts( env: Environment, *, k: int = 8, # 每个任务重复采样次数 tasks: Optional[Sequence[Task]] = None, # 任务子集（按身份从 env.tasks 挑选） model: Optional[Model] = None, # 模型覆盖（必须是 Model 实例，不接受字符串） concurrency: int = 4, # 并发尝试数 ) -> EnvironmentRunResult:

几个值得注意的执行语义：

每次尝试完全隔离：每个 attempt 运行在"新鲜的内存 DB + 新的 rollout 用户 ID + 关闭响应缓存"的 Agent 副本上，记忆、会话、用户态互不污染；同时切断写路径（记忆捕获、知识/学习写入、会话摘要写入、save_response_to_file），但保留知识检索等只读路径。这保证了统计数据回答的是"这个 Agent 本身行不行"，而不是"被上一次尝试污染的 Agent 行不行"。
指纹（fingerprint）：运行开始时对环境指纹（任务、评分器、工具、提示词字符串与标志等，版本前缀envfp2）和策略指纹（模型类、ID、provider、base_url 与请求参数）分别做 sha256，并盖在结果上。env_matches()保证只有同环境的结果才能diff()，None指纹永不匹配。
错误风暴（error-storm）早停：单任务运行且k超过证据窗口时，若开头一批尝试全部以同类型错误失败，则中止整个运行并标记stopped_early="error-storm"，避免坏配置白白烧掉所有采样。
同步入口的限制：run_rollouts不能从正在运行的事件循环中调用（应改用arun_rollouts）；同步评分器运行在线程中，超时无法中断其执行体，这是文档明示的残余语义。

评分端使用CodeScorer（详见 _03_code_scorer/ 的 README），它接受一个命名函数，返回布尔值、0~1 的分数或完整的Score。示例中的answer_matches(run, expected)即把结构化输出run.content.value与期望值做精确相等比较。

五、结果对象与学习区判读

run_rollouts返回EnvironmentRunResult，其关键派生指标（见 runner.py）：

指标	含义
`n_scored`/`n_unscored`	被评分的尝试数 / 未评分尝试数（超时不算错误答案，不计入统计）
`n_passed`/`pass_rate`	通过数 / 通过率（`n_passed / n_scored`）
`in_learning_zone`	`0 < n_passed < n_scored`——有失败也有通过，每次失败都有可对照的通过样本
`summary()`	CI 契约，返回固定键的摘要字典
`diff(baseline)`	与基线结果做逐任务 delta 对比，输出 improved / regressed
`save()`/`load()`	结果 JSON 落盘与回读，供"改完重跑看变化"

网格的可视化判读逻辑很直白：全满行 = 已被掌握；全空行 = 需要换干预手段；部分行 = 学习区。示例打印的n_passed/n_scored正是把每个任务的网格行压缩成一行文本。

六、测试日志解读：校准过程与通过率分布

TEST_LOG.md 记录了 2026-07-20 在 Agno 2.7.4 上针对gpt-5.5（经OpenAIResponses）的真实回归结果。这份日志同时是任务集校准（calibration）的方法论示范：

6.1 basic.py：PASS

12/12 次尝试全部被评分，0 次未评分；
观察到的通过率：easy-product4/4（1.00）、chained-product-a1/4（0.25）、dual-chain-sum0/4（0.00）；
校准过程：第一版校准在易行和两个独立链式行上都饱和到 4/4，于是用"复合双链求和"任务替换掉一个独立链式行后重跑——这正是任务集作为可靠性评测工具的核心用法：一个全满的网格不是有用的可靠性示例，必须让难度梯度产生分歧（disagreement）。

6.2 from_jsonl.py：PASS

从 JSONL 严格加载全部 3 行任务，12/12 次尝试被评分，0 次未评分；
观察到的通过率：easy-product4/4（1.00）、chained-product-a3/4（0.75）、dual-chain-sum0/4（0.00）；
日志特别指出：独立链式行提供了真实的"部分通过带"——即中间难度的任务恰好落进学习区（0 < pass_rate < 1），这是评测数据最有价值的部分。

6.3 with_metadata.py：PASS

通过元数据挑选两个原始任务对象作为校准行，8/8 次尝试被评分，0 次未评分；
观察到的通过率：chained-product-a3/4（0.75）、chained-product-b3/4（0.75）；
校准过程：第一版元数据切片只产生 4/4 和 0/4 两行（要么全过要么全挂），因此被重新校准为包含两个独立的中间难度候选，使两个任务都落入学习区。

综合来看，这份日志验证了任务集的三大能力：id让每个网格行可命名、可对齐；JSONL 让任务集可像数据一样被版本控制与评审；metadata让"校准切片"成为可重复的筛选操作。而"校准到分歧"的过程，本身就是评测可靠性的最佳实践——当任务难度梯度设计得当时，通过率矩阵会清晰暴露模型能力的分界线。