分類: AI 技術

  • 你的 Agent 不是防火牆:實務安全設計指南

    你的 Agent 不是防火牆:實務安全設計指南

    📌 本文重點

    • Agent 不能當安全邊界
    • 安全控制應落在工具層與基礎設施層
    • 透過分層權限、限額與 Guard 才能安全上線
    • 不要把 production credential 直接塞給 Agent

    很多團隊把「多代理、自動化工作流」直接接到真實帳號、真實金流、真實網路資源上,心裡想的是:

    反正我在 prompt 裡有說「不要亂刪資料、不要轉太多錢」。

    這篇的核心結論是:Agent 絕對不是安全邊界。你不能用「模型會乖乖聽話」來代替 RBAC、限額、rate limit、審計 log 等基本控管。本文用幾個真實事故做反推,給出可以直接套用的安全設計範式與程式碼範例。


    重點說明

    1. 四種常見災難模式

    1. 把 Agent 當人來信任
      PocketOS 的案例:AI coding agent 在 9 秒內刪掉 production DB 和所有備份,原因不是模型「壞」,而是:
    2. 找到 credential
    3. 直接呼叫具破壞性的 delete_database() API
    4. 沒有任何外部限制與二次確認

    💡 關鍵: 真實事故顯示,只要工具層沒有保護,Agent 能在數秒內造成不可逆的系統毀損。

    1. 授予過大、靜態權限
    2. API key 直接給到 Agent:讀寫同一組 credential,沒有 scope、沒有 TTL。
    3. 只想讓 Agent 「查詢」交易紀錄,卻順便給了「轉帳」權限。

    4. 缺乏金額與成本上限

    5. DN42 案例:Agent 為了掃描網路,瘋狂建立雲資源,最後把操作者帳單刷爆。
    6. 銀行 0.01 歐轉帳案例:小額轉帳流程缺乏額外風控,被用來做 prompt injection、流程繞過。

    7. 沒有動作級審計與防護

    8. 沒有 trace:你只看到「Agent 跑了一下」,卻不知道它 call 了什麼 API、帶了什麼參數、花了多少錢。
    9. 無 Guard:任何 prompt 被投餵進來,Agent 都會原樣帶著敏感資料丟到模型或外部 API。

    關鍵結論:不要用 prompt 當防火牆,安全邊界必須落在『工具層 / 基礎設施層』。


    2. 安全設計的技術骨架:分層權限、沙箱、限額、Guard

    可以把 Agent 系統拆成四層來設計安全性:

    1. 工具層(Tool Layer)
    2. 只提供「安全封裝」過的 API 給 Agent。
    3. 明確區分 讀工具 與 寫/破壞性工具。
    4. 在破壞性工具外再包一層 Guard + Policy Engine。

    5. 執行層(Execution / Sandbox Layer)

    6. Agent 的程式碼 & 工具呼叫,在 容器 / sandbox 中跑,掛上:

      • network egress policy
      • resource quota(CPU / RAM / disk)
      • IAM role with least privilege
    7. 費用與風險控制層

    8. 金額上限:單次指令 / 單日 / 單用戶的金額 cap。
    9. 速率限制:API Gateway 上對 每個工具 設定 QPS / burst。
    10. 執行次數 / token 上限:避免長鏈式工具呼叫刷爆成本。

    11. 觀測與審計層(Observability & Audit)

    12. 每一次工具呼叫都寫入 結構化審計 log。
    13. 對異常模式(相同 IP 大量轉帳、長時間掃描某網段)做 alert。
    14. 在銀行/企業內部,審計 log 應能回溯:誰的 Agent、基於哪個工作流、何時、對哪個客戶做了什麼操作。

    3. 對你的專案的實際好處

    這些額外的安全設計,對開發者的好處非常直接:

    • 讓你敢開放真實權限給 Agent,而不是永遠卡在 demo 階段。
    • 降低「一次失誤全毀」的 blast radius:即使 Agent 爆走,最多刪一個 tenant 的測試資料,而不是全區 production。
    • 讓合規與內部風控願意放行:有審計、有限額、可追蹤,才有機會上銀行、金融、企業內部關鍵流程。

    💡 關鍵: 安全骨架讓你可以在控制可承受風險的前提下,真正把 Agent 用在 production,而不是停留在展示環境。


    實作範例

    1. 安全的 Tool Schema 設計:讀寫分離 + 強制二次確認

    以銀行轉帳為例,先把工具拆成:

    • get_account_balance(純讀)
    • create_transfer_draft(建立草稿,不真正扣款)
    • confirm_transfer(只接受人類確認 Token)
    // TypeScript: 安全版 tool schema
    
    export const tools = {
      get_account_balance: {
        description: "查詢指定帳戶餘額(唯讀)",
        input_schema: {
          type: "object",
          properties: {
            account_id: { type: "string" }
          },
          required: ["account_id"],
          additionalProperties: false
        },
        // 後端實作會強制用呼叫者的 user_id 做授權檢查
      },
    
      create_transfer_draft: {
        description: "建立轉帳草稿,不會真的送出,會回傳 draft_id 與風險評分",
        input_schema: {
          type: "object",
          properties: {
            from_account: { type: "string" },
            to_account: { type: "string" },
            amount: { type: "number", minimum: 0.01 },
            currency: { type: "string", enum: ["EUR", "USD", "TWD"] },
            note: { type: "string" }
          },
          required: ["from_account", "to_account", "amount", "currency"],
          additionalProperties: false
        }
      },
    
      confirm_transfer: {
        description: "確認既有轉帳草稿,只接受人類確認 token",
        input_schema: {
          type: "object",
          properties: {
            draft_id: { type: "string" },
            user_confirmation_token: { type: "string" } // 只從前端 UI 注入,Agent 拿不到
          },
          required: ["draft_id", "user_confirmation_token"],
          additionalProperties: false
        }
      }
    } as const
    

    重點:

    • Agent 只能從模型側呼叫 get_account_balance / create_transfer_draft。
    • confirm_transfer 的 user_confirmation_token 必須來自人類 UI(例如 SMS OTP / 硬體 token),不透過模型。
    • 小額轉帳(例如 0.01 歐)仍要經過風險評分,避免被用來當作 prompt injection 的側信道。

    2. 在 API Gateway / RBAC 層包住 Agent 工具調用

    以下是假想的 API Gateway(以 Kong / Envoy 風格)設定,針對 Agent 工具做 角色 + 限額 控制:

    # gateway-routes.yaml
    
    routes:
      - name: agent-read-tools
        paths: ["/agent-tools/read"]
        methods: ["POST"]
        plugins:
          - name: jwt
            config:
              claims_to_verify: ["exp"]
              key_claim_name: "sub"  # 綁 agent instance id
          - name: acl
            config:
              whitelist: ["agent_read"]
          - name: rate-limiting
            config:
              minute: 120  # 每分鐘最多 120 次工具呼叫
    
      - name: agent-write-tools
        paths: ["/agent-tools/write"]
        methods: ["POST"]
        plugins:
          - name: jwt
          - name: acl
            config:
              whitelist: ["agent_write"]
          - name: rate-limiting
            config:
              minute: 10   # 寫操作極度限流
          - name: request-size-limiting
            config:
              allowed_payload_size: 64  # 防止一次送進超大批次破壞性操作
    

    配合後端 RBAC:

    // Node.js pseudo code: 在工具 handler 中做細粒度 RBAC + 金額限制
    
    function assertWritePermission(ctx: RequestContext, maxAmount: number) {
      if (!ctx.roles.includes("agent_write")) {
        throw new ForbiddenError("Agent has no write permission");
      }
    
      const amount = ctx.body.amount ?? 0;
      if (amount > maxAmount) {
        throw new ForbiddenError("Amount exceeds agent limit");
      }
    }
    
    app.post("/agent-tools/write/transfer", (req, res) => {
      const ctx = getContextFromRequest(req);
    
      // 例如每個 Agent 最高 50 EUR,超過必須走人類流程
      assertWritePermission(ctx, 50);
    
      // ... call internal transfer service
    });
    

    實際好處:你可以很放心地說「Agent 可以幫你轉帳」,但確定它永遠不會幫你一次轉出 10 萬,只會在可承受的風險範圍內操作。


    3. Guard 與敏感資料掃描:避免 Agent 自動外洩機密

    參考 Cursor 等實作,你可以在「呼叫模型前」加上三道 Guard:

    1. Input Guard:掃描要送進模型的內容,找出 API key / 密碼,紅標或遮罩。
    2. Output Guard:掃描模型輸出,要是模型要求「貼上你的私鑰」,直接攔截。
    3. Tool Guard:在執行工具前檢查參數是否違反政策(例如掃描內網、批次刪資料)。

    簡單的 Input Guard 例子:

    # Python pseudo code: input guard
    import re
    
    SECRET_PATTERNS = [
        re.compile(r"sk-[A-Za-z0-9]{32,}"),   # API key 格式
        re.compile(r"-----BEGIN PRIVATE KEY-----[\s\S]+?-----END PRIVATE KEY-----"),
    ]
    
    def redact_secrets(text: str) -> str:
        redacted = text
        for p in SECRET_PATTERNS:
            redacted = p.sub("[REDACTED_SECRET]", redacted)
        return redacted
    
    # 在送給 LLM 前
    prompt = redact_secrets(user_input + context_snippets)
    

    注意:不要只在前端掃,Agent 自己組裝的 context(例如程式碼、log、設定檔)也要過一遍 Guard,不然它會自己把 .env 塞進去。


    4. 審計與異常偵測:之後一定會被問到的東西

    在銀行或企業內部,合規與內審會問的問題通常是:

    • 這筆錯誤的轉帳 / 刪除操作是 哪個 Agent 做的?
    • 它當時看到什麼 context?是誰觸發的?
    • 是否有類似行為持續發生?

    你需要的是動作級審計 log:

    {
      "timestamp": "2026-06-13T09:01:23Z",
      "agent_id": "agent-123",
      "user_id": "user-456",
      "workflow_id": "payroll-v2",
      "tool_name": "create_transfer_draft",
      "input": {
        "from_account": "...masked...",
        "to_account": "...masked...",
        "amount": 42.5,
        "currency": "EUR"
      },
      "risk_score": 0.78,
      "policy_decision": "allowed",
      "cost_estimate": {
        "cloud_cost": 0.0004,
        "fee": 0.1
      }
    }
    

    這些 log 可以餵給 SIEM / 內部風控系統,針對:

    • 某 Agent 在短時間內大量建立轉帳草稿。
    • 某工作流突然開始頻繁掃描不應存取的網段(DN42 類似情境)。

    直接做告警或自動降權(例如暫時停用該 Agent 的 write tool)。

    💡 關鍵: 有結構化審計與異常偵測,才能在出事後追責與調整政策,而不是只能事後猜測。


    建議與注意事項

    1. 把「模型可以做什麼」當成風控產品,而不是單純開發功能

    實務上可以這樣落地:

    • 先設計 policy,再設計 tool:例如「Agent 單次轉帳上限 50 EUR、一日累計 200 EUR」,然後才決定工具 API。
    • 把工具視為「風控之後的介面」,而非直接包內部 microservice。

    2. 不要把 production credential 直接塞進 Agent

    常見坑:

    • 在 .env 裡放 DB_URL_PROD,Agent 的 code tool 一掃專案就看到了。

    • 解法:

    • Agent 跑在 專用 service account + IAM role 上。
    • 只能打透過 Gateway / policy engine 包過的 API,不直接碰 DB / message queue。

    3. 把「人類在 loop 裡」當成正式設計的一部分

    • 高風險動作預設需要人類確認:
    • 破壞性工具強制 user_confirmation_token。
    • UI 上顯示 Agent 的建議,讓使用者點擊確認。
    • 這不是「很土」;在銀行監管語境裡,這叫做 four-eyes principle(雙人覆核),是你讓 Agent 能真正上線的關鍵。

    4. 上線前的簡版安全 checklist

    你可以直接拿這份清單對照專案:

    1. 工具層:
    2. [ ] 讀寫工具有清楚分離?
    3. [ ] 破壞性工具是否有二次確認 / 額外 policy?

    4. 權限與憑證:

    5. [ ] Agent 使用的 credential 是否有最小權限與有效期限?
    6. [ ] Agent 是否只能透過 Gateway / policy engine 存取內部服務?

    7. 費用與風險上限:

    8. [ ] 有設定單次 / 單日金額上限?
    9. [ ] 有 API rate limit / 執行次數 / token cap?

    10. Guard:

    11. [ ] 呼叫模型前是否做敏感資料掃描與遮罩?
    12. [ ] Tool 執行前是否跑過 policy check?

    13. 觀測與審計:

    14. [ ] 所有工具呼叫都有結構化 log?
    15. [ ] 有針對異常模式的 alert(大量轉帳、大量雲資源建立)?

    只要你把安全邊界畫在這些「實際可控的層」上,而不是畫在 prompt 上,你的 Agent 就能在真實環境裡幫忙做事,而不是在 9 秒內幫你把公司刪掉。

    🚀 你現在可以做的事

    • 審查現有 Agent 工具清單,將讀寫操作拆分並為破壞性工具加上二次確認機制
    • 在 API Gateway 為 Agent 加上角色、rate limit 與金額上限等策略,並實作細粒度 RBAC
    • 為所有 Agent 呼叫流程加入敏感資料 Guard 與結構化審計 log,串接既有監控/風控系統
  • DiffusionGemma 擴散式文字生成實戰指南

    DiffusionGemma 擴散式文字生成實戰指南

    📌 本文重點

    • DiffusionGemma 把推理瓶頸從記憶體帶寬轉為純算力
    • 在短文本任務上可達約 3–4 倍吞吐提升
    • 品質略遜自回歸 LLM,適合作為「快但不精」支線

    DiffusionGemma 解決的是一個很單純、但很痛的點:自回歸 LLM 在高 TPS / 低延遲場景下,推理效能很難再壓榨。Diffusion 式文字生成把瓶頸從記憶體帶寬移到純算力,讓你在同一張 GPU 上,以一次處理整段 token 的方式,換到最高約 4 倍的輸出速度──代價是文字品質會略輸主流 LLM。對有既有推理集群的團隊,這是一條可以平行拉起的新「推理線」,用來承接對質量沒那麼敏感的流量。

    💡 關鍵: DiffusionGemma 透過一次處理整段 token,實測有機會達到約 4 倍輸出速度,適合追求高 TPS 的場景。


    重點說明:Diffusion 式文字生成 vs 自回歸 LLM

    1. 核心機制:一次優化整段 token

    傳統自回歸 LLM:

    • 每步只生成 1 個 token,依賴 KV cache 重用過去注意力結果
    • 每往前一步,都需要讀寫大量 KV cache,記憶體帶寬 很快變成瓶頸

    DiffusionGemma:

    • 一次初始化 固定長度(目前約 256 token)的序列為噪聲
    • 經過多步 去噪迭代(Uniform State Diffusion),每一步都同時更新整段序列
    • 不做逐 token 自回歸,而是像圖片 diffusion 那樣,不斷 refine 整個「句子影像」

    結果:

    • 前向步數固定(例如 10~20 步),沒有「越長越慢」的線性 token-by-token 開銷
    • 所有 token 一起算,計算圖相對規整,KV cache 開銷大幅減少
    • 推理瓶頸從 HBM 帶寬轉成算力,H100 這種高 FLOPS 卡會特別吃香

    💡 關鍵: 固定步數、整段同時更新,讓 Diffusion 模型在長度固定的短文本上顯著減少記憶體瓶頸。

    2. 效能特性:TPS 變高,但 max length 有限

    從社群與官方數據:

    • 單卡 H100 上可達 ~1000 tokens/s,約同級自回歸模型的 3–4 倍
    • 目前序列長度主打 短序列區間(~256 token),不適合超長上下文
    • 吞吐量隨 batch size 比較線性地提升,適合高併發、標註型任務

    結論:短回答、標註、摘要、即時互動,是 DiffusionGemma 最自然的戰場;長對話、多輪推理、嚴謹產出,仍然交給主力自回歸 LLM。

    3. 品質與適用場景

    已知特性:

    • 語言流暢度 OK,但邏輯一致性、長段落結構弱於同級自回歸模型
    • 某些語言 / domain(特別是英文以外)會出現語氣不穩定、專有名詞錯誤
    • 具備「重新注入噪聲重寫」的機制,可以多次生成做 rerank / filter

    實務上的「速度 vs 品質」切分:

    • 可以用 DiffusionGemma 的場景:
    • 大量 資料標註 / 自動摘要(例如標註說明文字、標記類別、產生粗稿)
    • 內部工具:開發文件整理、會議記錄內部摘要
    • 即時互動:客服工具的「第一輪回覆草稿」、遊戲 NPC 對話草稿
    • 不要用 DiffusionGemma 當主力的場景:
    • 對內容正確性要求高的產出:法務、醫療、財務建議
    • 面向最終客戶的長篇行銷文、深度技術文章

    擬實作範例:在 Hugging Face / vLLM 拉起一條 Diffusion 線

    以下範例假設你已經有基本 HF / vLLM 環境,目標是:在現有集群中多掛一條 DiffusionGemma 推理服務,方便做 A/B。

    1. Hugging Face Transformers 部署與簡單壓測

    注意:實際模型 repo 名稱與 API 會以 Google / HF 官方釋出為準,以下以假想名稱 google/diffusion-gemma-26b 示意。

    from transformers import AutoTokenizer, AutoModelForCausalLM
    import torch, time
    
    MODEL_ID = "google/diffusion-gemma-26b"
    
    device = "cuda"
    
    tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
    model = AutoModelForCausalLM.from_pretrained(
        MODEL_ID,
        torch_dtype=torch.bfloat16,
        device_map="auto"
    )
    
    prompt = "用三點整理,說明為什麼擴散式文字生成在推理效能上可能優於自回歸 LLM:"
    inputs = tokenizer([prompt] * 16, return_tensors="pt", padding=True).to(device)
    
    # 假設 DiffusionGemma 暴露一個專用的 generation API,例如 use_diffusion=True
    start = time.time()
    outputs = model.generate(
        **inputs,
        max_new_tokens=128,
        do_sample=True,
        temperature=0.7,
        **{"use_diffusion": True, "num_diffusion_steps": 16}
    )
    end = time.time()
    
    texts = tokenizer.batch_decode(outputs, skip_special_tokens=True)
    print(texts[0])
    
    total_tokens = outputs.shape[1] * outputs.shape[0]
    print("TPS:", total_tokens / (end - start))
    

    工程重點:

    • torch_dtype=torch.bfloat16:在 A100/H100 上幾乎是 must,節省記憶體並吃到張量核心
    • batch_size 建議從 8–32 試起,觀察 TPS 和 latency;Diffusion 模型通常對大 batch 更友善
    • num_diffusion_steps:步數越少越快,但品質下降,這是你可以直接調的「品質/速度旋鈕」

    可以用相同 prompt,分別對:

    • DiffusionGemma(use_diffusion=True)
    • 對應大小的自回歸 Gemma 4 / Llama 3.1

    測:

    • 單請求 latency
    • 在 batch=16, 32 時的 TPS

    用最簡單的方式做「這台卡上我實際的吞吐/延遲比是多少」。

    💡 關鍵: 透過同卡 A/B 壓測,你能直觀看到 Diffusion 與自回歸模型在 TPS 與 latency 上的實際差異。

    2. vLLM 伺服器部署範例

    vLLM 已宣稱支援 DiffusionGemma 類模型,部署可以沿用既有流程,只是要注意 max length 與 scheduler 參數。

    啟動伺服器:

    vllm serve google/diffusion-gemma-26b \
      --dtype bfloat16 \
      --tensor-parallel-size 2 \
      --port 8009 \
      --max-model-len 256 \
      --gpu-memory-utilization 0.9
    

    簡易 A/B 路由(Python 偽碼):

    import requests
    
    def call_vllm(url, prompt, max_tokens=128):
        payload = {
            "model": "google/diffusion-gemma-26b",
            "prompt": prompt,
            "max_tokens": max_tokens,
            "extra_body": {
                "use_diffusion": True,
                "num_diffusion_steps": 16
            }
        }
        r = requests.post(url, json=payload, timeout=10)
        return r.json()["text"]
    
    prompt = "幫我產生一段 200 字內的產品說明草稿,主題:雲端備份服務"
    
    text = call_vllm("http://diffusion-gemma-host:8009/generate", prompt)
    print(text)
    

    實務設定建議:

    • --max-model-len 保守設在 256 或官方建議值,避免 OOM 或品質崩壞
    • Diffusion 特性讓你可以把 gpu-memory-utilization 拉高一點,但要壓測記憶體尖峰
    • 若你原本就有 vLLM 服務,只需要 額外掛一個新的 port,指向 DiffusionGemma,即可開始在 gateway 做 A/B

    建議與注意事項:多模型路由與常見坑

    1. 架構層:多模型路由策略

    實務上較穩的做法是:

    • 主力自回歸模型(例:Llama / Gemma 4)
    • 用於:高品質輸出、長上下文、多輪對話
    • DiffusionGemma 作草稿 / 預生成
    • 先用 DiffusionGemma 生成短答案 / 草稿(速度快)
    • 視需求:
      • 直接用於內部場景(標註、內部摘要)
      • 或交給主力 LLM 做 rewrite/refine,縮短主力模型的「思考時間」

    簡單路由邏輯(pseudo-code):

    def route_request(task_type, prompt):
        if task_type in ["internal_summary", "dataset_label", "first_draft"]:
            return call_diffusion_gemma(prompt)
        else:
            return call_main_llm(prompt)
    

    Gateway / API 層可以加上 header 或 task tag,決定是否走 Diffusion 線。

    2. 專用 GPU vs 混合佈署

    • 專用 GPU 服務:
    • 優點:容易調 batch / scheduler,壓到極致吞吐
    • 用途:批次標註、離線摘要、模型自訓練資料生成
    • 混合佈署(與主力 LLM 共用節點):
    • 優點:無需新增節點,只多開 vLLM service
    • 風險:記憶體 / SM 資源管理變複雜,容易因為排程不佳造成抖動

    如果你的集群已接近飽和,建議:

    • 先在 一小部分節點 拉起 Diffusion 線做壓測
    • 確認 TPS & 成本優勢後,再決定是否專門拉一小 pool 做「標註工廠」

    3. 目前實測常見坑

    1. 輸出穩定度

    2. 在同樣 prompt 下,DiffusionGemma 的輸出變異度通常會比自回歸高

    3. 建議:

      • 對重要任務做 n 次生成 + rerank(例如用主力 LLM 打分)
      • 或限制 temperature、top_p,改用 deterministic 設定測基線
    4. max length 與截斷問題

    5. 模型目前針對短序列設計,超長 prompt 或 max_new_tokens 易導致:

      • 品質崩壞(後面亂飄)
      • 直接 OOM 或 latency 飆高
    6. 建議:

      • 在 API gateway 做 輸入長度上限檢查
      • 對需要長輸出的任務,直接路由回主力 LLM
    7. 語言 / domain 弱項

    8. 英文效果通常最好;中文、程式碼、專業術語出錯率偏高

    9. 實務做法:
      • 中文/多語任務:先用 DiffusionGemma 生成英文草稿,再用主力 LLM translate + refine
      • 特定 domain(醫療、金融):不要讓 DiffusionGemma 直接面對終端用戶,最多做內部摘要

    總結:DiffusionGemma 在你專案裡的實際位置

    如果你的系統:

    • 已經有一套穩定的自回歸 LLM 服務
    • 又需要大量中等品質、短文本輸出(標註、摘要、內部工具)

    那麼 DiffusionGemma 是一條值得立刻拉起來 A/B 的「實驗線」:

    • 好處:在專用 GPU 上實測有機會撿到 3–4 倍 TPS,單位 token 成本下降
    • 代價:語言品質略弱、max length 限制大、輸出穩定度較差

    將它放在:

    • 多模型路由中的「快但不精」支線
    • 或 主力 LLM 的草稿生成前置

    就能在不牺牲核心體驗的前提下,把推理成本再往下壓一段,並為未來可能普及的擴散式文字架構預先打通工程路線。

    🚀 你現在可以做的事

    • 在現有 GPU 上用同一組 prompt,對 DiffusionGemma 與主力自回歸 LLM 做一次 TPS / latency 壓測 A/B
    • 在 gateway 或 API 層加入 task_type 路由邏輯,先讓內部標註與摘要流量導向 Diffusion 線
    • 在 vLLM 或 HF 環境中實際部署一條 google/diffusion-gemma-26b 服務,觀察一週內的實際成本與穩定度
  • 在 16GB 跑 35B MoE:Luce Spark 實戰

    在 16GB 跑 35B MoE:Luce Spark 實戰

    📌 本文重點

    • 以 bounded GPU cache 在 16GB GPU 上跑 33–35B MoE
    • 路由器會自我調整常駐熱門 experts,提升 cache 命中率
    • 適合互動式 chat/Agent,而非大規模批次推理

    在本地推理上,最大的痛點通常是:

    1. 想要更聰明的模型(>30B),但 GPU 只有 12–16GB;
    2. 傳統 offload 雖然能把參數塞進去,卻帶來可怕的 PCIe 來回搬運延遲,對互動式 chat 幾乎不可用。

    Luce Spark 的關鍵突破是:用一個 bounded GPU cache + 自我調整路由器 的 MoE 機制,只把「當下會被用到的 experts」熱載到 GPU,其他放在 RAM,以此在 16GB GPU 上穩定跑 33–35B MoE,而不付典型 offload 的延遲稅。對你的專案,最直接的好處是:

    • 在 單卡消費級 GPU 上,拿到 明顯優於 7B/8B dense 的品質;
    • 互動延遲仍在可接受範圍,適合 chat、Agent、RAG;
    • 一條命令即可啟動,不用自己刻複雜的 offload pipeline。

    💡 關鍵: 利用 bounded GPU cache,只常駐少量熱門 experts,就能在 16GB GPU 上實用地跑 33–35B MoE 模型,避免傳統 offload 帶來的大量延遲。


    重點說明:Luce Spark 的三個工程關鍵

    1. 只熱載 active experts 的 bounded GPU cache

    Luce Spark 的 MoE 層大致長這樣(概念化):

    class SparkMoELayer(nn.Module):
        def __init__(self, num_experts, top_k, gpu_cache_size):
            self.router = Router(num_experts, top_k)
            self.expert_store = RAMExpertStore(num_experts)
            self.gpu_cache = BoundedGPUCache(capacity=gpu_cache_size)  # 核心
    
        def forward(self, x):
            # 1) 路由計分,決定這個 batch 要用哪些 experts
            route_scores = self.router(x)
            active_experts = select_topk_experts(route_scores)
    
            # 2) 保證 active_experts 在 GPU cache 中
            for eid in active_experts:
                if not self.gpu_cache.has(eid):
                    weights = self.expert_store.load_from_ram(eid)
                    self.gpu_cache.insert(eid, weights)  # 可能觸發 eviction
    
            # 3) 執行 MoE 推理(此時所有 active_experts 都在 GPU)
            outputs = []
            for eid in active_experts:
                expert = self.gpu_cache.get(eid)
                outputs.append(expert(x))
            return aggregate(outputs, route_scores)
    

    工程重點:

    • GPU 只存一小部分高頻 expert(例如 8–16 個),其餘放在 RAM;
    • cache 超出容量時,按照 最近使用頻率/路由權重 做 eviction;
    • 這類 cache 的 hit 率會隨著路由器調整逐漸提高,形成 穩定的熱門 experts 集合。

    這和傳統 offload 整層權重 / 按 layer streaming 的差別在於:Luce Spark 是 按 expert 粒度換入換出,而且不追求「全都裝上 GPU」,而是追求 高 cache hit 的少數專家。


    2. 路由器如何自我調整常駐專家

    Luce Spark 的路由器一開始對各 expert 並不了解,剛啟動時會出現:

    • 錯誤路由 → 頻繁觸發 cache miss → RAM→GPU 搬運多,延遲抖動;
    • experts 使用分佈不穩定,GPU cache 很難收斂到固定常駐集合。

    Luce Spark 的做法是:

    1. 在線統計每個 expert 的路由命中頻率(可視為 usage counter);
    2. 在 routing logits 上加入 輕量的 regularization / bias,鼓勵高頻專家更常被選到,同時避免單一 expert 過載;
    3. 這些統計在執行過程中持續更新,並在 重啟時重新載入先前的 usage profile,實現「帶記憶的熱啟動」。

    用比較工程的方式想:

    • 路由器 ≈ 一個動態學習的 expert ranking 模型;
    • GPU cache ≈ 硬體受限的 LRU + 熱點優先策略;
    • 熱啟動後,熱門 experts 幾乎常駐 GPU,新請求的 cache hit 率自然很高。

    這就是為什麼官方會強調:不需要離線 calibration 或額外語料,因為路由器會自己靠線上流量學出一組適合你 workload 的常駐 experts。

    💡 關鍵: 路由器透過線上 usage 統計與熱啟動機制,會漸進式「記住」你的 workload,把常用 experts 固定留在 GPU。


    3. RAM↔GPU 交換對延遲/吞吐的實際影響

    Luce Spark 的核心 claim 是「without the offload tax」。這裡要拆成兩個維度看:

    1. 互動式 chat(低 batch、長對話)
    2. 熱啟動後,route 分佈趨於穩定,多數 token 都命中 GPU cache;
    3. 偶爾遇到新的語境時,才需要從 RAM 拉少量不常用 expert,上升的是 尾延遲 而不是平均延遲;
    4. 整體體感比傳統 offload 模式順很多,適合當 主力 chat/Agent backend。

    5. 高吞吐批次推理(大 batch、多併發)

    6. 每 batch 涉及的 experts 數量暴增,cache hit 率下降;
    7. RAM↔GPU 數據搬運變得頻繁,吞吐下降明顯;
    8. 如果你在做 大規模批次評估、生成資料集,Luce Spark 未必比直接上大卡跑 dense 模型划算。

    結論:

    • Luce Spark 更像是 「local chat/Agent 專用 MoE backend」,而不是通用批次推理引擎;
    • 如果你的 workload 是「大量同步批處理」而不是「真人互動」,不建議把 dense backend 全換成這類 MoE。

    實作範例:在 16GB 機器上跑 35B MoE

    以下以假想的 CLI / config 形式示意 Luce Spark 的操作風格,重點是 思路與關鍵參數,實際 API 請對照官方 repo。

    1. 啟動指令與最小設定

    假設你有一台:

    • GPU:RTX 3090 / 4080 / 4060 16GB
    • RAM:64GB(實務上建議 至少 48GB+,越大越穩)

    啟動 35B MoE 後端:

    luce-spark serve \
      --model qwen3.6-35b-a3b \
      --gpu-memory 14GiB \
      --gpu-cache-experts 8 \
      --router-profile-path ./router_state.json \
      --port 8000
    

    關鍵參數說明:

    • --gpu-memory:限制模型在 GPU 上的最大佔用,預留 1–2GB 給 KV cache / 系統;
    • --gpu-cache-experts:bounded GPU cache 容量,對應 同時常駐的 expert 數目;
    • --router-profile-path:路由器使用統計持久化路徑,幫你做「熱啟動」。

    如果是本地 HTTP API 模式,可以在前面再包一層,例如:

    uvicorn luce_spark.api:app --host 0.0.0.0 --port 8000
    

    2. 設定檔範例:控制 cache 策略與路由監控

    你可以用一份簡單的 YAML 控制 cache 行為與監控:

    model: qwen3.6-35b-a3b
    server:
      host: 0.0.0.0
      port: 8000
    
    spark:
      gpu_memory_limit_gib: 14
      gpu_cache:
        max_experts: 8          # 常駐 expert 數
        eviction_policy: lru    # 或: hot-score
        warmup_tokens: 20000    # 熱身期間不嚴格淘汰
    
    router:
      profile_path: ./router_state.json
      stats_window: 10000       # 計算 usage 的滑動窗口長度
      balance_penalty: 0.02     # 防止少數 expert 過載的正規化強度
    
    monitor:
      enable_prometheus: true
      metrics_port: 9100
      log_interval_sec: 5
    

    這類設定讓你:

    • 用 max_experts 明確控制 GPU cache 壓力;
    • 用 warmup_tokens 來降低冷啟時的抖動;
    • 用 balance_penalty 抑制路由分佈極端不均。

    3. 監控 GPU / RAM / 路由命中率

    常見監控指標(Prometheus 風格示意):

    luce_gpu_memory_used_bytes
    luce_gpu_cache_expert_count
    luce_gpu_cache_hit_ratio
    luce_router_expert_usage{expert_id="7"}
    luce_ram_usage_bytes
    luce_inference_latency_ms_bucket
    

    推薦實務做法:

    watch -n 1 nvidia-smi
    htop      # 監控 RAM / swap
    curl localhost:9100/metrics | grep gpu_cache
    

    要特別盯這幾件事:

    • RAM 使用率:接近 100% 時、或開始大量使用 swap,就要立刻調低 context 長度 / 並發數 / gpu_cache_experts;
    • luce_gpu_cache_hit_ratio:熱啟後應該穩定在高位(例如 >0.9),如果長期很低,代表 workload 太發散或 router 設定有問題;
    • per-expert usage:若少數 expert 的 usage 異常高,可能需要調整 balance_penalty 或減小 top_k。

    4. 在 RAG / Agent 流程中整合 MoE backend

    假設你已有一個典型的 Python RAG/Agent 服務,只要把原本的 dense LLaMA backend 換成 Luce Spark 的 HTTP endpoint 即可。

    簡易推理 client 範例:

    import requests
    
    LUCE_ENDPOINT = "http://localhost:8000/v1/chat/completions"
    
    def moe_chat(messages, temperature=0.7):
        payload = {
            "model": "qwen3.6-35b-a3b",
            "messages": messages,
            "temperature": temperature,
            # 對 MoE backend 特別有用的 hint
            "metadata": {
                "session_id": messages[0].get("session_id", "default"),
                "task_tag": "rag_qa"  # 可幫助 router 收斂某些專家
            }
        }
        r = requests.post(LUCE_ENDPOINT, json=payload, timeout=60)
        r.raise_for_status()
        return r.json()["choices"][0]["message"]["content"]
    
    # RAG pipeline 中直接替換這個 call
    answer = moe_chat([
        {"role": "user", "content": "根據上面的文件,說明 Luce Spark 的快取機制。"}
    ])
    

    幾個整合上的實務建議:

    • RAG 上下文長度:在 16GB 上使用 35B MoE,要注意 長 context + 大 batch 會同時吃掉 GPU RAM(KV cache)與系統 RAM(experts);
    • Agent 多工具呼叫:同一個 session 建議重用 session_id,讓路由器能對此類對話學出穩定的 expert set;
    • 混合架構:可以保留原本的 7B dense 作為 high-throughput worker,Luce Spark 35B MoE 只接「需要高品質回答」的請求。

    建議與注意事項:哪些專案值得換?

    1. 適合 / 不適合的 workload

    適合:

    • 本地 chatbot / Agent / RAG QA,以互動體驗為主;
    • 中小團隊、個人開發者,只有 1 張 12–16GB GPU,但想提升模型品質;
    • 對 tail latency 有一定容忍度,但不能接受 offload 帶來的「整體都變慢」。

    不適合:

    • 大規模 批次生成 / 資料標註 / 雙塔 embedding 推理;
    • 延遲要求極端嚴格,且流量型態非常 diverse 的線上服務;
    • RAM 不足(<32GB)或機器經常跑其他吃 RAM 的服務。

    2. 常見踩坑 & 規避方式

    1. 冷啟時延遲抖動嚴重
    2. 現象:剛開機的前幾千 tokens,latency 波動明顯;
    3. 緩解:

      • 先用腳本做一輪「暖身推理」,覆蓋幾個主力場景:
        bash
        python warmup.py --endpoint http://localhost:8000
      • 調整 warmup_tokens,在熱身期間減少 aggressive eviction;
      • 持久化 router_profile_path,重啟時載入。
    4. RAM 不足導致 swap,整體卡死

    5. 現象:nvidia-smi 看起來 GPU 利用率不高,但系統整體很慢;
    6. 緩解:

      • 嚴格限制 最大並發 / 每請求 context 長度;
      • 減少 gpu_cache_experts,讓單次 active experts 集合縮小;
      • 監控 swap 使用,一旦 swap 大於幾百 MB,就要降載或升級 RAM。
    7. 路由分佈不均,少數 experts 過載

    8. 現象:某些 expert usage 長期偏高,GPU cache 命中率不穩;
    9. 緩解:

      • 調升 balance_penalty 或引入 temperature scaling,讓路由更分散;
      • 檢查是否某類請求 pattern 過於集中(例如只有單一業務場景),必要時拆成不同服務。
    10. 誤以為可以完全替代 dense backend

    11. 建議:
      • 對於批量任務,仍保留 dense LLaMA 類模型 作為 batch worker;
      • 將 Luce Spark 定位成 「少量高價值請求」的 premium backend。

    💡 關鍵: Luce Spark 適合作為高品質旁路 backend,而不是全面取代所有 dense 模型的通用推理引擎。


    3. 是否值得遷移:一個簡單的決策準則

    可以用這三個問題評估:

    1. 你的主要 workload 是人機互動(chat/Agent/RAG)嗎?
    2. 你的機器 RAM 至少有 48GB,可以給模型用 32GB 以上嗎?
    3. 你願意接受前幾分鐘的熱身時間,以及偶發的尾延遲尖峰嗎?

    如果 答案至少有兩個是「是」,那麼把現有 dense LLaMA backend 加一個 Luce Spark 35B MoE sidecar,作為高品質路徑,是非常值得嘗試的升級。


    總結:Luce Spark 用 bounded GPU cache + 自我調整路由,把傳統 offload 的延遲稅轉化成「可控的少量 RAM↔GPU 交換」,讓 16GB GPU 也能實用地跑 35B MoE。對已經有 LLM backend 的團隊,最實際的做法不是「全部換掉」,而是把它當成一個 高品質 MoE 旁路,專門處理那些你不想交給 7B/8B dense 的關鍵請求。

    🚀 你現在可以做的事

    • 在你的 12–16GB GPU 機器上,仿照文中的 CLI 與 YAML,啟一個 qwen3.6-35b-a3b 的 Luce Spark 測試服務
    • 把現有 RAG/Agent 專案中的 dense backend 呼叫,替換成文中的 moe_chat() HTTP client,在部分流量上 A/B 測試效果
    • 使用 nvidia-smi、htop 與 Prometheus 指標,實際觀察 gpu_cache_hit_ratio、RAM 使用與尾延遲,調整 gpu_cache_experts 與 warmup_tokens 等參數
  • 用 ASSERT 替 AI Agent 寫單元測試

    用 ASSERT 替 AI Agent 寫單元測試

    📌 本文重點

    • ASSERT 用文字規格測「整個 Agent 流程」
    • 可 mock 工具與政策,做可預期的回歸測試
    • 非決定性輸出用「性質斷言」而非比字串

    多數團隊在做 LLM/Agent 開發時,測試痛點很具體:

    • 同一個 prompt 今天過、明天壞,沒有紅綠燈,只能祈禱
    • 多 Agent + 工具調用後,bug 出在「流程」不是「回答內容」,傳統單元測試很難 cover
    • 合規、安全團隊寫了一堆 policy,難以自動驗證 Agent 是否真的遵守

    微軟開源的 ASSERT 直接對準這個痛:用純文字規格定義 Agent 的預期行為,讓你像寫單元測試一樣寫回歸測試,從「這個答案正不正確」提升到「整個 Agent workflow 行為可預期」。


    重點說明

    1. 從「回答對不對」到「行為對不對」

    傳統 LLM 評估多半是:

    • 給一段 input
    • 看 output 文本是否符合 ground truth / rubric

    ASSERT 的思維是:

    測試的是 Task + Workflow:給定初始指令、工具與外部環境,整個 Agent 互動過程是否符合文字規格中的 行為斷言。

    它特別適合:

    • 多 Agent 協作(例如 Planner → Worker → Reviewer)
    • 有工具調用(DB 查詢、API call、程式執行)
    • 有政策/合規約束(不得外洩個資、不得跨區讀資料)

    💡 關鍵: ASSERT 把測試焦點從單次回答,提升到整個任務與 workflow 的行為是否符合規格。


    2. ASSERT 規格長什麼樣:文字規格 + 執行模型 + 斷言機制

    ASSERT 的核心是測試規格檔(YAML / JSON / 純文字皆可包裝),通常包含三部分:

    1. scenario:描述這次要跑的任務
    2. execution:怎麼把這個 scenario 丟給 Agent workflow
    3. assertions:要驗證哪些行為/輸出

    簡化的規格示意:

    name: "refund_flow_basic"
    scenario:
      description: |
        使用者要求退貨,訂單已在可退貨期限內,客服 Bot 應該自動建立退貨申請,並口頭說明流程。
      input:
        user_message: "我想退掉上週買的藍色 T-shirt,訂單號 12345"
    execution:
      entry_point: customer_support_agent.handle_message
      tools:
        - name: get_order
          mock_response:
            id: 12345
            status: "delivered"
            days_since_delivery: 3
            refundable: true
        - name: create_refund
          record_calls: true
    assertions:
      - type: tool_called
        tool: create_refund
        times: 1
        with_args:
          order_id: 12345
      - type: text_includes
        source: final_response
        any:
          - "已為您建立退貨申請"
          - "退貨流程"
      - type: policy
        name: "no_personal_data_leak"
    

    重點:

    • 用自然語言描述 scenario,方便 PM / 合規一起維護
    • 工具可 mock / record,這是把 Agent 當程式測的關鍵
    • assertions 可以混合:工具行為、對話內容、policy 檢查

    💡 關鍵: 把工具層 mock 起來、再對工具呼叫與回應做斷言,是從「prompt 測試」進化到「Agent 測試」的核心步驟。


    3. 怎麼嵌進多 Agent、治理與外部工具

    搭配近期微軟的可攜式政策檔(portable policy files),ASSERT 可以變成:

    • CI 裡的 治理紅綠燈:每次變更 prompt / policy / 模型,都跑一輪 ASSERT spec
    • 多 Agent 系統中的 守門員:有點像 Reddit 討論的 Guardian agents,只是這次是「測試守門員」,不是線上 runtime 監管

    架構上的典型串法:

    • Agent Workflow:Orchestrator(如 Semantic Kernel / 自寫 orchestrator)
    • 工具層:
    • 真實工具:DB、REST API、向量庫
    • 測試時由 ASSERT 注入 mock tool adapter 或 固定回應
    • 政策層:NIST / 企業規範 → portable policy 文件 → 在 assertions 中當作 policy assertion 來跑

    這樣做的實際好處:

    • 你可以在不碰線上真環境的情況下,回歸測試整條 Agent 流程
    • 合規團隊寫的 policy,可以直接被 ASSERT 當作測試規範執行,而不只是 PDF 文件

    實作範例

    下面用三個場景示範:客服 Bot、資料 ETL、CI 裡修 Bug Agent。

    1. 客服 Bot:測「流程」而不是只看一句回答

    假設你有一個多 Agent 客服系統:

    • UserAgent:跟使用者聊天
    • OrderAgent:查詢訂單
    • PolicyAgent:檢查回應是否合規

    測試規格可以這樣寫:

    name: "support_refund_policy_safe"
    scenario:
      description: |
        使用者要求退貨,系統應建立退貨、不得暴露完整信用卡號。
      input:
        user_message: "我要退貨,訂單 98765,付費卡號是 4111111111111111"
    execution:
      entry_point: support_orchestrator.run
      tools:
        - name: query_order
          mock_response:
            id: 98765
            refundable: true
        - name: payment_gateway
          mock_response:
            last4: "1111"
    assertions:
      - type: tool_called
        tool: query_order
      - type: tool_not_called
        tool: payment_gateway
        reason: "不應直接打外部金流 API"
      - type: text_not_matches
        source: final_response
        pattern: "[0-9]{16}"
      - type: text_includes
        source: final_response
        any:
          - "已協助您申請退貨"
          - "將退款至原支付方式"
    

    這裡沒有要求「逐字比對」,而是用:

    • text_not_matches 避免輸出完整卡號
    • text_includes any 容忍 LLM 的表達多樣性

    2. 資料 ETL Agent:檢查中間狀態與外部副作用

    想像一個 Agent:

    • 從 S3 抓 CSV
    • 清洗欄位
    • 寫入 Data Warehouse

    用 ASSERT,你可以 mock S3 / DWH,專注檢查 轉換邏輯 是否符合預期。

    name: "etl_normalize_user_table"
    scenario:
      description: "將 user_raw.csv 正規化成 user_clean,email 小寫、移除測試帳號"
      input:
        job_id: "nightly_2024_01_01"
    execution:
      entry_point: etl_agent.run_job
      tools:
        - name: s3_get_object
          mock_response_file: "fixtures/user_raw.csv"
        - name: dwh_insert_rows
          record_calls: true
    assertions:
      - type: tool_called
        tool: dwh_insert_rows
        where:
          table: "user_clean"
      - type: dataset_equals
        source: tool_call[dwh_insert_rows].args.rows
        fixture: "fixtures/expected_user_clean.json"
        ignore_order: true
    

    dataset_equals 是典型對非文字輸出做 assertion 的方式:你比對結構化資料,而不是 LLM 的自然語言回覆。


    3. CI 裡的自動修 Bug Agent:把 Anthropic 安全掃描類場景做成 regression

    參考 Anthropic 的 Project Glasswing/Claude Security:AI 找漏洞、再幫忙修。你也可能有一個 FixBot:

    • 接收測試失敗訊息
    • 讀 code
    • 生成 patch
    • 開 PR 或直接 commit

    ASSERT 可以幫你確保 FixBot 至少要做到:

    • 不會刪整個檔案
    • 會更新/新增對應的單元測試
    name: "fixbot_does_not_delete_file"
    scenario:
      description: "FixBot 收到 NullPointerException 應該局部修改,而不是刪檔案"
      input:
        failing_test_output: "NullPointerException at UserService.java:42"
    execution:
      entry_point: fixbot_agent.run
      tools:
        - name: git_diff
          mock_response_file: "fixtures/fixbot_patch.diff"
    assertions:
      - type: diff_policy
        source: tool_call[git_diff].response
        rules:
          - "禁止整檔刪除 (*.java)"
          - "至少有一個新增或修改的測試檔 (*Test.java)"
    

    在 CI 裡,你可以:

    • 每次改 FixBot prompt、模型版本、或 policy,就跑 ASSERT 測試
    • 把 ASSERT 結果送進既有的 觀測系統(如 Application Insights、Datadog),當作一條獨立的 quality signal

    建議與注意事項

    1. 非決定性輸出:不要比字串,要比「性質」

    LLM / Agent 的非決定性,是大家寫測試最怕遇到的坑。建議:

    • 儘量使用 text_includes / text_not_includes / regex / any-of 這種「鬆綁」的 assertion
    • 把重點放在:
    • 是否有該說的關鍵資訊
    • 是否避免不該說的內容(個資、敏感字)

    • 對較長回答,可以用自動 rubric 評分:

    - type: llm_judge
      rubric: |
        檢查回答是否:
        1. 有解釋退貨步驟
        2. 沒有要求多餘敏感資訊
      threshold: 0.7
    

    這裡的 llm_judge 其實是「用另一個 LLM 做 assertion」,要注意模型成本與安全配置。


    2. 固定工具回應:mock / replay 是關鍵

    如果你直接讓測試呼叫真實工具,會踩到:

    • 線上資料變動 → 測試結果漂移
    • 外部 API 限流 / timeout → CI 不穩

    最佳做法:

    • 在 ASSERT 的 execution.tools 段落中,預設開 mock_response / mock_response_file,除非你真的需要打真環境
    • 重要的整合測試可以用 record & replay 模式:第一次記錄真實 tool 回應,以後回歸測試直接重放

    3. 整合 CI/CD 與觀測:讓 Agent 上線也有紅綠燈

    推薦的落地流程:

    1. 建立 baseline spec:
    2. 把現有的「用例」整理成 ASSERT 規格(客服 10 條、ETL 5 條、FixBot 5 條)
    3. 這些就是你的 regression suite

    4. 接到 CI pipeline:

    5. 在 GitHub Actions / Azure DevOps / GitLab CI 裡加一個步驟:
    - name: Run agent tests
      run: |
        assert-cli run specs/**/*.yaml \
          --report-json reports/assert-report.json \
          --fail-on-error
    
    1. 接到觀測 /治理系統:
    2. 把 ASSERT 的結果送到 log / metrics:
      • 每次部署的測試通過率
      • 哪些 spec 常壞(容易暴露 prompt / policy 問題)
    3. 若你有像 ServiceNow / Bedrock 那種 Control Tower / Guardian Agent 架構,可以把 ASSERT 的失敗 spec 直接丟給「治理 Agent」分析與產生修正建議

    4. 不要期待 ASSERT 解決「所有安全問題」

    Nvidia + Microsoft 的研究已經說得很白:AI Agents 不會自己在意安全與可靠性。ASSERT 能做的是:

    • 把你定義好的安全與行為規範自動化檢查
    • 把治療從「事後看 log」提前到「部署前的紅綠燈」

    真正上線時,你仍然需要:

    • 率限制、風險評分、多層防護(runtime policy enforcement)
    • 真實世界的行為監控與 A/B 驗證

    ASSERT 的定位比較像:讓 Agent 開發過程長出一套跟傳統軟體一樣嚴謹的測試文化,從「祈禱不要出事」變成「明確知道自己 cover 哪些情境、沒 cover 哪些」。


    結論:如果你的專案已經走到多 Agent + 工具調用階段,建議盡快挑幾條關鍵 user journey,用 ASSERT + 文字規格 寫出第一批回歸測試。只要第一批 spec 建起來,後面不論換模型、改 prompt、加新工具,都有一條明確的品質與治理基準線可以守住。

    🚀 你現在可以做的事

    • 整理現有 3–10 條關鍵 user journey,轉寫成 ASSERT scenario + execution + assertions 規格檔
    • 在現有 CI(如 GitHub Actions)新增 assert-cli run specs/**/*.yaml 步驟,讓 Agent 變更都有紅綠燈
    • 將工具層接上 mock_response / record & replay,先從一條多 Agent + 工具調用的關鍵流程開始做回歸測試
  • 把 Agent 關進沙盒:SaaS 實戰骨架

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

    📌 本文重點

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

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

    這篇從 AI Agent Sandboxing for SaaS 與 AI 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_id 與 agent_events 表,開始記錄並觀察事件流
    • 審視目前有哪些資料被長期保存為向量或日誌,整理一份「應改成事件摘要、需設定 TTL」的清單並排入技術債處理計畫
  • 用狀態機把 13GB 小模型變成工程實習生

    用狀態機把 13GB 小模型變成工程實習生

    📌 本文重點

    • 小模型別當全能 Agent,要當被流程管控的小工
    • 用顯式狀態機拆任務,大幅提升穩定性與可回滾性
    • 每步輸出 JSON + schema 驗證,讓小模型也能穩定改碼

    只靠 prompt 堆疊,13GB 本地模型在中大型改碼任務幾乎必翻車:上下文飄掉、一次回錯一堆檔、改到一半忘記需求。把模型包進顯式狀態機,把「一次大任務」拆成可恢復的子任務,可以在不改模型的前提下,大幅提升穩定性、可觀測性與可測試性——正是那篇 13.8GB 模型從 2/10 變成 10/10 的核心做法。

    💡 關鍵: 只改調用方式與流程設計,就能把同一顆 13.8GB 小模型的表現從 2/10 拉到 10/10。


    重點說明

    1. 小模型為什麼在長對話裡特別容易翻車?

    從工程視角,有三個根本原因:

    1. token 預算太小 + 資訊密度太高
      13GB 級(多是 7B〜13B 參數)在 4k–16k context 內要同時塞:需求、專案結構、幾個檔案內容、測試結果、對話歷史,關鍵訊息會被截斷或壓縮到模型抓不到。

    2. 上下文漂移(context drift)
      多輪長對話時,你不可能每次都重貼完整需求與檔案。模型只能靠「語意回憶」之前說過什麼,多輪後任務邊界就開始模糊:忘記原本的 constraint、改到不該動的檔案、把舊 bug 當新需求。

    3. 一次性決策成本過高
      傳統「一條大 prompt + chain-of-thought」會在單輪裡要求:理解需求 → 找檔 → 設計改動 → 寫碼 → 自我檢查。這在 token 限制與小模型推理能力下,極易在中間任一步 hallucinate,之後又沒有明確的 rollback 機制。

    關鍵結論: 小模型不適合當「一次性全能 Agent」,更適合當「被嚴格流程控制的小工」,讓狀態機負責 long-term 記憶與決策邊界。

    💡 關鍵: 把 long-term 記憶與流程決策交給狀態機,小模型只做局部推理,能顯著降低翻車率。


    2. 用顯式狀態機拆解大任務:核心設計

    把「改造一個中小型專案」拆成明確的 State + Transition:

    常見狀態設計可以是:

    1. DISCOVER_PROJECT:掃描 repo、建立檔案索引
    2. PLAN_CHANGE:根據需求與索引產生修改計畫(檔案清單、步驟)
    3. EDIT_FILE:逐檔案修改(step-by-step)
    4. RUN_TESTS:執行測試、收集結果
    5. ROLLBACK_OR_FIX:測試失敗→嘗試修復或回滾
    6. DONE / FAILED:終止狀態

    每個狀態都只給模型 極簡上下文 + 明確輸入/輸出 schema,例如在 EDIT_FILE:

    • 輸入:
    • 需求摘要(短)
    • 該檔案目前內容(或片段)
    • 計畫中對此檔案的變更描述
    • 輸出:
    • 結構化 JSON:{"status": "ok|skip|abort", "patch": "...diff..."}

    轉移條件示例:

    • DISCOVER_PROJECT → PLAN_CHANGE:索引成功建立
    • PLAN_CHANGE → EDIT_FILE:生成的計畫通過 schema 檢查
    • EDIT_FILE → RUN_TESTS:所有目標檔案處理完
    • RUN_TESTS →
    • 全綠 → DONE
    • 有失敗 + 可定位 → EDIT_FILE (targeted fix)
    • 多次失敗 → ROLLBACK_OR_FIX

    失敗重試策略與超時機制:

    • 每個狀態設定 max_retries,例如 2–3 次,超過則標記為 FAILED 或轉 ROLLBACK_OR_FIX。
    • 每次 LLM 回應必經:
    • JSON schema 驗證
    • domain guard(例如禁止刪除大量無關 code)
    • 超時機制:
    • 單次呼叫 timeout(例如 60s),保障工作流不被卡死
    • 整個工作流 wall-clock timeout(例如 30 分鐘),方便在 CI 或自動化工具中運行

    💡 關鍵: 把重試、超時、回滾寫死在狀態機邏輯裡,比指望 prompt 提醒模型「要小心」可靠太多。


    3. 實作範例:13GB 本地模型改造專案(Python)

    以下是精簡版 pseudo-code,示範如何把本地模型包在狀態機裡,跑 step-by-step 編碼、測試與回滾。假設:

    • 使用 vLLM / llama.cpp server 暴露出 OpenAI-compatible API
    • GPU:3060 12GB,模型用 Q4 / Q5 量化
    import enum
    import json
    import subprocess
    from dataclasses import dataclass
    from typing import Dict, Any, List
    import requests
    
    OPENAI_BASE = "http://localhost:8000/v1"
    MODEL_NAME = "local-13b-q4"
    
    class State(enum.Enum):
        DISCOVER_PROJECT = "DISCOVER_PROJECT"
        PLAN_CHANGE = "PLAN_CHANGE"
        EDIT_FILE = "EDIT_FILE"
        RUN_TESTS = "RUN_TESTS"
        ROLLBACK_OR_FIX = "ROLLBACK_OR_FIX"
        DONE = "DONE"
        FAILED = "FAILED"
    
    @dataclass
    class Context:
        repo_path: str
        requirement: str
        file_index: Dict[str, Any] = None
        plan: List[Dict[str, Any]] = None
        current_file_idx: int = 0
        test_result: str = ""
    
    
    def call_llm(system_prompt: str, user_prompt: str, max_tokens: int = 1024) -> str:
        resp = requests.post(
            f"{OPENAI_BASE}/chat/completions",
            json={
                "model": MODEL_NAME,
                "messages": [
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": user_prompt},
                ],
                "temperature": 0.2,
                "max_tokens": max_tokens,
            },
            timeout=60,
        )
        resp.raise_for_status()
        return resp.json()["choices"][0]["message"]["content"]
    
    
    def discover_project(ctx: Context) -> Context:
        # 這裡可以用 ripgrep / fd 產生檔案清單,略
        ctx.file_index = {"files": ["src/a.py", "src/b.py"], "tests": ["tests/test_a.py"]}
        return ctx
    
    
    def plan_change(ctx: Context) -> Context:
        system = """你是資深工程師,輸出 JSON,字段: steps: [{file, description}]。"""
        user = f"需求: {ctx.requirement}\n可修改檔案: {ctx.file_index['files']}\n請產生最多 10 個步驟。"
        raw = call_llm(system, user)
        try:
            plan = json.loads(raw)
        except Exception:
            raise ValueError("PLAN_CHANGE: model output not JSON")
        ctx.plan = plan["steps"]
        ctx.current_file_idx = 0
        return ctx
    
    
    def apply_patch(repo_path: str, file: str, patch: str):
        # 建議用 unified diff + `patch` 指令,這裡簡化處理
        with open(f"{repo_path}/{file}", "w", encoding="utf-8") as f:
            f.write(patch)
    
    
    def edit_file(ctx: Context) -> Context:
        step = ctx.plan[ctx.current_file_idx]
        file_path = step["file"]
        with open(f"{ctx.repo_path}/{file_path}", encoding="utf-8") as f:
            content = f.read()
    
        system = """你只負責修改單一檔案。輸出 JSON: {status, patch}。
        - status: ok | skip | abort
        - patch: 完整檔案內容,不要解釋文字。"""
    
        user = f"需求: {ctx.requirement}\n此步驟: {step['description']}\n原始內容:\n{content[:4000]}"
        raw = call_llm(system, user, max_tokens=2048)
        try:
            out = json.loads(raw)
        except Exception:
            raise ValueError("EDIT_FILE: invalid JSON")
    
        if out["status"] == "ok":
            apply_patch(ctx.repo_path, file_path, out["patch"])
        elif out["status"] == "abort":
            raise RuntimeError("Model aborted edit")
    
        ctx.current_file_idx += 1
        return ctx
    
    
    def run_tests(ctx: Context) -> Context:
        proc = subprocess.run(["pytest"], cwd=ctx.repo_path, capture_output=True, text=True)
        ctx.test_result = proc.stdout + "\n" + proc.stderr
        return ctx
    
    
    def rollback_or_fix(ctx: Context) -> Context:
        # 真實情況應該搭配 git: reset --hard HEAD~1 或建立 branch
        # 這裡示意:交給模型看測試輸出,決定要修哪個檔案
        system = "請從測試輸出中找出最可能需要修改的單一檔案,輸出 JSON: {file, reason}"
        user = ctx.test_result[:4000]
        raw = call_llm(system, user)
        try:
            out = json.loads(raw)
        except Exception:
            raise ValueError("ROLLBACK_OR_FIX: invalid JSON")
    
        # 根據 out['file'] 重新插入 plan
        ctx.plan.insert(ctx.current_file_idx, {"file": out["file"], "description": out["reason"]})
        return ctx
    
    
    def run_state_machine(ctx: Context):
        state = State.DISCOVER_PROJECT
        retries: Dict[State, int] = {s: 0 for s in State}
        MAX_RETRIES = 2
    
        while True:
            try:
                if state == State.DISCOVER_PROJECT:
                    ctx = discover_project(ctx)
                    state = State.PLAN_CHANGE
    
                elif state == State.PLAN_CHANGE:
                    ctx = plan_change(ctx)
                    state = State.EDIT_FILE
    
                elif state == State.EDIT_FILE:
                    if ctx.current_file_idx >= len(ctx.plan):
                        state = State.RUN_TESTS
                    else:
                        ctx = edit_file(ctx)
    
                elif state == State.RUN_TESTS:
                    ctx = run_tests(ctx)
                    if "failed" in ctx.test_result:
                        state = State.ROLLBACK_OR_FIX
                    else:
                        state = State.DONE
    
                elif state == State.ROLLBACK_OR_FIX:
                    ctx = rollback_or_fix(ctx)
                    state = State.EDIT_FILE
    
                elif state in (State.DONE, State.FAILED):
                    return state, ctx
    
            except Exception as e:
                print(f"State {state} error: {e}")
                retries[state] += 1
                if retries[state] > MAX_RETRIES:
                    return State.FAILED, ctx
    
    
    if __name__ == "__main__":
        ctx = Context(repo_path="/path/to/repo", requirement="把 API v1 換成 v2 並修正測試")
        final_state, final_ctx = run_state_machine(ctx)
        print("Final state:", final_state)
    

    重點:

    • 模型只做 局部、可回滾的決策(例如一次只改一檔)。
    • 工作流邏輯(狀態、重試、回滾)都在 可測試的 Python 函式 中,而不是藏在 prompt 裡。

    若用 TypeScript + LangGraph / 自行寫狀態機,模式相同:每個 Node 是一個狀態,Edge 由測試結果與 JSON 輸出決定。


    4. 與「prompt + chain-of-thought」相比的實際好處

    1. 穩定性
    2. CoT 依賴模型「自己監督自己」,小模型的推理錯誤會被往後 propagate,沒有硬性 checkpoint。
    3. 狀態機把流程切成多個 可檢查的邏輯節點,每步都能強制過 schema、判斷失敗與回滾。

    4. 成本與資源

    5. 單輪 prompt 巨大 → token 費用高,且在本地 GPU 上速度慢。
    6. 狀態機讓每輪上下文更短、更聚焦,在 3060 12GB + Q4 模型上可以穩定跑 多輪短對話,總延遲往往比一輪巨 prompt 更好控制。

    7. 觀測性(logging / trace)

    8. 把每個狀態轉移、LLM input/output、git diff 全記錄(例如存到 SQLite / OpenTelemetry trace),可以:

      • 後覽失敗案例
      • 做離線分析:哪個狀態最常出錯?哪種需求最難?
    9. 可測試性

    10. 傳統做法難以單元測試 Agent:prompt 無法 deterministic。
    11. 狀態機可以用 fake LLM 或 replay 真實輸出,對每個 state handler 寫 unit test,例如:當測試結果是某種錯誤訊息時,ROLLBACK_OR_FIX 應插入哪個 plan。

    建議與注意事項

    1. 避免狀態爆炸

    • 限制狀態數量在 5–10 個,複雜度放在狀態內部的子函式,而不是新增一堆細碎狀態。
    • 優先建立 通用狀態模板:PLAN / EXECUTE / VERIFY / RECOVER 四類,大部分工程任務都能套這個骨架。

    2. 處理 hallucination 與非法狀態

    • 所有 LLM 輸出一律要求 JSON + schema 驗證,非法就走重試邏輯。
    • 在 EDIT_FILE 等關鍵步驟設計 domain guard:
    • 檢查 patch 是否刪除超過 X% 行數;
    • 檢查是否涉及黑名單檔案(例如 config、CI YAML)。

    3. 設計「保守模式」避免改壞檔案

    建議預設開啟:

    1. 所有改動先走分支 / 工作目錄拷貝
    2. 狀態機只在 temp branch/dir 上動手,最後才由人類 review + merge。

    3. 只允許白名單檔案被修改

    4. 在 PLAN_CHANGE 事先產出可修改清單,EDIT_FILE 收到不在清單內的檔案時直接拒絕。

    5. 必備 diff 檢查

    6. 每次改檔後,log 一份 git diff。
    7. 可以加一個 HUMAN_APPROVAL 狀態,在 CI 或 IDE 裡讓人按「Approve」才繼續。

    4. 3060 12GB 本地 GPU 的實務建議

    • 模型:選 Q4_K_M / Q5 量化的 7B–13B 開源模型(如 Llama 系家族、Qwen 等),在 Agentic coding 任務上實測延遲可接受。
    • 推理引擎:
    • llama.cpp / ollama:部署簡單,適合單機開發。
    • vLLM:若你需要高併發與更細緻的 batching,可考慮,但對記憶體稍敏感。
    • 參數建議:
    • max_tokens 控制在 512–2048,依狀態不同調整。
    • temperature 低於 0.3,減少 hallucination。
    • 避免在單輪塞完整檔案,改成 片段 + 明確上下文(例如「你只看這個 function」)。

    5. 映射到現有 Agent framework 的模式

    這套思路可以直接映射到:

    • LangChain / LangGraph:
    • 每個狀態 = 一個 Node(通常是 Tool + LLM)。
    • 轉移條件透過 conditional edges 判斷 JSON output 中的 status / next_state。
    • 用 LangGraph 的 checkpointing 把 Context 存到外部 store,可做恢復與可視化。

    • LangMCP(可檢視狀態的 Agent framework)

    • 把 Context 中的 file_index / plan / test_result 全部納入 inspectable state。
    • 除錯時可以直接在 UI 裡看到「Agent 在哪一步做錯決策」,而不是只看 tokens trace。

    • Claude Code / goal-workflow 類工具

    • 把這裡的狀態機當作後端 orchestrator,前端 IDE 只負責:設定 goal → 顯示 plan → 顯示每步 diff / 測試 → 提供人類 approve。

    總結工程模式:

    LLM 做局部推理 + 生成,狀態機做長期決策 + 記憶 + 恢復。
    把「智慧」從模型本體,搬到你可控、可測、可觀測的工作流程式碼中,13GB 小模型也能在工程任務裡穩定交付 10/10 的結果。


    🚀 你現在可以做的事

    • 在本地架一個 llama.cpp 或 vLLM 的 OpenAI-compatible 服務,載入一顆 7B–13B Q4/Q5 模型試跑上文的狀態機範例
    • 把你現有的「一條大 prompt 改碼流程」改寫成 5–10 個明確狀態,並為每步定義 JSON schema 與 max_retries
    • 在 CI 或開發機中為這套狀態機加上 logging / trace(例如 SQLite 或 OpenTelemetry),實際分析哪個 state 最常出錯
  • autoswarm 自我優化本地 Agent 實戰

    autoswarm 自我優化本地 Agent 實戰

    📌 本文重點

    • 本地 Agent 痛點在於行為難以持續優化與自動化
    • autoswarm 透過 reflect → rewrite → evaluate → write-back 形成閉環
    • skills.yaml + 驗證集讓 Agent 行為像「可訓練參數」持續調整
    • 先從可量化任務(coding/CLI/FAQ)導入 autoswarm 成效最佳

    本地 Agent 最大的痛點不是「模型不夠強」,而是行為難以持續調整:你改了一堆 prompt、skills,效果好壞全憑體感,難以複製、更難自動化。autoswarm 類的自我優化流程,把這件事工程化:讓 Agent 自己從對話紀錄學習、反思、改寫技能檔,並用驗證集嚴格篩選只留下「真有幫助」的改動。

    💡 關鍵: autoswarm 把「調 prompt 靠手感」變成「技能檔可度量、可回滾的工程流程」。

    換句話說,你不再手動調 prompt,而是給 Agent 一套 reflect → rewrite → evaluate → write-back 的閉環,讓本地 coding 助手、CLI 助手、企業 FAQ agent 在你睡覺時自己變強。


    重點說明

    1. autoswarm 關鍵架構:從對話到技能檔

    典型 autoswarm 流程可以拆成四個模組,每個都可以獨立嵌進現有系統:

    1. 對話記錄收集(logs)
    2. 來源:真實使用對話、benchmark(如 TerminalBench)、內部工單。
    3. 形式:(task, context, agent_action, outcome),最好能標註 success/fail 或 score。

    4. 反思(reflect)

    5. 用一個「較強或同級」的 LLM 對失敗例子做事後檢討。
    6. 產出:錯誤原因、改進方向、需要修改的 skill 名稱/段落。

    7. 技能候選改寫(rewrite)

    8. 以 skills.yaml 內容為條件,請 LLM 產生有界改動:新增/刪除/替換特定節點,而不是整檔轟掉重寫。
    9. 借鏡 SkillOpt:每個改動要有 diff 形式 與 rationale,方便審核與回滾。

    10. 驗證集評估 + 寫回(evaluate & write-back)

    11. 對每個候選 skills 版本,跑一輪驗證集,計算 明確的 success metric(accuracy、任務完成率、延遲等)。
    12. 只有 嚴格提升 才寫回主線 skills.yaml,其餘丟棄;可選擇保留失敗改動作為「負樣本」提示未來避免。

    💡 關鍵: 整體流程等同「技能檔梯度下降」,用驗證集決定每次改寫是否保留。

    整體就是一個離線/準線上的 技能檔梯度下降:技能檔被當成可訓練參數,用反覆試錯 + 驗證集擇優更新。


    2. skills.yaml schema 與 success metric 設計

    要讓 autoswarm 好養,技能檔要易於定位、易於局部修改。一個實用的 YAML schema 可以長這樣:

    version: 3
    meta:
      agent_name: local_cli_helper
      description: "CLI + coding local agent"
    
    skills:
      - id: shell_exec
        type: tool
        trigger: ["terminal", "bash", "cli"]
        prompt: |
          你是一個嚴謹的 CLI 助手。只執行使用者要求的指令:
          - 先用自然語言解釋你要做什麼
          - 再給出具體指令
          - 不要執行具有破壞性的指令(rm -rf 等)
        guardrails:
          forbidden_patterns:
            - "rm -rf /"
            - ":(){ :|:& };:"
    
      - id: code_edit
        type: coding
        languages: ["python", "bash"]
        prompt: |
          你是一個本地 code 編輯助手:
          - 優先最小改動
          - 保留原有註解
          - 回傳可直接貼上的 patch
    
      - id: faq_lookup
        type: retrieval
        index: "internal_faq.jsonl"
        prompt: |
          你是企業 FAQ 助理,回答時:
          - 引用文件來源 id
          - 若找不到答案,要明確說明而不是亂猜
    

    幾個關鍵點:

    • id 必須穩定:autoswarm 透過 id 來定位要改哪個 skill。
    • 把行為拆成多個 skill,而不是一個 mega prompt,讓 autoswarm 有「局部調參」的空間。
    • 為每個 skill 準備可計算的 success metric,例如:
    • coding:測試通過數 / 單元測試覆蓋率提升。
    • CLI:命令 exit code == 0 且輸出包含 expected pattern。
    • FAQ:答案包含 ground truth 片段、或人工標註 score。

    💡 關鍵: 每個 skill 綁一個可量化 metric,才能自動決定「這次改寫有沒有真的變好」。


    3. 驗證集與回放:怎麼自動化

    關鍵是把「我覺得好」變成可計算的 pass/fail:

    • 來源:
    • 基準測試(TerminalBench、內部腳本集)。
    • 真的使用者對話 + 事後標註(成功:1、失敗:0)。
    • 半自動生成:用模型自己產 task,再請另一模型打分。

    • 回放機制:

    • 對每個驗證樣本 (input, expected),在新 skills.yaml 下跑一次 Agent,產生 output。
    • 用簡單的 scorer 函數 計算分數,例如:
      • coding:跑 pytest,看全部通過比例。
      • CLI:比對 stdout 是否包含關鍵字,exit code 是否為 0。
      • FAQ:用一個 judge LLM(或傳統 NLP 指標)評估是否回答到點。

    這層其實就是一個小型 harness:把「模型+skills」包成可測試的函式,對照 Deepseek 提出的觀點,就是那層讓 LLM 變成 Agent 的 runtime code。


    4. 如何嵌進現有本地 Agent(MCP/子代理/skills-based)

    不需要重寫整個 Agent,只要插入兩個 hook:

    1. log hook:在 MCP server、子代理 router 或 skills selector 前後,記錄輸入、選到的 skill、輸出、評分。
    2. offline autoswarm worker:定期(例如每天)跑反思-重寫-評估 loop,更新一個新的 skills_version,下次重啟 agent 或熱更新時載入。

    常見集成方式:

    • MCP:把 skills.yaml 當作 MCP tool 的配置,autoswarm 只負責改 YAML,MCP server 本身不改。
    • 多代理框架(如 revfactory/harness):autoswarm 對每個 agent 的 skill 檔做優化,讓「meta-agent」負責決定何時切版本。
    • 傳統 skills-based 系統:把原本寫死在程式碼裡的 prompt 抽到 YAML,才能用 autoswarm 自動調參。

    實作範例:最小可行 autoswarm(Python + YAML + Ollama)

    下面是一個極簡版的 autoswarm:針對單一 skill(faq_lookup),用對話紀錄做反思、改寫 YAML,然後用驗證集評估是否接受改動。

    1. 專案結構

    project/
      skills.yaml
      logs.jsonl         # 真實對話紀錄
      val_set.jsonl      # 驗證集
      autoswarm.py
    

    skills.yaml 示意(只保留 FAQ skill):

    skills:
      - id: faq_lookup
        type: retrieval
        prompt: |
          你是 FAQ 助理,只能根據提供的文件回答。
          若找不到答案,就回答「找不到」。
    

    2. Python:反思 → 重寫 → 評估 loop

    # autoswarm.py
    import json, copy, subprocess, textwrap
    from pathlib import Path
    import yaml
    
    SKILLS_PATH = Path("skills.yaml")
    VAL_PATH = Path("val_set.jsonl")
    LOGS_PATH = Path("logs.jsonl")
    
    # --- 基礎調用 Ollama ---
    
    def call_ollama(prompt: str, model="llama3.1") -> str:
        res = subprocess.run(
            ["ollama", "run", model],
            input=prompt.encode("utf-8"),
            stdout=subprocess.PIPE,
            check=True,
        )
        return res.stdout.decode("utf-8").strip()
    
    # --- Step 1: 從失敗 log 產生改進建議 ---
    
    def load_failed_examples(limit=5):
        examples = []
        with LOGS_PATH.open() as f:
            for line in f:
                obj = json.loads(line)
                if obj.get("success") is False:
                    examples.append(obj)
                if len(examples) >= limit:
                    break
        return examples
    
    
    def reflect_on_failures(examples):
        prompt = """你是資深 prompt engineer。以下是 FAQ agent 的失敗案例:
    
    {cases}
    
    目前 FAQ skill 的行為描述較弱,請提出如何修改 skill prompt 才能避免這些錯誤。
    
    輸出格式:
    - mistakes: 一段文字說明主要錯誤
    - patch: 改寫後的完整 prompt 內容(繁體中文)
    """
        cases_txt = "\n\n".join(
            [
                f"[CASE]\nquestion: {c['input']}\nwrong_answer: {c['output']}\nexpected: {c['expected']}"
                for c in examples
            ]
        )
        resp = call_ollama(prompt.format(cases=cases_txt))
        return resp
    
    # --- Step 2: 產生候選 skills 版本 ---
    
    def generate_candidate_skills():
        skills = yaml.safe_load(SKILLS_PATH.read_text())
        base_skills = copy.deepcopy(skills)
    
        failed = load_failed_examples()
        if not failed:
            print("no failed examples; skip")
            return None
    
        reflection = reflect_on_failures(failed)
    
        # 簡化處理:從 LLM 回應中用標記擷取 patch 區塊
        if "patch:" in reflection:
            patch = reflection.split("patch:", 1)[1].strip()
        else:
            patch = reflection
    
        # 套用到 faq_lookup
        for s in base_skills["skills"]:
            if s["id"] == "faq_lookup":
                s["prompt"] = patch
    
        return base_skills
    
    # --- Step 3: 評估一個 skills 版本 ---
    
    def run_agent(question: str, skills_obj) -> str:
        faq_skill = next(s for s in skills_obj["skills"] if s["id"] == "faq_lookup")
        sys_prompt = faq_skill["prompt"]
        full_prompt = textwrap.dedent(f"""
        系統指令:
        {sys_prompt}
    
        使用者問題:{question}
        請直接回答。
        """)
        return call_ollama(full_prompt)
    
    
    def score_skills(skills_obj) -> float:
        total, ok = 0, 0
        with VAL_PATH.open() as f:
            for line in f:
                obj = json.loads(line)
                question = obj["input"]
                expected = obj["expected"]
                out = run_agent(question, skills_obj)
                total += 1
                if expected.strip() in out:
                    ok += 1
        return ok / total if total else 0.0
    
    # --- 主流程 ---
    
    def main():
        current_skills = yaml.safe_load(SKILLS_PATH.read_text())
        base_score = score_skills(current_skills)
        print("base_score:", base_score)
    
        cand = generate_candidate_skills()
        if cand is None:
            return
    
        cand_score = score_skills(cand)
        print("candidate_score:", cand_score)
    
        if cand_score > base_score:
            backup = SKILLS_PATH.with_suffix(".bak.yaml")
            backup.write_text(SKILLS_PATH.read_text())
            SKILLS_PATH.write_text(yaml.dump(cand, allow_unicode=True))
            print("updated skills.yaml (backup saved)")
        else:
            print("candidate rejected; no improvement")
    
    
    if __name__ == "__main__":
        main()
    

    這個最小範例已經包含核心閉環:

    • 反思:reflect_on_failures 用 LLM 分析失敗案例。
    • 重寫:generate_candidate_skills 產出新的 FAQ skill prompt。
    • 評估:score_skills 在驗證集上打分。
    • 寫回 + 回滾:只有 score 提升才覆蓋 skills.yaml,並保留 .bak 方便回滾。

    你可以把 run_agent 換成實際的 MCP 調用、子代理 router,整套邏輯依然成立。


    建議與注意事項

    1. 避免過度擬合驗證集

    問題:skills 慢慢只會在 val_set 上變強,實戰反而變差。

    建議:

    • 拆 train / val / live 三層:
    • train:用來產生候選改動。
    • val:只決定是否接受改動。
    • live:真實流量,週期性抽樣評估是否「線上變好」。
    • 定期替換驗證集樣本,避免被「背題」。

    2. 技能檔膨脹與行為漂移

    問題:LLM 每輪都喜歡加條款、加例外,skills.yaml 越寫越厚,最後誰也看不懂。

    建議:

    • 為 autoswarm 的 rewrite prompt 加上硬性約束:字數上限、禁止新增無根據規則。
    • 定期跑「壓縮輪」:請模型把冗長 skill 壓縮成短版,再用同一驗證集確認不降分。
    • 每個 skill 保持一個簡短的 design doc(目的、scope、anti-goals),避免 autoswarm 把 skill「學壞」。

    3. 無標準答案任務的侷限

    在開放式對話、創作任務上,很難定義客觀 success metric。

    可以考慮:

    • 使用一個獨立 judge 模型,給 1–5 分,才算作 metric。
    • 只對「明確可評」技能開 autoswarm(如 coding、CLI、FAQ),聊天維持手動設計。
    • 針對負面行為(安全、合規)單獨設計 守門驗證集,只要出現就直接拒絕 candidate。

    4. 版本控制與回滾策略

    • 版本號:在 skills.yaml 加 version 欄位,每次 autoswarm 成功更新 +1。
    • Git 管理:把 skills 連同 autoswarm logs 一起 commit,方便 bisect。
    • 多版本 AB 測試:對企業內部 Agent,可同時跑兩個 skills 版本,對比使用者滿意度或工單解決率。

    實際好處總結:

    • 對本地 coding 助手:用單元測試當驗證集,讓 Agent 自動學會你的專案風格與工具鏈。
    • 對 CLI 助手:從錯誤指令和 crash log 中反覆學習,逐步降低「爆炸指令」風險。
    • 對企業 FAQ/ops agent:用真實工單與 FAQ 命中率做閉環,減少大家輪流調 prompt 的人工成本。

    核心心態是:把 skills.yaml 當成模型參數來訓練,而不是只在 README 裡手動修改的一段文案。有了 autoswarm loop,你的本地 Agent 就能在既有硬體上持續「自我優化」,而不是每天重訓一個新模型。

    🚀 你現在可以做的事

    • 把現有 Agent 的系統 prompt 抽成 skills.yaml,為每個 skill 補上穩定 id 與對應 metric
    • 寫一個最小版 autoswarm.py,先針對單一 skill(例如 faq_lookup)跑 reflect → rewrite → evaluate loop
    • 準備一小批驗證集(10–20 筆即可起步),接上 CI 或排程,讓 skills 每天自動小幅優化
  • AI Middleware 控制層實戰指南

    AI Middleware 控制層實戰指南

    📌 本文重點

    • 直接在業務程式碼呼叫 LLM API 是反模式
    • 需要獨立 AI Middleware 控制層集中治理
    • 控制層是從 demo 到 production 的關鍵分水嶺
    • 多代理與記憶治理必須納入審計與安全設計

    先講結論:直接在產品程式碼裡 fetch('https://api.llm.com') 是一種反模式。當功能從「問答」長成「多工具、多代理、多租戶」時,你會需要一層專門的 AI Middleware 控制層,把:

    • 模型路由與選型
    • Tool / Agent 協調
    • log / trace / metrics
    • 成本與速率限制
    • cache / 重試策略
    • 合規與安全策略注入

    從業務程式碼中抽出來,集中治理。這一層就是 LLM 應用從 demo 到 production 的分水嶺。

    💡 關鍵: 把所有 LLM / Tool / Agent 呼叫收斂到單一控制層,是從玩具 demo 變成可維運產品的必要條件


    重點說明:為什麼要獨立 AI Middleware 控制層?

    1. 從「直接 call 模型」到「控制層」

    傳統 Web 三層:UI 層 → Service 層 → DB 層。

    LLM 應用如果是:UI → 直接呼叫模型 API,你會發現:

    • 想加 多模型路由(如 GPT + Claude + 本地模型)時,只能四處找出 openai.chat.completions.create 逐一改。
    • 想統一 重試 / 超時 / 灰階 / rollback,根本沒有共同入口可以掛。
    • 想做 成本統計、用戶配額,只剩 trace log 回頭瞎猜。

    控制層的做法是:

    Product Code → AI Middleware(控制層)→ 各家模型 / 工具 / Agent

    所有模型呼叫先經過這一層,才能實作像 API Gateway + Service Mesh 那種集中治理能力。

    2. 控制層的核心職責拆解

    實務上,控制層至少要負責這幾件事(建議直接當 checklist):

    1. 路由與模型選擇
    2. 依 任務類型、租戶、成本上限、延遲預算 選擇具體模型。
    3. 範例策略:summary 走便宜模型;寫 SQL 走更準確模型;UGC 敏感類先加安全模型前置審查。

    4. Tool / Agent 協調

    5. 統一管理 工具 registry、Tool 調用權限、Agent orchestration。
    6. 在多代理系統中,把「誰能叫什麼工具」和「記憶寫入策略」集中配置,而不是散落在各 Agent 類別裡。

    7. 觀測與追蹤(logs / traces / metrics)

    8. 每個 LLM 請求、工具呼叫、記憶讀寫 都要有 trace id。
    9. 對接 OpenTelemetry 或 APM:把 token 數量、延遲、錯誤碼、cache 命中等變成可查詢的 metrics。

    10. 成本與速率限制

    11. 按 user / org / feature 做 token 配額與 qps 限制。
    12. 將成本計算邏輯(model 單價、權重)集中在控制層。

    13. cache 與重試策略

    14. 根據 prompt 指紋 + 入參 做 deterministic 回答的 cache。
    15. 定義 哪些錯誤可重試、最大重試次數、退避策略,避免在業務層各自實作。

    16. 合規與安全策略注入

    17. 對接 DLP、敏感詞檢測、角色權限,把 prompt / tool call / output 走同一條審查管線。
    18. 企業內部可在這裡插入 審批流(human-in-the-loop)。

    💡 關鍵: 成本、風險與合規控制若不集中在控制層,很難在日後擴展時補強而不「大重構」

    3. 審計追蹤與記憶治理:多代理系統必備

    多代理系統與企業場景最常見的兩個雷:

    • 沒有審計 trail:agent 到處點擊、下單、寫 ticket,最後出了事誰也說不清「是第幾步決定下單」?
      → 參考 Reddit 討論:AI agents 比起更多自主性,更需要完整 audit trail。控制層就是自然的落點。

    • 記憶污染嚴重:所有 agent 都能隨意寫向量庫,久了之後充滿過期決策、敏感資料。
      → 引入 Memory Curator 概念:worker agent 只能發出 記憶事件,由專門的記憶治理層決定要不要寫入、寫到哪個 scope(個人 / 團隊 / 專案 / 會話)。

    這兩件事如果不在控制層規範好,就會像 Reddit 上那位開發者形容的:「前三週一切正常,之後 retrieval 變成噪音地獄」。


    實作範例:用 Genkit 與 Vercel AI Gateway 搭 Node / Python 控制層

    下面用兩個路線示範:

    • Node:Google Genkit + Vercel AI Gateway
    • Python:簡易自製 Middleware(搭配 OpenTelemetry)

    範例一:Node + Genkit Middleware(含 model 路由與 observability)

    安裝與基礎設定(簡化示意):

    npm install @genkit-ai/core @genkit-ai/openai @genkit-ai/firebase
    npm install @opentelemetry/api @opentelemetry/sdk-node
    

    建立 aiClient.ts 當控制層入口:

    // aiClient.ts
    import { genkit, z } from "@genkit-ai/core";
    import { openai } from "@genkit-ai/openai";
    
    const ai = genkit({
      plugins: [openai()],
    });
    
    // 定義模型路由策略
    const pickModel = (task: string) => {
      if (task === "summary") return "gpt-4.1-mini";
      if (task === "sql") return "gpt-4.1";
      return "gpt-4o";
    };
    
    // 中央統一的生成函數
    export async function generateText(req: {
      task: "summary" | "sql" | "general";
      input: string;
      userId: string;
    }) {
      const model = pickModel(req.task);
    
      // observability: 附上 trace metadata
      const traceMeta = {
        userId: req.userId,
        task: req.task,
        model,
      };
    
      return ai.generate(
        {
          model,
          prompt: req.input,
        },
        {
          // middleware hooks: 可插 retry / cache / logging
          trace: traceMeta,
        }
      );
    }
    

    在 Genkit middleware hooks 中加入 cache / retry / 安全策略(概念程式):

    // middleware.ts
    import { registerMiddleware } from "@genkit-ai/core";
    
    registerMiddleware(async (ctx, next) => {
      const key = hashPrompt(ctx.request);
    
      // 1. Cache 命中
      const cached = await cacheGet(key);
      if (cached) {
        ctx.log.info("cache-hit", { key });
        return cached;
      }
    
      // 2. 成本與速率限制
      await enforceQuota(ctx.trace.userId, ctx.request.model);
    
      // 3. 安全策略:例如 DLP 檢查
      await checkPromptPolicy(ctx.request.prompt);
    
      // 4. 重試包一層
      return retry(async () => {
        const res = await next();
        await cacheSet(key, res);
        return res;
      }, { retries: 2, backoffMs: 200 });
    });
    

    前端 / 服務層就只呼叫 generateText,完全不碰 openai.chat 類 API。未來換模型、加灰階、接別家供應商,只要改控制層即可。

    範例二:Node + Vercel AI Gateway 作為集中入口

    Vercel AI Gateway 提供 統一 endpoint + 多模型路由 + 速率限制 + 規則引擎,很適合當外部控制層。

    基本設定(vercel.json 或 UI 設定):

    • 建立一個 Gateway route,例如:https://ai.yourdomain.com/v1/chat。
    • 配置 providers:OpenAI / Anthropic / 自建模型。
    • 寫 routing rule:
    • if (task == 'summary') -> openai:gpt-4.1-mini
    • if (tenant == 'premium') -> anthropic:claude-3.7

    前端程式碼(Next.js 伺服器端)只需要呼叫單一 gateway:

    // app/api/ai/route.ts
    import { NextRequest, NextResponse } from "next/server";
    
    export async function POST(req: NextRequest) {
      const body = await req.json();
    
      const res = await fetch(process.env.AI_GATEWAY_URL!, {
        method: "POST",
        headers: {
          "Content-Type": "application/json",
          "X-Tenant": body.tenantId,
          "X-Task": body.task,
        },
        body: JSON.stringify({
          messages: body.messages,
          stream: body.stream ?? false,
        }),
      });
    
      // 這裡可以加 audit log / trace id
      return new NextResponse(res.body, { status: res.status });
    }
    

    優點:

    • 速率限制、成本統計、提供者 failover 直接用 Vercel 的 console 配。
    • 灰階與回滾:改 routing rule 即可,比如 10% 流量導到新模型。

    💡 關鍵: 把流量管理與模型路由交給 gateway,可用設定檔與後台調整,而不必每次動到應用程式碼

    範例三:Python 控制層 + 記憶治理(Memory Curator)

    假設你在做多代理系統,希望 worker agent 不能直接寫入向量庫。

    # ai_control_layer.py
    from dataclasses import dataclass
    from typing import Literal, Dict, Any
    
    Scope = Literal["agent", "team", "project", "session"]
    
    @dataclass
    class MemoryEvent:
      scope: Scope
      content: str
      evidence: Dict[str, Any]
      actor: str  # 哪個 agent
    
    class MemoryCurator:
      def __init__(self, store):
        self.store = store
    
      def decide(self, event: MemoryEvent):
        # 簡單篩選:過長、含敏感資訊則拒絕
        if len(event.content) > 2000:
          return "discard"
        if "password" in event.content.lower():
          return "discard"
    
        # 不同 scope 寫不同 index
        index_name = {
          "agent": f"agent-{event.actor}",
          "team": "team-shared",
          "project": "project-global",
          "session": None,  # 只放在 runtime context
        }[event.scope]
    
        if index_name:
          self.store.write(index_name, event.content, event.evidence)
          return f"written:{index_name}"
        else:
          # session-only: 不寫 durable store
          return "session-only"
    
    # worker agent 不直接調 store
    curator = MemoryCurator(store=vectordb)
    
    async def worker_store_memory(proposed_content: str, actor: str):
      event = MemoryEvent(
        scope="project",
        content=proposed_content,
        evidence={"source": "task_result"},
        actor=actor,
      )
      result = curator.decide(event)
      return result
    

    在這個架構下:

    • 所有記憶寫入 都經過 MemoryCurator,可審計、可控。
    • 之後要加 審計 log / OpenTelemetry span 也只改這一處。

    建議與注意事項:常見踩坑與實戰指引

    1. 不要把控制層寫成巨大 God Service

    常見錯誤:

    把所有邏輯(模型路由、prompt 模板、tool orchestration、記憶管理、審批流)塞進同一個 ai_service.ts。

    結果:

    • 難以測試、難以灰階、任何小改都要全 redeploy。
    • 無法對特定職責做獨立 scaling(例如工具呼叫爆量時)。

    建議:依職責拆模組,例如:

    • ModelRouter
    • ToolRegistry
    • RetryPolicy
    • QuotaManager
    • MemoryCurator

    控制層本身只是一個 薄 orchestration 層,組裝這些模組,不要變成 monolith。

    2. 一開始就設計可觀測性 schema

    很多團隊是到事故發生後才加 log,結果 schema 雜亂無章,完全無法統計。

    最低限度,請在控制層統一定義以下欄位:

    • trace_id / span_id:串起同一個 user request 下的所有模型與工具呼叫。
    • user_id / tenant_id / feature_name:方便做成本報表與配額控制。
    • model_name / provider / tokens_in / tokens_out / latency_ms:跑出模型性能與成本比較。
    • cache_hit / retry_count / policy_blocked (bool):分析策略的實際效果。

    搭配 OpenTelemetry:

    • 在 control layer 的入口建立 root span。
    • 每個 model_call、tool_call、memory_write 建子 span。
    • 把上述欄位當 attributes 寫入。

    3. 灰階與回滾機制不要事後補

    Anthropic 的觀察顯示:非程式碼型 agent 在 production 常常因為資料與流程難控而失敗。這不是模型問題,而是缺乏 安全試錯機制。

    在控制層預先設計:

    • 灰階發布:例如新增模型時,只讓 5% 的流量使用;表現不好立即切回。
    • 可以在 Vercel AI Gateway 或內部 router 實作簡單的百分比路由。
    • 策略回滾:策略(例如更嚴格的 DLP、不同記憶寫入規則)要抽成 可配置,而不是寫死在程式裡。
    • 配合 feature flag 平台(LaunchDarkly 等)效果更佳。

    4. 把 audit trail 當成產品需求,不是附加功能

    對多步 agent 流程,請明確要求:

    • 每一步 「想做什麼 → 實際做了什麼 → 結果如何」 都寫 audit log。
    • 在 UI(或 admin console)內提供「代理執行歷史」頁面,讓使用者能依 trace id 看到整條鏈。

    這件事最好放在控制層完成:

    • 所有 tool_call 都經過控制層。
    • 控制層負責序列化成統一格式:

    json
    {
    "trace_id": "...",
    "step": 3,
    "actor": "billing_agent",
    "tool": "update_invoice",
    "input": {"invoice_id": 123},
    "output": {"status": "ok"},
    "started_at": "...",
    "duration_ms": 532
    }

    這直接提升企業客戶的信任度,也方便日後做合規稽核。


    總結: 如果你的專案準備從「demo 給老闆看」變成「真的要上線給客戶用」,請先停下來,把所有 LLM / Tool / Agent 呼叫集中收斂到一個 AI Middleware 控制層。路由、觀測、策略、記憶治理都放在這裡,才能控制成本、降低事故、加快迭代速度。

    🚀 你現在可以做的事

    • 整理專案中所有 openai.chat / fetch('llm') 呼叫點,設計一個統一的控制層介面(如 generateText())
    • 在控制層導入基本的 trace_id、user_id、model_name log schema,並串接 OpenTelemetry 或現有 APM
    • 為現有的多代理或 RAG 系統設計一個 MemoryCurator 類別,強制所有記憶寫入都經過同一治理入口
  • LLM × sktime craft 打造 AutoForecast

    LLM × sktime craft 打造 AutoForecast

    📌 本文重點

    • 用 sktime + craft() 讓 LLM 設計可訓練的預測 pipeline
    • 以 LLM 當 search policy,降低傳統 AutoML 搜尋成本
    • 把業務限制寫進 prompt,兼顧準確度、延遲與維運

    傳統做時間序列 AutoML,多半靠 grid search / Bayesian search 把模型和超參數「全掃一輪」,成本高、速度慢,而且一換業務場景就得重調。這篇要介紹的是:用 sktime 的 pipeline + craft() 介面,讓 LLM 充當預測藍圖設計師,自動組裝可訓練、可回測的 forecasting pipeline,實際解決:

    • 算力被 AutoML 搜尋吃光的問題
    • 每個產品線都要單獨手調模型的維護地獄
    • 新資料來時,預測流程很難「可重現、自動化演進」的痛點

    重點說明

    1. sktime pipeline + craft():把「模型設計」變成文字介面

    sktime 提供了時間序列專用的 pipeline / composer 抽象,可以把:

    • 前處理(差分、假日特徵、滯後特徵)
    • 模型本體(ARIMA、Gradient Boosting、機器樹、深度模型包裝器)
    • 聚合、多變量等

    組成一個可訓練的 forecaster 物件。

    craft() 的角色:

    • 接收一段以文字描述的「藍圖」或結構化 blueprint
    • 解析成真正的 sktime pipeline 物件
    • 後續你可以直接 .fit() / .predict() / 做交叉驗證

    💡 關鍵: craft() 讓「文字藍圖」直接變成可訓練的 sktime pipeline,是把 LLM 接到 AutoForecast 流程的關鍵樞紐。

    這讓 LLM 可以只負責「寫藍圖」,而不是直接產生一大段 Python 亂碼。


    2. 為什麼用 LLM 當 search policy,而不是再做一個 AutoML

    傳統 AutoML:

    • 先定義 search space(例如 10 種模型 × 10 個超參數 × 若干取值)
    • Grid/Bayesian 搜尋會盲目探索大量組合

    LLM 驅動 AutoForecast 的思路:

    • 你提供:資料描述、目標、限制條件、白名單元件
    • LLM 輸出:一個「專家風格」的 pipeline blueprint
    • sktime craft():把 blueprint 變成真正的 estimator
    • 之後用交叉驗證 + 回測,打分這個藍圖好不好

    好處:

    • 搜尋空間更有結構:LLM 預先排除很多不合理組合
    • 成本可控:每次只訓練少數幾個「有合理解釋」的藍圖
    • 你可以 固化得分高的藍圖,變成穩定的產線預測器

    💡 關鍵: 相較於暴力掃描大量組合,讓 LLM 先縮小「合理藍圖集合」,能在相同算力下探索更有價值的模型設計。


    3. 把業務限制寫進 prompt:準確度只是其中一個目標

    在實務專案中,像房地產、銷售預測一樣,除了誤差小,你還會在乎:

    • 推理時間(每天要跑上千條 SKU / 房源)
    • 部署複雜度(不要依賴罕見套件或 GPU)
    • 解釋性(要能向業務說明為什麼預測變化)

    這些都可以在 prompt 中顯式告訴 LLM,例如:

    • 限定只能用某些 estimator:NaiveForecaster, ExponentialSmoothing, ElasticEnsemble, LightGBM-based forecaster...
    • 給出目標指標:sMAPE 或 MAE,並設計多目標(準確度 + latency)

    實作範例:銷售時間序列 LLM AutoForecast

    以下示範一個簡化版流程:

    • 場景:預測未來 3 個月每月團隊銷售額
    • 資料:月度歷史 revenue、房源數量、成交率、利率、季節 dummy 等

    假設你已經載入 sktime 與一個 LLM SDK(例如 Anthropic、OpenAI 等)。以下程式碼偏 pseudo,但結構可直接套入專案。

    1. 描述資料與目標,建立 LLM prompt

    schema_description = {
        "frequency": "M",  # 月度資料
        "target": "team_gci",  # 團隊佣金收入
        "horizon": 3,  # 預測 3 個月
        "exogenous_features": [
            "active_listings",     # 有效掛牌數
            "pending_transactions", # 待成交案件
            "mortgage_rate",      # 抵押貸款利率
            "seasonality_flags"   # 季節性特徵
        ],
        "constraints": {
            "max_pipeline_depth": 4,
            "allowed_estimators": [
                "NaiveForecaster",
                "ExponentialSmoothing",
                "ThetaForecaster",
                "LightGBMForecaster"
            ],
            "allowed_transformers": [
                "Detrender",
                "STLTransformer",
                "Lag",
                "DateTimeFeatures"
            ],
            "primary_metric": "sMAPE",
            "secondary_metric": "latency_ms",
            "max_fit_time_minutes": 15
        }
    }
    
    system_prompt = """
    你是一位時間序列預測專家,負責設計 sktime 預測 pipeline。
    
    要求:
    1. 只使用以下白名單中的 estimator 和 transformer。
    2. 避免過度複雜的 pipeline,深度不超過 4 層。
    3. 避免使用不存在的類別和參數,嚴格遵守 sktime API。
    4. 針對月度房地產團隊銷售數據,考慮趨勢與季節性。
    5. primary metric 是 sMAPE,次要考慮推理延遲,盡量用輕量模型。
    
    輸出格式:只輸出 JSON,欄位為 `pipeline_spec`,不可包含其他文字。
    `pipeline_spec` 要能被 sktime.craft() 解析。
    """
    
    user_prompt = f"資料與限制如下:\n{schema_description}\n請產生 pipeline_spec。"
    

    2. 呼叫 LLM,取得 pipeline blueprint

    from some_llm_client import LLM
    
    llm = LLM(api_key="...")
    
    response = llm.chat(
        system=system_prompt,
        user=user_prompt
    )
    
    blueprint = response["pipeline_spec"]  # 假設已解析 JSON
    print(blueprint)
    

    例:LLM 可能輸出類似(簡化)

    {
      "type": "TransformedTargetForecaster",
      "steps": [
        {"name": "detrend", "class": "Detrender", "params": {"forecaster": "NaiveForecaster"}},
        {"name": "stl", "class": "STLTransformer", "params": {"seasonal": 7}},
        {"name": "lag", "class": "Lag", "params": {"lags": [1, 2, 3, 6, 12]}},
        {"name": "model", "class": "LightGBMForecaster", "params": {"num_leaves": 31, "learning_rate": 0.05}}
      ]
    }
    

    💡 關鍵: 藉由 max_pipeline_depth = 4、白名單與 max_fit_time_minutes = 15 等約束,LLM 被強迫產出既合理又可在時限內完成訓練的藍圖。

    3. 用 craft() 轉成可執行 pipeline

    from sktime.craft import craft
    
    # 這裡的 blueprint 就是上一步 LLM 回傳的 JSON
    forecaster = craft(blueprint)
    
    print(type(forecaster))
    # e.g. <class 'sktime.forecasting.compose._pipeline.TransformedTargetForecaster'>
    

    如果 LLM 有亂給不存在的 class/參數,這一步會直接爆掉,所以建議外面包一層驗證:

    def safe_craft(blueprint):
        try:
            return craft(blueprint)
        except Exception as e:
            # 記錄錯誤,丟回給 LLM 做自我修正或直接丟棄該藍圖
            print("Invalid blueprint:", e)
            return None
    
    forecaster = safe_craft(blueprint)
    if forecaster is None:
        # 重新請 LLM 生成,或 fallback 到手寫 baseline
        ...
    

    4. 做時間序列交叉驗證與回測(避免 leakage)

    時間序列不能隨機 shuffle,必須用 滾動時間窗。

    from sktime.forecasting.model_selection import ExpandingWindowSplitter
    from sktime.performance_metrics.forecasting import mean_absolute_percentage_error
    
    cv = ExpandingWindowSplitter(
        initial_window=36,  # 例如先用 3 年訓練
        step_length=3,      # 每次往前滾 3 個月
        fh=[1, 2, 3]        # 評估 1-3 個月 horizon
    )
    
    CV_scores = []
    for train_idx, test_idx in cv.split(y):
        y_train, y_test = y.iloc[train_idx], y.iloc[test_idx]
        X_train, X_test = X.iloc[train_idx], X.iloc[test_idx]
    
        forecaster.fit(y_train, X=X_train)
        y_pred = forecaster.predict(fh=cv.fh, X=X_test)
    
        score = mean_absolute_percentage_error(y_test, y_pred)
        CV_scores.append(score)
    
    print("CV MAPE:", sum(CV_scores) / len(CV_scores))
    

    這個流程可以放在一個 LLMBlueprintForecaster 類別裡,作為你的 AutoForecast 前端代理:

    1. 給資料描述與限制
    2. LLM 產生藍圖
    3. craft() 轉成 pipeline
    4. 用時間窗交叉驗證打分
    5. 挑最佳藍圖,固化成產線模型

    建議與注意事項

    1. LLM 亂組 class / 參數:一定要有 validator

    常見問題:

    • 寫出不存在的 class 名稱(如 XGBoostForecaster 明明沒這個)
    • 傳錯 參數名稱 或型別

    實務上建議:

    • 自行維護一份 白名單 registry:ALLOWED_ESTIMATORS, ALLOWED_TRANSFORMERS,包含合法 class 與參數 schema
    • LLM 輸出後先做 schema validation,不合法就直接丟棄或要求 LLM 修正

    2. 避免過度複雜 pipeline:限制深度與組合數

    LLM 很容易產生「看起來很專業」的 pipeline:一堆 transformer 疊來疊去,訓練時間爆炸還容易 overfit。

    做法:

    • 在 prompt 裡明寫:max_pipeline_depth、禁止嵌套某些昂貴 transformer
    • 在 validator 裡硬限制步數,例如 len(steps) <= 4
    • 將 fit time / memory 也當成 約束條件,訓練時加上 timeout + 監控

    3. 嚴格避免 leakage:時間切割一律「只看過去」

    坑點:

    • 把整段資料做標準化 / 滯後特徵時,無意間用到未來資訊

    避免方式:

    • 一律用 sktime 的 transformer + forecaster pipeline,讓 transform 在 fit 只看到 train window
    • cross-validation 必須用 ExpandingWindowSplitter / SlidingWindowSplitter 類型
    • 在 prompt 裡提醒:不得使用未來的統計量(例如整體均值)來處理訓練資料

    4. 在專案中落地:把 LLM 當 search policy,而不是 oracle

    建議實務流程:

    1. Search policy:LLM 只負責提案藍圖,不直接上產線
    2. 離線評估:用固定的 backtest 配置(split、metric、timeout)評估每個藍圖
    3. 固化最佳藍圖:將 blueprint JSON 連同該版本資料 schema 一起存入 repo
    4. CI 自動回歸:
    5. 新資料 schema / 分布變化時,自動對舊藍圖重新訓練 + 打分
    6. 可以定期讓 LLM 在新 constraint 下重新產生藍圖,與舊版本對比
    7. 可重現性:所有 LLM 輸出(prompt + response)都要 versioning(例如存到 S3 / Git LFS),確保每個産線模型的來歷可追溯

    關鍵結論:

    • craft() + LLM = 可控的 AutoForecast 工具鏈,你掌控 search space 與評估邏輯
    • 把 LLM 當作「有經驗的建模同事」,而不是神諭;所有藍圖都要在 sktime 的嚴格回測與 CI 下過關,才能進產線

    這樣,在實際銷售 / 房地產等時間序列場景中,你可以以相對低成本,不斷迭代更好的預測流程,同時維持可重現與可維運的工程品質。


    🚀 你現在可以做的事

    • 在專案中安裝並載入 sktime,試著手動呼叫 craft() 建一個簡單 pipeline
    • 依照文中範例,實作一份包含 max_pipeline_depth 與白名單的 LLM prompt,讓 LLM 先產出一版 pipeline_spec
    • 為產出的 blueprint 加上 safe_craft() + 時間序列交叉驗證,建立一個最小可用的 LLMBlueprintForecaster 原型
  • Claude 私有化:自托管沙箱與 MCP 隧道實戰

    Claude 私有化:自托管沙箱與 MCP 隧道實戰

    📌 本文重點

    • 模型與編排托管在 Anthropic,工具與資料留在你 VPC
    • 透過 self‑hosted sandbox 把程式執行權收回內網
    • 用 MCP tunnel 只開安全出站連線打通內網工具
    • 權限設計與審計要靠嚴格的 tools schema 與網路邊界

    典型企業場景:你想讓 Claude Managed Agent 幫你跑 CI/CD、讀內網 Git、打內部 REST / Postgres API,但 不能開公網入站、不能把 Git 暴露出去。這次的 self‑hosted sandboxes + MCP tunnels 更新,基本上就是:

    模型與編排留在 Anthropic,工具執行與資料存取拉回你自己的 VPC,且只用安全出站連線。

    實務上等於多了一種選擇:不用把模型拉進內網自建 inference,也能在嚴格邊界內讓 Agent 控制 CI、讀 repo、查 DB。


    重點說明:架構與安全邊界怎麼畫

    1. Orchestrator vs Tools:誰在外面、誰在裡面?

    Claude Managed Agents 大致拆成兩層:

    1. Orchestrator(Anthropic 端托管)
    2. 理解使用者意圖、規劃步驟、決定要叫哪些工具。
    3. 透過 MCP 協定 呼叫你定義的 tools(MCP server)。

    4. Tools / MCP servers(你自己控制)

    5. 例如:git、CI/CD runner、Postgres、內部 REST API。
    6. 可以跑在 self‑hosted sandbox 裡(受控容器/VM),或直接在你內網 VPC。

    💡 關鍵: Orchestrator 永遠只看得到你暴露出的 MCP tools 與其 I/O,真正的 Git / DB 資料面與執行權都留在你 VPC。

    關鍵:Orchestrator 看不到你的 Git/DB,只能透過你暴露出的 MCP tools 操作,權限由你決定。


    2. MCP 協定與 server:最小必須心智模型

    MCP server 就是一個會講 JSON‑RPC over stdio / WebSocket 的後端,向 Agent 宣告自己有哪些工具。概念類似「強 typing 的 function calling」。

    一個簡化版的 MCP server schema(YAML)可能長這樣:

    name: corp-dev-tools
    version: 0.1.0
    
    tools:
      - name: git_read_file
        description: 讀取內部 Git repo 某個檔案內容
        input_schema:
          type: object
          required: [repo, path, ref]
          properties:
            repo: { type: string }
            path: { type: string }
            ref:  { type: string }
    
      - name: ci_trigger_pipeline
        description: 觸發 CI pipeline,只能在 allowlist 專案上執行
        input_schema:
          type: object
          required: [project, branch]
          properties:
            project: { type: string }
            branch:  { type: string }
    

    Agent 端會看到 明確的工具清單與參數結構,再決定何時呼叫。


    3. Self‑hosted sandbox:把「程式執行權」留在自己這邊

    self‑hosted sandbox 解決的是:「我不想讓 Anthropic 直接在他們 infra 上跑 shell / Python,請改在我 VPC 裡跑」。

    實作上可以是:

    • 一個專用 k8s namespace 或 Fargate / VM,跑 Anthropic 提供的 sandbox runtime。
    • Agent 要執行程式碼(例如跑單元測試、lint、git 操作),會透過安全通道把 code / 指令丟到這個 sandbox。

    典型邊界設計:

    • sandbox 只能出站連到:
    • 你的 Git / CI / DB / 內部 API
    • Anthropic 的 MCP tunnel endpoint
    • 不能直連其他敏感系統(或至少預設 deny)。

    實際好處: Agent 可以幫你跑 pytest、打 CI webhook、改 Git branch,
    但所有「可以執行程式碼的環境」都在你的控制下,方便做防火牆、資安掃描、審計。


    4. MCP tunnels:如何在不開洞的情況下連到內網

    MCP tunnel 解決:「我的 MCP server 在 VPC 裡,如何讓 Anthropic 托管的 Agent 打到它,但我不願意開入站 443?」

    典型拓撲:

    [Anthropic Orchestrator]
            ^
            | (MCP over tunnel)
            v
    [MCP Tunnel Client in VPC] ----> [MCP Server + Sandbox]
              (outbound TLS)
    

    關鍵特性:

    • 只有 VPC → Anthropic 的出站連線(類似反向隧道)。
    • 隧道建立後,Anthropic 端像是在打本地 MCP server,但實際流量是經由你建立的 mutual TLS / token 隧道轉回來。
    • 容易套用零信任思路:每條隧道都視為一個 identity,綁定最小權限的一組 tools。

    💡 關鍵: 只用出站隧道與 mTLS,就能在不開任何入站 port 的前提下,把內網工具安全地接給 Managed Agent 用。


    實作範例:VPC 內最小可行架構

    場景:

    • VPC 內:
    • 一台 MCP server + sandbox Pod/VM。
    • 連得上 GitLab、Postgres、內部 https://api.intra。
    • 出站:允許打到 Anthropic 的 MCP tunnel endpoint。

    1. MCP server:Git + Postgres + REST API

    以 Node.js 為例(pseudo‑code):

    import { MCPServer } from "@anthropic-ai/mcp-sdk";
    import { execSync } from "node:child_process";
    import { Client } from "pg";
    import fetch from "node-fetch";
    
    const gitAllowlist = ["service-a", "service-b"];
    
    const server = new MCPServer({ name: "corp-dev-tools" });
    
    server.tool("git_read_file", async ({ repo, path, ref }) => {
      if (!gitAllowlist.includes(repo)) {
        throw new Error("repo not allowed");
      }
      const base = "/srv/git/" + repo;
      const content = execSync(`git --git-dir=${base} show ${ref}:${path}`, {
        encoding: "utf8",
      });
      return { content };
    });
    
    server.tool("db_read_customer", async ({ id }) => {
      const client = new Client({
        host: process.env.PG_HOST,
        user: "readonly_agent",
        password: process.env.PG_PWD,
        database: "app",
        ssl: true,
      });
      await client.connect();
      const res = await client.query("SELECT id, name, status FROM customers WHERE id=$1", [id]);
      await client.end();
      return { rows: res.rows };
    });
    
    server.tool("call_internal_api", async ({ path, method, body }) => {
      if (!path.startsWith("/public-agent/") || method !== "POST") {
        throw new Error("not allowed");
      }
      const resp = await fetch(`https://api.intra${path}`, {
        method,
        headers: { "Authorization": `Bearer ${process.env.AGENT_TOKEN}` },
        body: JSON.stringify(body ?? {}),
      });
      const json = await resp.json();
      return { status: resp.status, data: json };
    });
    
    server.listen();
    

    重點:

    • gitAllowlist:避免 Agent 任意讀所有 repo。
    • Postgres 使用 readonly_agent 帳號,限制只讀特定 schema。
    • 內部 API 降到 /public-agent/ 子路徑 + 專用 token。

    2. MCP tunnel client:出站連上 Anthropic

    實際指令會以官方 CLI / container 為主,概念配置類似:

    anthropic-mcp-tunnel \
      --mcp-url=http://localhost:8000 \
      --agent-id=corp-ci-agent \
      --tls-cert=/etc/mcp/cert.pem \
      --tls-key=/etc/mcp/key.pem \
      --anthropic-endpoint=https://mcp-tunnel.anthropic.com \
      --tags=env:prod,scope:ci
    

    在 Anthropic 控制台,你會:

    • 建立一個 Managed Agent:corp-ci-agent。
    • 只綁定這條 tunnel 曝露的 corp-dev-tools MCP server。
    • 開啟前人工審閱 / 部分工具 auto‑approve(視風險)。

    3. ACL 與工具權限設計

    簡單的 policy‑as‑code 思路:

    agent: corp-ci-agent
    allowed_tools:
      - git_read_file
      - ci_trigger_pipeline
      - db_read_customer
      - call_internal_api
    
    constraints:
      git_read_file:
        repos: ["service-a", "service-b"]
        max_file_size_kb: 256
    
      db_read_customer:
        max_rows: 1
    
      call_internal_api:
        allowed_paths:
          - "/public-agent/deploy"
          - "/public-agent/status"
    

    你可以在 MCP server 裡讀這個 YAML,做額外校驗。不要把 ACL 寫死在 prompt 裡,防禦提示注入要靠程式碼與網路邊界。


    建議與注意事項:安全坑與實務整合

    1. 工具權限過大 = Agent RCE 風險

    OWASP Agent Top 10 已經把 工具濫用 / 權限濫用 列為前幾名風險。常見錯誤:

    • 一個 tool 可以執行任意 shell、對任何 DB 下任意 query。
    • Agent 可以打到整個內網,而不是只打 CI / Git / API Gateway。

    建議:

    • 一個 tool 做一件小事,強 schema,避免 free‑form SQL / shell。
    • DB 使用 只讀 + row‑level / column‑level policy。
    • 網路上用 安全群組 / SG 把 sandbox 能打的 IP 段鎖死。

    2. 審計 / Logging:要記「自然語言意圖 + 工具調用」

    很多企業只有 infra log,缺少「Agent 為什麼要做這件事」的上下文。

    建議最低標準:

    • 針對每次工具呼叫記錄:
    • user_id / session_id
    • Agent 看到的 自然語言任務描述(可脫敏)
    • 工具名稱 + input 參數(敏感欄位做 partial redaction)
    • 執行結果摘要 / status code

    這樣在事後對齊 OWASP 事件分析時,才能把「提示注入 → 工具濫用 → 資料外洩」串成一條 timeline。


    3. 網路與認證:守住 MCP 隧道與 secrets

    重點:把隧道視為一個高價值通道,跟 VPN 一樣認真看待。

    具體建議:

    • 隧道一律走 mTLS,cert 由內部 CA 或雲端 CA 管理。
    • 隧道 client 的 API token / cert 放在 Vault / KMS,sandbox 上只拿短期 lease。
    • 工具裡 不要回傳 secrets(例如整個 JWT 與 DB 密碼)到 Agent,必要時只在 server 端使用。
    • 若擔心 API key 滲透,在 sandbox 層加 egress proxy,對外送出的 HTTP header 做檢查 / scrub。

    4. 與 SOAR/服務目錄/Secrets 管理整合的實務問題

    實務上會遇到:

    • SOAR / ticket 系統:
    • 建議把「開 ticket / 查告警 / 執行 playbook」封裝成 MCP tools,權限沿用既有 RBAC。

    • 服務目錄(例如 Backstage):

    • MCP tools 可以讀服務目錄 API,讓 Agent 知道 repo 屬於哪個團隊、能不能改 config。

    • Secrets 管理(Vault/KMS):

    • 不要給 Agent 直接讀 Vault 的能力;改由 MCP tool 在 server 端解密,對 Agent 只暴露結果(或再加工)。

    5. 和「模型拉進內網自建 inference」的取捨

    Managed Agent + self‑hosted sandbox + MCP tunnel:

    • 優點:
    • 不用自己跑 LLM cluster,只負責工具、網路、權限。
    • 快速接雲端最新模型(含之後像 Mythos 這種安全模型的企業版)。
    • 合規上:資料只經由 tools 進出,你可以精準監控。

    • 缺點:

    • Orchestrator 還是在 Anthropic,那邊仍會看到 工具 I/O 摘要。
    • 對極端資料主權(完全不能出域)的場景不適合。

    自建 inference:

    • 優點:
    • 完整掌控模型與權限,所有 token 在你網段內。

    • 缺點:

    • 要自己做 Agent orchestration、tooling、guardrail、OWASP Top 10 風險防護。
    • 成本與維運門檻高。

    如果你目前已在雲上、允許「模型在外、資料在內」,這次的 Claude Managed Agents 私有化能力 是一個相對平衡的折衷:

    把最麻煩的 LLM 與 Agent orchestration 交給 Anthropic,把最敏感的程式執行與資料權限留在 VPC,用 MCP + sandbox 畫清楚邊界。

    🚀 你現在可以做的事

    • 在現有 VPC 內起一個簡單 corp-dev-tools MCP server(照文中 Node.js 範例改成你公司的 Git / DB / API)
    • 部署 anthropic-mcp-tunnel 類似的隧道 client,實測只用出站連線即可讓 Managed Agent 操作內網工具
    • 寫一份 YAML ACL(如文中 policy‑as‑code 範例),把 repo / DB / API 權限具體收斂後再開放給 Agent 使用