RLHF - Kerwin AI Information

📌 本文重點

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 成可以重放的環境，你就能：

收集真實軌跡（trajectory）
在離線環境重放並評分
用 bandit / RL 訊號挑策略，而不是只看「模型句子好不好看」

3. Agentic RL 的基本構成

在工程上，你可以簡化為四個組件：

環境建模：統一封裝所有工具、API、資料庫，形成一個 Environment 介面
任務分解與規劃：讓模型先產生高階 plan，再逐步執行（而不是一口氣亂調用工具）
行動–觀察–反饋迴圈（loop）：每一步都記錄 (state, action, observation, reward)
長期回報設計：不是只有「這句回覆好不好」，而是「整個 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 的模式：

Shadow run：
新 Agent 在真實流量中「旁路」執行，但結果只記錄、不生效
比較新舊策略在同一批任務上的成功率、成本、風險
行為 diff（behavioral diff）：
對同一 input，收集舊策略與新策略的完整軌跡
比較：action 分布、工具使用頻率、完成時間、錯誤率
限權 sandbox：
新策略只允許讀取/寫入 sandbox DB 或「模擬帳號」
在確認行為安全後，再逐步放開至真實資源
Quota + kill switch：
設定每分鐘/每天的最大任務數、最大危險操作數（如更新信用卡）
發現異常行為立即切回穩定策略或人工處理

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

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

收集軌跡：
先在現有 Agent loop 中，完整記錄 (state, action, obs, reward)
reward 初期可以是簡化版：成功 / 失敗 / 人工接管 / 投訴
設計可重放環境：
把所有外部工具 access 透過統一的 Environment API 封裝
支援「重放模式」：用日誌中的 API response 而不打真實 service
離線 policy ranking / bandit：
用多種 prompt / temperature / tool-selection 策略跑同樣任務，離線評分
選出表現最好的作為新的 default policy，再進一步微調模型
逐步引入 RL 元件：
先做 Contextual Bandit：在不同任務類型選不同策略
再考慮 full RL：對整個 episode 的 policy 做梯度更新

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

🚀 你現在可以做的事

盤點現有 Agent 流程，開始在每一步記錄完整 (state, action, obs, reward) 軌跡

把所有外部 API/DB 操作封裝成統一的 Environment 介面，加入重放模式

為代表性的任務場景設計一版簡單的長期 reward，做離線 policy ranking 來選策略

標籤: RLHF

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