標籤: 企業級自動化

  • 把 Agent 關進沙盒:SaaS 實戰骨架

    把 Agent 關進沙盒:SaaS 實戰骨架

    📌 本文重點

    • Agent 要被關在嚴格 sandbox 與工具層裡
    • 記憶要分層,記流程不記祕密資訊
    • 用事件流與回放讓 Agent 可觀察、可控

    在 SaaS 裡塞一個 AI Agent,難點不是「會不會寫 prompt」,而是如何讓它在有限權限下,持久又安全地幫你自動化真實工作流程。沒 sandbox、沒記憶設計的 Agent,只適合做 demo:一旦上線,就會變成「拿著 admin key 的高智商腳本小孩」。

    這篇從 AI Agent Sandboxing for SaaSAI Agent Memory for SaaS 的思路出發,拆成你實作時一定會遇到的四個骨架:

    1. 權限與邊界:sandbox + 能力分級 + 審計/回放
    2. 記憶設計:短期 vs 長期組織記憶 + 何時忘記
    3. 資料模型與基礎設施:event sourcing + 任務關聯 + RAG 整合
    4. 開票/CRM 更新 Agent 實作雛形與踩坑清單

    重點說明


    1. Sandboxing:把 Agent 關在「業務安全區」裡

    目標:讓 Agent 有用,但永遠拿不到 root 权限

    💡 關鍵: 先設計權限邊界,再讓 Agent 介入,才能避免它變成拿著 admin key 的「高智商腳本小孩」。

    核心做法:

    1. 能力分級(建議至少三層)
    2. read-only:只能查詢 / 檢索(查訂單、查發票、查 CRM)
    3. scoped-write:限制在特定資源 + 明確條件(只能建立 invoice 草稿、只能改自己 owner 的 lead)
    4. admin-like:極少數動作(例如退款、刪除發票),預設關閉,需人工審批或 feature flag

    5. 工具層 sandbox(而非讓 LLM 直呼 DB / 外部 API)

    6. 對 LLM 暴露的是受控工具 API,例如:AgentTools.create_invoice_draft,而不是 POST /invoices 原始 API
    7. 工具層做 參數校驗、權限檢查、rate limit、審計 log

    8. 可回放測試 / 審計 log

    9. 每次 Agent 決策,記錄:
      • tool_call(名稱 + 參數)
      • 結果摘要(避免 log 泄露敏感資料)
      • 關聯 user_id / org_id / conversation_id / task_id
    10. 可以在 staging 用「回放同一串 event」重跑一遍,驗證升級後模型或 prompt 不會炸庫。

    2. 記憶設計:記住工作流,不記住祕密

    實務上可以拆成三層記憶:

    1. 短期上下文(working memory)
    2. 單次任務/對話的上下文,存在 conversation_state 或臨時向量 store
    3. 存活時間:幾分鐘到幾小時,任務結束後視情況壓縮成事件摘要

    4. 長期組織記憶(org memory)

    5. 公司政策、常見流程、產品價目表、範本回覆
    6. 存在 RAG + metadata(org_id, version, valid_from, valid_to)
    7. 修改政策時不覆蓋舊文,而是加新版本 + 標記舊版過期

    8. 個人偏好 / 使用者設定

    9. 比如:某 Sales 喜歡用英文回 mail、預設稅率 5%
    10. 存在 user_preferences 表或 key-value store,與 org policy 分離

    「何時該忘?」幾個實務策略:

    • 預設不把 user prompt 原文存成長期記憶,只保存「必要摘要 + 事件」,例如:
    • ❌ 存「請幫我開票給 XX 公司,統編 12345678,地址是…」
    • ✅ 存「2025-06-01 開立發票 INV-001, buyer=XX 公司, amount=10,000, owner=user_123」
    • retention policy
    • 短期記憶(對話內容)保留 30 天,之後只留聚合統計 / 匿名化摘要
    • 向量記憶可定期跑 job:找到稠密但從未被命中的 embedding → 刪除或降精度存儲

    💡 關鍵: 記憶層只存「去敏的業務事件」,既符合隱私需求,又保留足夠資訊讓 Agent 持續學習與優化。


    3. 資料模型與基礎設施:把 Agent 行為變成事件流

    為了可觀察、可回放,建議用輕量的 event sourcing 思路

    • agent_sessions:一次使用者啟動 Agent 的 session
    • agent_tasks:對應一個業務任務(例如「為 ticket#123 建立 invoice」)
    • agent_events:細顆粒度事件(tool call、LLM decision、error)

    搭配:

    • conversation_id:對話 thread ID(多輪聊天)
    • task_id:業務任務 ID(可以跨多個對話)
    • org_id / user_id:用來分庫、分 tenant、做權限控制

    與現有 DB/RAG 的整合方式

    • 業務資料留在原本的 transactional DB
    • Agent 不直接 query DB,而是走你包好的 BusinessAPI 或工具層 microservice
    • 長期記憶 / 知識庫:用 RAG(可參考 jamwithai/production-agentic-rag-course 的 patterns),但:
    • 純「查詢」→ read-only 工具
    • 「根據 RAG 結果改資料」→ 一律走 scoped-write 工具並寫 event log

    💡 關鍵: 把 Agent 所有操作轉成事件流,才能事後追蹤、審計與在 staging 做「重放實驗」。


    實作範例:開票/CRM 更新 Agent 雛形

    下面用 pseudo code 展示一個典型「讀 ticket → 建發票草稿 → 更新 CRM」的 sandbox + memory schema。


    1. 工具層 sandbox 定義

    // 工具層:只暴露給 Agent 這些「安全操作」
    
    interface AgentContext {
      orgId: string;
      userId: string;
      role: 'read_only' | 'scoped_write' | 'admin';
      taskId: string;
    }
    
    class AgentTools {
      constructor(private ctx: AgentContext) {}
    
      // 讀取支援 ticket(read-only)
      async getSupportTicket(ticketId: string) {
        assertRole(['read_only', 'scoped_write', 'admin'], this.ctx.role);
        const ticket = await TicketService.getById(this.ctx.orgId, ticketId);
        await AgentAudit.log({
          type: 'tool_call',
          tool: 'getSupportTicket',
          ctx: this.ctx,
          input: { ticketId },
          outputSummary: { status: ticket.status }, // 避免 log 敏感內容
        });
        return ticket;
      }
    
      // 建立發票「草稿」而非正式發票(scoped-write)
      async createInvoiceDraft(payload: {
        ticketId: string;
        customerId: string;
        amount: number;
        currency: string;
      }) {
        assertRole(['scoped_write', 'admin'], this.ctx.role);
    
        // 額外安全檢查:金額上限、防重複開票
        if (payload.amount > 10000) throw new Error('amount_exceeds_limit');
        await BusinessRules.ensureNoDuplicateDraft(
          this.ctx.orgId,
          payload.ticketId,
        );
    
        const invoice = await InvoiceService.createDraft({
          ...payload,
          orgId: this.ctx.orgId,
          createdBy: this.ctx.userId,
        });
    
        await AgentAudit.log({
          type: 'tool_call',
          tool: 'createInvoiceDraft',
          ctx: this.ctx,
          input: payload,
          outputSummary: { invoiceId: invoice.id },
        });
    
        return invoice;
      }
    
      // 更新 CRM:只允許更新部分欄位
      async updateCrmLead(leadId: string, patch: { status?: string }) {
        assertRole(['scoped_write', 'admin'], this.ctx.role);
        const safePatch = pick(patch, ['status']); // 避免 Agent 任意改 email 等敏感欄位
    
        const lead = await CrmService.updateLead(this.ctx.orgId, leadId, safePatch);
        await AgentAudit.log({
          type: 'tool_call',
          tool: 'updateCrmLead',
          ctx: this.ctx,
          input: { leadId, patch: safePatch },
          outputSummary: { status: lead.status },
        });
    
        return lead;
      }
    }
    

    2. Agent 任務流程(記憶與事件流)

    // 啟動一個 Agent 任務:從 ticket 開票 + 更新 CRM
    
    async function runInvoiceAgent(params: {
      orgId: string;
      userId: string;
      ticketId: string;
    }) {
      const taskId = await AgentTaskStore.create({
        orgId: params.orgId,
        userId: params.userId,
        type: 'INVOICE_FROM_TICKET',
        status: 'running',
      });
    
      const ctx: AgentContext = {
        orgId: params.orgId,
        userId: params.userId,
        role: 'scoped_write',
        taskId,
      };
    
      const tools = new AgentTools(ctx);
    
      // event sourcing:每一步都寫入 agent_events
      await AgentEventStore.append({
        taskId,
        type: 'task_started',
        payload: { ticketId: params.ticketId },
      });
    
      // 1) LLM 讀 ticket + 商業規則摘要(短期記憶)
      const ticket = await tools.getSupportTicket(params.ticketId);
    
      const policyDocs = await OrgPolicyRAG.search({
        orgId: params.orgId,
        query: '開立發票規則',
        topK: 3,
      });
    
      const llmInput = buildPrompt({ ticket, policyDocs });
    
      const llmDecision = await LLM.chatCompletion({
        model: 'gpt-4.1-mini',
        tools: [
          { name: 'createInvoiceDraft', schema: InvoiceDraftSchema },
          { name: 'updateCrmLead', schema: CrmPatchSchema },
        ],
        messages: [
          { role: 'system', content: SYSTEM_PROMPT },
          { role: 'user', content: llmInput },
        ],
      });
    
      await AgentEventStore.append({
        taskId,
        type: 'llm_decision',
        payload: safeDecisionLog(llmDecision),
      });
    
      // 2) 根據 LLM 決策安全執行工具
      const result = await ToolExecutor.run(llmDecision, tools);
    
      // 3) 將任務摘要存入長期「事件記憶」(去敏 + 可查詢)
      await AgentMemoryStore.saveTaskSummary({
        orgId: params.orgId,
        taskId,
        type: 'INVOICE_TASK_SUMMARY',
        summary: buildTaskSummary({ ticket, result }),
        // 設定過期策略:例如 180 天後自動清除
        expiresAt: dayjs().add(180, 'day').toDate(),
      });
    
      await AgentTaskStore.update(taskId, { status: 'completed' });
    
      return result;
    }
    

    3. Memory Schema(簡化版)

    -- 任務層級摘要,作為長期「安全記憶」
    CREATE TABLE agent_task_memory (
      id            BIGSERIAL PRIMARY KEY,
      org_id        VARCHAR(64) NOT NULL,
      task_id       VARCHAR(64) NOT NULL,
      type          VARCHAR(64) NOT NULL,
      summary_json  JSONB NOT NULL,   -- 已去識別 / 去敏的摘要
      created_at    TIMESTAMP NOT NULL DEFAULT now(),
      expires_at    TIMESTAMP NULL,
      INDEX idx_org_type_created (org_id, type, created_at)
    );
    
    -- 事件流,用於回放與審計
    CREATE TABLE agent_events (
      id            BIGSERIAL PRIMARY KEY,
      org_id        VARCHAR(64) NOT NULL,
      task_id       VARCHAR(64) NOT NULL,
      event_type    VARCHAR(64) NOT NULL, -- tool_call / llm_decision / error ...
      payload       JSONB NOT NULL,
      created_at    TIMESTAMP NOT NULL DEFAULT now(),
      INDEX idx_task_created (task_id, created_at)
    );
    

    建議與注意事項


    1. 常見踩坑

    1. 讓 Agent 拿到全庫 query 能力
    2. 例如暴露 run_sql(query) 這種工具 → 等於給 LLM 一把 DB root key
    3. 建議:只提供具體業務操作工具get_invoice_by_id / create_invoice_draft),不提供自由 SQL / 任意 filter

    4. 把 user prompt 直接當長期記憶存

    5. 風險:
      • 敏感資訊(住址、email、信用卡後四碼)被永久 index
      • 未來 RAG 檢索時把別人對話調出來
    6. 解法:只存事件摘要(例如:某天完成一筆開票),prompt 原文只能在短期 log / 加密 log 中保留,並設明確 retention

    7. 沒有 rollback / dry-run 機制

    8. Demo 時一切完美,上線後改個 prompt 就開始亂開票
    9. 建議:

      • 預設跑在 dry-run / shadow mode:只寫 event,不真正寫 DB,由人審批
      • 對高風險操作(刪除、退款)設計 雙階段提交流程:Agent 產生建議 → 人按下「Apply」才真正執行
    10. 把政策寫死在 prompt

    11. 政策一變,所有 Agent 行為都過期,但你不知道是哪個版本出的錯
    12. 建議:政策存 RAG / config store,prompt 只說「請依據最新的 org policy 回應」,並在 log 記錄使用的 policy_version_id

    2. 實戰建議(可直接用在專案裡)

    1. 先只讓 Agent 操作「草稿」資源
    2. 如範例:createInvoiceDraft,由人類在 UI 裡確認後再正式開票
    3. 這個模式在導入初期可以快速建立信任,也方便收集訓練資料

    4. 每個 Agent 任務都要有 task_id + org_id + user_id

    5. 方便之後做:

      • per-org 行為分析
      • 問題排查:「這張錯誤發票是哪個 Agent 任務生成的?」
      • 回放測試:「重跑這個 task,看新版模型會不會做出不一樣決策」
    6. 記憶層要先畫邊界,再決定用什麼向量庫

    7. 問自己三件事:
      • 哪些東西必須記一輩子(例如:已開立的發票、客戶同意條款紀錄)
      • 哪些只需要短期記憶(例如:這週正在處理的 ticket 狀態)
      • 哪些不該記(例如:一次性敏感資訊)
    8. 然後才決定:哪些用 transactional DB、哪些進向量庫、哪些只當 log 放 object storage + TTL

    9. 用事件流做 A/B 測試與回放

    10. 有了 agent_events 後,可以:
      • 在 staging 重播同一串事件,切不同模型 / prompt
      • 比較產生的 tool call 是否差異過大
      • 逐步從 demo 模式 → 實際寫入模式

    整體來說,把 Agent 裝進 SaaS,不是再多寫幾個工具函式,而是要把它當「受控的自動化子系統」來設計:

    • 用 sandbox 做權限邊界
    • 用多層記憶管理上下文與風險
    • 用事件流與回放讓它可觀察、可演進、可 debug

    一旦這套骨架打好,你的 SaaS 就可以從「有個聊天盒子」升級成「能自己處理開票、更新 CRM、遵守政策的半自動業務夥伴」。


    🚀 你現在可以做的事

    • 在現有 SaaS 服務中先列出所有「只允許草稿」的業務操作,設計對應的 scoped_write 工具層 API
    • 為你的 Agent 任務加上 task_id / org_id / user_idagent_events 表,開始記錄並觀察事件流
    • 審視目前有哪些資料被長期保存為向量或日誌,整理一份「應改成事件摘要、需設定 TTL」的清單並排入技術債處理計畫