agno Environments 训练数据加载器（Trainer Loader）：在 SFT JSONL 与训练器之间构建可验证的交接边界

导读

本文围绕 agno 仓库 cookbook/environments/_12_trainer_loader/ 目录下的两个示例（basic.py 与 validate_messages.py）及其测试记录，讲解如何把强化学习环境（Environment）中滚动（rollout）产生的对话数据，导出为标准「对话式 SFT JSONL」并在训练器消费边界上完成加载与校验。读完本文，你将掌握 run_rollouts → learning_zone() → to_sft_jsonl 的完整导出链路、可移植 SFT 行的精确格式约束、ExportReport 各计数器的语义，以及如何用校验脚本在把数据交给任何训练器适配器之前拦截非法行。

背景：数据从环境滚动到训练系统的必经之路

在 agno 的 environments 模块中，一次典型的训练数据生产流程由 runner.py 中的 run_rollouts(env, *, k=8, tasks=None, model=None, concurrency=4) 驱动：它接受一个 Environment（由 agent、tasks、scorer 组成，定义于 environment.py），对每个任务执行 k 次尝试，再交给 scorer 判定通过/失败。

但「生成了包含对话的文件」并不等于「可以开始训练」。_12_trainer_loader 这一节正是把持住加载器边界（loader boundary）：它只负责读取、验证、加载消息数组，绝不启动任何训练作业。这一设计意图在该目录 README.md 中写得非常明确——"These examples validate and load message rows only; they do not start a training job."，对应的测试记录 TEST_LOG.md 也逐条给出了两个示例的 PASS 状态与实测结果。

训练器加载器的定位：用在哪、为什么这么用

按 README.md 的指引，这一组示例的使用时机是：

• 前置依赖：先运行 _11_export_provenance/ 完成带溯源（provenance）的导出，再进入本节，把生成的文件接到独立的训练系统上；
• 刻意忽略溯源边车：加载器只消费文本 JSONL，故意忽略 <path>.meta.json 溯源边车（sidecar）；边车文件应归档用于审计，而训练器只读取文本数据；
• 后续流程：如果要做逐轮回归验证（在生成下一批数据集前确认环境配置没有漂移），继续前往 _13_saved_baselines/。

两个示例的分工：

文件	职责
basic.py	导出通过的学习区（learning zone）尝试，然后加载其 messages 数组
validate_messages.py	在把行交给任何训练器特定适配器之前，强制校验可移植行结构（row shape）与允许的角色

basic.py：导出通过样本并加载消息数组

bASIC.py 的完整流程如下：

1. 定义结构化输出与计分器：用 pydantic 定义 Answer(BaseModel)（value: int），并用 CodeScorer(exact_value) 包装一个 (run, expected) -> bool 的精确值比较函数。CodeScorer 定义于 code.py，接受任意 (run, expected) -> bool | float | Score 形式的可调用对象。
2. 构造环境：Environment(name="trainer-loader-basic", agent=agent, tasks=(...), scorer=CodeScorer(exact_value))，其中两个任务是纯算术推演题（大数相乘 + 数字和 + 模运算），expected 分别为 20944939 与 76998482。
3. 执行滚动：run_rollouts(env, k=6)，即每个任务滚动 6 次。
4. 取出学习区：result.learning_zone() 是 runner.py 提供的过滤器，只保留「既有通过尝试又有失败尝试」的任务——这是 SFT 数据生产的核心素材。
5. 导出并加载：to_sft_jsonl(zone, output_path) 写入 JSONL，随后 load_message_rows() 逐行 json.loads(line)["messages"] 取出消息数组，并用断言 len(message_rows) == report.n_written 保证「加载器收到的消息数组数 = 实际写入行数」，最后打印 loader received N message arrays 并声明停在加载器边界、未发生训练。

validate_messages.py：可移植行的结构校验

validate_messages.py 的核心是 validate_sft_rows(path) 函数，它对每一行（一个 {"messages": [...]} 对象）执行以下断言：

• 顶层键严格等于 {"messages"}（set(row) == {"messages"}）；
• messages 非空；
• 至少存在一条 role == "user" 的消息；
• 最后一条消息必须是 assistant（row["messages"][-1]["role"] == "assistant"）；
• 每条消息的键严格等于 {"role", "content"}；
• 角色必须是 system / user / assistant 三者之一；
• content 必须是非空字符串（isinstance(content, str) and content.strip()）。

该脚本的其余骨架与 basic.py 一致（同样的 Answer、CodeScorer、run_rollouts(env, k=4) 与 to_sft_jsonl），只是任务换成了 product-a 与 product-c。这种「先断言、后交给适配器」的模式，正是 README.md 所说"enforce the portable row shape and allowed roles before handing rows to any trainer-specific adapter"。

可移植 SFT JSONL 格式：为什么「少即是多」

导出器实现位于 sft.py，其文档字符串阐述了格式设计的关键哲学：

Tinker、Together、Fireworks 与 OpenAI 都接受 {"messages": [{"role", "content"}]} 这个核心结构。它们在两个轴上分叉——工具表示（tool representation）与损失加权（loss weighting）——而这个导出器两者都不输出，因此文件是通过「省略」而非「翻译」实现可移植的。

这意味着：

• 禁止添加多余键：不要「好心」地加上 tools、weight 或 trainable 键。最严格的消费者做的是集合严格相等检查（不是丢弃未知键），任何一个多余键都会让整个文件被拒收——一个键的失误会让所有下游消费者同时失败。
• 溯源放在边车文件：分数与指纹（fingerprint）无处可放，因此写入 <path>.meta.json 边车。边车含 env_fingerprint、policy_fingerprint、report（各跳过计数器）、options（only_passed）与 lines（逐行 task_id / attempt_index / score 溯源列表）。

消息构建规则

_conversation_from() 决定一条候选对话是否可导出，规则如下：

• 消息来源是 run.messages，但剔除 from_history=True 的历史消息；
• 保留角色限于 {system, user, assistant}，且内容为非空字符串；
• system 消息会被保留——它携带诱导模型输出格式的指令；
• assistant 文本取最终一条 assistant 消息的原始 Message.content 字符串，逐字节忠实于模型输出，绝不序列化 run.content（在 output_schema 下它是 pydantic 模型，其 str() / model_dump_json() 都会产生模型从未生成过的文本）；
• 对话必须至少包含一条 user 消息，否则返回 None（不可导出）。

分类与跳过逻辑（_classify）

每条候选按顺序应用跳过检查，首个命中即生效，且每个计数器只计一次：

1. 未评分（score is None）：不算候选，永远到不了导出器；
2. 失败（only_passed=True 时）：计 n_skipped_failed；
3. 工具调用超限（tool_call_limit_hit）：计 n_skipped_limit_hit——「在拒绝工具调用之后得出的答案」被视为受胁迫的答案；
4. 使用了工具（run.tools 非空）：计 n_skipped_tool_runs——交集格式没有工具表示，只导出最终答案而不导出产生它的工具轨迹，会教会模型「不用工具也能答」；
5. 无可导出文本：计 n_skipped_no_text。

ExportReport 的全部字段为 n_written、n_skipped_failed、n_skipped_tool_runs、n_skipped_limit_hit、n_skipped_no_text、n_dropped_over_cap。

上限与确定性

严格消费者的硬性上限定义在 _validate.py：

• MAX_CONVERSATIONS = 320（对应消费端 BATCH_SIZE(8) * MAX_STEPS(40)）；
• MAX_DATASET_BYTES = 1024 * 1024（1 MiB，UTF-8 编码后字节数）；
• 超限行按发射顺序从尾部丢弃并计数，绝不静默截断；
• 发射顺序为「任务顺序、任务内尝试顺序」，在任何并发下都确定；
• 写入时 newline="" 关闭平台换行翻译——在 Windows 上若被翻译成 CRLF，会让恰好贴满字节上限的文件超限，破坏 sha256 固定的确定性。

官方校验器 validate_sft_jsonl

_validate.py 还提供一个 validate_sft_jsonl(path) -> int 函数，它严格复刻了最严苛消费端的验收逻辑（vendored Tinker acceptance check）：文件 ≤ 1 MiB、非空、按 "\n" 切分（splitlines() 会额外切分 U+2028/U+2029/U+0085，而 json.dumps(ensure_ascii=False) 会原样发射这些字符，故必须用 split("\n")）、行数 ≤ 320、每行是合法 JSON、顶层键严格等于 {"messages"}、消息键严格等于 {"role", "content"}、角色白名单、内容非空字符串、至少一条 user、末条为 assistant。返回对话数，遇到首个违规即抛 ValueError。validate_messages.py 中的断言集正是这一验收规则的 Python 直译。

测试记录解读：实测结果与校准过程

TEST_LOG.md 记录了 2026-07-20 使用 OpenAIResponses(id="gpt-5.5", reasoning_effort="low") 实测的结果：

basic.py — PASS

• 行为：导出通过的学习区尝试，只加载其消息数组，在任何训练操作之前停止；
• 结果：product-a 通过 3/6（0.50），product-d 通过 6/6（1.00），加载器收到 3 个消息数组，未启动任何训练；
• 校准记录：最初的任务集是 product-a + product-b、k=4，但两行都在 4/4 上饱和（全通过，学习区取不到「有通过有失败」的任务，导出为空）。因此把 product-b 换成 product-d，并把 k 提到 6，才记录下这次 PASS。这印证了 learning_zone() 的构造性质——它只保留同时含通过与失败尝试的任务，若某任务全通过则无法进入学习区。

validate_messages.py — PASS

• 行为：校验可移植 SFT 行的顶层键、允许角色、user 与最终 assistant 的存在性、非空文本内容；
• 结果：product-a 通过 1/4（0.25），product-c 通过 4/4（1.00）。唯一导出的通过行满足全部加载器形状断言，未启动训练。

这两条记录还给出了一条实用的任务难度校准经验：当单次滚动全部通过时，学习区会空转；应通过替换任务或调整 k 值，让任务集落在「部分通过、部分失败」的区间，才能生产出有区分度的 SFT 训练数据。

运行方式与前置条件

按 README.md，运行命令为：

python cookbook/environments/_12_trainer_loader/basic.py
python cookbook/environments/_12_trainer_loader/validate_messages.py

• 需要 OPENAI_API_KEY 环境变量；
• 所有示例均通过 OpenAIResponses 使用 gpt-5.5（reasoning_effort="low"）；
• 输出文件写入各自脚本同目录下的 data/generated/（trainer_input.jsonl 与 validated.jsonl），并附带对应的 .meta.json 溯源边车。

关键结论速查

1. 边界意识：生成 SFT JSONL ≠ 发起训练；加载器示例刻意止步于「加载消息数组」。
2. 可移植格式：{"messages": [{"role": "system|user|assistant", "content": "非空字符串"}]}，顶层与消息级均严格集合相等，禁止任何多余键。
3. 跳过语义：失败、工具超限、带工具运行、无文本分别计入 ExportReport 的不同计数器，超上限的行从尾部丢弃并计数。
4. 溯源分离：分数与指纹存于 .meta.json 边车，供审计使用；训练器只消费文本 JSONL。
5. 学习区特性：learning_zone() 只保留有通过也有失败的任务，任务全通过时导出为空——需要校准任务难度或 k 值。
6. 校验闭环：validate_messages.py 的断言与 _validate.py 的 validate_sft_jsonl 完全对齐，是交给任何训练器适配器前的最后一道闸门。

如需在生成下一批数据集之前做逐轮回归验证，可继续参考 _13_saved_baselines/ 的示例。