在 Amazon SageMaker AI 中做多轮强化学习:从可信环境到外部评测

2026-07-03 45 预计阅读时间: 1 分钟
来源: aws.amazon.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:10 分钟

多轮强化学习的难点不只是“让模型多说几轮”。一旦 Agent 可以连续行动,环境状态、奖励归因、评测方式和监控指标都会变得更敏感。要把训练做可靠,关键是先让训练环境可信,再用独立评测验证结果,并把奖励函数对齐到最终任务,而不是只奖励表面行为。

训练环境要先变成可信仪表盘

多轮 RL 的训练环境不是一个简单的输入输出函数。它需要在每一轮保存状态、执行动作、返回观测,并在合适的时候终止 episode。如果环境本身不稳定,后面的奖励设计和模型调参都会被噪声带偏。

一个可靠环境通常要做到:

  • 状态可复现:同一个 seed、同一组初始条件能复现相近轨迹。
  • 终止条件清晰:不要让 Agent 无限对话,也不要过早截断关键步骤。
  • 失败可解释:工具调用失败、格式错误、超时、无效动作要有明确标记。
  • 日志够细:至少记录每轮 observation、action、reward、done、错误类型。

可以这样实践:先用一个极小的多轮环境验证训练闭环,再接入 SageMaker 训练作业。下面示例模拟一个“多轮客服 Agent”:Agent 最多问 3 次问题,目标是收集用户意图并给出解决方案。

# multi_turn_env.py
from dataclasses import dataclass, field
from typing import Dict, List, Tuple

@dataclass
class TicketEnv:
    max_turns: int = 3
    turn: int = 0
    facts: Dict[str, bool] = field(default_factory=lambda: {
        "intent_identified": False,
        "account_checked": False,
        "solution_given": False,
    })
    history: List[Dict[str, str]] = field(default_factory=list)

    def reset(self) -> str:
        self.turn = 0
        self.facts = {
            "intent_identified": False,
            "account_checked": False,
            "solution_given": False,
        }
        self.history = []
        return "用户:我的账单有问题。"

    def step(self, action: str) -> Tuple[str, float, bool, Dict[str, str]]:
        self.turn += 1
        action_lc = action.lower()
        reward = -0.05
        info = {"error": ""}

        if "账单" in action or "bill" in action_lc:
            self.facts["intent_identified"] = True
            reward += 0.3
        if "账户" in action or "account" in action_lc:
            self.facts["account_checked"] = True
            reward += 0.3
        if "解决" in action or "refund" in action_lc or "solution" in action_lc:
            self.facts["solution_given"] = True
            reward += 0.5

        if len(action.strip()) < 4:
            reward -= 0.4
            info["error"] = "too_short"

        done = self.turn >= self.max_turns or self.facts["solution_given"]
        if done and all(self.facts.values()):
            reward += 1.0
        elif done and not self.facts["solution_given"]:
            reward -= 0.5

        observation = f"第 {self.turn} 轮后状态:{self.facts}"
        self.history.append({"action": action, "observation": observation})
        return observation, reward, done, info

if __name__ == "__main__":
    env = TicketEnv()
    print(env.reset())
    for action in ["我来确认账单问题", "请提供账户信息", "我会给出解决方案"]:
        obs, reward, done, info = env.step(action)
        print({"obs": obs, "reward": reward, "done": done, "info": info})
        if done:
            break

运行:

python multi_turn_env.py

这个环境很简陋,但它能暴露多轮 RL 的核心问题:奖励是否被单轮动作误导?终止条件是否合理?日志是否能解释一次失败轨迹?在上 SageMaker AI 训练前,先用这种最小环境把接口和数据结构跑通,通常能节省大量排障时间。

外部评测不能和训练奖励混在一起

训练奖励告诉你 Agent 在环境里拿了多少分,但它不等于真实任务质量。尤其在多轮场景里,模型可能学会“刷奖励”:重复触发关键词、拖延对话、避开高风险动作,却没有真正完成任务。

更稳妥的做法是设置外部 evaluation:

  • 使用与训练分离的任务集。
  • 用独立脚本或服务回放完整多轮轨迹。
  • 统计任务成功率、平均轮数、工具失败率、无效动作率。
  • 保留人工抽样检查,避免自动指标掩盖体验问题。

可以这样设计一个评测输入文件:

[
  {
    "case_id": "billing_001",
    "user_goal": "用户想确认异常账单并获得处理方案",
    "expected_facts": ["intent_identified", "account_checked", "solution_given"],
    "max_turns": 3
  },
  {
    "case_id": "refund_001",
    "user_goal": "用户想申请退款并确认下一步",
    "expected_facts": ["intent_identified", "solution_given"],
    "max_turns": 4
  }
]

评测脚本可以独立于训练作业运行,只读取模型输出轨迹并打分。这样训练奖励变化时,你仍然有一个稳定标尺,判断模型是否真的更接近业务目标。

奖励函数要奖励“最终任务”,不是奖励漂亮动作

多轮 Agent 最容易掉进局部奖励陷阱。比如客服场景中,“问澄清问题”是好行为,但如果奖励太高,Agent 可能一直问问题;“调用工具”很重要,但如果每次调用都加分,它可能过度调用。

奖励设计可以按三层拆开:

  • 终局奖励:任务完成、用户目标达成、答案可执行。
  • 过程奖励:必要信息被收集、关键工具被正确调用。
  • 惩罚项:无效动作、重复提问、超出轮数、格式错误、危险输出。

一个可改造的奖励函数示例:

def compute_reward(state, action, done):
    reward = -0.02  # 每轮轻微成本,鼓励高效完成

    if action.get("type") == "invalid":
        reward -= 0.5

    if action.get("type") == "ask_clarifying_question" and not state.get("intent_identified"):
        reward += 0.2

    if action.get("type") == "tool_call" and action.get("tool") == "account_lookup":
        if state.get("intent_identified") and not state.get("account_checked"):
            reward += 0.3
        else:
            reward -= 0.2  # 避免无意义工具调用

    if action.get("type") == "final_answer":
        if state.get("account_checked") and state.get("solution_given"):
            reward += 1.0
        else:
            reward -= 0.6

    if done and state.get("task_success"):
        reward += 2.0

    return reward

重点不是把奖励函数写复杂,而是让它表达真正的任务偏好:少走弯路、正确使用工具、在有限轮数内完成用户目标。

多轮之后,监控指标也要换一套

单轮模型常看准确率、loss 或单次回答得分;多轮 RL 需要观察轨迹质量。训练时建议至少监控:

  • episode reward:整体奖励是否上升,是否突然塌陷。
  • task success rate:独立评测任务成功率是否同步提升。
  • average turns:平均轮数是否变长,是否出现拖延策略。
  • invalid action rate:格式错误、非法工具调用是否下降。
  • reward hacking signals:高训练奖励但低外部评测分,通常意味着奖励被钻空子。

在 SageMaker AI 中,可以把训练脚本输出结构化日志,再由 CloudWatch 或评测流水线消费。可以这样打印 JSON Lines,方便后续聚合:

import json
import time

def log_episode_metrics(job_name, episode_id, reward, success, turns, invalid_actions):
    print(json.dumps({
        "timestamp": int(time.time()),
        "job_name": job_name,
        "episode_id": episode_id,
        "episode_reward": reward,
        "task_success": success,
        "turns": turns,
        "invalid_action_rate": invalid_actions / max(turns, 1),
    }))

这些指标要一起看。只看 reward,可能错过奖励黑客;只看成功率,可能忽略成本和轮数;只看平均轮数,又可能惩罚了确实需要澄清的复杂任务。

落地清单:先小后大,先评测后扩容

多轮 RL 在 SageMaker AI 上落地时,不建议一开始就追求大规模训练。更稳的路径是:

  • 先实现一个可复现、可回放、可解释的训练环境。
  • 把外部评测从第一天接入,不要等模型“看起来不错”之后再补。
  • 奖励函数优先表达终局成功,再补过程奖励。
  • 每次只调整一个主要变量:环境、奖励、模型策略或训练参数。
  • 监控轨迹级指标,重点排查 reward hacking、无效动作和轮数膨胀。

多轮强化学习的收益来自 Agent 能连续推理和行动,但风险也来自连续性本身。环境、奖励、评测、监控四件事闭环之后,SageMaker AI 才能真正成为可靠的训练平台,而不是一台更昂贵的试错机器。


相关推荐