ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

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

在线训练奖励信号偏严,TaoToken 把 GRPO 的 judge 调用接上 1. GRPO 奖励信号偏严时先别急着换 judge把评分依据任务自适应再切到 TaoTokenGRPO 奖励信号偏严时先做两件事把 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 切到 TaoTokenBase URL 与 Key 的最小闭环训练期奖励 judge 的调用量通常有几个特征并发高、单次 prompt 长、输出要求结构化、失败重试多。如果每个 rollout 都调一次外部 judgetoken 成本和延迟会直接决定 GRPO 能不能长时间跑。因此第一步不是改模型而是让调用链路可控。先到 TaoToken 官网 创建 API Key。申请完成后环境变量里统一放三件套export TAOTOKEN_API_KEYYOUR_API_KEY export TAOTOKEN_BASE_URLhttps://taotoken.net/api export JUDGE_MODEL你从模型对话页选择的模型 ID注意Base URL 是工具配置不加 UTM固定用https://taotoken.net/apiPython 端用 OpenAI 兼容 SDK 调 TaoToken 的最小 judge 如下。这里假设你已经有轨迹帧文本或截图描述实际训练时可以把截图转成多模态消息也可以先用 UI 树、动作历史和关键帧描述做文本 judge。import os import json import time from openai import OpenAI client OpenAI( api_keyos.environ[TAOTOKEN_API_KEY], base_urlhttps://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( modelJUDGE_MODEL, messages[ {role: system, content: system_prompt}, {role: user, content: user_prompt}, ], temperature0, max_tokens512, response_format{type: json_object}, timeout60, ) 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, encodingutf-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( modelJUDGE_MODEL, messages[{role: user, content: prompt}], temperature0, max_tokens256, response_format{type: json_object}, timeout60, ) 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, encodingutf-8) as f: f.write(json.dumps(row, ensure_asciiFalse) \n) return reward这段逻辑里要注意几个工程点奖励偏严时不要直接改 reward 阈值。先看日志里successfalse但evidence很强的样本判断是细准则加戏还是轨迹帧漏了关键证据。细准则生成要可弃权。如果每条指令都强行补两条会退化成冗长 checklist。prompt 里必须允许返回空数组。judge 输出要结构化。success用于成功率对照score用于 GRPO rewardreason/evidence/rubric_hits用于排查。失败回退不要默认给高奖励。judge 调用异常时给 0 是安全的但要在日志里标记error否则你会误以为模型真的失败。缓存高频指令。同一指令下多个 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, encodingutf-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, indexFalse)如果你更习惯 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 模型奖励均值任务成功率失败误判成功平均延迟平均 Tokenbaseline静态通用模板原 judgeadapt_v1粗准则 细准则TaoToken judgeadapt_v2细准则可弃权 关键帧精选TaoToken judge表格里的数字不要照搬论文也不要用未经验证的外部结论。你应该用自己的任务集、自己的轨迹池、自己的 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 Codesettings.json 配置路径通常是~/.claude/settings.json写入{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_API_KEY: YOUR_API_KEY, ANTHROPIC_MODEL: 你从模型对话页选择的模型 ID } }如果你用环境变量也可以export ANTHROPIC_BASE_URLhttps://taotoken.net/api export ANTHROPIC_API_KEYYOUR_API_KEY export ANTHROPIC_MODEL你从模型对话页选择的模型 IDClaude Code 的文档入口在文末 CTA 区配置前可以先确认字段名和当前版本要求。Codexconfig.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_KEYYOUR_API_KEY如果你的 Codex 版本使用responseswire API则按版本说明调整核心是 provider 指向 TaoTokenBase URL 用https://taotoken.net/apiKey 用YOUR_API_KEY。CC Switch三件套配置CC Switch 里新增自定义供应商时填三件套即可供应商名称TaoTokenBase URLhttps://taotoken.net/apiAPI KeyYOUR_API_KEY默认模型从 TaoToken 模型对话页选择如果你还没有 Key先去 TaoToken 官网 创建再回到 CC Switch 填三件套。这样 Claude Code、Codex、训练脚本可以共用同一套 Key但配置格式不要互相套用。7. 常见排障奖励偏严、JSON 不合法、超时与 Token 失控奖励 judge 接上 TaoToken 后常见问题通常不是“接口不通”而是输出不可用或信号偏严。可以按下面表排查。症状可能原因处理大量成功轨迹被判 0细准则过度严苛或 judge 自行加戏检查细准则是否能在指令中找到依据启用弃权降低 temperature在 prompt 中禁止额外要求judge 经常漏判成功粗准则太泛或关键帧没保留增加类别级检查步骤保留提交、发送、保存附近帧加入初始帧和终止帧JSON 解析失败模型输出多余文本强制response_format{type:json_object}失败重试日志记录 raw output训练速度慢judge 串行调用、上下文太长并发调用缓存限制帧数先预生成细准则压缩轨迹摘要Token 消耗失控全量轨迹喂入、细准则冗长最多两条细准则关键帧精选限制 max_tokens按 rubric_version 统计 token成功率对照不可信日志缺字段或评估集变化固定 task_id、rubric_version、judge_model每次评估用同一批任务judge 调用失败后奖励异常异常回退处理不当失败给 0 并记录 error不要给高奖励增加重试和超时尤其要注意“奖励偏严”和“奖励噪声”是两件事。偏严是系统性地把成功判失败噪声是同一轨迹多次打分不一致。排查时可以先对同一批轨迹重复 judge 三次看success和score的方差。如果方差大先降 temperature、固定模型、加强 JSON schema如果方差小但假阴性多再改 rubric。8. 下一步模型对话 → Coding Plan → 创建 Key → Claude Code 文档把 GRPO 的奖励 judge 接上 TaoToken 后推荐按这个路径推进先到 模型对话 确认可用模型选一个适合做 reward judge 的模型 ID。如果训练和调试需要更长周期查看 Coding Plan把日常调用和训练期 judge 调用分开管理。到 创建 API Key 创建YOUR_API_KEY填入环境变量和训练脚本。需要配置 Claude Code 时对照 Claude Code 文档 设置settings.json和ANTHROPIC_*。最后用本文的日志脚本跑一轮奖励信号日志与任务成功率对照确认偏严是否下降、Token 是否可控、延迟是否可接受。在线强化学习里的奖励信号不是“调一次就完事”的接口而是训练系统的一部分。先把 judge 调用切到可观测的 TaoToken再把评分表从通用模板升级成任务自适应最后用日志和成功率说话你得到的就不只是一套更细的评分表而是一条可复现、可对照、可优化的 GRPO 奖励流水线。
返回列表