agno Environments 学习区 SFT 数据导出实战：从 Rollout 到可消费的对话式 JSONL 数据集

agno Environments 学习区 SFT 数据导出实战：从 Rollout 到可消费的对话式 JSONL 数据集

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

导读

在 agno 的环境（Environments）体系中，_10_export_sft位于"学习区（learning zone）筛选"与"数据导出"之间的关键环节：它把已经跑完多轮 rollout、被评分器标记为通过/失败的对话尝试，筛选出处于中间难度带（learning zone）的任务，并只把其中通过的纯文本对话导出为一行一个 JSON 对象的对话式 SFT JSONL 文件。本文以 cookbook/environments/_10_export_sft/README.md 及其三个示例脚本为骨架，结合 agno 源码深入讲解run_rollouts、learning_zone()、to_sft_jsonl的底层实现、导出格式的严格约束、空学习区的防御性处理，以及它们在整个"数据集生成而非模型训练"工作流中的定位。

前置知识：Export SFT 在 Environments 流水线中的位置

从 cookbook/environments/_10_export_sft/README.md 可知，导出 SFT 数据应放在_06_learning_zone/之后使用：

• _06_learning_zone/先展示哪些任务处于真实的中间带通过率；
• _10_export_sft再把处于学习区任务中的通过尝试导出为数据集；
• 导出的 JSONL 旁会生成一个<path>.meta.jsonsidecar 文件，记录分数与指纹信息，下一章_11_export_provenance/专门分析这个 sidecar。

需要特别强调的是：导出只生成数据集，不会训练任何模型。另外，带有工具调用（tool-bearing）的对话在这个纯文本格式中无法被忠实表示，因此会被跳过，工具相关的验证请参考_17_tool_reliability/。

核心 API 全景：Environment、run_rollouts、learning_zone、to_sft_jsonl

三个示例脚本都遵循同一个四步工作流，涉及的 API 均来自agno.environments与agno.scorer：

run_rollouts：同步入口与并发语义

run_rollouts定义在 libs/agno/agno/environments/runner.py，签名如下：

def run_rollouts( env: Environment, *, k: int = 8, tasks: Optional[Sequence[Task]] = None, model: Optional[Model] = None, concurrency: int = 4, ) -> EnvironmentRunResult:

• k是每个任务的尝试次数，默认为 8；示例脚本中分别使用了k=4与k=6；
• concurrency控制并发尝试数，默认 4；
• 它是arun_rollouts的同步门面，内部用asyncio.run执行；
• 超时语义沿用异步实现：尝试协程会在env.timeout_seconds到达时被取消；
• 特殊限制：run_rollouts不能在已有运行事件循环的环境中调用（会抛出RuntimeError），此时应改用await arun_rollouts。

EnvironmentRunResult提供summary()（返回带固定键的 CI 契约字典，包含env、k、n_attempts、pass_rate、env_fingerprint、policy_fingerprint等）和errors()（按任务 id 分组的错误信息）等能力。

learning_zone：按任务而不是按尝试筛选

learning_zone()的实现（libs/agno/agno/environments/runner.py）返回一个只保留学习区任务的EnvironmentRunResult副本：

def learning_zone(self) -> "EnvironmentRunResult": return replace( self, task_results=tuple(task_result for task_result in self.task_results if task_result.in_learning_zone), )

而"学习区"的判定标准非常明确（libs/agno/agno/environments/runner.py）：

@property def in_learning_zone(self) -> bool: return 0 < self.n_passed < self.n_scored

即：该任务有通过的评分尝试，也有失败的评分尝试——任务既没有饱和（全部通过）也没有完全无望（全部失败），因此每一次失败都能找到一次通过尝试作为对照。未评分尝试（如超时）不参与统计，也绝不会被当作 0 分强算进通过率。由于保留的是同一个结果对象的副本（指纹一致），summary()、网格视图和导出器都能直接作用于学习区子集。

to_sft_jsonl：把通过的对话写成可消费的 JSONL

to_sft_jsonl定义在 libs/agno/agno/environments/exporters/sft.py：

def to_sft_jsonl( result: EnvironmentRunResult, path: Union[str, Path], *, only_passed: bool = True, ) -> ExportReport:

关键设计点：

1. only_passed=True是默认值。因为learning_zone()选中的任务天然"既有通过又有失败"，如果不按尝试过滤，就会把错误答案写进监督数据集。源码注释说得非常直白："learning_zone()selects tasks;only_passedselects attempts within them: SFT wants both."——两者缺一不可。
2. 输出格式为对话式 JSONL：每行一个 JSON 对象{"messages": [{"role", "content"}, ...]}，角色仅允许system/user/assistant，content必须是非空字符串，且至少包含一条 user 消息、最后一条必须是 assistant 消息（见 libs/agno/agno/environments/exporters/_validate.py）。
3. 容量上限严格对齐最严格的消费端：最多 320 条对话、单文件 1 MiB（utf-8 编码后）——上限来自消费端的BATCH_SIZE (8) * MAX_STEPS (40)。超限的行按发射顺序从尾部丢弃并计入n_dropped_over_cap，绝不静默截断。
4. 系统消息会被保留：它承载了诱导出该输出的格式指令；assistant 文本则取模型实际生成的Message.content原文（逐字节一致），绝不是对run.content的序列化——在output_schema场景下 pydantic 模型的str()/model_dump_json()会产生模型从未生成过的文本。
5. 发射顺序确定：按任务顺序、任务内按尝试顺序输出，并发场景下也保持确定性。
6. 文件写入使用newline=""：禁用平台换行符转换，避免 Windows 上 CRLF 转换悄悄把一个刚好达到字节上限的文件推过 1 MiB 上限，破坏基于 sha256 的确定性。

导出器会同步写出 sidecar 文件<path>.meta.json，其中包含env_fingerprint、policy_fingerprint、report（各计数）、options（如only_passed）以及lines（每条已写对话的task_id、attempt_index、score）——这就是后续_11_export_provenance/要分析的溯源数据。

ExportReport：每个候选去向都有计数

to_sft_jsonl返回的ExportReport（libs/agno/agno/environments/exporters/sft.py）包含六个计数，且各计数之和等于候选总数（仅指已评分尝试；未评分尝试在任一only_passed取值下都不算候选）：

_classify（libs/agno/agno/environments/exporters/sft.py）按固定顺序执行跳过检查，命中即计数并跳过——这些规则在构造上就有重叠（例如触发工具上限的尝试同时也是工具型运行），不固定顺序就会重复计数。值得注意的两条规则：

• 工具调用上限命中（tool_call_limit_hit）会被跳过："拒绝工具调用之后才给出的答案，是迫于压力下的答案"；
• 工具型运行（run.tools非空）会被跳过：交集格式没有工具表示法，只输出最终答案而不输出产生它的工具轨迹，会训练模型"不用它实际使用的工具也能作答"。

三个示例脚本逐一遍历

basic.py：最小完整闭环

basic.py 演示了完整的"跑 → 筛 → 导出"链路：

• 两个任务product-a、product-b均为需要先做大数乘法再按素数（65521 / 65537）取模计算的复合算术题，输出通过Answer（value: int）结构强制为整数；
• 评分器使用CodeScorer(exact_value)，其中exact_value(run, expected)检查run.content.value == expected；
• run_rollouts(env, k=4)跑 4 轮，result.learning_zone()取学习区；
• 若学习区为空则打印提示并中止；否则to_sft_jsonl(zone, output_path)导出到data/generated/train.jsonl，并打印n_written（写出的通过对话数）与n_skipped_failed（跳过的失败尝试数）。

passed_only.py：验证失败尝试被排除

passed_only.py 专门验证"学习区选中任务之后，默认导出器会把任务内失败的尝试剔除"：

• 使用更难的任务组合（product-a与product-d，模数提升到 524287 / 99991），k=6；
• 导出后直接读取生成的 JSONL 文件，逐行json.loads解析，并断言report.n_written == n_passed == len(rows)三者相等——即写出的行数、学习区内的通过尝试数、文件中实际行数完全一致；
• 同时打印n_skipped_failed展示被剔除的失败数。

empty_zone_guard.py：空学习区的防御

empty_zone_guard.py 处理一个真实且常见的场景：空选择是正常结果，必须在下游把空文件误认为"数据集生成完成"之前检查它。

脚本的防御策略：

1. 首先删除可能存在的陈旧产物（output_path与 sidecar 路径上的.meta.json），保证"无操作导出绝不在对外承诺的路径上留下上次运行的数据集"；
2. 任务组合中刻意加入easy-anchor（17 × 23 = 391 的简单整数题）作为"非学习区锚点"，与两道难题混跑；
3. 真实运行后若学习区非空则正常导出；若为空则断言output_path与 sidecar均不存在，防止空文件残留被误读为有效数据；
4. 最后用dataclasses.replace(result, task_results=())构造一个合成空选择来演练分支，并明确"导出器在此有意不被调用"。

测试日志解读：验证结果与校准过程

TEST_LOG.md 记录了 2026-07-20 使用OpenAIResponses(id="gpt-5.5", reasoning_effort="low")完成的三组验证：

• basic.py（PASS）：两个算术任务各跑 4 次，学习区两行均被选中，只导出通过的文本对话。product-a通过 1/4（0.25），product-b通过 3/4（0.75），导出器写出 4 条对话、跳过 4 次失败尝试。
• passed_only.py（PASS）：默认导出器在学习区任务内只保留通过尝试。product-a通过 4/6（0.67），product-d通过 5/6（0.83），JSONL 恰好包含 9 条通过尝试，3 条失败尝试被排除。日志还记录了校准过程：最初的任务组合（product-a、product-c、k=4）在两行上双双饱和到 4/4，于是用更难的product-d替换product-c并把 k 提到 6，才记录到这次 PASS——这正是学习区机制要求的"中间带难度"校准。
• empty_zone_guard.py（PASS）：清理陈旧产物后导出真实学习区选择，再演练显式空选择。最终实跑：easy-anchor通过 4/4（1.00，饱和非学习区）、product-a通过 1/4（0.25，学习区）、product-b通过 4/4（1.00，饱和非学习区）。真实选择写出 1 行，合成空选择被检测并跳过；并断言未来真实运行为空时，data/generated/guarded.jsonl及其.meta.json不会残留。

运行与前置条件

按 README 的指引，三个脚本依次执行即可：

python cookbook/environments/_10_export_sft/basic.py python cookbook/environments/_10_export_sft/passed_only.py python cookbook/environments/_10_export_sft/empty_zone_guard.py

前置条件：

• 需要OPENAI_API_KEY环境变量；所有示例都通过OpenAIResponses使用gpt-5.5，并在示例中采用reasoning_effort="low"以控制推理成本；
• 导出目录（data/generated/）由导出器自动创建（output_path.parent.mkdir(parents=True, exist_ok=True)）；
• 若在已有事件循环的异步环境中运行，请改用agno.environments提供的ato_sft_jsonl与arun_rollouts异步版本。

数据流总结与注意事项

从构建环境到拿到可消费数据集，完整数据流为：

Environment(Task×N, scorer) --run_rollouts(k)--> EnvironmentRunResult --learning_zone()--> 学习区任务子集 --to_sft_jsonl(only_passed=True)--> train.jsonl + train.jsonl.meta.json

几个必须记住的工程要点：

1. 导出≠训练：本模块只产出数据集文件，不触发任何训练流程；
2. 学习区选任务、only_passed选尝试：两者组合才是干净的监督数据；
3. 严格 schema：顶层只能有messages，消息只能有role和content，多一个未知键会导致最严格的消费端整文件拒绝（严格集合相等，不会丢弃多余键）——所以不要"好心"添加tools、weight、trainable之类的键；
4. 工具型与触发工具上限的尝试一律跳过，文本缺失、超限的尝试各有独立计数；
5. 空学习区是正常结果：在把 JSONL 当作"已生成数据集"交给下游之前，必须检查学习区是否为空，并防止陈旧产物残留误导后续步骤。

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