生產級 Agent 架構：狀態機、持久化與重放

狀態機定義

from enum import Enum

class State(str, Enum):
    PENDING       = 'pending'
    THINKING      = 'thinking'
    TOOL_RUNNING  = 'tool_running'
    NEED_CONFIRM  = 'need_confirm'
    DONE          = 'done'
    FAILED        = 'failed'

TRANSITIONS = {
    State.PENDING:      {State.THINKING},
    State.THINKING:     {State.TOOL_RUNNING, State.NEED_CONFIRM, State.DONE, State.FAILED},
    State.TOOL_RUNNING: {State.THINKING, State.FAILED},
    State.NEED_CONFIRM: {State.TOOL_RUNNING, State.FAILED},
}

事件溯源：只追加，不覆寫

不要在資料庫裡存「當前狀態」然後不斷更新。改成記錄事件序列，當前狀態由事件重放得出。這樣你自然獲得完整稽核軌跡與重放能力。

# agent_events 表：只 INSERT，永不 UPDATE
{
  'run_id': 'r_8f2a', 'seq': 7, 'ts': '2026-08-29T10:22:31Z',
  'type': 'tool_result',
  'payload': {'tool': 'search_orders', 'ok': True,
              'result_hash': 'a91c…', 'latency_ms': 320},
  'state_after': 'thinking',
}

def rebuild(run_id):
    events = db.fetch_events(run_id)          # 依 seq 排序
    state = initial_state()
    for e in events:
        state = apply_event(state, e)
    return state

為什麼要能重放

• 除錯：可以完整重現使用者遇到的那一次執行。
• 回歸測試：把歷史執行當測試案例，改動後重放看行為是否改變。
• 稽核：受監管的場景需要證明「當時系統依據什麼做出決定」。
• 成本分析：離線重算不同策略下的成本，不必上線硬試。

工具結果要存原文

重放時如果重新呼叫外部 API，結果可能已經不同（訂單狀態變了、資料被刪了），重放就失去意義。所以工具的原始回傳要存下來，重放模式下從儲存讀取而非真的呼叫。

def dispatch(name, args, ctx):
    if ctx.replay:
        return ctx.recorded_result(name, args)   # 重放：讀存檔
    result = TOOLS[name](**args)
    ctx.record(name, args, result)               # 正常：呼叫並記錄
    return result

並行與冪等

分散式環境中，同一個 run 可能被兩個工作程序同時撿起。用資料庫的樂觀鎖處理：事件寫入時帶上預期的 seq，衝突就放棄本次執行。

-- seq 為 (run_id, seq) 唯一鍵，衝突即代表有人先寫入
INSERT INTO agent_events (run_id, seq, type, payload, state_after)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (run_id, seq) DO NOTHING
RETURNING id;

逾時與孤兒任務

• 每個 run 記錄 lease_until，工作程序定期續租。
• 背景任務掃描過期未續租的 run，標記為可重新撿取。
• 重試次數超過上限的 run 進入死信佇列，等人工處理。

漸進式落地

不必一開始就上事件溯源。務實的順序是：先把訊息串存進資料庫（可恢復）→ 加上狀態欄位（可查詢）→ 改成事件序列（可重放）。每一步都能單獨帶來價值。