第 24 章：把 Harbor 环境接入强化学习
训练任务刚启动，监控面板就出现了一个令人振奋的现象：平均 Reward 持续上升。几小时后，独立评测却没有改善。进一步排查发现，训练侧把缺失 Reward 当成了零，把验证集任务也送进了 rollout 队列；更糟的是，部分样本只有 completion token，没有与之同源的 prompt token 和旧策略 logprob。优化器确实在更新，但它优化的并不是团队以为的数据契约。
本章解决的是 rollout 系统与 Harbor 之间的工程接缝。Harbor 负责把 Task 解析成隔离的 Environment，运行 Agent，调用 Verifier，并把 token、轨迹、Reward 和异常写进 Trial/Job 结果；训练系统负责组批、优势估计、损失、反向传播、优化器和权重发布。中间的 Adapter 必须拒绝不完整样本，而不是为了“不断流”猜测默认值。
本章继续锁定 Harbor v0.18.0、提交 527d50deb63a5d279e8c20593c18a2cbc7f61f9e。示例只运行配置模型和确定性契约测试；没有启动 Docker、没有调用模型，也没有执行梯度更新或声称任何训练收益。
读完本章，你应该能够：
- 解释 Environment、Task、Trial、Job、Rollout 与 Reward 在 RL 流水线中的位置；
- 用
Job.create()/Job.run()实现一个批量 rollout Adapter； - 验证 token、logprob、Reward、异常和训练/评测 split 的正负向契约；
- 处理稀疏 Reward、reward gaming、旧数据复用和独立评测；
- 明确 Harbor v0.18.0 与外部训练框架各自负责什么。
24.1 先画边界：Harbor 不是优化器
对 Agentic RL，一次 episode 通常不是一次聊天补全，而是 Agent 在一个任务环境中的完整尝试：读取指令、进行若干模型调用、执行工具、改变环境状态，最后由 Verifier 给出 Reward。Harbor 的自然执行单位正是 Trial。单步 Trial 按“Agent 执行—Artifact 收集—Verifier”顺序运行；Job 则把 Task、Agent 和 attempt 展开为多个 Trial，并用队列控制并发。12
可以把职责边界写成下面这条数据流：
训练系统
│ 选择策略权重、任务组、每题采样数
▼
Harbor Adapter
│ JobConfig(tasks, agent, environment, n_attempts, concurrency)
▼
Harbor Job -> Trial -> Environment <-> Agent
│
└-> Verifier -> Reward
▲
│ TrialResult(agent_result, verifier_result, exception_info)
▼
Adapter 契约门禁
│ prompt/completion token、old logprob、Reward、来源、策略版本
▼
训练系统
└-> advantage / loss / backward / optimizer / checkpoint
其中 Harbor v0.18.0 的 AgentContext 能保存输入、缓存和输出 token 计数、成本、rollout_details 与元数据；RolloutDetail 实际声明的是逐轮 prompt_token_ids、completion_token_ids、logprobs 和 provider-specific extra。它没有独立的 loss_masks 字段。34 因此，训练侧若需要 loss mask，应从“哪些 token 是策略生成的 completion”这一明确规则构造，并把规则版本化；不能声称 mask 是 Harbor 已经返回的字段。
Harbor 不负责以下工作：
- 计算 advantage、return 或 KL 项；
- 实现 PPO、GRPO 或其他训练目标；
- 切分 optimizer minibatch、梯度累积或参数更新；
- 同步训练权重到推理服务；
- 判断旧 rollout 是否仍满足某个算法的 on-policy/off-policy 条件；
- 替团队决定哪些 Task 可以用于训练、验证或最终测试。
这不是能力缺失，而是接口边界。PPO 的原始描述本身就是在“与环境交互采样”和“对替代目标做多轮 minibatch 优化”之间交替；Harbor适合承担前半段的可复现环境执行，后半段仍属于训练系统。5
24.2 以锁定源码为准，而不是照抄浮动示例
Harbor v0.18.0 随附的 RL 文档建议用 TaskConfig 作为 rollout batch 对象，并用 TrialConfig 或 JobConfig 配置运行。这个方向仍然成立；但同页示例导入了该版本不存在的 OrchestratorConfig，还直接调用已被弃用的 Job(config=...) 构造方式。67 v0.18.0 的真实字段是 JobConfig.n_concurrent_trials，正确入口是先 await Job.create(config)，再 await job.run()。
锁定的 Harbor Cookbook 还包含一个名为 harbor_rl 的示例，它明确依赖 harbor.rl 的功能分支/PR；而 v0.18.0 安装包没有 harbor.rl 模块。8 所以本章不使用其中的 RLEnvironment.step() 接口，也不把它写成 v0.18.0 的稳定能力。下面这段探针是在锁定源码环境中的真实核验方法：
cd /path/to/harbor-v0.18.0
uv run --frozen python - <<'PY'
from importlib.util import find_spec
from harbor.models.job.config import JobConfig
print("top_level_concurrency=", "n_concurrent_trials" in JobConfig.model_fields)
try:
from harbor.models.job.config import OrchestratorConfig
except ImportError:
print("orchestrator_config_importable=False")
print("harbor_rl_spec=", find_spec("harbor.rl"))
PY
本章复验得到：顶层并发字段存在，OrchestratorConfig 不可导入，find_spec("harbor.rl") 返回 None。这只是 API 存在性探针，不运行 Environment。
还有一个容易混淆的重名：harbor.models.trial.config.TaskConfig 是 Job/Trial 使用的任务引用，可指向本地路径、Git 位置或包；harbor.models.task.config.TaskConfig 则是任务目录中 task.toml 的任务内容模型。前者进入 Job，后者在 Task 下载和解析后决定 Environment、Verifier、Step 等行为。910 本章 Adapter 导入的是前者。
24.2.1 Environment 是生命周期，不只是一个容器地址
训练框架常把 Environment 抽象成 reset()/step()；在本书基线里，更准确的理解是“由 Trial 管理的一组有生命周期的执行能力”。Trial 创建时加载 Task、初始化 Agent 与 Environment，准备阶段启动环境、运行健康检查、上传 Skill 并安装 Agent；执行结束或异常恢复后停止环境、持久化结果并发出结束事件。Agent 调用拿到的是 BaseEnvironment，可以在其中执行命令和传输文件，但这些动作仍处于 Trial 的超时、用户、网络策略和日志上下文中。11
这一区别影响 RL Adapter 的恢复设计。训练 worker 进程崩溃后，不能只靠一个“环境 ID”假定 episode 可继续；要先检查 Job 目录中的 config.json、lock.json、Trial 目录和最终 result.json，再让 Harbor 按相同配置判断续跑。相反，训练系统自己的 optimizer step、梯度缓存和权重版本不在 Trial 生命周期内，必须由训练框架单独做 checkpoint。
Task 中的 [environment] 描述任务所需的 OS、资源、健康检查、网络基线和工作目录；Job/Trial 的 EnvironmentConfig 选择实际 provider，并可施加资源覆盖、挂载、环境变量和删除策略。Trial 创建 Environment 时会同时传入运行配置与 Task Environment 配置。12 由此可以推断：同一个任务在不同 provider 上运行时，任务语义应保持不变，但启动延迟、挂载能力和网络策略切换能力可能不同。上线前应为实际 provider 跑契约测试，不能因为 Docker 路径通过就把云端后端视为等价实测。
24.3 把 Rollout 写成可拒绝的数据契约
Terminus-2 在 collect_rollout_details=True 时会要求底层 LLM 收集 token ID 与 logprob，并在 Agent 结束时把主对话和 subagent 段放进 AgentContext.rollout_details；该选项默认关闭，而且源码明确提示上下文 summarization 会使 rollout detail 不完整。13 LiteLLM 路径还可能遇到 provider 拒绝 return_token_ids：v0.18.0 会去掉被拒绝的参数后重试，并警告 token ID 不再可用。14
因此，“设置了开关”不等于“样本可训练”。最小契约至少应包含：
| 字段 | 正向条件 | 负向处理 |
|---|---|---|
| Trial 状态 | exception_info is None | 拒绝；不得改写为 Reward 0 |
| Reward | 指定 key 存在、为数值且有限 | 拒绝缺失、布尔值、NaN、Inf |
| rollout 段 | 当前 Adapter 明确支持一种拓扑 | 遇到多个段或 multi-step 时拒绝并另写适配器 |
| prompt token | 每轮非空整数序列 | 不用本地 tokenizer 猜回 provider token |
| completion token | 与 prompt 轮数一致 | 不拼接不等长轮次 |
| old logprob | 每个 completion token 一个有限值 | 不用 0 填充 |
| split | rollout 入口只接受 train | eval/test 立即失败 |
| 策略身份 | checkpoint、tokenizer、renderer、采样参数可追溯 | 不把仅有 model_name 当权重证明 |
这里的 old logprob 是本章面向 PPO/重要性比率训练的 Adapter 契约，不是所有 RL 算法的普遍必选字段。若训练目标不消费旧策略 logprob，应另行定义并测试它所需的最小字段；不能为了复用本章代码伪造全零 logprob，也不能反过来声称 Harbor 要求所有训练框架都使用 PPO 式数据。
Harbor Verifier 优先解析 reward.json，否则解析 reward.txt；没有文件、空文件或解析失败会产生不同异常，而不是一个合法的零分。15 Adapter 也应保持这一区别。零分表示 Verifier 成功执行并判为零；缺失 Reward 表示数据生产失败。
本章还采取一个保守限制：训练 rollout 只接受一个 RolloutDetail。这是教学 Adapter 的契约，不是 Harbor 的全局限制。Terminus-2 可能把主对话、summarization 或 subagent 信息表达为多个段；若训练系统要支持这些拓扑，就必须定义策略归属、段顺序、重复上下文和 loss mask，而不是盲目 flatten。
24.4 最小可测试 Adapter
在配套项目根目录创建 rl_adapter.py。RLTask 把 split 与 Harbor 的任务引用绑定；generate() 只能消费训练任务并返回 token 级 rollout，evaluate() 只能消费评测任务且只返回 Reward 记录。run_id 由调用方提供，避免一次迭代意外续跑另一次 Job。
from __future__ import annotations
from collections.abc import Awaitable, Callable, Sequence
from dataclasses import dataclass
from math import isfinite
from pathlib import Path
from typing import Any, Literal
from harbor.job import Job
from harbor.models.agent.name import AgentName
from harbor.models.environment_type import EnvironmentType
from harbor.models.job.config import JobConfig
from harbor.models.trial.config import AgentConfig, EnvironmentConfig, TaskConfig
Phase = Literal["train", "eval"]
RunJob = Callable[[JobConfig], Awaitable[Any]]
class ContractError(ValueError):
pass
@dataclass(frozen=True)
class RLTask:
key: str
split: Phase
config: TaskConfig
@dataclass(frozen=True)
class PolicyRollout:
trial_name: str
task_name: str
prompt_token_ids: tuple[tuple[int, ...], ...]
completion_token_ids: tuple[tuple[int, ...], ...]
old_logprobs: tuple[tuple[float, ...], ...]
reward: float
@dataclass(frozen=True)
class EvaluationRecord:
trial_name: str
task_name: str
reward: float
async def run_harbor_job(config: JobConfig) -> Any:
job = await Job.create(config)
return await job.run()
class HarborRLAdapter:
def __init__(
self,
*,
jobs_dir: Path,
model_name: str,
environment_type: EnvironmentType = EnvironmentType.DOCKER,
reward_key: str = "reward",
run_job: RunJob | None = None,
) -> None:
self.jobs_dir = jobs_dir
self.model_name = model_name
self.environment_type = environment_type
self.reward_key = reward_key
self._run_job = run_job or run_harbor_job
def build_config(
self,
specs: Sequence[RLTask],
*,
phase: Phase,
run_id: str,
attempts: int,
concurrency: int,
) -> JobConfig:
if not specs:
raise ContractError("empty task batch")
if not run_id or Path(run_id).name != run_id or run_id in {".", ".."}:
raise ContractError("run_id must be one path component")
if attempts < 1 or concurrency < 1:
raise ContractError("attempts and concurrency must be positive")
if len({spec.key for spec in specs}) != len(specs):
raise ContractError("duplicate task key; use attempts for repeated sampling")
leaked = [spec.key for spec in specs if spec.split != phase]
if leaked:
raise ContractError(f"{phase} batch contains wrong split: {leaked}")
return JobConfig(
job_name=run_id,
jobs_dir=self.jobs_dir,
n_attempts=attempts,
n_concurrent_trials=concurrency,
quiet=True,
environment=EnvironmentConfig(type=self.environment_type),
agents=[
AgentConfig(
name=AgentName.TERMINUS_2.value,
model_name=self.model_name,
kwargs={
"collect_rollout_details": phase == "train",
"enable_summarize": False,
},
)
],
tasks=[spec.config for spec in specs],
)
async def generate(
self,
specs: Sequence[RLTask],
*,
run_id: str,
attempts: int,
concurrency: int,
) -> list[PolicyRollout]:
config = self.build_config(
specs,
phase="train",
run_id=run_id,
attempts=attempts,
concurrency=concurrency,
)
result = await self._run_job(config)
self._check_count(result, len(specs) * attempts)
return [self.to_policy_rollout(trial) for trial in result.trial_results]
async def evaluate(
self,
specs: Sequence[RLTask],
*,
run_id: str,
attempts: int,
concurrency: int,
) -> list[EvaluationRecord]:
config = self.build_config(
specs,
phase="eval",
run_id=run_id,
attempts=attempts,
concurrency=concurrency,
)
result = await self._run_job(config)
self._check_count(result, len(specs) * attempts)
return [
EvaluationRecord(
trial_name=trial.trial_name,
task_name=trial.task_name,
reward=self._reward(trial),
)
for trial in result.trial_results
]
def _check_count(self, result: Any, expected: int) -> None:
actual_total = getattr(result, "n_total_trials", None)
trials = getattr(result, "trial_results", None)
if actual_total != expected or not isinstance(trials, list):
raise ContractError(
f"incomplete job result: expected={expected}, total={actual_total}"
)
if len(trials) != expected:
raise ContractError(
f"incomplete trial list: expected={expected}, actual={len(trials)}"
)
def _reward(self, trial: Any) -> float:
if getattr(trial, "exception_info", None) is not None:
raise ContractError(f"trial {trial.trial_name} has an exception")
verifier = getattr(trial, "verifier_result", None)
rewards = getattr(verifier, "rewards", None)
if not isinstance(rewards, dict) or self.reward_key not in rewards:
raise ContractError(f"trial {trial.trial_name} has no {self.reward_key!r}")
value = rewards[self.reward_key]
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise ContractError(f"trial {trial.trial_name} reward is not numeric")
value = float(value)
if not isfinite(value):
raise ContractError(f"trial {trial.trial_name} reward is not finite")
return value
def to_policy_rollout(self, trial: Any) -> PolicyRollout:
reward = self._reward(trial)
if getattr(trial, "step_results", None):
raise ContractError("multi-step trial needs a step-aware rollout adapter")
agent = getattr(trial, "agent_result", None)
details = getattr(agent, "rollout_details", None)
if not isinstance(details, list) or len(details) != 1:
raise ContractError("expected exactly one rollout detail")
detail = details[0]
if not isinstance(detail, dict):
raise ContractError("rollout detail must be a mapping")
prompts = self._nested_ints(detail.get("prompt_token_ids"), "prompt")
completions = self._nested_ints(
detail.get("completion_token_ids"), "completion"
)
logprobs = self._nested_floats(detail.get("logprobs"), "logprobs")
if not (len(prompts) == len(completions) == len(logprobs)):
raise ContractError("turn counts differ")
if any(len(tokens) != len(logps) for tokens, logps in zip(completions, logprobs)):
raise ContractError("completion/logprob lengths differ")
return PolicyRollout(
trial_name=trial.trial_name,
task_name=trial.task_name,
prompt_token_ids=prompts,
completion_token_ids=completions,
old_logprobs=logprobs,
reward=reward,
)
@staticmethod
def _nested_ints(value: Any, label: str) -> tuple[tuple[int, ...], ...]:
if not isinstance(value, list) or not value:
raise ContractError(f"missing {label} token rows")
rows: list[tuple[int, ...]] = []
for row in value:
if (
not isinstance(row, list)
or not row
or any(isinstance(item, bool) or not isinstance(item, int) for item in row)
):
raise ContractError(f"invalid {label} token row")
rows.append(tuple(row))
return tuple(rows)
@staticmethod
def _nested_floats(value: Any, label: str) -> tuple[tuple[float, ...], ...]:
if not isinstance(value, list) or not value:
raise ContractError(f"missing {label} rows")
rows: list[tuple[float, ...]] = []
for row in value:
if not isinstance(row, list) or not row:
raise ContractError(f"invalid {label} row")
converted: list[float] = []
for item in row:
if isinstance(item, bool) or not isinstance(item, (int, float)):
raise ContractError(f"invalid {label} value")
item = float(item)
if not isfinite(item):
raise ContractError(f"non-finite {label} value")
converted.append(item)
rows.append(tuple(converted))
return tuple(rows)
生产接入时，model_name 应指向当前采样服务支持的模型身份，且还要在 Adapter 外部清单记录精确 checkpoint digest、tokenizer/renderer 版本、温度、top-p、随机种子策略和训练迭代号。Harbor 的 lock.json 会记录 Harbor 信息、Task digest、Agent/Environment/Verifier 配置与并发重试等重放输入；它并不是训练 checkpoint 或 optimizer 状态清单。16
enable_summarize=False 是为了让这个最小 Adapter 保持单段线性契约，不是通用最佳实践。它可能增加上下文超限风险；真实系统应在“禁用压缩并限制长度”和“实现可证明正确的多段 Adapter”之间明确选择。
24.5 用正负向测试验收 Adapter
在同目录创建 test_rl_adapter.py。这些对象是教学性构造的 Trial 形状，不模拟 Docker 或模型，只验证边界。
from pathlib import Path
from types import SimpleNamespace
import pytest
from harbor.models.trial.config import TaskConfig
from rl_adapter import ContractError, HarborRLAdapter, RLTask
def trial(name: str = "t1", reward: float = 0.75):
return SimpleNamespace(
trial_name=name,
task_name="diagnose-api",
exception_info=None,
step_results=None,
verifier_result=SimpleNamespace(rewards={"reward": reward}),
agent_result=SimpleNamespace(
rollout_details=[
{
"prompt_token_ids": [[10, 11]],
"completion_token_ids": [[20, 21]],
"logprobs": [[-0.2, -0.3]],
}
]
),
)
def adapter(**kwargs):
return HarborRLAdapter(
jobs_dir=Path("jobs/rl"),
model_name="hosted/model-placeholder",
**kwargs,
)
def spec(split="train"):
return RLTask(
key=f"diagnose-api:{split}",
split=split,
config=TaskConfig(path=Path(f"tasks/{split}/diagnose-api")),
)
def test_valid_rollout():
row = adapter().to_policy_rollout(trial())
assert row.reward == 0.75
assert row.completion_token_ids == ((20, 21),)
def test_eval_task_cannot_enter_training_batch():
with pytest.raises(ContractError, match="wrong split"):
adapter().build_config(
[spec("eval")], phase="train", run_id="iter-001", attempts=2, concurrency=2
)
@pytest.mark.parametrize("bad_reward", [float("nan"), float("inf"), True])
def test_bad_reward_is_rejected(bad_reward):
with pytest.raises(ContractError):
adapter().to_policy_rollout(trial(reward=bad_reward))
def test_missing_reward_is_not_zero():
value = trial()
value.verifier_result.rewards = {}
with pytest.raises(ContractError, match="has no"):
adapter().to_policy_rollout(value)
def test_token_logprob_mismatch_is_rejected():
value = trial()
value.agent_result.rollout_details[0]["logprobs"] = [[-0.2]]
with pytest.raises(ContractError, match="lengths differ"):
adapter().to_policy_rollout(value)
def test_multi_step_needs_an_explicit_adapter():
value = trial()
value.step_results = [SimpleNamespace()]
with pytest.raises(ContractError, match="step-aware"):
adapter().to_policy_rollout(value)
@pytest.mark.asyncio
async def test_generate_builds_expected_harbor_batch():
seen = []
async def fake_run(config):
seen.append(config)
return SimpleNamespace(
n_total_trials=2,
trial_results=[trial("t1"), trial("t2")],
)
rows = await adapter(run_job=fake_run).generate(
[spec("train")], run_id="iter-001", attempts=2, concurrency=1
)
assert len(rows) == 2
assert seen[0].n_attempts == 2
assert seen[0].n_concurrent_trials == 1
assert seen[0].agents[0].kwargs["collect_rollout_details"] is True
从锁定 Harbor 源码环境安装项目依赖后执行：
uv run pytest -q test_rl_adapter.py
本章把正文两个 Python 代码块提取到临时目录后实际执行，结果为 9 passed；测试没有创建 Harbor Job 目录，因为注入的是 fake_run。此外，锁定源码中 Reward、token、Job lock、并发和 task TOML 的 125 个定向测试也全部通过。完整命令和未执行边界记录在本章写作报告中。
警告：示例的
hosted/model-placeholder不能直接运行。不要为通过配置测试而把真实 API Key 写进代码或 Job 配置文件；运行时通过受控环境变量注入，并检查持久化结果是否已脱敏。
24.6 三种“批量”不能混为一谈
RL 系统里至少有三种批量：
- 任务批（task batch）：本轮选择哪些
RLTask； - 每题采样组（rollout group）：
n_attempts让同一个 Task/Agent 产生多少次独立 Trial； - 优化器批（optimizer minibatch）：训练系统如何切 token、做 padding、累积梯度。
Harbor 的 Job 展开式是 attempt × task × agent。示例固定一个 Agent，所以期望 Trial 数是 len(specs) × attempts；Adapter 对 n_total_trials 和最终列表长度做双重校验。n_concurrent_trials 只是同时执行多少个 Trial 的上限，不是训练 batch size，也不保证 Trial 按输入顺序完成。172
并发调大前应分别测量四类瓶颈：Environment 创建速率、模型服务限流、Verifier 吞吐和 Artifact/日志 I/O。仅把 n_concurrent_trials 改大，可能让模型服务重试、云沙箱配额或本机磁盘先成为瓶颈。Harbor 的 AgentConfig.n_concurrent 还能为单个 Agent 或共享 concurrency group 设置 Job 总并发之下的子上限；它不能超过 n_concurrent_trials。18
token 也不能只看 n_output_tokens 总数。RolloutDetail.prompt_token_ids 的每一行是该轮完整 prompt（可能含此前历史），而 completion 才是该轮策略生成部分。直接把所有 prompt 行首尾拼接，会重复历史；正确的序列化取决于训练框架的 renderer 和多轮格式。Adapter 因此保留逐轮嵌套结构，让训练侧显式完成 flatten、mask 和长度裁剪。
长度门禁至少应记录：完整序列 token 数、可训练 completion token 数、被截断原因和策略。超长样本不能静默截断 old logprob 数组；要么在 rollout 前限制最大轮数/输出长度，要么让训练侧对 token、mask、logprob 同步裁剪并写出规则。
24.6.1 n_attempts 是采样，retry 是故障恢复
n_attempts=K 会在 Job 展开阶段创建 K 份不同 Trial 配置，是训练者主动请求的 K 次采样。RetryConfig.max_retries 则由 TrialQueue 在某个 Trial 返回异常后决定是否用同一 TrialConfig 重试；成功或达到上限后，队列只返回最终那份 TrialResult。默认排除列表包括 Agent/Verifier 超时、Reward 文件缺失/为空/解析失败和 API 用量限制等异常。19
二者不能一起计入 rollout group size。假设一题配置四个 attempts，其中一次因环境创建错误被重试两次，训练 batch 仍应最多得到四条最终合法 rollout，而不是六条。被重试删除的中间目录属于故障证据；如需分析基础设施稳定性，应从 Job 的 retry 统计和外部日志取数，不能把失败尝试当作额外策略探索。反过来，若希望同一策略对同一任务多采样，应该增加 n_attempts，而不是故意制造可重试异常。
在 RL 生产中还应审查默认 retry 策略是否符合成本和数据语义。例如 provider 短暂连接错误可以重试，但模型已经生成完 token、只在结果回传时断线的场景可能造成重复计费和不可见的重复 episode。稳妥做法是给每次外部模型请求使用可追溯 request/session ID，在训练清单中记录重试计数，并让 Adapter 只接收 Harbor 最终返回的 TrialResult。
24.7 稀疏 Reward：先改善证据，不要伪造密集反馈
很多 Harbor Task 只有 episode 末尾的 Verifier Reward。这是典型的稀疏反馈：失败 Trial 告诉我们最终目标未满足，却不直接指出哪次工具调用造成失败。应对方式不是把“命令退出 0”“写过某个文件”随意改成正 Reward；一旦代理指标与真实目标有偏差，优化会放大这个偏差。
更稳妥的工程顺序是：
- 先保证最终 Verifier 确实覆盖任务目标；
- 把可独立验证的阶段做成 multi-step Task，每步由真实测试评分；
- 使用多维
reward.json保存 correctness、安全、效率等不同语义； - 由训练系统明确把多维 Reward 标量化，保留原维度供审计；
- 对失败轨迹保留异常类型、终端证据和 Verifier 日志，但不要把异常自动等同于合法零分。
Harbor multi-step Task 可以用 min_reward 在某步未达阈值时停止后续步骤，并用 mean 或 final 形成 Trial 级 Reward；mean 对已有 VerifierResult 的各步按 key 求均值，缺 key 按 0，完全没有 VerifierResult 的步骤不进分母。20 这是一种任务执行和聚合机制，不是 token 级 credit assignment。若要训练每步策略，必须把 step_results[i].agent_result、该步 Reward 与对应 token 对齐；本章最小 Adapter 正因没有定义这一点而拒绝 multi-step 训练样本。
24.7.1 先版本化 Reward 语义，再选择标量化公式
VerifierResult.rewards 是字符串到数值的映射，它没有替使用者定义“越大越好”、量纲、合法范围或各维度之间的优先级。一个名为 latency_ms 的数值若直接与 correctness 相加，优化方向甚至可能反了。本章 Adapter 只读取明确指定的 reward_key，既不裁剪到 [0, 1]，也不把其他 key 自动求平均，这是有意的保守选择。
生产项目应维护 reward-schema.json 或等价清单，至少写出每个 key 的含义、方向、范围、缺失策略、计算代码 digest 和 schema 版本。标量化放在训练侧的纯函数中，输入保留原始多维 Reward，输出同时记录公式版本。例如可以把 correctness 设为硬门禁，只有满足后才计算效率项；也可以把安全维度设为否决项。具体公式是任务设计决策，不是 Harbor 默认语义。
验收时要为每个维度准备边界夹具：正确结果、明显错误、空输出、超大输出、NaN/Inf、缺 key、Verifier 异常以及试图操纵评分器的输入。若公式或 Verifier 更新，旧 rollout 的 Reward 不应原地覆盖；应生成带新 schema 版本的派生记录，保留旧值以便比较漂移。这样才能区分“策略变好”和“尺子变了”。
24.8 Reward gaming 是 Verifier 威胁模型
Reward 上升只说明模型更擅长最大化当前测量。关于 AI safety 的经典问题分类把 reward hacking 列为“目标函数错误”导致的实际风险之一；它与“监督太昂贵而无法频繁评价”是不同问题。21 对 Harbor Task，常见攻击面包括修改测试、预写 Reward 文件、读取参考答案、利用 Verifier 未覆盖的等价类、污染 sidecar 状态，以及通过网络取得受保护数据。
防护清单应落到 Task 设计，而不只写一句“防作弊”：
- Agent 镜像和工作目录中不包含 Oracle solution 或隐藏答案；
- Verifier 测试在 Agent 阶段后注入，关键测试不对 Agent 可读；
- 高价值任务使用 separate Verifier 环境，并只传入最小 Artifact；
- Reward 同时检查结果正确性与禁止的副作用；
- 对测试文件、Verifier 文件和关键输入记录 digest；
- 对高 Reward 轨迹做对抗抽检，检查是否绕过目标；
- 独立评测使用 Agent 从未接触的任务变体和 Verifier 实现。
Harbor v0.18.0 支持 shared/separate Verifier 模式；single-step 在 separate 模式下会先收集 Artifact、停止 Agent Environment，再运行独立 Verifier。22 这能缩小直接篡改测试环境的攻击面，但不能自动证明 Verifier 完整，也不能防止 Artifact 本身携带利用载荷。安全边界仍需任务作者定义输入白名单、大小限制、解析隔离和结果交叉检查。
注意：训练过程反复访问同一个 Verifier，本身会让策略逐步发现其盲点。最终评测若复用同一实现，只能测“对这个 Verifier 的适应”，不能单独证明真实任务质量。
24.9 数据复用与独立评测
每条 rollout 至少应绑定以下外部清单字段：
run_id
task_key + task_digest + split
harbor_version + harbor_commit
agent_name + agent_config
policy_checkpoint_digest
tokenizer + renderer version
sampling parameters
reward_schema_version + verifier_digest
created_at
Harbor lock.json 可覆盖其中一部分运行输入，但 checkpoint digest、reward schema 的组织语义和训练迭代仍需要训练系统补充。保存这些字段后，数据才能按策略版本分桶，而不是把几轮不同策略生成的样本混成一个“最新 batch”。
对 on-policy 算法，旧数据能否复用由算法及其重要性修正规则决定；Adapter 不应自行决定。PPO 允许对一批采样数据做多个 epoch 的 minibatch 更新，不等于任意历史 rollout 都可以无限复用。5 如果训练框架接受离线或 off-policy 数据，应在接口上显式标记 behavior policy，并验证 old logprob 与当前 tokenizer 对齐。
训练、开发和最终评测应按任务身份或来源组切分，而不是按 Trial 随机切分。同一 Task 的多个 attempts、轻微改写、同仓库相邻 issue 或共享测试模板都可能高度相关；随机按 rollout 分行会把近重复样本放到两侧。RLTask.split 的运行时门禁只能阻止显式标错，真正的 split 生成还要在上游按 group key、许可证和污染规则完成，并把 manifest 设为只读。
独立评测进程应满足四点：
- 不向训练生成器返回 token 或单题隐藏诊断；
- 不用 eval Reward 选择当前 step 的梯度更新；
- 使用独立 Job 目录、固定 task manifest 和 Verifier digest；
- 报告每个任务/运行的分布和不确定性，不只报告一个点估计。
深度 RL 实验研究反复指出，算法与环境的随机性会让少量运行的点估计难以解释；更可靠的做法是标准化实验条件并报告不确定性。23 对少量高成本运行，相关研究进一步建议区间估计和性能分布，而不是只比较均值或中位数。24 这些是通用 RL 评测建议，不是 Harbor 自动生成的统计保证。
24.9.1 把“可重放”拆成三种问题
工程会议里常说“这个实验可以 replay”，但至少需要区分三种含义。第一种是执行重放：相同 Task digest、Agent 配置、Environment 和 Verifier 能再次启动；Harbor 的 lock 文件主要服务这一层。第二种是采样重放：相同 checkpoint、tokenizer、renderer、采样参数和随机源能否产生相同 token；外部推理服务、并发调度和非确定性算子可能让答案是否定的。第三种是训练重放：相同 rollout 顺序、padding、loss、混合精度、分布式规约和 optimizer 状态能否得到同一权重；这一层完全超出 Harbor Job 的保证。
因此，复现实验失败时应按层排查。Task digest 不同，先停止比较并修复数据版本；Task 相同但 token 不同，检查 checkpoint、renderer、采样和服务端实现；token 相同但 loss/权重不同，再进入训练框架的数值与分布式诊断。把三层都压成一个 seed=42 字段，会让故障定位失去证据。
数据复用也要按这三层管理。缓存完整 Environment snapshot 可以降低启动成本，但必须证明 snapshot 没带入上个 episode 的状态；复用 token rollout 可以节省采样成本，但必须满足算法对 behavior policy 的要求；复用已计算的 advantage 或张量则进一步绑定 Reward、value model 和归一化批次。默认只复用最靠前、最容易审计的层，越靠近优化器，失效条件越多。
24.10 上线前的两级验收
先做静态与无环境验收：
# 1. 配置模型可解析，split 门禁有效
uv run pytest -q test_rl_adapter.py
# 2. 锁定 Harbor 版本，不允许解析到其他安装
uv run python - <<'PY'
import harbor
from importlib.metadata import version
print(version("harbor"), harbor.__file__)
PY
# 3. 训练清单与评测清单的 group key 不相交
uv run python tools/check_split_manifest.py manifests/train.json manifests/eval.json
第三条命令中的 tools/check_split_manifest.py 是项目自有门禁，本章未提供也未执行；它应比较规范化 task/group key、内容 digest 和来源组，而不只比较文件名。
再做受控真实 rollout 验收：
- 先用一个无凭据的小 Task 和一个 attempt；
- 检查
jobs/<run_id>/config.json、lock.json、Jobresult.json与每个 Trial 结果； - 逐轮验证 prompt/completion token 和 logprob 长度；
- 人工核对 Reward 0、缺失 Reward、Agent 超时和 Verifier 超时的区别；
- 故意把一个 eval Task 传入
generate()，确认在创建 Job 前失败； - 故意让 provider 不返回 token ID，确认 Adapter 拒绝；
- 最后才增加 attempts 和 concurrency，并设置成本、时间、失败率和磁盘上限。
只有这些门禁通过，才把 PolicyRollout 转给具体训练框架。若训练框架需要扁平 token、attention mask、loss mask、advantage 或张量设备信息，应在下一层 Adapter 中完成，并为转换写独立测试。
24.11 本章小结
- Harbor 的 Trial 是 Agentic episode 的自然容器；Job 负责批量展开和并发，不负责梯度更新。
- v0.18.0 应使用
await Job.create(config)和顶层n_concurrent_trials；随附 RL 文档和 Cookbook 中存在超出该版本的旧/未来接口。 collect_rollout_details=True只是采集请求，provider 仍可能不给 token；Adapter 必须验证逐轮 token 与 logprob。- 缺失 Reward、异常和合法零分是三种状态，不能合并。
n_attempts、运行并发和 optimizer minibatch 是不同概念。- 稀疏 Reward 的解决方向是改进可验证证据和任务分解，不是发明代理分数。
- 训练与评测 split、策略 checkpoint、tokenizer 和 Verifier digest 必须在 Harbor 结果之外继续版本化。
24.12 练习
- 扩展测试，让
generate()面对n_total_trials正确但trial_results少一项时失败，并确认错误消息包含 expected/actual。 - 为多维 Reward 增加一个显式标量化策略，例如同时要求
correctness >= 1，再从efficiency计算辅助项；保留原始维度，不覆盖 Verifier 输出。 - 为 multi-step Task 设计
StepPolicyRollout，说明每步 token、Reward、异常和最终聚合 Reward 的对应关系；在契约明确前不要实现 flatten。 - 构造 train/eval 各两个 Task，其中一对共享同一上游仓库或模板。设计 group key，使这对任务不能跨 split。
- 设计一次 reward-gaming 红队演练：允许 Agent提交恶意 Artifact，列出 separate Verifier 仍需防护的解析、资源和语义攻击面。