在线训练奖励信号偏严，TaoToken 把 GRPO 的 judge 调用接上

1. GRPO 奖励信号偏严时，先别急着换 judge：把评分依据任务自适应，再切到 TaoToken

GRPO 奖励信号偏严时，先做两件事：把 judge 判分标准改成任务自适应，并把 judge 调用切到 TaoToken。拿 Key 走 TaoToken 官网，Base URL 用 https://taotoken.net/api。在线强化学习里，真正消耗 Token 的往往不是策略模型采样，而是训练期奖励 judge：每条轨迹、每个 rollout 都要被 VLM judge 看一遍，如果评分标准太粗，它会把“其实成功”的轨迹判 0；如果标准太隐式，它又会自己加戏，把格式、措辞、界面细节当成硬性失败条件。GUI Agent 的 reward model 最近被讨论很多，核心矛盾并不只在“judge 模型够不够强”，而在 judge 打分前有没有拿到一条贴合当前任务的成功标准。

ADAPT RUBRIC 这类工作的启发是：把判分标准拆成“类别级粗准则 + 实例级细准则”。粗准则解决这一类任务该看什么，比如通信类要核对收件人、正文和发送动作；创建/修改类要确认对象是否真正落地。细准则只补当前指令独有的约束，比如某个数值、某个可见范围、某个标签、某个格式。最后把粗准则作为主体，把细准则作为任务专属块追加，再精选轨迹帧交给 VLM judge 输出成功或失败。这个思路对工程很友好：judge 模型本身可以不换，改的是喂给它的判分依据和轨迹上下文。对正在跑 GRPO 的团队来说，落地路径可以更直接：先把奖励 judge 的 API 调用切到 TaoToken，拿到可观测的 token、延迟和 reward 日志；再把 rubric 从静态模板升级成任务自适应；最后用奖励信号日志和任务成功率对照，验证偏严问题是否真的缓解。

本文不写成论文摘要复述，而是按可跟做的配置步骤来：拿 Key、配 Base URL、写最小 judge、构建粗到细 rubric、接 GRPO reward function、记录日志、做成功率对照，并把 Claude Code、Codex、CC Switch 的配置分开写清楚。

2. 把训练期奖励 judge 切到 TaoToken：Base URL 与 Key 的最小闭环

训练期奖励 judge 的调用量通常有几个特征：并发高、单次 prompt 长、输出要求结构化、失败重试多。如果每个 rollout 都调一次外部 judge，token 成本和延迟会直接决定 GRPO 能不能长时间跑。因此第一步不是改模型，而是让调用链路可控。

先到 TaoToken 官网 创建 API Key。申请完成后，环境变量里统一放三件套：

export TAOTOKEN_API_KEY="YOUR_API_KEY" export TAOTOKEN_BASE_URL="https://taotoken.net/api" export JUDGE_MODEL="你从模型对话页选择的模型 ID"

注意：Base URL 是工具配置，不加 UTM，固定用：

https://taotoken.net/api

Python 端用 OpenAI 兼容 SDK 调 TaoToken 的最小 judge 如下。这里假设你已经有轨迹帧文本或截图描述，实际训练时可以把截图转成多模态消息，也可以先用 UI 树、动作历史和关键帧描述做文本 judge。

import os import json import time from openai import OpenAI client = OpenAI( api_key=os.environ["TAOTOKEN_API_KEY"], base_url="https://taotoken.net/api", ) JUDGE_MODEL = os.environ.get("JUDGE_MODEL", "你的模型ID") def call_reward_judge(instruction: str, trajectory_summary: str, rubric_text: str) -> dict: system_prompt = ( "你是一个 GUI Agent 轨迹奖励验证器。" "只根据用户指令和给定评分表判断轨迹是否成功。" "不要补充评分表中没有的额外要求。" "必须输出 JSON，字段为 success、score、reason、evidence、rubric_hits。" "success 为布尔值，score 为 0 到 1 之间的数值。" ) user_prompt = f""" 【用户指令】 {instruction} 【评分表】 {rubric_text} 【轨迹摘要与关键帧】 {trajectory_summary} 请输出 JSON： {{ "success": true/false, "score": 0/1, "reason": "简短说明", "evidence": ["关键证据1", "关键证据2"], "rubric_hits": ["命中的评分项"] }} """ start = time.time() resp = client.chat.completions.create( model=JUDGE_MODEL, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], temperature=0, max_tokens=512, response_format={"type": "json_object"}, timeout=60, ) latency_ms = int((time.time() - start) * 1000) content = resp.choices[0].message.content result = json.loads(content) result["_latency_ms"] = latency_ms result["_usage"] = { "prompt_tokens": getattr(resp.usage, "prompt_tokens", None), "completion_tokens": getattr(resp.usage, "completion_tokens", None), "total_tokens": getattr(resp.usage, "total_tokens", None), } return result

这段代码的重点不是“调通一个模型”，而是把奖励 judge 的输入、输出、延迟和 token 用量固定下来。后面的日志对照、偏严排查、成本优化，都依赖这些字段。没有这些字段，你只能看到任务成功率涨跌，却不知道是 rubric 变了、模型变了，还是轨迹帧选得不对。

如果你还没有 Key，直接去 TaoToken 创建 API Key 页面创建；模型选择可以先从 模型对话 页确认可用模型 ID，再填到JUDGE_MODEL。

3. 复刻 ADAPT RUBRIC 的最小流程：粗准则检索、细准则生成、融合打 0/1

不要一上来就期望完整复现论文。工程落地可以先做最小闭环：本地维护一份类别级粗准则库，用任务路由器选类别，再让模型生成最多两条实例级细准则，最后融合成一个 judge prompt。

第一步，粗准则库不要写成散落 prompt，放到 YAML 或 JSON 里，便于版本管理：

# rubric_library.yaml communication: name: 通信与消息发送 steps: - 确认收件人、频道或账号是否正确 - 核对正文、附件、链接是否与指令一致 - 确认最终动作是发送，而不是仅打开草稿 pitfalls: - 只停留在编辑页未发送 - 收件人或频道选错 - 可见范围、标签、@对象不符合指令 output_format: 消息发送成功且关键字段一致 create_modify: name: 创建、修改与保存 steps: - 确认目标对象被创建或被修改 - 核对字段值、数量、日期、状态 - 确认保存或提交动作完成 pitfalls: - 只填写未保存 - 修改错对象 - 字段值大小写、单位、格式错误 general: name: 通用兜底 steps: - 判断用户指令中的核心目标是否完成 - 判断是否存在明确未完成的必要动作 pitfalls: - 把中间状态当成最终状态 - 忽略指令中的否定条件

第二步，任务路由器可以先做轻量版：用小模型分类，或者直接关键词加规则兜底。它的目标不是 100% 准确，而是减少“所有任务共用一套通用模板”的过度泛化。

import yaml with open("rubric_library.yaml", "r", encoding="utf-8") as f: RUBRIC_LIB = yaml.safe_load(f) def route_category(instruction: str) -> str: text = instruction.lower() communication_words = ["发送", "邮件", "消息", "通知", "群", "频道", "收件人", "标签"] modify_words = ["创建", "新建", "修改", "保存", "提交", "更新", "设置"] if any(w in instruction for w in communication_words): return "communication" if any(w in instruction for w in modify_words): return "create_modify" return "general" def get_coarse_rubric(category: str) -> dict: return RUBRIC_LIB.get(category, RUBRIC_LIB["general"])

第三步，细准则生成。它只补当前指令独有的约束，最多两条。硬约束要写进 prompt：每条必须能在指令里找到依据；最多两条；如果不需要就返回空数组。这样可以避免 judge 自己加戏，也能减少“过度严苛”导致的假阴性。

def generate_fine_rubrics(instruction: str, category: str, coarse: dict) -> list[str]: prompt = f""" 你是评分表细化器。请根据用户指令、任务类别和粗准则，生成最多 2 条实例级细准则。 要求： 1. 每条细准则必须能在用户指令里找到明确短语、数值、范围或约束作为依据。 2. 不要重复粗准则，不要加入指令没有要求的额外条件。 3. 如果当前指令不需要额外细项，返回空数组。 4. 只输出 JSON：{{"fine_rubrics": ["..."]}} 用户指令：{instruction} 任务类别：{category} 粗准则：{coarse} """ resp = client.chat.completions.create( model=JUDGE_MODEL, messages=[{"role": "user", "content": prompt}], temperature=0, max_tokens=256, response_format={"type": "json_object"}, timeout=60, ) data = json.loads(resp.choices[0].message.content) return data.get("fine_rubrics", [])[:2]

第四步，融合与轨迹精选。粗准则作为主体，细准则作为独立任务块追加。轨迹不要全量喂，保留初始帧、终止帧，以及文本输入、提交、状态切换附近的帧。这样既省钱，也让 judge 不容易被几十帧界面变化干扰。

def build_rubric_text(coarse: dict, fine_rubrics: list[str]) -> str: lines = ["【粗准则】"] lines.append(f"类别：{coarse['name']}") lines.append("检查步骤：" + "；".join(coarse["steps"])) lines.append("常见坑点：" + "；".join(coarse["pitfalls"])) if fine_rubrics: lines.append("") lines.append("【当前指令专属细准则】") for i, item in enumerate(fine_rubrics, 1): lines.append(f"{i}. {item}") return "\n".join(lines) def select_key_frames(frames: list[dict], max_frames: int = 10) -> list[dict]: if len(frames) <= max_frames: return frames selected = [frames[0], frames[-1]] middle = frames[1:-1] key_actions = {"input", "submit", "click_submit", "state_change", "send", "save"} for frame in middle: if frame.get("action") in key_actions: selected.append(frame) if len(selected) >= max_frames: break if len(selected) < max_frames: remain = [f for f in middle if f not in selected] selected.extend(remain[: max_frames - len(selected)]) return selected[:max_frames]

到这里，judge 的输入就从“任务完成没”变成“类别边界 + 当前指令要点 + 关键轨迹证据”。judge 模型不需要重训，你只是把题出得更准。

4. GRPO 奖励函数怎么接：把 judge 输出变成可训练 reward

在 GRPO 里，reward function 是训练循环的一部分。它不能只是“调一下模型”，还要处理超时、重试、缓存、并发、失败回退和日志。建议把 judge 调用封装成独立模块，策略模型只消费 reward 数值。

一个可用的 reward function 骨架如下：

import hashlib import json import os import time REWARD_LOG = "reward_log.jsonl" def stable_hash(text: str) -> str: return hashlib.sha256(text.encode("utf-8")).hexdigest()[:16] def reward_fn(sample: dict, rubric_version: str = "adapt_v1") -> float: instruction = sample["instruction"] frames = sample.get("frames", []) category = route_category(instruction) coarse = get_coarse_rubric(category) fine_rubrics = generate_fine_rubrics(instruction, category, coarse) rubric_text = build_rubric_text(coarse, fine_rubrics) selected_frames = select_key_frames(frames) trajectory_summary = "\n".join( f"[{i}] action={f.get('action')} text={f.get('text', '')} state={f.get('state', '')}" for i, f in enumerate(selected_frames) ) row = { "task_id": sample.get("task_id"), "step": sample.get("step"), "instruction_hash": stable_hash(instruction), "category": category, "rubric_version": rubric_version, "coarse_id": category, "fine_rubrics": fine_rubrics, "frame_count": len(selected_frames), "judge_model": JUDGE_MODEL, "timestamp": time.time(), } try: result = call_reward_judge(instruction, trajectory_summary, rubric_text) reward = float(result.get("score", 0.0)) row.update({ "reward": reward, "success": bool(result.get("success")), "reason": result.get("reason", ""), "evidence": result.get("evidence", []), "rubric_hits": result.get("rubric_hits", []), "latency_ms": result.get("_latency_ms"), "usage": result.get("_usage", {}), "error": None, }) except Exception as e: reward = 0.0 row.update({ "reward": reward, "success": False, "reason": "judge_call_failed", "evidence": [], "rubric_hits": [], "latency_ms": None, "usage": {}, "error": repr(e), }) with open(REWARD_LOG, "a", encoding="utf-8") as f: f.write(json.dumps(row, ensure_ascii=False) + "\n") return reward

这段逻辑里要注意几个工程点：

1. 奖励偏严时不要直接改 reward 阈值。先看日志里success=false但evidence很强的样本，判断是细准则加戏，还是轨迹帧漏了关键证据。
2. 细准则生成要可弃权。如果每条指令都强行补两条，会退化成冗长 checklist。prompt 里必须允许返回空数组。
3. judge 输出要结构化。success用于成功率对照，score用于 GRPO reward，reason/evidence/rubric_hits用于排查。
4. 失败回退不要默认给高奖励。judge 调用异常时给 0 是安全的，但要在日志里标记error，否则你会误以为模型真的失败。
5. 缓存高频指令。同一指令下多个 rollout 可能重复 judge，可以用 instruction_hash + rubric_version + trajectory_hash 做本地缓存。

如果你希望进一步降低成本，可以把“粗准则检索 + 细准则生成”合并成一次调用，或者在训练前批量预生成细准则，训练中只调用最终 judge。消耗 Token 的大头仍然是训练期奖励 judge，所以日志里一定要按rubric_version、judge_model、frame_count分组统计。

5. 奖励信号日志与任务成功率对照：本地可复现的评估表

想验证“偏严是否缓解”，不能只看训练曲线。建议每 N 个 step 固定抽样一批任务，分别用旧模板 judge 和自适应 rubric judge 打分，再独立跑任务成功率评估。产出至少包含两张表：奖励信号日志表、任务成功率对照表。

奖励日志可以用 JSONL，也可以用本地 SQLite。下面是一个本地聚合脚本：

import json import pandas as pd rows = [] with open("reward_log.jsonl", "r", encoding="utf-8") as f: for line in f: if line.strip(): rows.append(json.loads(line)) df = pd.DataFrame(rows) df["total_tokens"] = df["usage"].apply(lambda x: x.get("total_tokens") if isinstance(x, dict) else None) summary = df.groupby(["rubric_version", "judge_model"]).agg( reward_mean=("reward", "mean"), success_rate=("success", "mean"), avg_latency_ms=("latency_ms", "mean"), avg_total_tokens=("total_tokens", "mean"), count=("reward", "count"), ).reset_index() print(summary) summary.to_csv("reward_signal_summary.csv", index=False)

如果你更习惯 SQL，可以在本地 SQLite 中执行：

-- 先在本地将 reward_log.csv 导入 sqlite，再执行聚合 SELECT rubric_version, judge_model, AVG(reward) AS reward_mean, AVG(CASE WHEN success THEN 1.0 ELSE 0.0 END) AS success_rate, AVG(latency_ms) AS avg_latency_ms, COUNT(*) AS sample_count FROM reward_log GROUP BY rubric_version, judge_model;

任务成功率对照表建议这样记录：

表格里的数字不要照搬论文，也不要用未经验证的外部结论。你应该用自己的任务集、自己的轨迹池、自己的 judge 模型跑出来。可复现的关键是固定四件事：

• 固定评估任务集和 rollout 数量；
• 固定 judge 模型和 temperature；
• 固定 rubric 版本号；
• 固定日志字段和聚合脚本。

只有这四件事固定，奖励信号日志与任务成功率对照才有意义。否则你看到的提升可能来自采样随机性，而不是 rubric 改进。

6. Claude Code、Codex、CC Switch 配置：让调试和训练脚本共用同一套 Key

奖励 judge 调试时，你可能还会用 Claude Code、Codex 或 CC Switch 来辅助查看配置、跑脚本、管理供应商。这里必须区分工具链：Claude Code 使用settings.json和ANTHROPIC_*环境变量；Codex 使用config.toml，不要混用ANTHROPIC_*；CC Switch 则把它当成供应商三件套来配。

Claude Code：settings.json 配置

路径通常是~/.claude/settings.json，写入：

{ "env": { "ANTHROPIC_BASE_URL": "https://taotoken.net/api", "ANTHROPIC_API_KEY": "YOUR_API_KEY", "ANTHROPIC_MODEL": "你从模型对话页选择的模型 ID" } }

如果你用环境变量，也可以：

export ANTHROPIC_BASE_URL="https://taotoken.net/api" export ANTHROPIC_API_KEY="YOUR_API_KEY" export ANTHROPIC_MODEL="你从模型对话页选择的模型 ID"

Claude Code 的文档入口在文末 CTA 区，配置前可以先确认字段名和当前版本要求。

Codex：config.toml 配置

Codex 不要用ANTHROPIC_*。在~/.codex/config.toml或项目配置中写：

model = "你的模型ID" model_provider = "taotoken" [model_providers.taotoken] name = "TaoToken" base_url = "https://taotoken.net/api" env_key = "TAOTOKEN_API_KEY" wire_api = "chat"

然后设置：

export TAOTOKEN_API_KEY="YOUR_API_KEY"

如果你的 Codex 版本使用responseswire API，则按版本说明调整；核心是 provider 指向 TaoToken，Base URL 用https://taotoken.net/api，Key 用YOUR_API_KEY。

CC Switch：三件套配置

CC Switch 里新增自定义供应商时，填三件套即可：

• 供应商名称：TaoToken
• Base URL：https://taotoken.net/api
• API Key：YOUR_API_KEY
• 默认模型：从 TaoToken 模型对话页选择

如果你还没有 Key，先去 TaoToken 官网 创建，再回到 CC Switch 填三件套。这样 Claude Code、Codex、训练脚本可以共用同一套 Key，但配置格式不要互相套用。

7. 常见排障：奖励偏严、JSON 不合法、超时与 Token 失控

奖励 judge 接上 TaoToken 后，常见问题通常不是“接口不通”，而是输出不可用或信号偏严。可以按下面表排查。

尤其要注意“奖励偏严”和“奖励噪声”是两件事。偏严是系统性地把成功判失败；噪声是同一轨迹多次打分不一致。排查时可以先对同一批轨迹重复 judge 三次，看success和score的方差。如果方差大，先降 temperature、固定模型、加强 JSON schema；如果方差小但假阴性多，再改 rubric。

8. 下一步：模型对话 → Coding Plan → 创建 Key → Claude Code 文档

把 GRPO 的奖励 judge 接上 TaoToken 后，推荐按这个路径推进：

1. 先到 模型对话 确认可用模型，选一个适合做 reward judge 的模型 ID。
2. 如果训练和调试需要更长周期，查看 Coding Plan，把日常调用和训练期 judge 调用分开管理。
3. 到 创建 API Key 创建YOUR_API_KEY，填入环境变量和训练脚本。
4. 需要配置 Claude Code 时，对照 Claude Code 文档 设置settings.json和ANTHROPIC_*。
5. 最后用本文的日志脚本跑一轮奖励信号日志与任务成功率对照，确认偏严是否下降、Token 是否可控、延迟是否可接受。

在线强化学习里的奖励信号不是“调一次就完事”的接口，而是训练系统的一部分。先把 judge 调用切到可观测的 TaoToken，再把评分表从通用模板升级成任务自适应，最后用日志和成功率说话，你得到的就不只是一套更细的评分表，而是一条可复现、可对照、可优化的 GRPO 奖励流水线。