標籤: LLM 優化

  • 在 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 等參數
  • 用狀態機把 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 每天自動小幅優化
  • SmallCode 架構:讓 4B 模型也能帶專案

    SmallCode 架構:讓 4B 模型也能帶專案

    📌 本文重點

    • 小模型不適合搭配太多零碎 tools
    • 用少量 compound tools 封裝整個工作流
    • 加上可控 improvement loop 提升穩定性
    • 本地 4B 模型也能跑實用 coding agent

    在本地用 4B Gemma 這種小模型帶一個中大型 repo,傳統做法是「LLM + 一堆小工具」:讀檔、寫檔、跑測試、grep 全部拆成獨立 tool。結果大多數人遇到的狀況是:上下文爆炸、工具呼叫瘋狂往返、推理鏈斷裂,小模型最後只會在錯誤訊息上打轉。SmallCode 的重點就是把這些多步操作 封裝成少量高階 compound tools,加上一個可控的 improvement loop,讓 4B 模型也能穩定完成多檔案、反覆修改的任務。


    重點說明

    1. 為什麼「LLM + 一堆小工具」在小模型上會崩盤

    從工程視角,小模型掛點原因其實很單純:

    1. 上下文爆炸:
    2. 每呼叫一次 read_file、search、run_tests,LLM 就要重新看到整段對話 + 工具 I/O。
    3. 小模型 context 小、壓縮能力差,於是早期決策被擠掉,任務計畫完全失憶。

    💡 關鍵: 小模型的 context 與壓縮能力有限,多工具頻繁往返會迅速擠掉關鍵決策,導致任務中途失憶。

    1. 頻繁往返 + 推理鏈斷裂:
    2. 多工具設計變成:LLM → read_file → 回答 → write_file → 回答 → run_tests ...
    3. 每一步都要模型自己想「下一步該用什麼工具」,對小模型來說元認知成本太高,很容易在第 3、4 步就跑偏。

    4. 錯誤訊息解析太細碎:

    5. 錯誤訊息來了:run_tests → 報一堆 stacktrace → LLM 要自己決定要再 read_file 哪幾個檔案。
    6. 4B 模型常常讀錯檔或只讀到片段,最後修 bug 幻覺化。

    SmallCode 的做法是:把「讀檔 → 修改 → 測試」這個典型工作流視為一個原子操作,交給 compound tool 內部處理,LLM 只需要決定「要不要再試一次?」。


    2. Compound Tools:把多步工作流變成一個 API

    設計思路可以簡化成三件事:

    1. 把多步操作拉到工具內執行
    2. 典型例子:edit_and_test:
      • 根據指定 glob pattern 讀檔
      • 在本地套用 LLM patch(或簡單模板)
      • 寫回檔案
      • 跑 pytest / npm test,收集輸出
    3. 對 LLM 而言,這整件事就是 一次工具呼叫 + 一個結構化結果。

    4. 清楚定義輸入 / 輸出 Schema

    輸入 Schema(JSON Schema 或 Pydantic)範例:

    python
    class EditAndTestInput(BaseModel):
    goal: str # 自然語言:『讓 tests/test_api.py 全部通過』
    target_files: List[str] # ['src/api.py', 'tests/test_api.py']
    test_command: str = "pytest -q"
    max_edits: int = 3

    輸出 Schema:

    “`python
    class EditSummary(BaseModel):
    file: str
    diff: str # unified diff

    class EditAndTestOutput(BaseModel):
    success: bool
    edits: List[EditSummary]
    test_output: str
    error_summary: Optional[str]
    “`

    重點是讓小模型只要關注:有沒有成功?改了哪些檔?錯在哪裡?

    1. 在一次工具呼叫中完成工作流

    Tool handler(Python 假想範例):

    “`python
    def edit_and_test_tool(payload: EditAndTestInput) -> EditAndTestOutput:
    # 1. 收斂上下文:只讀需要的檔案內容
    files = {f: Path(f).read_text() for f in payload.target_files}

       # 2. 呼叫同一個或更小的 LLM 產生 patch(可本地或子進程)
       patch = call_llm_generate_patch(goal=payload.goal, files=files)
       edits = apply_patch_to_files(patch)
    
       # 3. 寫回檔案
       for e in edits:
           Path(e.file).write_text(e.new_content)
    
       # 4. 跑測試
       ok, test_output = run_command(payload.test_command)
    
       return EditAndTestOutput(
           success=ok,
           edits=[
               EditSummary(file=e.file, diff=e.diff)
               for e in edits
           ],
           test_output=test_output,
           error_summary=None if ok else summarize_test_output(test_output),
       )
    

    “`

    好處:主 agent 模型只要發出一個 edit_and_test 呼叫,不用自己 orchestrate read_file、write_file、run_tests,大幅降低 思考步數 和 context 髒亂度。


    3. Improvement Loop:可控的自我改進迴圈

    SmallCode 成效好的關鍵是:讓 retry 變成一級公民,而不是「失敗就結束」。

    1. 失敗偵測策略
    2. 以 compound tool 的輸出為主,不讓模型自己猜:
      • success == False
      • test_output 中出現關鍵字(FAILED、Traceback、AssertionError)
    3. 這樣 loop controller 可以用硬邏輯判斷下一步,不依賴 LLM 理解每一行 stacktrace。

    4. 重試策略

    簡單可用的架構:

    “`python
    MAX_ATTEMPTS = 5

    for attempt in range(1, MAX_ATTEMPTS + 1):
    tool_result = edit_and_test_tool(input_payload)

       if tool_result.success:
           break
    
       feedback = build_feedback_prompt(tool_result)
       input_payload.goal = feedback  # 將錯誤摘要回餵給模型
    

    “`

    build_feedback_prompt 可以把 error_summary + 失敗測試名稱打包,讓下一輪 LLM 更聚焦。

    1. 避免無限 loop 的做法
    2. 硬限制:MAX_ATTEMPTS、最大 wall time。
    3. 檢查 edits 是否變化:若連續兩次 diff 幾乎一樣(甚至相同 hash),就早停。
    4. error signature 去重:同一個 assertion / stacktrace 重複出現 N 次就停止,標記為「需要人介入」。

    這樣,4B 模型只要能理解「這次錯在哪裡、我還有幾次機會」,就足以在迴圈中持續收斂。


    4. 在本地 4B 模型上的部署實務

    以 Gemma 2 4B 為例,用 llama.cpp / Ollama 都可以吃得很順,但細節會影響體驗:

    1. 記憶體與延遲粗估
    2. 4B Q4_K_M:VRAM 約 3–4 GB;Q6 大概 5–6 GB。
    3. context 8k、推理 1 token ≈ 10–40 ms,視 CPU/GPU 而定。
    4. agent 架構推薦:主模型用中量化(Q4/Q5)+ MTP(若硬體支援),工具內部幫忙 patch 的模型可以更小或更低位元。

    💡 關鍵: Gemma 2 4B 在 Q4 量化、約 3–4 GB VRAM 和 8k context 下即可順暢運行,適合作為本地實用 coding agent 的基礎。

    1. llama.cpp 整合範例

    bash
    # 轉 GGUF 後
    ./llama-cli \
    -m gemma-2-4b-q4_k_m.gguf \
    -c 8192 \
    -ngl 35 \ # offload 到 GPU layer 數
    --temp 0.2 \
    --top-p 0.9

    在你的 agent server(Python)裡包一層簡單的 HTTP:

    “`python
    from llama_cpp import Llama

    llm = Llama(model_path=”gemma-2-4b-q4_k_m.gguf”, n_ctx=8192)

    def chat(messages):
    return llm.create_chat_completion(
    messages=messages,
    tools=TOOLS_SCHEMA, # compound tools 定義
    )
    “`

    1. Ollama 整合範例

    yaml
    # Modelfile
    FROM gemma2:4b
    PARAMETER temperature 0.2
    PARAMETER num_ctx 8192

    Python 呼叫:

    “`python
    import requests, json

    def ollama_chat(messages, tools=None):
    payload = {“model”: “gemma2-4b”, “messages”: messages}
    if tools:
    payload[“tools”] = tools
    r = requests.post(“http://localhost:11434/api/chat”, json=payload)
    return r.json()
    “`

    注意:不要貪心開太大 context,4B 小模型在 16k context 上的品質掉得很明顯,8k 左右通常較穩定。


    實作範例:簡化版 SmallCode 架構

    以下是一個「可以直接改造」的最小可行架構:

    # 1. 定義 compound tools
    TOOLS_SCHEMA = [
        {
            "type": "function",
            "function": {
                "name": "edit_and_test",
                "description": "Edit target files to satisfy goal and run tests.",
                "parameters": EditAndTestInput.model_json_schema(),
            },
        },
    ]
    
    # 2. Agent 迴圈
    
    def coding_agent(task_description: str):
        messages = [
            {"role": "system", "content": "你是嚴謹的資深工程師,專注讓測試通過。"},
            {"role": "user", "content": task_description},
        ]
    
        for attempt in range(1, 6):
            resp = ollama_chat(messages, tools=TOOLS_SCHEMA)
            choice = resp["message"]
    
            if "tool_calls" not in choice:
                # 當成總結
                return choice["content"]
    
            for tool_call in choice["tool_calls"]:
                if tool_call["function"]["name"] == "edit_and_test":
                    args = json.loads(tool_call["function"]["arguments"])
                    result = edit_and_test_tool(EditAndTestInput(**args))
    
                    # 記錄 tool 結果
                    messages.append({
                        "role": "tool",
                        "name": "edit_and_test",
                        "content": result.model_dump_json(),
                    })
    
                    if result.success:
                        messages.append({
                            "role": "user",
                            "content": "測試已通過,請簡要總結你做了什麼修改。",
                        })
                        final = ollama_chat(messages)
                        return final["message"]["content"]
    
        return "多次嘗試仍未通過測試,請人工檢查。"
    

    實際好處:
    – 你不需要讓 4B 模型自己 orchestrate 所有 file ops,只決定「目標」和「是否繼續嘗試」。
    – 迴圈與成功判斷在 host 程式碼內可觀測、可監控,易於 debug。


    建議與注意事項

    1. tool 太細 vs 太粗的 trade-off
    2. 太細:read_file、write_file、run_tests 分開 → 小模型迷路。
    3. 太粗:一個 tool 內藏太多隱含狀態 → 工具結果難以解釋與監控。
    4. 建議:以「一次迴圈可解決的一個明確目標」為單位設計 compound tool,例如:edit_and_test_suite、add_feature_and_generate_tests。

    5. 錯誤訊息解析策略

    6. 不要把整個 test log 丟給小模型,先在 host 端做:
      • 只保留最後一個錯誤 block
      • 摘要檔名、行號、錯誤訊息
    7. 這樣可以避免 4B 模型被 1k tokens 的 noisy log 淹沒。

    8. 測試覆蓋率不足導致的幻覺修 bug

    9. 如果只有 1–2 個測試,小模型會傾向「只讓這兩個測試過」而破壞其他邏輯。
    10. 實務上:

      • 優先在 agent pipeline 前補上最小必要測試
      • 或在 tool 裡檢查 diff 是否只動到相關檔案/區塊(heuristic)。
    11. 如何監控與記錄 agent 決策

    12. 每一次迴圈記錄:
      • 使用的 tool 名稱 + 輸入參數
      • diff 摘要(檔名、行數、行數變化量)
      • test command 與結果(pass/fail、耗時)
    13. 建議:

    text
    logs/
    session-2025-05-21T12-34-56Z/
    step-01-request.json
    step-01-edit_and_test-input.json
    step-01-edit_and_test-output.json
    step-01-diff.patch

    之後你可以回放整個 session,分析為什麼某一輪跑偏,進而調整 tool schema 或 system prompt。


    總結:如果你想在本地 4B 模型上做實用的 coding agent,不要再堆滿十幾個零碎 tools。把關鍵工作流收斂成少量 compound tools,加上一個明確可控的 improvement loop,再配合 llama.cpp / Ollama 的輕量部署,就能把原本只敢交給 GPT-級別模型的任務,下放到可自託管的小模型上。


    🚀 你現在可以做的事

    • 在現有 agent 專案中,把零碎的 read_file / write_file / run_tests 整合成一個 edit_and_test compound tool
    • 用 Gemma 2 4B + Ollama 或 llama.cpp 在本地啟一個 8k context、Q4 量化的測試環境
    • 為你的 repo 實作一個最小版 improvement loop,限制 MAX_ATTEMPTS 並記錄每次 diff 與測試結果
  • 自我優化 LLM Stack 實戰架構

    自我優化 LLM Stack 實戰架構

    📌 本文重點

    • 用結構化 trace 做 LLM observability
    • 以多模型路由平衡成本、延遲、質量
    • 用真實流量自動微調與 A/B 測試
    • 建立安全可控的自動優化閉環

    手動挑模型、改 prompt、算預算,做到上線後你會發現:每個路徑都在燒錢,而且調一次就壞一次。這篇的結論很直接:

    把「觀測 → 評分 → 路由 → 微調」做成閉環,你的 LLM Stack 會自己變便宜、變準、變穩定,而不是靠工程師加班微調。

    下面用一個可落地的架構,示範:
    – 要記哪些欄位才能做 LLM observability
    – 怎麼設計 線上多模型路由(成本 / 延遲 / 質量三者權衡)
    – 用真實流量做 持續微調 + 線上 A/B 測試
    – 如何在 安全可控 的前提下讓這個 loop 自動跑


    重點說明

    1. 觀測是自我優化的資料 API:要記什麼?

    你要的不是 log,而是可以訓練 & 決策的 結構化 trace。一筆 LLM 呼叫最少要記:

    • 請求層級欄位
    • trace_id:關聯前後多次呼叫
    • tenant_id / user_id:用於分群 & 權限
    • task_type:如 summarize, classify, tagging(路由和微調的最重要欄位)
    • 模型與成本欄位
    • model_name:如 gpt-4.1, local-7b-v1
    • input_tokens / output_tokens
    • cost_usd:用 provider 單價事後計算
    • latency_ms:end-to-end 延遲
    • 內容與品質欄位
    • prompt, completion(支援 PII 遮蔽)
    • quality_score:0–1 或 0–100,可來自:
      • 人工評分
      • 規則(例如是否通過 JSON schema)
      • LLM-as-judge 模型給分
    • hallucination_flag / safety_flag:是否被檢測為幻覺或違規

    💡 關鍵: 把每次 LLM 呼叫記成可查詢的結構化 trace,而不是散亂 log,才能支撐路由、微調與監控三種決策。

    這些欄位之後會被用在:
    – 自動模型路由(根據歷史質量 + 成本)
    – 持續微調(從高信心樣本抽訓練資料)
    – 質量監控(模型版本切換時是否退步)

    像 Torrix 這類自託管 observability 工具已經把大部分欄位幫你設計好了,你只要在程式碼層接上 proxy 或 SDK 即可。


    2. 多模型路由:把成本 / 延遲 / 質量變成可調參數

    目標:對每一類請求,自動選擇「在 SLA 內成本最低、且質量不低於門檻」的模型。

    常見做法:
    1. 用 embedding 對請求做 clustering,找到「相似任務族群」
    2. 在每個 cluster 裡統計:每個 model_name 的平均 quality_score, cost_usd, latency_ms
    3. 設計一個路由 scoring 函數:

    ( \text{score} = w_q · q – w_c · \log(1+cost) – w_l · \log(1+latency) )

    • w_q, w_c, w_l 是你可調的權重(例如 B2B 產品就偏質量,內部工具偏成本)

    在線上:
    – 每次請求先預測 cluster(根據 task_type + embedding)
    – 查表得到該 cluster 下每個模型的歷史 score
    – 選擇 score 最高模型,加上一點探索策略(epsilon-greedy / UCB)確保新模型有被試用機會

    實際好處:
    – 把「今天要不要全站切到新模型?」變成連續微調權重的線上學習問題
    – 你只要設定業務指標(每月預算、延遲 SLA),Router 會幫你在可接受範圍內壓成本


    3. 真實流量驅動的持續微調 + A/B 測試

    你不需要標一大堆資料,反而是:
    – 利用線上的 quality_score + hallucination_flag 自動篩樣本
    – 抽取高信心樣本給 7B/8B 模型微調
    – 再把微調後模型放回 Router 做灰度 A/B 測試

    做法可以類似 Reddit 那個案例:
    – 第 1–3 週:用 GPT-4/5.x 當 teacher,產生高品質標註
    – 第 4 週起:用這些資料微調 7B 模型接管特定 task(例如 classify / tagging / summarize)
    – 然後透過 Router 把低風險請求(內部標註、非生死決策)逐步導到 7B 模型

    💡 關鍵: 把旗艦模型當 teacher,用真實流量訓練 7B/8B 模型,可以做到品質接近但成本只剩個位數百分比。

    這樣可以做到「95% 與旗艦模型一致,但成本是 2%」的效果。


    實作範例

    下面用一個簡化的 Python 範例,示範:
    – 接上 Torrix 之類 observability
    – 寫一個最小可用的 router
    – 基於線上資料做粗略的微調樣本抽取與 A/B 測試策略


    1. 設計 Trace 結構與上報(以 Torrix HTTP proxy 為例)

    import requests
    import time
    
    TORRIX_PROXY_URL = "http://localhost:8787/proxy"  # Torrix 的 HTTP proxy
    
    MODELS = {
        "fast": "gpt-4o-mini",
        "strong": "gpt-4.1",
        "cheap_local": "local-7b-v1",
    }
    
    
    def call_llm(model_key: str, prompt: str, task_type: str, meta: dict):
        start = time.time()
    
        payload = {
            "model": MODELS[model_key],
            "messages": [{"role": "user", "content": prompt}],
            "temperature": 0.2,
            # 重要:附上自訂 metadata,方便 observability / 分群
            "metadata": {
                "task_type": task_type,
                "user_id": meta.get("user_id"),
                "tenant_id": meta.get("tenant_id"),
            },
        }
    
        # 經由 Torrix proxy 轉發,Torrix 會自動記錄 token, cost, latency 等
        resp = requests.post(TORRIX_PROXY_URL, json=payload)
        resp.raise_for_status()
    
        latency_ms = (time.time() - start) * 1000
        data = resp.json()
    
        return {
            "completion": data["choices"][0]["message"]["content"],
            "latency_ms": latency_ms,
            # token / cost 會在 Torrix 裡算,因此這裡只做最小回傳
        }
    

    實務上你還會再寫一個 async wrapper,確保不堵住整個 API。


    2. 最小可用 Router:基於 task_type + 歷史表現

    假設我們在背景 job 定期從 observability DB 撈聚合數據,產生一個 routing table:

    # 假設這個表是 batch job 每 5 分鐘更新一次
    # 由 observability 系統依 task_type + model 聚合而來
    ROUTING_TABLE = {
        # task_type: {model_key: {"q": quality, "c": cost, "l": latency_ms}}
        "summarize": {
            "fast": {"q": 0.92, "c": 0.002, "l": 800},
            "strong": {"q": 0.96, "c": 0.01,  "l": 1200},
            "cheap_local": {"q": 0.90, "c": 0.0004, "l": 950},
        },
        "classify": {
            "fast": {"q": 0.94, "c": 0.002, "l": 700},
            "cheap_local": {"q": 0.93, "c": 0.0004, "l": 600},
        },
    }
    
    # 路由權重:可透過環境變數或管理介面動態調
    W_Q = 1.0  # 質量
    W_C = 3.0  # 成本敏感度
    W_L = 0.5  # 延遲敏感度
    
    
    def select_model(task_type: str, explore_eps: float = 0.05) -> str:
        import math, random
    
        # 探索: 以小機率隨機挑一個模型,給新模型累積資料機會
        if random.random() < explore_eps:
            return random.choice(list(MODELS.keys()))
    
        stats = ROUTING_TABLE.get(task_type)
        if not stats:
            # 沒有歷史資料時的 fallback 策略
            return "fast"  # 或者直接用強模型保守處理
    
        best_score, best_model = -1e9, None
        for model_key, v in stats.items():
            q, c, l = v["q"], v["c"], v["l"]
            score = W_Q * q - W_C * math.log(1 + c) - W_L * math.log(1 + l)
            if score > best_score:
                best_score, best_model = score, model_key
    
        return best_model or "fast"
    
    
    def handle_user_request(prompt: str, task_type: str, meta: dict):
        model_key = select_model(task_type)
        res = call_llm(model_key, prompt, task_type, meta)
        return res["completion"]
    

    這樣你的 API 層就已經有一個可學習的 router,之後只要讓 batch job 持續更新 ROUTING_TABLE 即可。


    3. 用真實流量抽訓練資料 + A/B 測試策略(偽碼)

    下面的 pseudo code 示意:
    – 如何從 observability DB 抽出高品質樣本
    – 微調本地 7B 模型
    – 灰度放量到 router

    # 1. 從 trace DB 抽樣本(例如從 Torrix 的 SQLite / exports)
    # SELECT prompt, completion, quality_score
    # FROM traces
    # WHERE task_type = 'classify'
    #   AND quality_score >= 0.9
    #   AND hallucination_flag = 0
    #   AND model_name IN ('gpt-4.1', 'gpt-4.5')
    # LIMIT 100_000;
    
    # 2. 整理成 SFT 資料格式
    # {"messages": [{"role": "user", "content": prompt},
    #               {"role": "assistant", "content": completion}]}
    
    # 3. 用你習慣的框架(例如 LlamaFactory / axolotl)做 SFT
    
    # 4. 微調完得到 local-7b-v2,先只在 router 裡給 5% 流量:
    # - 在 ROUTING_TABLE 中,classify 下新增 cheap_local_v2 的統計
    # - select_model() 的 epsilon-greedy 會開始給它少量流量
    
    # 5. 定期比較:
    # - cheap_local_v2 vs cheap_local_v1 vs fast 在同一 task_type 的 quality_score
    # - 只有當 v2 的質量穩定 >= v1,才逐步提高 v2 的預設權重
    

    關鍵是:所有決策依賴線上真實質量分數,而不是人工測幾個 prompt。


    建議與注意事項

    1. 資料隱私:觀測≠把所有東西存一份

    • 對 Prompt / Completion 要做:
    • PII 遮蔽(email、電話、身分證、住址)
    • 對敏感欄位做 hash / tokenization,只保留足夠做分群的特徵
    • 如果你使用像 Torrix 這樣的自託管工具:
    • 優先把 SQLite / volume 放在私有網段,避免開放到公網
    • 對離線匯出的 trace 做加密儲存 & 權限控管

    實際風險:一旦 trace 被外流,不只是 prompt 洩漏,連你使用了哪些模型、成本結構都會被看光。


    2. 評分標準漂移(Evaluation Drift)

    當你改了:
    – LLM-as-judge 的版本
    – 質量打分 rubric(例如原本只看正確性,後來加入安全性)

    你歷史的 quality_score 就不再可比。建議:

    • 在 trace 裡加上 evaluator_version:
    • evaluator_model_name
    • rubric_version(JSON schema 或 hash)
    • 做趨勢分析時,同一條圖上只放同 evaluator_version 的資料
    • 如果要重跑評分,記得把舊分數保留一份,避免回溯分析被污染

    3. 模型切換導致行為不穩定

    多模型路由會遇到一個常見坑:
    – 業務邏輯假設「回傳格式永遠一樣」
    – Router 為了省錢,幫你換成另一個模型
    – 結果 JSON schema 不穩、排序不同、偶爾講幹話 → 下游全部爆掉

    緩解方式:
    – 在 observability 層記錄:schema_valid_flag(是否通過 JSON schema 驗證)
    – 對格式敏感的任務,在 Router 做:
    – 只允許通過 schema 驗證率 > 某門檻的模型
    – 或硬性綁定單一模型,先解決穩定性再談成本
    – 切換模型時先在只讀場景做 shadow traffic:
    – 用新模型跑同一批請求,但不回給使用者,只記分數
    – 分數穩定後再逐步放量


    4. 安全可控的自動 loop:永遠保留手煞車

    即使是自我優化架構,也要留:
    – 全局開關:一個環境變數就可以把 router 關掉,全部打到保守模型
    – 模型白名單:router 只能從白名單裡選,避免誤打到測試中的模型
    – 預算上限:
    – observability 層記累計 cost_usd
    – 一旦超過日/月預算,強制把高單價模型設為 offline

    搭配這些保護,你才敢讓自動 loop 長期自己跑,而不用每天盯帳單。

    💡 關鍵: 有全局開關、白名單與預算上限等「手煞車」,才能放心讓自動優化長期在線運行。


    總結:

    • LLM observability 不是畫漂亮 dashboard,而是提供可訓練 + 可決策的結構化 trace。
    • 多模型路由 把成本 / 延遲 / 質量變成可調參數,用線上真實質量分數自動選模型。
    • 用真實流量微調小模型 + A/B 測試,可以在特定任務上達到旗艦模型 90–95% 的效能,成本卻只要幾%。
    • 同時注意資料隱私、評分標準漂移、模型切換穩定性,並保留手動「手煞車」,你就能讓 LLM Stack 在安全邊界內自己變強、自己變便宜。

    🚀 你現在可以做的事

    • 列出並實作文中提到的 trace 欄位,接上現有 LLM 呼叫流程
    • 寫一個簡單的 select_model(),用歷史 quality_score + cost_usd 做最小可用路由
    • 從線上流量中抽樣高 quality_score、低 hallucination_flag 的請求,整理成 SFT 資料集準備微調小模型
  • RL 訓練版 Prompt Cache 7.5x 提速解析

    RL 訓練版 Prompt Cache 7.5x 提速解析

    📌 本文重點

    • 長 prompt / 短 response RL 訓練會浪費 >90% 計算
    • 把推理用 KV/prefix cache 思路搬進帶梯度訓練可大幅提速
    • 在 Qwen3.5-4B 上實測最高約 7.5x throughput 提升

    長 prompt、短 response 的 RLHF/RLAIF 任務(例如對話評分、工具調用評分)有一個非常痛的點:每個樣本都在重算同一段 prompt。對 1000-token prompt、100-token response 的場景,你實際上有 >90% 的 FLOPs 在白白重褾。這篇要講的是:如何把推理時的 KV/prefix cache 思路搬進帶梯度的 RL 訓練,在 Qwen3.5-4B 上實測最高拿到 7.5x 速度提升,並給你一套可以直接落地的工程實作方案。

    💡 關鍵: 在長 prompt / 短 response 場景中,重用 prompt 前向計算可將大部分重複 FLOPs 直接省掉,帶來數倍級 throughput 提升。


    重點說明

    1. 為什麼 RL 訓練會浪費那麼多計算?

    典型的 RLHF/RLAIF 術次資料形態:

    • prompt:系統 + 多輪對話 + 任務描述(幾百到上千 tokens)
    • response:模型生成或候選回答(幾十到一兩百 tokens)

    多數開源 RL engine(包括許多自寫 pipeline)會:

    [ prompt tokens ][ response tokens ]
      T_prompt           T_resp
    

    對每一個樣本、每一次 rollout / gradient step,都從頭跑整條序列,雖然 prompt 完全相同,只是 response 不同。這會帶來幾個直接影響:

    1. GPU 利用率被長 prompt 綁死:
    2. 你以為自己 batch size 是 64,其實「有效」只有在 response 段,前面 90% 的計算是在重放。
    3. batch 設計被 context 長度限制:
    4. 1000+ token prompt 會吃掉大部份 memory,導致你無法疊大 batch,只能靠 gradient accumulation,進一步增加 step latency。
    5. RL 特有放大器:
    6. 同一個 prompt 下可能要算多個候選 response、policy/value 多頭、不同 reward function,全都從 prompt 重新 forward 一次。

    因此,只要你是「長 prompt / 短 response」型任務,任何一點在 prompt 端節省的 FLOPs,都是純利潤。


    2. 把 KV/prefix cache 搬進訓練:核心思路

    推理時我們早就習慣用 KV cache/prefix cache:

    1. 先跑一次 prompt,存下每層的 key/value(或 hidden states)。
    2. 生成 response 時,只計算增量 token,復用前綴。

    在訓練中要做到類似的事情,難點在於:

    • 我們需要 完整的 computation graph(for backprop)。
    • 不能只存數值(像推理那樣),還要讓 autograd 知道這些值是可導的。
    • 不能打壞 attention:response 的 attention 要能看見 prompt token 的 hidden states。

    一種工程上可行的做法(簡化描述):

    1. 把序列拆成兩段圖:prompt graph + response graph。
    2. prompt 部分:
    3. 前向一次,拿到 prompt hidden states(例如每層的 h_prompt)與最後一層的 cache-like 表示。
    4. 保留其 computation graph(不 detach),但不馬上 backward。
    5. response 部分:
    6. 再跑一次 LLM,但將 prompt 當成固定 prefix 傳入,使 response token 的 attention 能看到這些 prefix hidden states。
    7. 在 PyTorch 裡可以透過自訂 forward 函數,把 prompt hidden states 塞回 attention 模組,類似手動實作 prefix cache。
    8. loss 計算只對 response tokens 做(例如 policy loss、value loss),但梯度會沿著 response→prompt 的 graph 反傳,保證不破壞訓練正確性。

    關鍵是:

    • 只對 prompt 前向一次,但仍然讓 prompt 參與梯度更新。
    • 對同一 prompt 的多個 response,重複使用一份 prompt hidden states(甚至在一個批次中共享)。

    在 Qwen3.5-4B 上,reddit 實測:

    • prompt : response ≈ 10:1(例如 1000:100)
    • RL 任務:長對話 + 短完成
    • 快取後在長 prompt/短 response 工作負載下 最高取得 ~7.5x step throughput 提升(取決於實際長度比與 IO/通信開銷)。

    💡 關鍵: 當 prompt 與 response 長度比約 10:1 時,只重算 response 部分可在實測中帶來約 7.5 倍 step throughput 提升。


    3. 什麼任務最吃紅利?

    根據 Qwen3.5-4B 測試經驗與工作負載特性,大致可以這樣判斷:

    1. 長 prompt / 短 response(T_prompt / T_resp ≥ 4)
    2. 如:對話 RLHF 評分(用戶上下文很長,模型答覆很短)。
    3. 工具調用評分:所有工具 schema + log 作為 prompt,再對短 decision 進行 RL。
    4. 部分代碼 RL:整個大檔案為 prompt,模型只改一小段。
    5. 這類場景通常可以拿到 3x–7.5x 的實際提速。

    6. 中 prompt / 中 response(T_prompt / T_resp ≈ 1)

    7. 如:通用問答 RLHF(prompt 只有一兩句,回答較長)。
    8. 提速有限,約 1.2x–2x,且實作複雜度可能不值。

    9. 短 prompt / 長 response(T_prompt / T_resp < 1)

    10. 基本沒紅利,甚至會因複雜控制流、多段 graph 而變慢。

    實務上可以用一條 thumb rule:

    如果你平均的 prompt token 數是 response 的 3 倍以上,就應該認真評估導入。

    💡 關鍵: 當 T_prompt 至少約為 T_resp 的 3 倍時,引入訓練版 prompt cache 通常才有顯著性價比。


    實作範例

    以下示例是 PyTorch 為主,偏 pseudo code,但結構與實務工程接近。

    1. 資料結構與 DataLoader 改寫

    我們先把一個 RL batch 明確拆成 prompt / response:

    # 每個樣本:
    # prompt_ids: [T_p]
    # resp_ids:   [T_r]
    
    class RLDataset(torch.utils.data.Dataset):
        def __getitem__(self, idx):
            item = self.data[idx]
            return {
                "prompt_ids": item.prompt_ids,   # 長
                "resp_ids": item.resp_ids,       # 短
                "reward": item.reward,           # 或 advantage
            }
    
    
    def collate_fn(batch):
        # padding & batch 組合
        prompt_ids = pad_sequence([b["prompt_ids"] for b in batch], batch_first=True)
        resp_ids   = pad_sequence([b["resp_ids"]   for b in batch], batch_first=True)
    
        # 生成對應 mask
        prompt_attn_mask = (prompt_ids != pad_token_id)
        resp_attn_mask   = (resp_ids   != pad_token_id)
    
        return {
            "prompt_ids": prompt_ids,
            "resp_ids": resp_ids,
            "prompt_mask": prompt_attn_mask,
            "resp_mask": resp_attn_mask,
            "reward": torch.tensor([b["reward"] for b in batch]),
        }
    

    2. 模型 forward:拆成 prompt graph + response graph

    假設你有一個可插拔的 LLM 模型 model,我們新增兩個關鍵 API:

    • model.forward_prompt(...):只跑 prompt,返回 hidden states(及必要 cache)。
    • model.forward_response_with_prefix(...):給定 prefix hidden states,跑 response。
    class RLPromptCacheModel(nn.Module):
        def forward_prompt(self, input_ids, attention_mask):
            # 返回每層的 hidden,或最後一層即可
            # 重要:不要 detach,保持 grad
            outputs = self.transformer(
                input_ids=input_ids,
                attention_mask=attention_mask,
                output_hidden_states=True,
            )
            return outputs.hidden_states  # list[Layer][B, T_p, H]
    
        def forward_response_with_prefix(self,
                                         resp_ids,
                                         resp_mask,
                                         prompt_hidden_states,
                                         prompt_mask):
            # 這裡需要改造 attention:
            # 讓每層 self-attention 的 KV = [prompt, resp]
            # 可以在每層 module 裡寫一個 hook,或實作 custom attn。
            outputs = self.transformer_with_prefix(
                resp_ids=resp_ids,
                resp_mask=resp_mask,
                prefix_hidden_states=prompt_hidden_states,
                prefix_mask=prompt_mask,
            )
            return outputs.last_hidden_state
    

    核心點:transformer_with_prefix 要做到:

    • 對於每層的 self-attention:
    • query 來自 response tokens;
    • key/value 為 [prefix_hidden_states; resp_hidden];
    • 這讓 response token 能正常 attend 到 prompt,並保持完整 graph。

    實務上可以參考 FlashAttention / prefix-tuning 的實作方式,直接拼接 prefix hidden 作為額外 token,再控制 mask:

    def transformer_with_prefix(...):
        # 假設我們把 prefix & response 在 time 維度上串起來
        # 注意這裡是邏輯串接,實際可用 concat + mask 控制
        concat_hidden = torch.cat([prefix_hidden, resp_emb], dim=1)  # [B, T_p+T_r, H]
        concat_mask   = torch.cat([prefix_mask, resp_mask], dim=1)   # [B, T_p+T_r]
    
        # 交給原本的 transformer 做 self-attention
        outputs = self.base_transformer(
            hidden_states=concat_hidden,
            attention_mask=concat_mask,
        )
        # 只取 response 對應位置的輸出
        resp_hidden_out = outputs.last_hidden_state[:, -resp_len:, :]
        return resp_hidden_out
    

    3. Loss 計算與 RL head

    以 policy gradient 為例,我們只對 response token 做 loss:

    prompt_hs = model.forward_prompt(batch["prompt_ids"], batch["prompt_mask"])  # list[L]
    
    resp_logits = model.forward_response_with_prefix(
        batch["resp_ids"],
        batch["resp_mask"],
        prompt_hs,
        batch["prompt_mask"],
    )
    
    # policy head
    logits = policy_head(resp_logits)  # [B, T_r, V]
    log_probs = F.log_softmax(logits, dim=-1)
    
    # 只對實際採樣到的 token 做 loss
    # 假設 resp_ids 是我們的 action
    token_logp = log_probs.gather(-1, batch["resp_ids"].unsqueeze(-1)).squeeze(-1)
    
    # 依 RL 演算法計算 advantage 等
    loss = -(token_logp * advantage_mask).sum() / num_valid_tokens
    loss.backward()
    

    因為 prompt_hs 沒有被 detach,梯度會沿著 response 部分回傳到 prompt 部分,等效於一次走完整個序列,但 prompt 只 forward 一次。


    4. 與 gradient checkpointing / mixed precision / DDP 整合

    • gradient checkpointing:
    • 可以只對 response graph 開啟 checkpoint,prompt graph 一般不需要再切。
    • 若 prompt 特別長,可在 prompt 段也設 checkpoint,但要注意不要把 cache 給破壞(照 layer 切即可)。

    • mixed precision (AMP/Fp16/bf16):

    • 保持 prompt & response forward 使用同一個 torch.cuda.amp.autocast 區塊。
    • prompt cached hidden 和 response 的精度必須一致,避免 dtype mismatch。

    • DDP/FSDP:

    • 基本原則:prompt forward 也在每個 rank 上做一次,不要跨 rank 共用 hidden,避免額外通信。
    • 對 FSDP 來說,prompt hidden 是 activation,照樣會被 shard/rebuild,不需要特別處理。
    • 注意 loss scale 及 no_sync() 區段,確保多 step accumulation 時 prompt/response 的 backward 一致。

    建議與注意事項

    1. 常見坑

    1. 快取導致樣本 shuffle 不均
    2. 若你把「相同 prompt 的多個 response」綁在一起,容易造成某些 prompt 被過度訓練。
    3. 建議在 dataset 層維持 樣本級 shuffle,不要把 prompt 當成硬分桶,或定期重組 group。

    4. mask 錯誤導致梯度泄漏

    5. 如果 attention mask 沒處理好,可能出現:response token 看到未來 token,或不同樣本互相看到彼此的 prompt。
    6. 尤其在 concat prefix 時,要確認:

      • padding token 完全被 mask 掉;
      • prefix 與 response 的因果 mask 正確(response 不該看到未來 response)。
    7. policy / value head 不一致

    8. 很多 RL pipeline 會同時跑 policy head + value head。
    9. 如果你只對 policy 路徑用 prompt cache,而 value 還在跑 full sequence,
      會導致兩邊的 feature distribution 不一致。
    10. 建議:兩個 head 共用同一套 prompt+response 拆圖邏輯,或至少在 feature 塊對齊。

    2. 什麼時候值得導入?

    你可以簡單做一個估算:

    • 計算平均 T_prompt / T_resp。
    • 估算你的訓練 step 中,有多少時間是花在 forward(相對於通信/IO)。
    • 目標提速 ≈ T_total / (T_resp + T_prompt / cache_reuse_factor)。

    若粗算下來:

    • 理論加速 > 2x,且你目前的 RL 訓練被 FLOPs-bound(非 IO-bound),那導入很可能值得。
    • 若你被 data loading 或 reward 模型 inference 卡住,則先優化 pipeline 再考慮這一層。

    3. 實務指引(TL;DR)

    • 優先導入場景:
    • RLHF/RLAIF 的對話評分、工具調用評分、長上下文 code RL。
    • prompt 長度是 response 的 3–10 倍。
    • 使用 Qwen3.5-4B 或相近大小模型,GPU 計算是主要瓶頸。

    • 預期收益:

    • 實測可達 3x–7.5x throughput 提升。
    • 允許你把 batch 撐大,減少 gradient accumulation,進一步提高 GPU 利用率。
    • 相同 GPU 成本下,能多跑數倍 rollout 或更長訓練步數。

    • 導入步驟建議:

    • 先在小 batch 上實作 forward_prompt + forward_response_with_prefix,只做 sanity check。
    • 確認與原 full sequence 訓練的 loss/梯度差異在可接受範圍(數值抖動為正常)。
    • 再導入 DDP/FSDP + AMP,逐步拉大 batch 測 throughput。
    • 監控 loss 曲線與最終 RL reward,確認沒有明顯退化。

    只要你的 RL 任務落在「長 prompt / 短 response」區間,RL 訓練版 prompt cache 幾乎就是一次性的大幅成本折扣;對正在做 RLHF/RLAIF 的團隊,值得花 1–2 週工程時間好好實作一版。


    🚀 你現在可以做的事

    • 在現有 RLHF/RLAIF 代碼中量測平均 T_prompt / T_resp,判斷是否達到導入門檻(≥3)
    • 在一個小型實驗中實作 forward_prompt 與 forward_response_with_prefix,對比 full sequence 訓練的 loss/梯度
    • 在實際 Qwen3.5-4B 或現用模型上開啟 prompt cache 實驗,記錄 throughput 與成本變化,評估是否全面導入
  • 低延遲語音 AI 架構實戰解析

    低延遲語音 AI 架構實戰解析

    📌 本文重點

    • 目標是在 200–400ms 內提供高品質雙向語音互動
    • 關鍵在通訊管線穩定與三模型 streaming 並行
    • 難點是 tail latency 與企業網路環境下的可用性
    • 先用 WebRTC/WS + 開源模型做 MVP 再優化

    要把GPT‑5 級推理塞進即時通話,最大痛點是:總延遲必須壓在 200–400ms 內,還要撐住大量並發。這篇用 OpenAI 近期的語音架構做藍本,從通訊層、模型層、系統層拆解,並給出一個用「常見雲 + WebRTC/WS + 開源語音模型」的最小可行方案(MVP),協助你評估:

    • 專案能不能做到「類 GPT-Realtime-2」的體驗
    • 目前架構要改哪些地方
    • 延遲預算怎麼抓、實測怎麼調

    重點說明:三層拆解你應該先想清楚什麼

    1. 通訊層:WebRTC / Streaming API 管線

    核心結論:語音幀越小、管線越穩定,LLM 才有操作空間。

    關鍵設計:

    • 雙通道設計:
    • WebRTC:負責雙向音訊(RTP),盡量 P2P,失敗時回退 TURN/Relay
    • WebSocket / gRPC streaming:負責把編碼後音訊送進推理後端,收 TTS 音訊回來
    • 幀大小與編碼:
    • 單向 20ms 幀是常見折衷(Opus 20ms/packet),RTT + 解碼後約 40–60ms
    • OpenAI 類似設計:低 bit‑rate Opus / 自家 codec + 小幀 + 伺服端聚合
    • 回退策略:
    • NAT/防火牆下 P2P 常失敗,要有 ICE + TURN,並在 handshake 階段就降級到「WebRTC → TURN relay → 伺服器」模式

    💡 關鍵: 前端小幀 + 後端穩定回退(ICE + TURN)是把延遲壓到 200–400ms 的第一道門檻

    2. 模型層:語音↔文字↔推理鏈路的裁剪與並行

    核心結論:不要把語音→文字→LLM→TTS 串成一條同步鏈,要做 streaming 並行。

    典型 full pipeline:

    1. 語音 ASR:Speech → Text
    2. 文字 LLM:Text → Response tokens
    3. TTS:Text → Speech

    在低延遲場景下可以這樣優化:

    • 早啟動推理:
    • ASR 用 streaming 模式,每 100–200ms 一個 partial transcript
    • 當句子結構「大致明朗」時(看到疑問詞或語尾),就把目前 transcript 丟給 LLM,不用等語音結束
    • 分段生成 + streaming tokens:
    • LLM 開啟 streaming,邊出 token 邊餵給 TTS
    • 控制 max_tokens / per-turn token budget,避免一次生成長篇大論造成尾端延遲
    • TTS 緩衝策略:
    • 不要等完整句子才播,通常 150–300ms 音訊緩衝就可以開始播放
    • 但也不能太小,避免「一卡一卡」;常見做法是根據 網路 jitter + 解碼時間 動態調整

    OpenAI 最新的 GPT-Realtime-2 / -Translate / -Whisper 其實就是把這條鏈路收斂成幾個特化模型,讓內部共享語音表徵與推理能力,減少中間編碼/解碼開銷。你在自建時不一定能做到單一多模態模型,但至少要做到三模型 streaming 並行。

    💡 關鍵: 把 ASR、LLM、TTS 並行 streaming,通常能把首次開口時間從秒級壓到 300ms 左右

    3. 系統與部署層:分布式推理與 tail latency 控制

    核心結論:平均延遲不難,難的是 99th percentile。

    設計要點:

    • 模型路由與排程:
    • 輕量語音模型(ASR/TTS)可以 多實例 + 每 GPU 多 worker,吃滿 GPU
    • 重 LLM 建議走 集中式推理叢集 + router,依 session 粘性綁定同一实例
    • GPU 利用率:
    • 啟用 batching + token 并行,但要對語音場景限縮 batch size,避免增加 tail latency
    • 長回應可切段生成:先生成 1–2 秒的語音對應文字,再補充後半段
    • tail latency 監控:
    • 重要指標:錄音開始 → 第一個回傳音訊 的 p50/p95/p99
    • 以 tracing 把 pipeline 切開:上傳 / ASR / LLM queue / LLM compute / TTS / 下行網路
    • 一旦 p99 失控,先檢查 排程佇列(排隊時間)而不是模型本身

    💡 關鍵: 真正破壞體驗的是 p99 延遲而不是平均值,監控時要把 queue time 單獨拉出來看


    實作範例:一個最小可行架構(MVP)

    架構概覽

    • 前端:Browser WebRTC(音訊 capture) + WebSocket(控制訊息)
    • 後端:
    • Signaling + API Gateway:Node / Go 都可
    • gRPC streaming 到推理服務
    • 推理服務:Python + 開源 Whisper streaming + 開源 LLM + VITS/TTS(示意)

    前端:WebRTC + WebSocket 管線

    // 簡化版:建立 WebRTC 傳音訊 + WS 傳控制
    const pc = new RTCPeerConnection({
      iceServers: [{ urls: 'stun:stun.l.google.com:19302' }]
    });
    
    const ws = new WebSocket("wss://your-api.example.com/signaling");
    
    async function startCall() {
      const stream = await navigator.mediaDevices.getUserMedia({ audio: true });
      stream.getAudioTracks().forEach(t => pc.addTrack(t, stream));
    
      pc.onicecandidate = e => {
        if (e.candidate) ws.send(JSON.stringify({ type: 'candidate', data: e.candidate }));
      };
    
      pc.ontrack = e => {
        // 收到 TTS 音訊(伺服端用 WebRTC 回推)
        const audio = document.getElementById('remote') as HTMLAudioElement;
        audio.srcObject = e.streams[0];
      };
    
      const offer = await pc.createOffer({ offerToReceiveAudio: true });
      await pc.setLocalDescription(offer);
    
      ws.onopen = () => {
        ws.send(JSON.stringify({ type: 'offer', data: offer }));
      };
    
      ws.onmessage = async (msg) => {
        const { type, data } = JSON.parse(msg.data);
        if (type === 'answer') {
          await pc.setRemoteDescription(data);
        }
      };
    }
    

    注意:

    • 幀大小主要在伺服端設定 Opus encoder(例:20ms),前端維持預設即可
    • 若 WebRTC 被防火牆擋住,signaling 伺服端要下指令讓前端切換為 WebSocket 直接上傳 PCM/Opus 模式

    後端:gRPC streaming 推理服務(示意)

    假設有一個 SpeechService 接收 Opus 幀,回傳已編碼好的音訊幀:

    // speech.proto
    service SpeechService {
      rpc Converse(stream AudioFrame) returns (stream AudioFrame) {}
    }
    
    message AudioFrame {
      bytes data = 1;      // Opus 或 raw PCM
      int64 timestamp = 2; // client capture ts
    }
    

    伺服端 Python(簡化,忽略實際音訊處理細節):

    class SpeechService(servicer_pb2_grpc.SpeechServiceServicer):
        async def Converse(self, request_iterator, context):
            # 1) 啟動 ASR/LLM/TTS 協程
            asr_queue = asyncio.Queue()
            llm_queue = asyncio.Queue()
            tts_queue = asyncio.Queue()
    
            async def asr_worker():
                async for frame in request_iterator:
                    # 解碼 Opus -> PCM -> ASR partial text
                    text_partial = asr_model.transcribe_stream(frame.data)
                    await llm_queue.put(text_partial)
    
            async def llm_worker():
                async for partial in llm_queue:
                    # 送入 LLM streaming,邊出 token 邊丟給 TTS
                    async for chunk in llm.stream(partial, max_tokens=64):
                        await tts_queue.put(chunk.text)
    
            async def tts_worker():
                async for txt in tts_queue:
                    # 生成短語音片段(200–300ms)
                    audio_bytes = tts_model.synthesize(txt)
                    yield speech_pb2.AudioFrame(
                        data=audio_bytes,
                        timestamp=int(time.time() * 1000)
                    )
    
            await asyncio.gather(asr_worker(), llm_worker(), tts_worker())
    

    實務上你會:

    • 用 更細緻的 queue 協調(包含會話 ID、句子邊界)
    • 控制 llm.stream 的 token 長度與 stop 條件,避免超長句
    • 在 TTS 部分先緩衝幾個 frame,再開始透過 WebRTC/WS 推到 client

    延遲 budget 規劃(示意)

    在穩定網路下可以先抓:

    • 上行錄音 + 傳輸:40–80ms(20ms 幀 + RTT)
    • ASR streaming:40–80ms(小模型 + GPU)
    • LLM 推理:80–150ms(取決於 token 數與模型大小)
    • TTS 生成 + 下行傳輸:60–120ms

    整體 p50 目標:220–350ms 第一個回應音訊開始播放。

    優化策略:

    • 第一輪回應:用 較短回應模板(像「嗯、好的,我來看一下…」)快速回覆,爭取後面長推理時間
    • 持續對 ASR/LLM/TTS 做 A/B test:看哪一段是主要瓶頸,優先調那裡的 model size / batch / GPU 排程

    建議與注意事項:幾個常見坑

    1. NAT / 防火牆導致 WebRTC P2P 失敗

    • 坑點:只測局域網或開放網路,實際部署到企業網路立刻掛掉
    • 建議:
    • 一開始就部署 TURN server,並在前端暴露 ICE 連線狀態,回報到後端
    • 若連線失敗,API 層切到 純 WebSocket 音訊通道,雖然成本高但能保證可用性

    2. 語音切段過粗,造成「打斷感」

    • 坑點:以 1 秒幀或整句才送 ASR,LLM 只在句尾發言,對話像對講機
    • 建議:
    • 200ms 以內 的 audio chunk;ASR 使用 partial result callback
    • 根據語氣停頓(VAD)+ 標點預測,判斷何時啟動 LLM 回應

    3. TTS 緩衝策略不當,導致「一卡一卡」

    • 坑點:
    • 緩衝太短 → 網路 jitter 就卡
    • 緩衝太長 → 首次開口延遲拉高
    • 建議:
    • 先以 250ms 緩衝 做 baseline,再按實測 jitter 動態調整在 200–400ms
    • 在 client 端維護一個小 buffer,利用 AudioContext / Web Audio API 自行排程播放,而不是一次性丟給 <audio>

    4. GPU 利用率 vs tail latency 的拉扯

    • 坑點:最大化 batch size 很爽,但 p99 延遲爆炸
    • 建議:
    • 把語音場景的推理服務與一般 chat/RAG 分開,語音路徑限制 max batch size
    • 使用 token-level scheduling(類似 OpenAI 做法),避免長上下文會話拖累短 query

    對專案的實際好處

    • 如果你現在線上只有「按鈕錄音 → 傳檔 → 回文字」,這套設計可以讓你在 1–2 週內做出可 demo 的雙向語音助理
    • 用 WebRTC/WS + 開源模型的 MVP,可以先驗證:
    • 使用者對 latency 敏感度
    • 需要多強的推理(是否真的要 GPT‑5 級推理)
    • 實際 GPU 成本與擴展上限
    • 後續要接上 OpenAI 類似 GPT-Realtime-2 的託管服務時,這套三層思路與接口方式幾乎可以直接沿用,只是把內部 ASR/LLM/TTS 換成單一多模態 API。

    🚀 你現在可以做的事

    • 在現有專案中畫出完整語音管線,標註各節點預估延遲(上傳、ASR、LLM、TTS、下行)
    • 用 WebRTC + WebSocket 加上任一開源 Whisper + TTS,實作一個能雙向講話的最小 demo
    • 部署基本的 tracing(例如在每一階段打 log),實測並記錄「錄音開始 → 第一個回應音訊」的 p50/p95/p99 數據
  • Claude 永續 Agent Warm-Cache 實戰

    Claude 永續 Agent Warm-Cache 實戰

    📌 本文重點

    • 全上下文重送會讓長期 Agent 在成本與延遲上崩盤
    • 用 Warm-Cache 三層快取可把成本壓到約 1/8
    • 短期 context + 向量庫分層記憶可兼顧長期記憶與成本
    • 嚴格工具邊界與審計是讓 Claude Agent 能上線的關鍵

    在 Discord 上跑一個長期管理 AWS 基礎設施與程式碼的 Claude Agent,如果每次請求都把 全上下文重送,你很快就會發現兩個殘酷事實:token 費用爆炸、延遲高到用不下去。實測數據來看,透過 Warm-Cache + 分層記憶架構,可以把成本壓到原本的 1/8 左右,P95 latency 也從 10+ 秒壓到 3 秒內,而且邏輯與安全性更可控。

    💡 關鍵: 透過結構化快取與記憶分層設計,可以同時把成本壓到約 1/8,並把 P95 延遲從 10 秒級降到 3 秒內,讓長期 Agent 實際可用。


    重點說明

    1. 為什麼「全上下文重送」會崩盤?

    典型實作:

    • 每個 Discord 訊息 → 直接呼叫 /v1/messages
    • 把 完整對話歷史 + 工具定義 + 系統提示 一起丟進去

    問題:

    1. token 費用幾乎線性成長:對話越長,每次重送的 tokens 越多,長期 Agent 變成「每句話都在重付歷史學費」。
    2. 延遲被序列化成本綁死:100K context 每次 encode / decode 都是固定開銷,沒做 cache 再快的模型也救不了。
    3. 易爆 context:聊久一點就逼近上限,被系統自動截斷,Agent 出現「金魚記憶」。

    結論:永續 Agent 若不做 Prompt Caching,本質上不具備經濟可行性。

    💡 關鍵: 對長期 Agent 而言,不做 Prompt Caching 意味著 token 成本和延遲會隨時間線性惡化,最終失去經濟可行性。


    2. Warm-Cache 三層設計:工具、系統提示、歷史

    核心想法:把「幾乎不變」的部分從請求中抽出來,讓 Claude 的 Prompt Caching 真正生效,同時在你自己的系統再加一層 cache。

    三層結構:

    1. 工具定義層(Tools Cache)
    2. 例如 AWS 管理、Git 操作、MemPalace 查詢等工具定義
    3. 用 穩定的 ID + 版本號 來標記(例如 aws_tools:v3)
    4. 實作:

      • 本地用 JSON 檔 或 TypeScript enum 管理
      • 對 Claude 端利用 prompt_cache_key(概念上,可用 system prompt 方式固定)
    5. 系統提示層(System Prompt Cache)

    6. 定義 Agent 的角色、邊界、倫理規則(例如只能操作 Private VPC 而非公網)
    7. 變動頻率低,但會跟版本、環境(staging/prod)綁定
    8. 推薦:用 template + 版本號,例如 discord_infra_agent:v5

    9. 歷史記錄層(Conversation Cache)

    10. 只快取「近期對話 + 工具呼叫結果」的短期記憶
    11. 長期記憶丟給向量庫(MemPalace / 自建 Milvus / PGvector),避免塞爆 context
    12. 每個 channel / user 維護一個 sliding window,例如最近 30 則訊息

    典型資料結構(TypeScript):

    type CacheKey = string; // e.g. "tools:aws:v3", "sys:discord_agent:v5"
    
    interface WarmCacheEntry {
      version: string;
      contentHash: string;
      serialized: string;   // 已處理過、可直接拼進 messages 的 JSON 字串
      updatedAt: number;
    }
    
    class WarmCache {
      private store = new Map<CacheKey, WarmCacheEntry>();
    
      get(key: CacheKey): WarmCacheEntry | undefined {
        return this.store.get(key);
      }
    
      set(key: CacheKey, entry: WarmCacheEntry) {
        this.store.set(key, entry);
      }
    }
    

    版本管理與失效策略:

    • 工具或系統提示改版 → 直接 變更 version(v3→v4),讓舊 cache 自然失效
    • 每次啟動時計算一遍 contentHash,若 hash 改變但 version 沒變,log 出警告避免「隱性分叉」

    3. 長期記憶:MemPalace + 短期上下文的分層設計

    要讓 Agent 在 Discord 長期「記得」你的 AWS 結構、服務慣例,又不把所有東西塞進 context,做法是:

    1. 短期記憶(Context Window)
    2. Warm-Cache 上的歷史層,只保留最近 N 回合(例如 30)
    3. 專門服務「連續對話」與「工具呼叫之前的局部上下文」

    4. 長期記憶(向量庫 / MemPalace)

    5. 把:
      • 專案 README
      • 關鍵 AWS 架構說明
      • 常見 Runbook / SOP
    6. 全部 embed 成向量,存進 MemPalace / 其他向量庫

    7. 查詢流程:

    8. 使用者問問題 →

    9. 先以「channel + user + 問題」做 embedding,去 MemPalace 找 Top-K 相關記憶片段
    10. 把這些片段壓縮後,丟進當次 system 或 user message 的前置 context

    簡單 Python 記憶層(SQLite + 向量庫 ID)示意:

    import sqlite3
    
    conn = sqlite3.connect("memory.db")
    cur = conn.cursor()
    
    cur.execute("""
    CREATE TABLE IF NOT EXISTS long_term_memory (
      id INTEGER PRIMARY KEY,
      user_id TEXT,
      channel_id TEXT,
      vector_id TEXT,   -- 真正的向量存在 MemPalace / pgvector
      summary TEXT,
      created_at INTEGER
    );
    """)
    
    # 檢索時:先從 MemPalace 拿相關 vector_id,再 join 回 summary
    

    好處:

    • context 永遠保持在一個可以預估的上限
    • 記憶可審計、可搜索,而不是全埋在 opaque 的 token 流裡

    實作範例

    1. Node.js:Claude Warm-Cache middleware

    以下是假想的 middleware,包裝 /v1/messages 呼叫,示意如何組合三層快取與向量記憶:

    import { claudeClient } from "./claude";
    import { WarmCache } from "./warmCache";
    import { fetchMemories } from "./memPalace";
    
    const cache = new WarmCache();
    
    export async function handleDiscordMessage(ctx: {
      channelId: string;
      userId: string;
      message: string;
      history: any[]; // 最近 N 則對話
    }) {
      const toolsKey = "tools:aws:v3";
      const sysKey = "sys:discord_infra_agent:v5";
    
      const tools = cache.get(toolsKey) ?? buildAndCacheTools(toolsKey);
      const systemPrompt = cache.get(sysKey) ?? buildAndCacheSystem(sysKey);
    
      const longTerm = await fetchMemories(ctx.userId, ctx.channelId, ctx.message);
    
      const messages = [
        { role: "system", content: systemPrompt.serialized },
        { role: "user", content: buildUserContent(ctx.message, longTerm) },
        ...ctx.history
      ];
    
      const res = await claudeClient.messages.create({
        model: "claude-3.7-sonnet",
        max_tokens: 1024,
        tools: JSON.parse(tools.serialized),
        messages
      });
    
      return res;
    }
    

    關鍵點:

    • tools、systemPrompt 都是快取後的 序列化結果,避免每請求重組
    • history 控制在固定長度,長期記憶透過 fetchMemories 注入

    2. Claude 系統 Prompt 模板(安全與邊界)

    你是一個在 Discord 裡專門協助管理 AWS 基礎設施與程式碼庫的 Agent。
    
    嚴格規則:
    - 只能透過提供的工具存取資源,禁止自行連線外部網路。
    - 所有操作必須限制在指定的 AWS Account 與 VPC,禁止新增具有公開網路權限的資源。
    - 若使用者要求執行具破壞性的操作(刪庫、清 bucket、關閉整個叢集),必須:
      1. 先以自然語言解釋風險與影響。
      2. 要求使用者提供明確確認字串(例如 "CONFIRM_DELETE_PROD")。
      3. 仍應優先建議更安全的替代方案。
    
    審計要求:
    - 對每一次工具呼叫,以簡潔 JSON 描述操作意圖與參數,方便後續寫入 audit log。
    

    3. Redis-based 歷史快取(短期記憶)

    import redis
    import json
    
    r = redis.Redis(host="localhost", port=6379, db=0)
    
    HISTORY_LIMIT = 30
    
    def push_history(channel_id: str, message: dict):
      key = f"history:{channel_id}"
      r.lpush(key, json.dumps(message))
      r.ltrim(key, 0, HISTORY_LIMIT - 1)
    
    def get_history(channel_id: str):
      key = f"history:{channel_id}"
      return [json.loads(x) for x in r.lrange(key, 0, -1)][::-1]
    

    建議與注意事項

    1. 監控:請求數、token、P95 latency 要一起看

    至少打三個 metrics:

    • token_usage_total:區分 prompt / completion / cache-hit
    • request_latency_ms:P50 / P95 / P99,分 model / route
    • tool_invocation_count:看 Agent 是否頻繁誤用工具

    優化策略:

    • 發現 P95 延遲高但 token 不高 → 多半是工具 / 外部 API 慢
    • 發現 token 緩慢上升 → 歷史快取 window 太大、向量記憶注入過多

    2. MCP / 工具設計:少而精 + 嚴格邊界

    • 像 PullMD 那樣,利用 MCP 把「HTML 轉 Markdown」這種重複工作下沉到工具層,避免讓 LLM 直接吃原始 HTML,token 省很大。
    • 工具要:
    • 明確輸入輸出 schema
    • 在私有網路中運行(Docker / Kubernetes namespace)
    • 只開最小必要權限(IAM 最小權限 + security group 限制)

    3. 避免「刪庫跑路」:幾個實務守則

    1. 只給「建議權」不給「直接執行權」 在 production
    2. 例如:Agent 只能產生 Terraform / CloudFormation patch,由人類 review + apply。
    3. 所有破壞性操作都經過雙重 gate:
    4. system prompt 要求二次確認
    5. backend 還要檢查「環境 + 操作類型」,prod 一律走人工流程
    6. 完整審計 log:
    7. 記錄:使用者指令、模型輸出、工具參數、執行結果
    8. 存在 append-only storage(CloudWatch Logs / Loki / S3 + Object Lock)

    4. 部署拓撲:限制在私有網路

    • Discord Bot → Gateway → Agent 後端(VPC 內)→ MCP 工具(同 VPC)
    • 往外只有到 Claude API + 向量庫(若是 SaaS) 的 egress
    • 不讓 Agent 直接 hit 公網,避免「自己 curl 一個 random script 來跑」這類事故

    總結:

    • 用 Warm-Cache 三層快取(工具、系統、歷史)+ 分層記憶(短期 context + MemPalace 長期記憶),可以在實戰中穩定做到 成本 ≈ 1/8、P95 latency < 3s。
    • 關鍵不是「多堆一點 GPU」,而是把「一次性 prompt」變成「可重用的結構」,再加上嚴格邊界與審計,讓你的 Claude 永續 Agent 真正能上 production。

    把上面的 middleware + Redis + SQLite/向量庫實作搬進你的客服 bot、infra bot 或內部 Copilot,大部分情況下只需要換掉工具與系統 prompt,就能直接開始省錢又提速。

    🚀 你現在可以做的事

    • 在現有 Discord / Slack Bot 中,先實作一層 Warm-Cache,把工具定義與系統提示抽出並版本化
    • 建一個最小可行的向量庫(MemPalace 或 pgvector),將 README、架構文件與 Runbook 全部 embed 進去
    • 為 production 環境補上系統 prompt 邊界、工具權限縮減與審計 log pipeline,驗證一條完整安全鏈路
  • GPT-5.5 實戰:從舊 API 到 Agent 模型

    GPT-5.5 實戰:從舊 API 到 Agent 模型

    📌 本文重點

    • GPT-5.5 對複雜多步任務與程式碼生成穩定度提升
    • 成本約為 GPT-5.4 的兩倍,需搭配模型路由控費
    • 建議先讓 GPT-5.5 接手最痛的 10% 高複雜任務

    GPT-5.5 主要解決兩個老問題:複雜多步任務很難穩定跑完、以及 程式碼生成在實務專案中需要大量人工修補。代價是 API 價格約 翻倍,但在多步推理、跨工具協作(agentic)場景,實測能少掉 30–60% 的「人肉 orchestrator」工作。這篇從工程落地角度整理:何時值得升級、怎麼改最少程式碼、怎麼安全灰度上線。


    重點說明

    1. 能力與效益:什麼場景值得多付兩倍單價?

    基於官方說明與社群測試,GPT-5.5 / 5.5 Pro 相較 GPT-5.4 / GPT-4.x 的實務差異,可粗略量化成幾類:

    💡 關鍵: 若你有大量跨系統、多步驟任務,GPT-5.5 能實際減少 30–60% 人工編排成本,值得用較高單價換穩定度與省人力。

    1. 程式碼生成 / 除錯
    2. 專案級 refactor(多檔案、跨模組)成功率提升,一次生成即可可編譯 / 可跑的比例顯著增加。
    3. 能自己分解成「閱讀現有程式碼 → 擬方案 → 修改多個檔案 → 自我檢查」的多步流程。
    4. 若你現在常遇到:

      • 4.x 產出的 patch 無法編譯
      • RAG 上接錯 API、型別對不起來

      → 使用 GPT-5.5 Pro 當「主程式碼助手」通常物有所值。

    5. 多步任務編排 / Agent 能力

    6. GPT-5.5 對 tool calling 的規劃更積極:
      • 能自動決定「先查 DB → 再呼叫支付 API → 最後寄信」,而不是你手動 orchestrate。
      • 對含糊任務會先發問澄清,而不是直接亂調工具。
    7. 適合:客服自動處理、報表生成、跨系統自動化(CRM + 票務 + ERP)。

    8. 上下文與多模態

    9. 更長的 context window(依官方實際規格為準),對 RAG / 長文件總結,能減少 chunking 與多輪 query。
    10. 圖片 + 文字 + 結構化資料混合輸入時的理解更穩。

    不建議升級的場景:
    – 純 FAQ、簡單分類、模板生成(信件、固定格式回答)。
    – 已經用 4.x 跑得很穩,且沒有多工具協作需求。

    此時可維持舊模型,或只對「高價值任務」做路由到 GPT-5.5。


    2. API 變更與最小遷移清單

    以官方 changelog 與社群實測為基礎,整理從 GPT-5.4 / GPT-4.x → GPT-5.5 的常見差異(命名依照 OpenAI 既有慣例,實際以文件為準):

    1. 模型名稱與 context
    2. 一般能力:gpt-5.5(假設 context 最高 ~200k tokens 級別)。
    3. 高階版:gpt-5.5-pro(更快、更穩、較高 rate limit)。
    4. 最小變更:
      “`diff

      • model: “gpt-4.1-mini”
      • model: “gpt-5.5”
        “`
    5. Tool calling / JSON mode 行為

    6. 工具呼叫邏輯更 agentic:模型會「自己決定」何時用工具,而不是你硬塞指令。
    7. response_format 行為加強:
      • {"type": "json_schema"} 更嚴格遵守 schema,但也可能為滿足 schema 而「合理捏造」欄位。
    8. 工具呼叫格式仍是 tools + tool_choice,但推薦寫法:
      jsonc
      {
      "model": "gpt-5.5",
      "tools": [
      {
      "type": "function",
      "function": {
      "name": "get_user_profile",
      "parameters": {
      "type": "object",
      "properties": {
      "user_id": {
      "type": "string"
      }
      },
      "required": ["user_id"]
      }
      }
      }
      ],
      "tool_choice": "auto" // 讓 5.5 自行規劃
      }

    9. 安全策略與輸出

    10. 官方系統卡說明:安全防護更嚴格,對灰色內容更傾向拒絕或弱化。
    11. 實務影響:有些之前「勉強會答」的 debug / 測試資料,可能會被誤判為敏感,需要:

      • 加強 system prompt:強調是企業內部開發、無真實個資。
      • 避免在 prompt 中填入真實 PII,改用匿名 ID。
    12. 延遲與費用

    13. token 單價約為 5.4 的兩倍級別(需看官方表)。
    14. GPT-5.5 本身更快,但若大量 tool calling,整體延遲可能 抖動更大(因為多輪 HTTP)。

    💡 關鍵: 單價約為 5.4 的兩倍,但若只在高價值、多步任務上使用,整體成本未必增加,反而可能因少錯誤與少人工介入而下降。

    最小遷移清單:
    – [ ] 替換 model 名稱為 gpt-5.5 或 gpt-5.5-pro。
    – [ ] 檢查 tool 定義:補齊 parameters schema,避免舊寬鬆 schema 造成誤呼叫。
    – [ ] 若依賴 JSON 格式輸出,統一改用 response_format: { type: "json_schema" } 並加上 嚴格驗證。
    – [ ] 更新成本計算與限額:調整配額、降級策略。


    3. 把 5.5 的 agent 能力整進現有架構

    一個實用思路:不要讓 GPT-5.5 直接當「超級大腦」管所有東西,而是:

    現有後端 + 工具層不動,只是把「任務分解與工具選擇」交給 5.5 來做。

    常見架構:

    Client → API Gateway → Orchestrator Service →
      ├─ LLM (GPT-4.x / 5.4)
      ├─ Tool Services (DB / CRM / Payment / RAG)
      └─ Logging & Guardrails
    

    升級方式:在 Orchestrator 裡新增一個路徑:

    Orchestrator
      ├─ Simple flows → 4.x
      └─ Complex multi-step flows → 5.5 (tool auto)
    

    實作範例

    1. 基本遷移:從 GPT-4.1 到 GPT-5.5 + JSON Schema

    // Node/TS 假想範例
    import OpenAI from "openai";
    
    const client = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });
    
    async function generateInvoice(data: any) {
      const completion = await client.responses.create({
        model: "gpt-5.5",
        input: [
          {
            role: "system",
            content: "你是一個嚴格輸出 JSON 的後端服務,不要輸出解釋文字。",
          },
          {
            role: "user",
            content: `根據以下訂單資料產生發票 JSON:${JSON.stringify(data)}`,
          },
        ],
        response_format: {
          type: "json_schema",
          json_schema: {
            name: "InvoiceSchema",
            schema: {
              type: "object",
              required: ["invoice_id", "items", "total"],
              properties: {
                invoice_id: { type: "string" },
                items: {
                  type: "array",
                  items: {
                    type: "object",
                    required: ["name", "price"],
                    properties: {
                      name: { type: "string" },
                      price: { type: "number" },
                    },
                  },
                },
                total: { type: "number" },
              },
            },
            strict: true,
          },
        },
      });
    
      const json = JSON.parse(completion.output[0].content[0].text);
      return json;
    }
    

    好處:
    – GPT-5.5 在複雜訂單(折扣、稅金)時,更少漏欄位與型別錯誤。
    – strict: true 讓 schema 驗證更嚴格,搭配後端再做一次 JSON schema 驗證,可大幅降低格式 bug。


    2. Agentic tool calling:自動任務分解 + 多工具串接

    以下示範:用 GPT-5.5 當任務規劃器 + 工具選擇器,工具維持既有 microservice。

    const tools = [
      {
        type: "function",
        function: {
          name: "search_tickets",
          description: "查詢使用者未處理工單",
          parameters: {
            type: "object",
            properties: { user_id: { type: "string" } },
            required: ["user_id"],
          },
        },
      },
      {
        type: "function",
        function: {
          name: "create_ticket_reply",
          description: "對特定工單回覆訊息",
          parameters: {
            type: "object",
            properties: {
              ticket_id: { type: "string" },
              message: { type: "string" },
            },
            required: ["ticket_id", "message"],
          },
        },
      },
    ];
    
    async function handleSupportRequest(userId: string, query: string) {
      const res = await client.responses.create({
        model: "gpt-5.5",
        tools,
        tool_choice: "auto", // 讓 5.5 自己決定呼叫順序
        input: [
          {
            role: "system",
            content:
              "你是客服 Agent,可以呼叫工具查詢工單並回覆。遇到資訊不足時先提問澄清。",
          },
          { role: "user", content: `user_id=${userId}, 問題:${query}` },
        ],
      });
    
      // 實務上這裡要迴圈處理多輪 tool calls,以下簡化偽碼
      for (const output of res.output) {
        for (const item of output.content) {
          if (item.type === "tool_call") {
            const { name, arguments: args } = item.tool_call;
            const toolResult = await dispatchTool(name, args); // call your microservice
            // 把工具結果再丟回 5.5 讓它整合
          }
        }
      }
    }
    

    實際好處:
    – 過去你可能要在 Orchestrator 裡手寫流程:先 search_tickets,再挑一筆,然後叫模型產生回覆,再 create_ticket_reply。
    – 現在可以讓 GPT-5.5 自己決定要查幾次、要不要先澄清,你只需負責工具實作 + 安全閘。


    3. 成本優化與模型路由示意

    簡單的分層推理策略(Pseudo-code):

    async function routeLLMTask(task: Task) {
      // 1. 便宜模型先做分類 / 難度預估
      const difficulty = await estimateDifficultyWithMini(task);
    
      if (difficulty === "simple") {
        return callLLM({ model: "gpt-4.1-mini", task });
      }
    
      if (difficulty === "medium") {
        return callLLM({ model: "gpt-5.4", task });
      }
    
      // 真的複雜 / 高價值才用 5.5 Pro
      return callLLM({ model: "gpt-5.5-pro", task });
    }
    

    適用場景:
    – SaaS 產品內的「AI 助理」,各種請求混雜。
    – 有明顯高價值操作(下單、修改合約)與低價值操作(查 FAQ)。


    建議與注意事項

    1. 常見坑

    1. 自動工具過度呼叫
    2. GPT-5.5 在 tool_choice: "auto" 下偏好積極使用工具,可能導致:
      • 單次對話打爆你的 microservice rate limit。
    3. 建議:

      • 在 Orchestrator 加 工具呼叫次數上限(例如每次對話最多 5 次)。
      • 若超過,回傳一個「工具不可用」的 faux tool result,要求模型改用已有資訊回答。
    4. 推理時間抖動

    5. 多輪 tool calling 會導致延遲暴增(LLM 快,但你的工具慢)。
    6. 建議:

      • 對每個工具加 timeout;
      • 若工具 timeout,回傳明確錯誤給 LLM(例如 "status": "timeout"),讓它用降級策略回應。
    7. 輸出格式不穩 / schema 假資料

    8. json_schema 雖強,但 GPT-5.5 會為滿足 schema 而補齊不存在的欄位。
    9. 必做:
      • 後端再驗證一次 JSON schema,不要信任模型;
      • 對關鍵欄位(如金額、user_id)加入「只允許從工具輸入,不允許模型自由發明」的規則(可在 prompt 說明、也可在 runtime 檢查來源)。

    2. 灰度上線與降級策略

    建議 rollout 策略:

    1. 先鎖定 1–2 個「高價值 + 複雜」flow:
    2. 例如:整合多系統產生週報、客服自動處理退款申請。
    3. 開 feature flag:
    4. 部分租戶 / 內部帳號先用 GPT-5.5,其他維持 4.x。
    5. 監控三件事:
    6. 成單 / 解決率提升(而不是只看 token 使用量)。
    7. 平均與 P95 latency。
    8. 工具錯誤率與人工介入次數。
    9. 預設降級路徑:
    10. 若工具錯誤或 LLM 回傳不符合 schema,
      • 自動重試一次 GPT-5.5;
      • 仍失敗則降級到 GPT-5.4 或交由人工處理(打 label,順便收集資料)。

    💡 關鍵: 用 feature flag + 降級路徑灰度上線,可以在不影響主流程穩定性的前提下,逐步放大 GPT-5.5 的覆蓋範圍。


    結論:什麼時候立刻上 GPT-5.5?

    優先升級條件:
    – 你有大量「跨系統、多步驟」任務,目前靠工程師硬寫 orchestration 邏輯維持。
    – 你在做程式碼助手、IDE 插件、CI 上的自動修 bug / 重構,現有模型常產生半成品。

    不必急著升級:
    – 任務單步、邏輯簡單,或 4.x 已經穩定跑很久;
    – 成本壓力大,且沒有足夠監控來衡量 GPT-5.5 帶來的實際收益。

    合理的做法是:先用 GPT-5.5 接手最痛的 10% 任務,在舊架構外側加一層 agentic 能力,再決定是否全面遷移。

    🚀 你現在可以做的事

    • 先盤點系統中最複雜、跨多服務的 10% flow,評估是否改由 gpt-5.5 處理
    • 把現有 tools schema 補齊與收斂,為 tool_choice: "auto" 與 json_schema 做好準備
    • 實作一個簡單的模型路由器,先在測試環境導入 gpt-5.5-pro 並觀察錯誤率與延遲指標
  • Claude Opus 4.7 實作可控 Agent 平台

    Claude Opus 4.7 實作可控 Agent 平台

    📌 本文重點

    • Opus 4.7 更適合長上下文、多步任務與自我校對
    • 可作為常駐 coding / ops Supervisor Agent
    • 透過工具層與治理設計,接手真實 CI / PR pipeline

    Opus 4.7 解決的痛點很直接:以前你不太敢把多步任務完全丟給 LLM 自動跑——上下文記不住、自我檢查不足、程式碼改著改著就壞掉、Agent 亂調工具、成本爆炸。Opus 4.7 把這幾個點同時強化:長上下文 + 自我校對 + agentic coding,讓它不再只是「聊天模型」,而是可以放進持續運行 pipeline 的一個穩定元件。


    重點說明:Opus 4.7 對 Agent 能力的實質升級

    1. 長上下文 + 自我校對 = 多步任務可「放手」

    Opus 4.7 官方強調:

    • 更長上下文:可以在一次對話裡管理整個任務歷史、spec、log、既有程式碼與錯誤紀錄,減少你自己在應用層做 chunk & stitching。
    • 自我核查輸出:模型在回傳前會傾向先「檢查」自己的推論、程式碼或計算結果,等於內建一層 lightweight critic。

    💡 關鍵: 更長上下文搭配自我核查,實際上讓多步任務可以交給單一模型從頭管到尾,而不是切給一堆臨時腳本與輔助模型。

    對多步 pipeline(例如:分析 log → 找 root cause → 編輯程式碼 → 產 PR)最大的好處是:

    • 可以讓 Supervisor Agent 一次看到完整任務 timeline,而不是一段一段 patch;
    • 減少你在系統外再包一層「審題 / 校對」模型的需求(但關鍵步驟仍建議顯式加 guardrail)。

    2. Agentic Coding:如何讓 Opus 4.7 當常駐 coding / ops agent

    Opus 4.7 在 程式碼規劃 + 工具調用 上的品質明顯提高,實務上你可以讓它做:

    • 長期追蹤一組 repo 的變更,持續提出 refactor / bugfix 建議;
    • 自動跑 CI log 分析 → 開 issue → 出 patch → 發 PR;
    • ops 向:監看監控告警 → 初步診斷 → 呼叫 runbook 工具。

    💡 關鍵: 把「讀檔、改檔、跑測試、開 PR」封裝成工具給 Opus 4.7 用,它就能長期常駐在 repo / pipeline 中當實際執行者,而不是只當輔助聊天夥伴。

    核心設計重點:

    1. 工具調用(Tool use)
    2. 用 /v1/messages + tools 讓 Opus 負責選擇何時 call tool、填參數;
    3. 將「讀檔、改檔、跑 test、開 PR」抽成安全封裝的工具,不讓模型直接操作 Git。

    4. 程式碼修改與回滾

    5. 永遠透過 「diff-based API」 修改程式碼,而不是讓模型輸出整檔;
    6. 由工具層實作版本管理(Git branch/commit),模型只負責描述修改 intent。

    7. 安全護欄設計

    8. 用 角色+權限 限制工具:code_writer 不能直接 deploy,ops_agent 只能操作 sandbox;
    9. 重要操作強制走「人審 + MCP gateway」流程。

    3. 在多代理系統中的定位:Supervisor / Orchestrator 角色

    以「Supervisor Agent」架構來看(類似 Towards AI 提到的 blueprint):

    • Opus 4.7 很適合擔任 Supervisor / Orchestrator:
    • 拆解使用者目標 → 任務樹;
    • 安排子代理:搜尋 Agent、執行 Agent、評估/QA Agent;
    • 維護整個任務的 context & memory。

    • MCP 或你自建的工具層則提供:

    • 可治理的工具執行(權限、審計 log、版本管理);
    • 與企業內部 API / 資料庫 / CI/CD / issue tracker 的橋接。

    重點:Supervisor 不直接做所有事,而是負責問對問題、調對工具、把任務切給對的子代理。Opus 4.7 的長上下文 +較穩定規劃能力,剛好補上這個角色。


    實作範例:自動 triage issue → 修 bug → 開 PR

    以下示意使用 Claude API + MCP / 自建工具,流程:

    1. 新 issue 建立 → Webhook 觸發 Supervisor;
    2. Supervisor(Opus 4.7)分析 issue,決定是否可自動處理;
    3. 呼叫 repo 工具 找到相關檔案、測試;
    4. 呼叫 code-agent(也是 LLM 或工具) 產生 patch;
    5. 呼叫 git/pr 工具 開 PR,必要時標記需要人工 review。

    1. Supervisor 的 messages 結構

    POST /v1/messages
    {
      "model": "claude-3-opus-4.7",  // **關鍵:Supervisor 指定 Opus 4.7**
      "system": "你是Supervisor Agent,負責協調工具與子代理.\n" +
        "規則:\n" +
        "1. 僅在需要時使用工具,不要自己幻想結果。\n" +
        "2. 高風險操作(寫檔、開PR)前,一定先提出計畫並自我檢查。\n" +
        "3. 若不確定,改為標記需要人工審核.",
      "tools": [
        { "name": "repo_search", "input_schema": {"type": "object", ...} },
        { "name": "repo_read_file", "input_schema": {"type": "object", ...} },
        { "name": "repo_edit_file_diff", "input_schema": {"type": "object", ...} },
        { "name": "run_tests", "input_schema": {"type": "object", ...} },
        { "name": "create_pr", "input_schema": {"type": "object", ...} }
      ],
      "messages": [
        {
          "role": "user",
          "content": [
            {
              "type": "text",
              "text": "有新的 GitHub issue:#1234 API 回傳 500,log ID=abcd...\n" +
                       "請決定是否可以自動修復,若可以,直接走:分析→修改→測試→開PR。"
            }
          ]
        }
      ]
    }
    

    2. 工具呼叫:diff-based 程式碼修改

    在第一次 tool call 前,先讓 Supervisor 規劃:

    1. 讀取相關檔案與測試檔。
    2. 產生修改計畫與預期影響範圍。
    3. 套用最小 diff 修改程式碼。
    4. 執行相關測試。
    5. 測試通過後建立 PR,標記『自動修復,請人工 review』。
    

    接著模型會回覆一個 tool 呼叫(格式依 SDK):

    {
      "role": "assistant",
      "content": [
        {
          "type": "tool_use",
          "id": "toolu_1",
          "name": "repo_search",
          "input": {
            "query": "500 error log id abcd", 
            "max_results": 10
          }
        }
      ]
    }
    

    你的應用層接到後:

    1. 執行 repo_search(你實作的服務,可能包 Git grep / code search);
    2. 把結果以 tool_result 回餵:
    {
      "role": "tool",
      "tool_use_id": "toolu_1",
      "content": [{
        "type": "text",
        "text": "找到可能相關檔案: src/api/user.ts, src/service/userService.ts ..."
      }]
    }
    

    之後 Supervisor 可能呼叫:

    {
      "type": "tool_use",
      "name": "repo_edit_file_diff",
      "input": {
        "path": "src/service/userService.ts",
        "diff": "@@ -42,6 +42,10 @@\n- const result = await dao.getUser(id);\n+ const result = await dao.getUser(id).catch(e => {\n+   logger.error('getUser failed', { id, err: e });\n+   throw new HttpError(500, 'USER_LOOKUP_FAILED');\n+ });"
      }
    }
    

    這裡的關鍵是:

    • repo_edit_file_diff 工具會:
    • 把原始檔讀出來;
    • 套用 diff(可用 git apply 或自寫 patch 邏輯);
    • commit 到專用 branch;
    • 回傳新的 snippet / commit id;
    • 回滾 就交給 Git:若後續測試失敗,由 Supervisor 呼叫 git_reset_to_commit(另一個工具),或人工介面一鍵 rollback。

    3. MCP / 自建工具層設計示意(pseudo code)

    // MCP Gateway / 工具伺服器(Node 範例)
    
    import express from 'express';
    import { searchRepo, editFileWithDiff, runTests, createPR } from './infra';
    
    const app = express();
    app.use(express.json());
    
    function requireRole(role: string) {
      return (req, res, next) => {
        const callerRole = req.headers['x-agent-role'];
        if (callerRole !== role) return res.status(403).send('forbidden');
        next();
      };
    }
    
    app.post('/tools/repo_edit_file_diff', requireRole('supervisor'), async (req, res) => {
      const { path, diff } = req.body;
      // 審計 log
      console.log('[AUDIT] edit_file_diff', { path, by: 'supervisor' });
      const result = await editFileWithDiff(path, diff);
      res.json(result);
    });
    
    // 其他工具類似實作...
    
    app.listen(3001);
    

    Supervisor Agent 呼叫工具時,由你的 orchestrator 轉成 HTTP request,並加上 x-agent-role: supervisor 等標頭,實作權限與審計。


    建議與注意事項:成本、延遲與治理

    1. 成本與延遲控制

    • 長上下文 ≠ 無腦塞所有東西:
    • 為 Supervisor 設計 分層上下文:
      • 任務規格(長期保持);
      • 當前子任務狀態(中期);
      • 最近工具回傳、log(短期)。
    • 透過你的應用層做 context summarization / state store,不要每次把整個任務歷史丟進去。

    • 對長跑任務(例如多輪 patch + test):

    • 使用 max_tokens / tool_temperature 控制回覆長度與探索度;
    • 將「細節 log」放在外部存儲,必要時讓模型用工具查詢,而不是全部當成 prompt。

    2. 避免過度「自作主張」

    Opus 4.7 的推理與自我校對更強,副作用是它會更願意自己決定事情。控制方式:

    • 在 system 提前定義 contract:
    • 任務邊界:可修改哪些 repo / namespace;
    • 停機條件:
      • 測試連續失敗 N 次 → 停止修改,標記需要人工;
      • 連續無法重現 bug → 輸出完整調查報告,停止嘗試;
    • 審核步驟:
      • 高風險變更必須產生「變更說明 + 風險清單」給人看。

    範例 system 片段:

    你是後端修 bug 的 Supervisor Agent,遵守以下 contract:
    - 只能操作 repo `my-service`,不可動 infra repo。
    - 若連續 2 次修改導致測試新增失敗,立即停止,輸出調查報告,等待人工處理。
    - 不得直接部署,只能開 PR,並在 PR 描述中列出:問題原因、修改內容、風險與 rollback 方式。
    

    3. 對話風格變更對既有工作流的影響

    Opus 4.7 更傾向「先思考再回答」,你會看到:

    • 回覆結構更完整,但文字量可能變多,對老工作流可能:
    • 冗長說明擾亂你原本靠 pattern matching 的 parser;
    • 原本 prompt 期待的 JSON 結構會被「多講兩句」破壞。

    💡 關鍵: 若你的系統嚴重依賴固定輸出格式,必須明確要求「只輸出 JSON」,並用 schema 驗證與重試機制包住模型。

    建議:

    • 儘量讓模型輸出 單一結構化 payload,多餘說明放在欄位內:
    {
      "plan": "文字說明…",
      "actions": [ ... ],
      "needs_human_review": true
    }
    
    • 在 system 明確要求:僅輸出 JSON,不要額外文字,並在應用層加 schema 驗證,若解析失敗就回饋「格式錯誤,請重新輸出」。

    結論:Opus 4.7 的能力重點不是「更會聊天」,而是更適合當有邊界、有工具、有治理的 Agent 核心。只要把它放在 Supervisor 位置,搭配 MCP 或自建工具層,控制好 contract / 成本 / 權限,你就可以開始放心讓它接手一部分真實的 coding / ops pipeline,而不只是做輔助建議。

    🚀 你現在可以做的事

    • 在現有 LLM 應用中,先挑一條「分析 log → 修 bug → 開 PR」的小流程改由 Opus 4.7 當 Supervisor 嘗試落地
    • 設計一組 repo_search、repo_edit_file_diff、run_tests、create_pr 工具,並封裝成 MCP 或內部 HTTP 服務
    • 為 Opus 4.7 Supervisor 撰寫明確的 system contract(任務邊界、停機條件、審核規則),再逐步擴大可自動處理的任務範圍