從 RLHF 到 Agentic RL:讓 Agent 真正會做事

從 RLHF 到 Agentic RL:讓 Agent 真正會做事

📌 本文重點

  • RLHF 只優化單輪輸出,不管整體任務
  • Agentic RL 把產品流程當決策過程來學
  • 先設環境與 reward,再收集軌跡做離線 ranking
  • 安全疊代要用 shadow run 與 sandbox 控制風險

現有多數商用 LLM 都經過 RLHF 微調,所以模型看起來「很乖」,會照格式回答、會道歉、會避開風險。但當你要它接 CRM、排行程、幫客服真正自動處理問題時,常見狀況是:

  • 要嘛只完成第一步就說「完成了」
  • 要嘛在工具調用之間 瘋狂 loop,一直改 plan 不收斂

💡 關鍵: RLHF 主要優化「單輪回答好不好看」,而不是整個多步流程是否真正完成任務。

痛點就是:RLHF 只優化單輪輸出品質,沒有優化「整個任務流程」的長期回報。Agentic RL 的目的,就是把整個產品流程當成可學習的決策過程,讓 Agent 在真實業務環境裡學會:怎麼分解任務、怎麼多輪調用工具完成目標、怎麼在成本/風險下做權衡。


重點說明

1. RLHF 與 Agentic RL:優勢與侷限

RLHF 做到的事:

  • 把基礎模型變成「有禮貌、守規則、對齊人類偏好」的聊天助手
  • 著重在單輪回合的回覆是否人類覺得好(helpful/harmless)

但 RLHF 不做這些:

  • 不關心模型在一整個多步任務上的總表現(例如 10 步內完成退貨流程)
  • 不優化工具使用策略:什麼時候查 DB、什麼時候寫 note、什麼時候結束

Agentic RL 多做的事:

  • 把 Agent 放在明確定義的 環境(API、DB、外部系統)裡學習
  • 用 任務分解 + 規劃 + 行動–觀察–反饋迴圈 來建模整個流程
  • 以「長期回報」作為訓練目標:成功率、成本、風險、用戶滿意度

💡 關鍵: Agentic RL 的訓練目標是整個 episode 的「長期回報」,而不是單一回答的文句品質。


2. 把真實產品抽象成可訓練的 MDP

你要做的,是把現有產品流程抽象成一個 MDP / 決策流程:

  • 狀態(state):Agent 目前看到的上下文 + 工具回傳 + 用戶狀態
  • 例:客服場景 = 聊天紀錄 + 工單狀態 + CRM 查詢結果
  • 行動(action):Agent 可以做什麼工具調用/決策
  • 例:CALL_SEARCH_TICKET, UPDATE_CRM_FIELD, SEND_REPLY, CLOSE_CASE
  • 轉移(transition):環境如何回應(API response / 用戶反應)
  • 回報(reward):最後有沒有解決問題?是否超時?是否誤操作?

核心是:先把整個流程 formalize 成可以重放的環境,你就能:

  1. 收集真實軌跡(trajectory)
  2. 在離線環境重放並評分
  3. 用 bandit / RL 訊號挑策略,而不是只看「模型句子好不好看」

3. Agentic RL 的基本構成

在工程上,你可以簡化為四個組件:

  1. 環境建模:統一封裝所有工具、API、資料庫,形成一個 Environment 介面
  2. 任務分解與規劃:讓模型先產生高階 plan,再逐步執行(而不是一口氣亂調用工具)
  3. 行動–觀察–反饋迴圈(loop):每一步都記錄 (state, action, observation, reward)
  4. 長期回報設計:不是只有「這句回覆好不好」,而是「整個 episode 是否達成業務目標」

常見場景:

  • 自動化客服:episode = 一次工單的完整處理
  • 研究代做:episode = 從 query 到交付報告的完整流程
  • 程式碼 refactor:episode = 一次 PR 的修改 + 測試 + 說明
  • 電話 / 日程代理:episode = 一次預約成功或明確被拒絕

實作範例:簡化版 CRM Agent + 軌跡記錄

以下是一個極簡化的 Python 範例,示範:

  • 用 GPT 類模型(假設有 chat_completion API)當 policy
  • 外面包一層 Environment,提供 search_customer / update_crm 兩個工具
  • 用簡單 reward:更新成功 + 回覆合理 = 高分
  • 記錄軌跡(trajectory),再用 bandit 式 ranking 挑出較佳策略
import uuid
from typing import List, Dict, Any

# ===== 環境定義 =====

class CRMEnvironment:
    def __init__(self, crm_client):
        self.crm = crm_client

    def step(self, state: Dict[str, Any], action: Dict[str, Any]):
        """
        action 的結構示例:
        {
          "type": "tool" | "respond" | "finish",
          "tool_name": "search_customer" | "update_crm",
          "params": {...},
          "reply": "給使用者的訊息"
        }
        """
        obs, reward, done = {}, 0.0, False

        if action["type"] == "tool":
            if action["tool_name"] == "search_customer":
                obs["customer"] = self.crm.search(action["params"]["email"])
            elif action["tool_name"] == "update_crm":
                success = self.crm.update(
                    customer_id=action["params"]["id"],
                    fields=action["params"]["fields"],
                )
                obs["update_success"] = success
        elif action["type"] == "respond":
            # 實際上這裡會寫入聊天系統
            obs["reply_ack"] = True
        elif action["type"] == "finish":
            done = True

        # 這裡只示意:如果更新成功且已 finish,給正向 reward
        if obs.get("update_success") and action["type"] == "finish":
            reward = 1.0
        return obs, reward, done


# ===== Policy(LLM Agent) =====

def llm_policy(model, state: Dict[str, Any]) -> Dict[str, Any]:
    """使用 LLM 決定下一步 action。"""
    system_prompt = """你是一個 CRM 自動化代理,
    只能使用以下操作:search_customer(email), update_crm(id, fields), respond(user_message), finish。
    請一步一步完成查詢客戶並更新 CRM,最後用 finish 結束。
    請以 JSON 輸出下一步 action。
    """

    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": state["task_description"]},
        {"role": "assistant", "content": str(state["history"])},
    ]

    resp = model.chat_completion(messages=messages)
    # 假設模型已被 prompt 成只輸出 JSON
    action = resp.to_json()  # pseudo-code
    return action


# ===== 執行 episode 並記錄軌跡 =====

def run_episode(env: CRMEnvironment, model, task_description: str):
    trajectory = []
    state = {"task_description": task_description, "history": []}

    for step_idx in range(10):  # 簡單防止無限 loop
        action = llm_policy(model, state)
        obs, reward, done = env.step(state, action)

        transition = {
            "state": state.copy(),
            "action": action,
            "obs": obs,
            "reward": reward,
        }
        trajectory.append(transition)

        # 更新 state
        state["history"].append({"action": action, "obs": obs})

        if done:
            break
    return trajectory


# ===== 離線 ranking:挑選較佳策略 =====

def offline_policy_ranking(trajectories: List[List[Dict[str, Any]]]):
    # 簡單 bandit:以 episode 總 reward 排序
    scored = []
    for traj in trajectories:
        total_r = sum(t["reward"] for t in traj)
        scored.append((total_r, traj))
    scored.sort(key=lambda x: x[0], reverse=True)
    return scored


# 用法示意
# env = CRMEnvironment(crm_client)
# trajectories = [run_episode(env, model_v1, "請幫這位客戶更新聯絡電話") for _ in range(100)]
# best_traj = offline_policy_ranking(trajectories)[0]
# 接下來可以分析 best_traj 對應的 LLM prompt / hyperparameters,
# 或作為離線 RL 資料再微調一個專用 Agent。

這個範例刻意簡化幾件事,但核心工程訊息是:

  • 把 tools/環境封裝成獨立的 Environment,而不是讓 LLM 直接打 API
  • 每一步記錄 transition,為後續離線學習準備資料
  • 先從 offline policy ranking / bandit 開始,不必一上來就做 policy gradient

💡 關鍵: 即使只用簡單的 bandit 排序,你也能開始選擇「整體流程表現更好」的策略,而不只是調句子風格。


建議與注意事項

1. Reward 設計:避免「為了拿高分做壞事」

常見 reward 組合:

  • 成功率:工單是否完成、預約是否成功
  • 用戶滿意度:CSAT、NPS、或簡化為「無投訴 + 無人工接管」
  • 成本:API 次數、模型 token 消耗、處理時間
  • 風險:是否觸碰敏感欄位、是否違反合規(GDPR、PCI 等)

做法建議:

  • 用線性組合:reward = w1*success - w2*cost - w3*risk
  • 對某些行為設硬性負 reward:如改動金融數據、刪除客戶資料
  • 對「過度延長對話」「工具瘋狂重試」設懲罰,避免 loop 不收斂

2. 常見坑

  • 代理重覆 loop 不收斂:
  • 沒有明確的 finish 條件 或 max_step 限制
  • reward 沒有懲罰冗長或失敗重試

  • sim2real 落差:

  • 在模擬環境學到的策略,放到 production 會因 API latency、錯誤率全變樣
  • 建議:影子執行(shadow run) + 回放真實日誌,逐步縮小差距

  • reward hacking:

  • 代理學會「只更新容易成功的欄位」或「不碰難案」,表面成功率高但業務壞掉
  • 需對「長期未處理案件」、「人工接管頻率」給負向訊號

  • 日誌與隱私合規:

  • 軌跡中常含 PII、金融資訊
  • 務必:匿名化、遮罩敏感欄位、分區存放訓練資料,並設 access control

3. 在 production 上安全疊代:如何不把系統玩壞

可以參考 shadow run + 行為 diff 的模式:

  1. Shadow run:
  2. 新 Agent 在真實流量中「旁路」執行,但結果只記錄、不生效
  3. 比較新舊策略在同一批任務上的成功率、成本、風險

  4. 行為 diff(behavioral diff):

  5. 對同一 input,收集舊策略與新策略的完整軌跡
  6. 比較:action 分布、工具使用頻率、完成時間、錯誤率

  7. 限權 sandbox:

  8. 新策略只允許讀取/寫入 sandbox DB 或「模擬帳號」
  9. 在確認行為安全後,再逐步放開至真實資源

  10. Quota + kill switch:

  11. 設定每分鐘/每天的最大任務數、最大危險操作數(如更新信用卡)
  12. 發現異常行為立即切回穩定策略或人工處理

4. 如果你現在在做 Agent 產品,短期可以先做哪些「Agentic RL 化」?

不需要一開始就上全套 RL pipeline,可以循序漸進:

  1. 收集軌跡:
  2. 先在現有 Agent loop 中,完整記錄 (state, action, obs, reward)
  3. reward 初期可以是簡化版:成功 / 失敗 / 人工接管 / 投訴

  4. 設計可重放環境:

  5. 把所有外部工具 access 透過統一的 Environment API 封裝
  6. 支援「重放模式」:用日誌中的 API response 而不打真實 service

  7. 離線 policy ranking / bandit:

  8. 用多種 prompt / temperature / tool-selection 策略跑同樣任務,離線評分
  9. 選出表現最好的作為新的 default policy,再進一步微調模型

  10. 逐步引入 RL 元件:

  11. 先做 Contextual Bandit:在不同任務類型選不同策略
  12. 再考慮 full RL:對整個 episode 的 policy 做梯度更新

結論:如果你的 Agent 現在只是「很會聊天但不會做事」,優先事項不是再加更多工具,而是設計可重放的 environment、明確的 reward,開始收集軌跡並做離線 ranking。這就是從 RLHF 走向 Agentic RL 的第一步。


🚀 你現在可以做的事

  • 盤點現有 Agent 流程,開始在每一步記錄完整 (state, action, obs, reward) 軌跡
  • 把所有外部 API/DB 操作封裝成統一的 Environment 介面,加入重放模式
  • 為代表性的任務場景設計一版簡單的長期 reward,做離線 policy ranking 來選策略

留言

發佈留言

發佈留言必須填寫的電子郵件地址不會公開。 必填欄位標示為 *