分類: AI 技術

  • 企業級 RAG 實戰架構與評估全攻略

    企業級 RAG 實戰架構與評估全攻略

    📌 本文重點

    • 企業級 RAG 關鍵在檢索架構與權限設計
    • 混合檢索 + 重排序是實務標配
    • metadata/ACL 必須在檢索前就生效
    • 評估指標需同時涵蓋正確性與延遲

    在企業環境裡,RAG 的真正價值是:讓 LLM 能安全地用內部最新知識、可控地減少幻覺、並可以用工程方法迭代優化。本文從工程視角拆解:向量資料庫選型、檢索策略、chunking 與 metadata 設計、多租戶與權限控制、線上/離線評估指標,以及常見坑與解法。


    重點說明

    1. 檢索策略:不只向量,BM25 + 重排序才是實務標配

    企業知識庫多來源、多格式,單純向量檢索很容易 miss 關鍵字或出現語義偏移。建議:

    1. 混合檢索(Hybrid Search):
    2. 先用 BM25 或全文搜尋(Postgres tsvector、Elasticsearch、OpenSearch)做初篩
    3. 再用向量相似度做語義排序,避免只靠 embedding 導致關鍵領域術語被忽略

    4. 重排序(Rerank):

    5. 對 top-50 結果使用 cross-encoder / reranker 模型重新排序(如 bge-reranker-large)
    6. 好處:在不放大向量庫負載的情況下,顯著提升 answer correctness

    💡 關鍵: 透過「先 BM25 初篩、再向量排序、最後 rerank」三段式檢索,可以在不犧牲效能的前提下,大幅提升答案正確率與穩定性。

    實務上你可以:

    • 用 pgvector 存 embedding + Postgres 原生全文檢索
    • 或用 Milvus 做向量檢索,搭配獨立搜尋服務(Elastic / OpenSearch)做 BM25

    2. Chunking 與 Metadata:為檢索設計,而不是為分段而分段

    錯誤的 chunking 會直接降低 context recall,企業常見問題是「段太小、沒有結構」。建議:

    1. 混合策略 chunking:
    2. 先以語義斷點(標題、小節)切大塊,再用字數/token 限制微調
    3. 典型配置:512–1024 tokens + 128–256 tokens overlap

    4. metadata schema 是檢索與權限的核心:至少包含:

    5. tenant_id: 多租戶隔離
    6. doc_id, section_id: 追蹤來源與回溯
    7. source_system: Slack / GDrive / Confluence / Jira…
    8. visibility_tags / acl: 權限控制(角色、群組、文件 owner)

    好處:

    • chunk 不只是「段落」,而是帶有權限、業務上下文的最小檢索單位
    • 評估失敗案例時可以精準定位是哪個 chunk / doc 出問題

    3. 多租戶與權限:在「檢索前」把不該看的東西砍掉

    語義向量檢索會天然繞過傳統 RBAC 的關鍵字邊界,必須在檢索前加上身分綁定的 filter:

    • Identity-bound pre-retrieval filters:查詢向量庫前,先用 tenant_id + ACL 建立 filter
    • 所有檢索 API 都要支援條件:WHERE tenant_id = $tenant AND acl @> $user_roles

    關鍵結論:

    • 權限控制不能只做在「生成後」,因為一旦檢索到不該看的 context,就算你遮罩輸出,也已經有資料洩漏風險
    • 向量庫層一定要有 硬隔離策略(租戶切庫 / 切 collection / 至少切 partition)

    💡 關鍵: 權限過濾一定要在向量檢索「之前」就生效,否則只在輸出層遮罩,其實已經完成資料洩漏。


    4. 評估與監控:不只看 BLEU,更要看「能不能在生產上 debug」

    建議至少三類指標:

    1. Answer Correctness(生成品質)
    2. LLM 自評(使用 judge model)、人工標註小樣本、或用 rule-based 準確率

    3. Context Recall(檢索品質)

    4. 離線 benchmark:準備一組「問題 + 標準文件」pair,測量 top-k 是否包含正確文件

    5. Latency(服務體驗)

    6. 分段量測:檢索延遲、重排序延遲、LLM 生成延遲

    好處:

    • 可以清楚區分是檢索出錯、權限漏控,還是 LLM 幻覺
    • 為迭代(換 embedding 模型、調 chunking、調 rerank)提供可量化目標

    實作範例:Node + pgvector + 開源 embedding 的簡單企業知識庫

    下面用一個簡化的架構示範:

    • 向量庫:Postgres + pgvector
    • Embedding 模型:BAAI/bge-base-en(你可以換成中文模型)
    • 後端:Node.js (TypeScript)

    1. 資料庫 schema 設計

    -- pgvector 安裝後,建立知識庫表
    CREATE TABLE kb_chunks (
      id             BIGSERIAL PRIMARY KEY,
      tenant_id      TEXT NOT NULL,
      doc_id         TEXT NOT NULL,
      section_id     TEXT,
      source_system  TEXT NOT NULL,
      content        TEXT NOT NULL,
      embedding      VECTOR(768) NOT NULL,
      visibility_tags TEXT[], -- e.g. ['legal', 'finance']
      acl_roles      TEXT[],  -- e.g. ['legal_team', 'admin']
      created_at     TIMESTAMP DEFAULT NOW()
    );
    
    CREATE INDEX idx_kb_chunks_tenant ON kb_chunks (tenant_id);
    CREATE INDEX idx_kb_chunks_acl_roles ON kb_chunks USING GIN (acl_roles);
    CREATE INDEX idx_kb_chunks_visibility_tags ON kb_chunks USING GIN (visibility_tags);
    CREATE INDEX idx_kb_chunks_embedding ON kb_chunks USING ivfflat (embedding vector_l2_ops) WITH (lists = 100);
    

    重點:

    • 用 ivfflat + lists 調優檢索速度
    • 用 acl_roles / visibility_tags 作為 pre-retrieval filter 的基礎

    2. Chunking 與寫入管線(Python 伺服端工具)

    from transformers import AutoTokenizer, AutoModel
    import psycopg2
    
    MODEL_NAME = "BAAI/bge-base-en"
    MAX_TOKENS = 800
    OVERLAP = 200
    
    # 省略連線與模型初始化
    
    def chunk_document(text: str, tenant_id: str, doc_id: str, source_system: str, acl_roles: list[str]):
        tokens = tokenizer.encode(text)
        chunks = []
        start = 0
        while start < len(tokens):
            end = min(start + MAX_TOKENS, len(tokens))
            chunk_tokens = tokens[start:end]
            chunk_text = tokenizer.decode(chunk_tokens)
            chunks.append(chunk_text)
            start += MAX_TOKENS - OVERLAP
        return chunks
    
    def embed(texts: list[str]):
        # 簡化:batch embedding
        inputs = tokenizer(texts, padding=True, truncation=True, return_tensors="pt")
        with torch.no_grad():
            outputs = model(**inputs)
        embeddings = outputs.last_hidden_state[:, 0, :].cpu().numpy()  # CLS pooling 或改用 mean pooling
        return embeddings
    
    # 寫入 pgvector 的 SQL 略
    

    好處:

    • chunking 有明確 token 參數,可調試
    • metadata 與 ACL 一起寫入,避免「後面再補權限」的錯誤做法

    3. Node RAG Pipeline:帶權限的混合檢索 + 重排序

    import { Pool } from 'pg';
    import { getEmbedding } from './embeddingClient'; // 呼叫 Python 或直接用 Node 模型
    
    const pool = new Pool({ /* db config */ });
    
    interface UserContext {
      tenantId: string;
      roles: string[];
    }
    
    export async function ragAnswer(query: string, user: UserContext) {
      const embedding = await getEmbedding(query); // 返回 Float32Array 長度 768
    
      // 1. 帶 ACL 的向量檢索
      const vectorSql = `
        SELECT id, content, doc_id, section_id, source_system,
               1 - (embedding <=> $1::vector) AS score
        FROM kb_chunks
        WHERE tenant_id = $2
          AND acl_roles && $3::text[]
        ORDER BY embedding <-> $1::vector
        LIMIT 50;
      `;
    
      const vectorRes = await pool.query(vectorSql, [embedding, user.tenantId, user.roles]);
    
      // 2. 這裡可以加 BM25 / tsquery 做 keyword 初篩(略)
    
      // 3. 用 reranker 模型重排序(虛擬碼)
      const reranked = await rerank(query, vectorRes.rows.map(r => r.content));
      const topContexts = reranked.slice(0, 5).map(r => r.content);
    
      // 4. 組 prompt 給 LLM
      const prompt = `你是公司內部助理,回答必須只根據提供的內容。
    
    [檢索到的內容]
    ${topContexts.join('\n---\n')}
    
    [問題]
    ${query}
    
    請根據上面內容回答,若資料不足請明確說「目前知識庫沒有相關資訊」。`;
    
      const answer = await callLLM(prompt); // 例如 OpenAI / 本地 LLM
    
      return {
        answer,
        contexts: topContexts,
        debug: {
          retrievedCount: vectorRes.rowCount,
          tenantId: user.tenantId,
        },
      };
    }
    

    關鍵 API / 參數:

    • embedding <-> $1::vector:pgvector 距離運算,建議用 L2 或 cosine
    • acl_roles && $3::text[]:在檢索階段就做 ACL filter(pre-retrieval)
    • LLM prompt 強制「無資料要明說」,降低幻覺

    4. 簡單線上評估與監控

    可以加一個中介層紀錄:

    await logMetrics({
      tenantId: user.tenantId,
      query,
      latencyMs,
      retrievedCount: vectorRes.rowCount,
      model: 'bge-base-en',
      llmModel: 'gpt-4o',
    });
    

    後續離線跑:

    • 對一批標註問題跑 RAG,請 judge LLM 給出 correctness score(1–5)
    • 比較不同 embedding / chunking / rerank 策略的分數與延遲,做 A/B 測試

    建議與注意事項

    1. 幻覺依然存在:RAG 不是「關幻覺開關」

    • 即使檢索正確,LLM 仍可能「補細節」或「推理過頭」
    • 解法:
    • 明確在 prompt 中說:「不在 context 裡的資訊一律不要編造」
    • 在輸出層加 rule-based 檢查(如法律/財務答案一定要附來源文件 ID)

    2. 檢索結果不穩定:embedding / chunking / rerank 三者缺一不可

    • 換 embedding 模型時,一定要重新做離線 benchmark,不要只看 demo 感覺
    • chunking 調整需同時看:
    • context recall(有沒有檢索到正確文件)
    • latency(chunk 變多會拖慢向量檢索)

    💡 關鍵: 每次調整 embedding 或 chunking,都要用標註資料評估「召回率 + 延遲」,而不是只憑主觀 demo 感受做決策。


    3. 權限洩漏:不要相信「應用層自己會控」

    • RBAC 必須深入到向量庫 query 層
    • 尤其是把 Slack / GDrive / Jira 接在一起作企業 search 時:
    • 預設策略是「拒絕」,只對明確授權內容建索引
    • 建議對敏感資料(legal / M&A 文件)做 單獨 collection / database 隔離

    4. 迭代路線:從 PoC 到生產級

    1. PoC 階段:單租戶、簡單 chunking、只向量檢索
    2. Beta 階段:加 metadata schema、權限 filter、簡單監控
    3. 生產階段:
    4. Hybrid search(BM25 + 向量)
    5. reranker 模型
    6. 線上指標 + 離線 benchmark(answer correctness / context recall / latency)
    7. 權限審計與合規(log 每次檢索的 doc_id 與 user_id)

    核心結論:企業級 RAG 的關鍵不在「模型選哪個」,而在於:檢索架構、權限設計與可評估性是否工程化落地。只要這三件事做穩,模型與向量庫都可以迭代替換,而整個系統仍保持穩定、可回溯、可持續優化。


    🚀 你現在可以做的事

    • 在現有 Postgres 專案中安裝 pgvector,建立含 tenant_id 與 acl_roles 的 kb_chunks 表
    • 準備一批「問題 + 標準答案文件」pair,離線跑一次 context recall/latency benchmark
    • 將現有 RAG 應用的權限邏輯下沉到向量庫查詢層,加入 tenant_id + ACL 的 pre-retrieval filter
  • Eval-Driven Agents:讓 LLM 代理變成可維運系統

    Eval-Driven Agents:讓 LLM 代理變成可維運系統

    📌 本文重點

    • Evals 是 Agent 的 CI/CD Gate,讓更新可量化
    • 四個支柱讓 Agent 從黑盒變成可維運系統
    • 一週內可導入最小可行的 eval pipeline

    LLM Agent 最大的痛點很直接:一改 prompt 或策略,整體效果「好像」有變好,但沒人說得出到底好多少、哪裡變差、能不能安全上線。 Eval-Driven Agents 的核心,就是把 evals 當成 Agent 的 CI/CD,讓代理不再是黑盒魔法,而是可以版本管理、回溯、觀察與持續優化的軟體系統。


    重點說明

    1. Evals 是 Agent 的 CI/CD Gate

    把 Agent 當成服務在維護,就不能只靠「體感」判斷更新好壞。核心做法是:

    • 為每種任務設計 評估集(eval set) 與 指標(metrics)
    • 每次改 prompt / 工具 / 策略 / 模型,都先在 replay data 上跑一輪 eval
    • 只有達到門檻(類似單元測試全部通過)才允許 rollout

    💡 關鍵: 先建立穩定比較機制,再來追求模型效果,才能在特定場景量化「+8% 成功率」這種改善。

    關鍵不是追求完美模型,而是建立 穩定的比較機制,讓你敢說:這次改動在「退款流程」場景成功率 +8%,且「錯誤升級」場景沒退步。


    2. 四個支柱:讓 Agent 變得可維運

    圍繞「evals 是 CI/CD」這個核心,一個 production-grade Agent 至少要有四個支柱:

    1. 評估集與指標設計:為非確定性輸出定義 pass/fail 與容錯區間
    2. 任務切成多個「用例」:例如 FAQ 回答、工單分類、報表生成
    3. 每個用例定義:成功條件、允許誤差、關鍵失敗模式
    4. 指標不只看單一 aggregate score,而要區分場景與錯誤類型

    5. 資料與結構化回饋:用 Pydantic / schema 把工具調用、記憶與決策結構化

    6. 每次 Agent 執行都產出可重放的 trace:輸入、工具調用、模型回應、最終決策
    7. 讓 eval runner 可以重播舊版本 vs 新版本,逐步比較

    8. 觀察性與追蹤:整合 OpenTelemetry + Grafana Tempo 做分散式追蹤

    9. 將整條 agent workflow 變成 trace:每一次 tool call、一段 prompt 改動都看得到
    10. 問問題變得具體:「為什麼這次多調用了三個工具?」、「哪一段 prompt 改壞了成功率?」

    11. Eval-loop 與版本管理:把 eval 接進部署管線

    12. 類似單元測試 gate:每次改 prompt / 策略 / 模型先跑 eval
    13. 區分 實驗 eval(探索新策略)與 回歸 eval(保護既有能力)

    3. 這對你的專案有什麼實際好處?

    如果你現在有一個已上線但很脆弱的 Agent:

    • 降低改壞風險:每次調 prompt 不再是賭運氣,而是有數據護欄
    • 讓 debug 有方向:看到哪個場景、哪個工具路徑在退化,而不是「整體感覺怪怪的」
    • 更好控成本:配合 trace,知道哪裡 token 浪費最多(工具過度調用、不必要長上下文)
    • 團隊協作更順:PM/ML/Backend 看同一套 eval 報表,不再靠各自的 demo 體驗說服對方

    實作範例

    以下用簡化版 Python + Pydantic + OpenTelemetry,示範如何把一個脆弱 Agent,在一週內升級到有 eval pipeline 的狀態。


    1. 用 Pydantic 結構化 Agent Trace

    先定義 agent 的執行結果與步驟:

    from pydantic import BaseModel, Field
    from typing import List, Literal, Optional
    
    class ToolCall(BaseModel):
        name: str
        input: dict
        output: dict
        latency_ms: float
    
    class AgentStep(BaseModel):
        step_type: Literal["plan", "tool", "reflect", "final"]
        prompt: str
        response: str
        tool_call: Optional[ToolCall] = None
    
    class AgentTrace(BaseModel):
        trace_id: str
        user_input: str
        steps: List[AgentStep]
        final_output: str
        metadata: dict = Field(default_factory=dict)
    

    好處:

    • 每次執行都可重播:你可以拿 user_input + steps,在新版本模型上重跑,生成新的 trace
    • 易於比較舊版 vs 新版:對齊同一個 trace_id,逐步看哪個 step 不同

    在現有 Agent 中,只需要在 orchestrator 把每次決策與工具調用寫進這個 AgentTrace,就有一份可用作 eval 的資料。


    2. 定義 Eval 任務與指標:非確定性也能 Pass/Fail

    假設你有一個客服 Agent,需要回答退款相關問題,我們定義一個簡單的 eval:

    from pydantic import BaseModel
    
    class RefundEvalCase(BaseModel):
        case_id: str
        user_input: str
        expected_keywords: list[str]  # 例如 ["退款條件", "處理時間"]
        must_not_include: list[str]   # 例如 ["保證立即退款"]
    
    class EvalResult(BaseModel):
        case_id: str
        passed: bool
        score: float
        missing_keywords: list[str]
        bad_phrases: list[str]
    

    簡單的 eval runner:

    def run_refund_eval(agent_fn, cases: list[RefundEvalCase]) -> list[EvalResult]:
        results = []
        for case in cases:
            output = agent_fn(case.user_input)
            lower_output = output.lower()
    
            missing = [k for k in case.expected_keywords if k.lower() not in lower_output]
            bad = [p for p in case.must_not_include if p.lower() in lower_output]
    
            # 線性打分:關鍵字命中率 - 違規懲罰
            keyword_score = 1 - len(missing) / max(len(case.expected_keywords), 1)
            penalty = 0.3 * len(bad)
            final_score = max(0.0, keyword_score - penalty)
    
            results.append(EvalResult(
                case_id=case.case_id,
                passed=(final_score >= 0.8),  # **容錯區間**:>= 0.8 視為可接受
                score=final_score,
                missing_keywords=missing,
                bad_phrases=bad,
            ))
        return results
    

    💡 關鍵: 將非確定性輸出轉成「>= 0.8 即通過」的標準,讓 CI 能用 pass/fail 自動守門。

    這裡重點不是打分公式有多精緻,而是:

    • 先把「成功條件」具體化:有哪些必講的資訊、哪些不能亂承諾
    • 為每個 eval case 保留 error breakdown:缺失關鍵字 vs 違規措辭
    • 讓 CI 上只看 passed 率,而錯誤細節回報給開發者/PM 作 prompt 調整依據

    3. 用 OpenTelemetry + Grafana Tempo 做 Agent Trace

    把每次 Agent 流程變成可視化 trace,方便查為何多調用了三個工具、是哪一段 prompt 變壞。

    簡化版埋點(以 OpenTelemetry Python 為例):

    from opentelemetry import trace
    from opentelemetry.sdk.trace import TracerProvider
    from opentelemetry.sdk.trace.export import BatchSpanProcessor
    from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
    
    # 初始化
    provider = TracerProvider()
    exporter = OTLPSpanExporter(endpoint="http://tempo:4318/v1/traces")
    provider.add_span_processor(BatchSpanProcessor(exporter))
    trace.set_tracer_provider(provider)
    tracer = trace.get_tracer(__name__)
    
    # 在 Agent orchestrator 中
    
    def run_agent(user_input: str) -> AgentTrace:
        with tracer.start_as_current_span("agent_run") as span:
            span.set_attribute("agent.user_input", user_input)
    
            trace_model = AgentTrace(trace_id="...", user_input=user_input, steps=[])
    
            # Step 1: plan
            with tracer.start_as_current_span("plan_step") as s:
                plan_prompt = make_plan_prompt(user_input)
                plan_resp = call_llm(plan_prompt)
                s.set_attribute("llm.tokens", plan_resp.usage.total_tokens)
    
            # Step 2: tool call example
            with tracer.start_as_current_span("tool_call:search_order") as s:
                tool_input = {"order_id": "123"}
                tool_output = search_order(tool_input)
                s.set_attribute("tool.latency_ms", 42.0)
    
            # ...其餘步驟
    
            return trace_model
    

    好處:

    • 在 Grafana Tempo 裡可以完整看到 agent_run 的 timeline
    • 搭配 token 計費(參考 The Economics of Agents 那篇),可以計出 每個步驟的成本與貢獻
    • 當成功率下降時,你能具體問:是 plan_step 的 tokens 被砍太多,還是 tool_call latency 飆高導致超時?

    4. 把 eval 接進部署管線(CI/CD Gate)

    最後把 eval 變成 CI 裡的一個 stage:

    # .github/workflows/agent-eval.yml
    name: agent-eval
    
    on:
      pull_request:
        paths:
          - "agents/**"
          - "prompts/**"
    
    jobs:
      run-evals:
        runs-on: ubuntu-latest
        steps:
          - uses: actions/checkout@v4
    
          - name: Set up Python
            uses: actions/setup-python@v5
            with:
              python-version: "3.11"
    
          - name: Install deps
            run: pip install -r requirements.txt
    
          - name: Run regression evals
            run: python evals/run_regression.py
    
          - name: Check thresholds
            run: python evals/check_thresholds.py
    

    check_thresholds.py 裡只做幾件事:

    import sys
    from evals.results import load_results
    
    REGRESSION_MIN_PASS_RATE = 0.9  # **回歸 eval 門檻**
    
    if __name__ == "__main__":
        results = load_results("artifacts/regression.json")
        pass_rate = results.pass_rate()
    
        if pass_rate < REGRESSION_MIN_PASS_RATE:
            print(f"Regression eval failed: pass_rate={pass_rate}")
            sys.exit(1)  # CI 失敗,禁止合併
        else:
            print(f"Regression eval passed: pass_rate={pass_rate}")
    

    在實務上你會分兩組 eval:

    • 回歸 eval:保護現有功能,不允許退化
    • 實驗 eval:探索新功能、新策略,不當成 blocking gate,但會留報表做決策

    💡 關鍵: 將 REGRESSION_MIN_PASS_RATE 設為 0.9 這類門檻,讓 CI 自動阻擋明顯退化的改版。


    建議與注意事項

    1. 一週內最小成本導入 eval pipeline 的路線圖

    如果你現在有一個「已上線但很脆弱」的 Agent,可以這樣在一週內切入:

    Day 1-2:蒐集與結構化 trace

    • 在現有 Agent 增加最薄的 AgentTrace 結構(如上 Pydantic 範例)
    • 將最近 1-2 週的真實請求轉成 trace,存到資料庫或物件儲存(S3 / SeaweedFS)

    Day 3-4:定義 1-2 個核心場景的 eval

    • 選擇對業務影響最大的 1-2 個流程(例如退款、升級、帳單說明)
    • 用簡單規則+少量人工標註,做出第一版 pass/fail 判準與 score
    • 不求完美,只要能在版本之間穩定比較就夠

    Day 5-7:接進 CI,建立第一個 gate

    • 在 PR pipeline 裡加入 run_regression.py + check_thresholds.py
    • 門檻先設寬一點,只要不要明顯退化就放行
    • 讓團隊開始習慣:改 prompt / 策略前先跑 eval,之後再慢慢收緊標準

    2. 常見坑與避雷建議

    1. 過度依賴人工標註

    2. 你不需要為每個回應都人工打分,容易拖垮迭代速度

    3. 建議:少量標註 + 規則/模型輔助,人工只介入灰色地帶

    4. 用單一 aggregate 分數掩蓋失敗模式

    5. 整體 score 變高不代表沒有嚴重退化,例如:一般 FAQ 變強,但退款場景大幅下降

    6. 建議:按 場景 / 任務類別分開看指標,並保留 error breakdown

    7. 沒有分離「實驗 eval」與「回歸 eval」

    8. 把所有 eval 都當 blocking gate,會讓團隊不敢做大幅創新

    9. 建議:

      • 回歸 eval:保護既有能力,作為 CI gate
      • 實驗 eval:以 dashboard & report 呈現,不直接 block 部署
    10. 只評估最終輸出,忽略中間決策與工具路徑

    11. 這會讓你無法回答「為什麼這次多調用了三個工具?」

    12. 建議:在 trace 中至少記:每個 step 的 prompt、回應、工具調用與耗時,並用 OpenTelemetry 對應到 visualize 的 trace

    13. 沒把成本納入 eval 指標

    14. Agent 問題常常不是性能,而是「單位經濟」:同樣成功率但 token 成本翻倍

    15. 建議:在 eval 報告中加上 每個場景的平均 token / latency / tool calls 次數,作為優化目標之一

    核心結論:

    把 evals 當成 Agent 的 CI/CD,不是為了追求「更高的神奇效果」,而是讓你敢在生產環境持續迭代,而不會每次改動都賭運氣。從今天開始,先把你的 Agent 當成一個 可維運的軟體系統:有 trace、有 eval、有 gate,其他的魔法再說。這樣你才能在之後的模型升級、工具新增、記憶架構重構時,保持速度又不失控。

    🚀 你現在可以做的事

    • 在現有 Agent 專案中加入 AgentTrace 結構,開始紀錄並保存每次執行的 trace
    • 為一個關鍵業務場景(如退款流程)撰寫首批 5–10 個 RefundEvalCase 並實作 run_refund_eval
    • 在 CI(如 GitHub Actions)新增 agent-eval workflow,實作 check_thresholds.py 並先將 REGRESSION_MIN_PASS_RATE 設為 0.9
  • 為 AI Agent 建長期記憶:Rust 實戰

    為 AI Agent 建長期記憶:Rust 實戰

    📌 本文重點

    • 單純丟向量庫不足以支撐可運維的長期記憶
    • 將 Events / Sessions / Facts 結構化並分層檢索
    • 以 Rust 記憶服務提供 vendor-agnostic 的統一記憶層

    AI Agent 開到一定規模後,「把聊天記錄丟進向量資料庫」很快就不夠用:

    • 記憶膨脹導致成本失控、查詢變慢
    • 多個 Agent、不同 LLM 共用資料時互相污染
    • 想換模型或供應商時,歷史記憶幾乎不能重用

    這篇從記憶層設計問題拆解起,用 Rust 開源專案 akitaonrails/ai-memory 當主線,示範如何把「長期記憶」升級成可運維的基礎設施,而不是一堆臨時向量。


    重點說明:把「記憶」變成明確的基礎設施

    💡 關鍵: 把記憶從「單一向量庫」拆成 Events / Sessions / Facts,可以同時兼顧語義檢索與結構化查詢,讓長期記憶真正可維護、可重用。

    1. 資料結構:從 chat log 到 events / sessions / facts

    粗糙作法:

    • 每輪對話做 embedding → 丟進向量庫 → 用 semantic search 回撈

    問題:

    • 事件順序感消失:只剩相似度,沒有「先後」與「上下文」
    • 無法表達持久事實:像使用者偏好、系統狀態,只是散落在多個對話片段

    較好的設計是拆成三類:

    • Events:原始互動紀錄
    • 例:user_message, tool_call, agent_decision
    • 保留時間戳與來源,適合做 audit 和重播
    • Sessions:一次任務或一段對話的邊界
    • 例:session_id、agent_id、status=completed
    • 讓你知道「這次訂房流程」完整發生了什麼
    • Facts:可重用、可更新的持久知識
    • 例:使用者偏好、系統配置、業務規則
    • 需要版本與狀態(active, deprecated)

    這樣一來,semantic search 用在 Events/Facts 的內容檢索,structured query 用在 Sessions/Facts 的條件篩選,組出可靠的上下文再餵給 LLM,而不是全靠相似度。


    2. 檢索策略與 vendor-agnostic 介面

    實際上你會需要兩層 API:

    • 語義檢索層:
    • 例如:search_events(query, top_k)、search_facts(query, filters)
    • 背後可接不同 embedding provider(OpenAI, local model 等),但對上層 Agent 暴露的是穩定的介面

    • 結構化查詢層:

    • 例如:get_session(session_id)、list_facts(owner_id, kind="preference")
    • 通常走資料庫索引(Postgres, SQLite),不牽涉向量搜尋

    akitaonrails/ai-memory 正是要提供這種「供應商無關的記憶層」,讓你可以:

    • 今天用 OpenAI,明天換到本地模型
    • 多個 Agent framework(LangChain, LlamaIndex, 自家的 SDK)共用同一套記憶服務

    3. Rust 記憶服務 + RAG / 工具調用整合

    長期記憶一旦變成基礎設施,就有三個技術要求:

    • 效能:高頻讀寫、多 Agent 並行
    • 安全:記憶裡必然有 PII 和敏感業務資訊
    • 跨語言可接:Python、Node、Go、甚至 CLI agent 都要用

    Rust 在這裡的角色:

    • 提供一個高效、型別安全的記憶核心(ai-memory)
    • 對外以 FFI / HTTP / IPC 三種方式暴露 API
    • Agent 框架只需要呼叫類似 store_event / query_memory 的介面,就能把記憶接進 RAG pipeline 或工具調用流程

    實作範例:用 ai-memory 建一個共用長期記憶服務

    以下以虛構的 ai-memory 介面示意,重點放在設計思路而不是精確函式名稱。

    1)定義記憶 schema,接上現有 Agent

    先在 Rust 端定義核心結構:

    // memory_schema.rs
    
    #[derive(Debug, Clone)]
    pub enum EventKind {
        UserMessage,
        AgentReply,
        ToolCall,
        ToolResult,
    }
    
    #[derive(Debug, Clone)]
    pub struct Event {
        pub id: String,
        pub session_id: String,
        pub agent_id: String,
        pub kind: EventKind,
        pub content: String,
        pub metadata: serde_json::Value,
        pub created_at: chrono::DateTime<chrono::Utc>,
    }
    
    #[derive(Debug, Clone)]
    pub struct Fact {
        pub id: String,
        pub owner_id: String,    // user_id 或 system
        pub kind: String,        // preference, policy, profile
        pub content: String,
        pub embedding: Option<Vec<f32>>,  // for semantic search
        pub version: i32,
        pub status: String,      // active, deprecated
    }
    

    對 Python Agent 來說,只需要一個薄封裝,把每輪互動寫入記憶:

    # agent_memory.py
    
    class MemoryClient:
        def __init__(self, base_url: str):
            self.base_url = base_url
    
        def store_event(self, session_id, agent_id, kind, content, metadata=None):
            payload = {
                "session_id": session_id,
                "agent_id": agent_id,
                "kind": kind,
                "content": content,
                "metadata": metadata or {},
            }
            requests.post(f"{self.base_url}/events", json=payload)
    
        def search_facts(self, query, owner_id=None, kind=None, top_k=5):
            params = {
                "query": query,
                "owner_id": owner_id,
                "kind": kind,
                "top_k": top_k,
            }
            resp = requests.get(f"{self.base_url}/facts/search", params=params)
            return resp.json()["results"]
    

    Agent loop 中的使用方式:

    memory = MemoryClient(base_url="http://localhost:8080")
    
    # 每輪對話寫入 event
    memory.store_event(
        session_id=session_id,
        agent_id="support-bot-v2",
        kind="UserMessage",
        content=user_input,
    )
    
    # 在回應前查詢長期偏好
    facts = memory.search_facts(
        query="使用者的語言偏好與通知設定",
        owner_id=user_id,
        kind="preference",
        top_k=3,
    )
    
    context = format_facts_for_prompt(facts)
    response = llm.chat(prompt=build_prompt(user_input, context))
    

    這樣你的 Agent 邏輯完全不需要知道底層用的是哪家向量資料庫,也不綁在單一 LLM provider。


    2)Rust 記憶服務:FFI / HTTP / IPC 三種接法

    假設 ai-memory 提供核心 crate ai_memory_core,我們可以用三種方式包出去。

    HTTP 服務(最通用)

    優點:任何語言都能接;缺點:有網路 overhead。

    // http_server.rs
    
    use ai_memory_core::{MemoryStore, SearchQuery};
    use axum::{routing::get, routing::post, Json, Router};
    
    async fn create_event(Json(payload): Json<CreateEventRequest>) -> Json<EventResponse> {
        let mut store = MemoryStore::global();
        let event = store.store_event(payload.into())?;
        Json(EventResponse::from(event))
    }
    
    async fn search_facts(Json(query): Json<SearchFactsRequest>) -> Json<SearchFactsResponse> {
        let store = MemoryStore::global();
        let results = store.search_facts(SearchQuery::from(query))?;
        Json(SearchFactsResponse { results })
    }
    
    pub fn app() -> Router {
        Router::new()
            .route("/events", post(create_event))
            .route("/facts/search", get(search_facts))
    }
    

    命令列 Agent 或後端服務只要跑一個共用 memory server,就可以用 HTTP 存取。

    FFI(嵌入到 Python / Node 進程)

    優點:效能好、延遲低;缺點:需維護 binding。適用高頻工具調用型 Agent。

    // lib.rs (Rust)
    
    #[no_mangle]
    pub extern "C" fn store_event_ffi(json_payload: *const c_char) -> *const c_char {
        // 解析 JSON,呼叫 MemoryStore,回傳 JSON 字串
    }
    

    Python 側使用 ctypes 或 pyo3 包一層,暴露同樣的 store_event / search_facts 介面,對應前面的 MemoryClient。

    IPC(同機不同進程,高安全場景)

    可以用 Unix socket + protobuf 或 Cap’n Proto:

    • 優點:比 HTTP 更輕量,適合同機多服務共用記憶
    • 缺點:部署稍複雜,需額外 tooling

    設計上只要確保三種接法都共用同一套核心 API(MemoryStore),就能在不同專案中任意選擇實作方式,而不改動 Agent 邏輯。


    3)版本升級、資料遷移、多模型共用記憶庫

    長期記憶真正困難在於 「持續演化」。

    Schema 版本管理

    在 Fact 結構上掛版本欄位:

    pub struct Fact {
        pub id: String,
        pub owner_id: String,
        pub kind: String,
        pub content: String,
        pub version: i32,       // schema/version
        pub status: String,
        pub embedding: Option<Vec<f32>>,  
    }
    

    當你需要新增欄位或改變結構時:

    • 新寫入用 version = 2
    • 舊資料由 migration job 緩慢升級
    • 查詢 API 接受 min_version / max_version 作為過渡策略

    多模型共用同一記憶庫

    不同 LLM 對同一條記錄的解讀可能不同,所以要在 metadata 明確標注來源模型:

    pub struct Event {
        pub model_name: Option<String>,   // gpt-4o, llama-3-70b 等
        pub agent_id: String,
        // ...
    }
    

    策略上:

    • Facts 儘量由工具或人類決策產生,減少「模型幻覺」寫入持久記憶
    • 檢索時可加 filter:model_name in ["gpt-4o", "internal-rule-engine"],避免用某些品質較差模型產生的事件來推論

    建議與注意事項:把坑提前填好

    💡 關鍵: 控制 embedding 範圍、處理 PII、標記 embedding 模型,是讓長期記憶在成本、隱私與準確度間取得平衡的三個關鍵。

    1. 記憶膨脹與成本控制

    問題:

    • 所有對話都 embedding → 向量庫數量爆炸,成本跟查詢延遲一起上升

    建議:

    • 只對重要 Events / Facts 做 embedding,例如:完成一個任務時產生 summary fact
    • 針對長期 session,定期做 conversation summarization,保留摘要而不是 raw log
    • 對向量庫設計 TTL 或冷/熱層級:冷資料只保留摘要向量

    2. 隱私與合規(PII / 敏感資料)

    問題:

    • 長期記憶通常含姓名、電話、訂單資訊等 PII

    建議:

    • 設計 PII-aware schema:把 PII 拆出獨立欄位,便於加密與 masking
    • 在寫入前跑簡單的 PII 檢測 rule:
    • 例:電話號碼、email pattern 直接用工具抽出 → 存入安全欄位
    • 對查詢 API 加上角色權限(例如 owner_id+role=internal_support)

    3. 語義檢索漂移與模型更新

    問題:

    • 換 embedding 模型後,舊向量的語義分佈不同,semantic search 精度變差

    簡單防線:

    • 在向量旁存 embedding_model 欄位
    • 新模型上線後:
    • 新增資料用新 embedding
    • 舊資料分批 re-embed,或只重算活躍 Facts
    • 用 offline evaluation:定義一組標準 query + expected hits,監控 semantic search 成效

    4. 不同模型對同一紀錄的解讀不一致

    問題:

    • Model A 把某次對話解讀為「使用者喜歡簡訊通知」,Model B 覺得是「偏好 email」

    建議:

    • 把「推論」與「事實」分開存:
    • Fact:明確的、可驗證的偏好(使用者自行設定)
    • Inference:模型的猜測,需經多次交叉驗證或人工確認才升級為 Fact
    • 在 prompt 中區分:
    • 「已確認偏好」 vs. 「推測偏好」

    結語:把長期記憶升級成基礎設施

    對成熟的 AI 專案來說,長期記憶不再是「多塞幾個向量」的 hack,而是需要設計、版本與治理的系統。利用像 akitaonrails/ai-memory 這種以 Rust 實作的開源方案,你可以:

    • 為所有 Agent 建立一個 供應商無關、可演化的記憶層
    • 清楚區分 Events / Sessions / Facts,同時用語義與結構化查詢做精準檢索
    • 以 HTTP / FFI / IPC 等方式,在不同語言與框架中重用同一記憶庫

    這些設計一開始多花一點功夫,換到的是更穩定的成本、更易維護的架構,以及在多 Agent、多模型環境裡,真正能「記住」使用者與業務脈絡的系統。

    🚀 你現在可以做的事

    • 到 GitHub 搜尋並閱讀 akitaonrails/ai-memory 專案原始碼與文件
    • 在現有 Agent 專案中先導入 Events / Sessions / Facts 基本 schema,重構記憶寫入流程
    • 實作一個簡單的 HTTP memory server,讓至少兩個不同語言或框架的 Agent 共用同一記憶層
  • Ultrafast GPT-5.6 實戰:14 倍加速怎麼用

    Ultrafast GPT-5.6 實戰:14 倍加速怎麼用

    📌 本文重點

    • Ultrafast 模式專攻高頻互動的延遲瓶頸
    • 同一模型家族內可用路由動態切換速度層
    • 需在 SDK 與 gateway 層抽象速度與成本
    • Ultrafast 僅適合特定低延遲、高價值場景

    Ultrafast 模式解決的痛點很直接:高頻互動場景的延遲瓶頸。以前你要不是砸更多錢堆實例、要不就犧牲回應品質;現在 OpenAI 把 推論速度做成三層產品(Standard / Fast / Ultrafast),開發者可以在同一套 API 與模型家族裡,用路由策略動態換檔,對齊不同用例的 SLO、成本與體驗,而不用再自己做一堆複雜的模型拆分與基礎設施擴縮容。


    重點說明

    1. 三層速度 / 定價:如何影響架構與 SLO

    OpenAI 對 GPT-5.6 Sol 提供三種模式(名稱以概念為主,實際 key 可能會略有出入):

    • Standard:預設速度,成本最低,延遲中等
    • Fast:約數倍加速,成本略高,適合互動產品主路徑
    • Ultrafast:宣稱最高 14× 加速、最高約 750 tokens/s 輸出,成本最高,針對極低延遲需求

    💡 關鍵: 透過 Standard / Fast / Ultrafast 三層速度,開發者可以用同一套 API 針對不同用戶與場景調整 SLO 與成本,而不用拆模型或重做基礎設施。

    對服務設計的直接影響:

    1. SLO 要分級:不要一套 SLO 打天下。可以明確定義:
    2. Free / basic 用戶:p95 2–3 秒內回應(Standard/Fast)
    3. Pro / enterprise:p95 0.5–1 秒內首 token(Ultrafast)
    4. 成本與速度綁定:把「速度」變成 config,而不是寫死在產品邏輯。同一套業務流程可以根據 user tier / request 類型選擇 不同 mode,而不是複製一套 API client。
    5. 多租戶隔離:高價的 Ultrafast 不應被低價客戶打爆。要在 gateway 層對不同模式做 獨立併發 / QPS 限流,避免一個 tenant 把所有 Ultrafast capacity 用光。

    2. Cerebras 硬體、吞吐與併發規劃

    Ultrafast 是跑在 Cerebras 專用硬體上,直接後果:

    • 單請求輸出速度極快(最高 750 tok/s),流式模式下體感差異巨大。
    • 批次處理 vs 流式響應要重新平衡:
    • 傳統 GPU 場景下,你可能會用較大的 batch 來吃滿 GPU
    • Ultrafast 下,單請求已經很快,過大 batch 反而拉高首 token 延遲

    💡 關鍵: 在 Ultrafast 上追求大 batch 已經不是核心,反而要優化首 token 延遲與流式體感,重新設計互動與批次任務的平衡。

    建議策略:

    • 互動型(chat, tools, game)→ 走流式,以首 token 延遲為優先。
    • 批次生成(離線摘要、報表)→ 仍可 batch,但不一定需要 Ultrafast,用 Standard/Fast 成本更合理。

    在多租戶情境下,伺服端可以這樣規劃:

    • 對每種模式維護獨立的 併發窗口與隊列:
    • concurrency.standard、concurrency.fast、concurrency.ultrafast
    • 配合 token rate 限制:
    • 以「每租戶每分鐘最大 tokens」而不是只看 QPS,避免 prompt 過長把 Cerebras 打爆。

    3. 哪些情境適合 / 不適合 Ultrafast

    特別適合:

    1. 即時客服 / 聊天機器人
    2. 要求打字中就開始看到回覆、上下文短到中等,流式 + Ultrafast 直接改善體感。
    3. 語音 / 遊戲對話
    4. 語音聊天、NPC 對話,延遲 > 500ms 就很明顯,Ultrafast 可以把首 token 拖到人類可接受區間。
    5. 代理工作流中的多步工具調用
    6. 每步都要 LLM 思考 + call tool,如果每步縮短 5–10 倍,整個 workflow latency 會非常明顯地下降。
    7. 長對話、工具調用場景
    8. context 長但每輪生成量中等,Ultrafast 有助於掩蓋長 prompt 帶來的延遲感。

    不適合 / 需謹慎:

    1. 重推理、高精度任務(複雜 codegen、數學推理、法律合約草擬)
    2. Ultrafast 主要是同一模型的加速推理模式,不是「更聰明」,在此類任務中瓶頸往往是思考層面而非純 IO,速度優勢有限,成本反而偏高。
    3. 高單次成本任務(超長上下文、長文生成)
    4. 一次就幾萬 tokens,Ultrafast 只會讓你更快花完錢。對這種任務 Standard/Fast 更合理。

    💡 關鍵: Ultrafast 適合高價值、對延遲極敏感的互動任務,不適合超長上下文或高精度重推理工作,否則只會加速燒錢。


    4. 應用層「延遲感知路由」:動態選 Standard/Fast/Ultrafast

    核心思路:把「速度選擇」抽象成一個 routing decision,而不是在業務碼 scattered 寫死。

    常用維度:

    • request 類型:"chat" | "tool_call" | "batch_summarize" | "voice" 等
    • 用戶等級:free | pro | enterprise
    • 成本閾值:每 request 最大可接受預估成本(可由長度 * 單價估)

    簡單決策矩陣例子:

    • voice 或 realtime_game → Ultrafast(僅限 pro/enterprise)
    • chat + enterprise → Fast / Ultrafast(看負載與成本)
    • batch_summarize → Standard

    實作範例

    以下假設有一個類似 gpt-5.6-sol-ultrafast 的 model 名稱(實際以官方為準)。

    1. 基本 API 呼叫:切換模式

    Node.js(使用官方風格 client)

    import OpenAI from "openai";
    
    const client = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });
    
    // 封裝一個工廠函式,統一進入點
    export async function callModel({
      mode,          // 'standard' | 'fast' | 'ultrafast'
      messages,
      stream = false,
    }) {
      const modelMap = {
        standard: "gpt-5.6-sol",
        fast: "gpt-5.6-sol-fast",
        ultrafast: "gpt-5.6-sol-ultrafast",
      } as const;
    
      const model = modelMap[mode];
    
      const response = await client.chat.completions.create({
        model,
        messages,
        stream,
        // 關鍵:對 Ultrafast 場景,適當降低 max_tokens,避免成本飆升
        max_tokens: mode === "ultrafast" ? 256 : 1024,
      });
    
      return response;
    }
    

    重點:
    – 把模式抽象成 mode,不要在各個業務 service 裡到處寫 "gpt-5.6-sol-ultrafast"。
    – 未來更換硬體 / 新增另一次速度層,只需改 modelMap 即可,避免 breaking changes 滲透全碼庫。

    Python:簡易延遲感知路由

    from typing import Literal, Dict
    from openai import OpenAI
    
    client = OpenAI()
    
    SpeedMode = Literal["standard", "fast", "ultrafast"]
    
    MODEL_MAP: Dict[SpeedMode, str] = {
        "standard": "gpt-5.6-sol",
        "fast": "gpt-5.6-sol-fast",
        "ultrafast": "gpt-5.6-sol-ultrafast",
    }
    
    
    def decide_speed_mode(request_type: str, user_tier: str, est_tokens: int) -> SpeedMode:
        # 粗略成本估:假設 ultrafast 單 token 價格 ~ 3x standard(示意)
        # 真實值請用官方定價
        if request_type in {"voice", "realtime_game"}:
            return "ultrafast" if user_tier in {"pro", "enterprise"} else "fast"
    
        if request_type == "chat":
            if user_tier == "enterprise":
                return "fast"
            if est_tokens > 4000:
                return "standard"
            return "standard"
    
        if request_type == "batch_summarize":
            return "standard"
    
        return "standard"
    
    
    def call_gpt(messages, request_type: str, user_tier: str, est_tokens: int, stream: bool = False):
        mode = decide_speed_mode(request_type, user_tier, est_tokens)
        model = MODEL_MAP[mode]
    
        resp = client.chat.completions.create(
            model=model,
            messages=messages,
            stream=stream,
            max_tokens=min(est_tokens, 1024 if mode == "ultrafast" else 4096),
        )
        return resp
    

    2. 流式回應:心跳、重連與部分輸出

    流式(SSE / WebSocket)在 Ultrafast 下會遇到幾個新問題:

    • 速度太快導致前端渲染抖動:一次性大量 token 噴到前端,DOM 大量重繪。
    • 連線中斷時,部分輸出要不要保留?怎麼重試?

    伺服器端(Node)伺服器串流示意:

    // 假設在 Express handler 裡
    
    app.post("/chat-stream", async (req, res) => {
      const { messages, requestType, userTier } = req.body;
    
      // 設置 SSE header
      res.writeHead(200, {
        "Content-Type": "text/event-stream",
        "Cache-Control": "no-cache",
        Connection: "keep-alive",
      });
    
      const estTokens = 512; // 可依 prompt 長度 + UX 預期估算
    
      const mode = decideSpeedMode(requestType, userTier, estTokens); // 從前面的邏輯抽出
    
      const stream = await client.chat.completions.create({
        model: MODEL_MAP[mode],
        messages,
        stream: true,
      });
    
      let lastEventAt = Date.now();
    
      for await (const chunk of stream) {
        lastEventAt = Date.now();
        const delta = chunk.choices[0]?.delta?.content || "";
        res.write(`data: ${JSON.stringify({ type: "chunk", delta })}\n\n`);
      }
    
      // 安全結束
      res.write(`data: ${JSON.stringify({ type: "end" })}\n\n`);
      res.end();
    
      // 簡易心跳:若長時間沒有 chunk,可定期送心跳
      setInterval(() => {
        const now = Date.now();
        if (now - lastEventAt > 2000) {
          res.write(`data: ${JSON.stringify({ type: "heartbeat" })}\n\n`);
          lastEventAt = now;
        }
      }, 2000);
    });
    

    實作重點與坑:

    • 心跳機制:Ultrafast 大多數請求都會很快完成,但在長 prompt 或工具呼叫等待時,前端若長時間沒收到事件會以為掛了。定期送 heartbeat 可避免誤判。
    • 部分輸出策略:
    • 若連線中斷,但前端已接到 80% 內容,通常 不要重新發同樣 prompt,而是提示用戶「是否補完」或在後端追蹤 offset(稍複雜)。
    • 若是關鍵指令(例如 code patch)中斷,則 必須在後端標記這次呼叫為失敗,不應自動套用部分結果。

    3. 成本計費鉤子與觀測

    在 gateway 層把 token 使用量、所用模式、user_tier 全部打 log,才能做之後的調整。

    簡化版計費鉤子(Node):

    interface BillingInfo {
      userId: string;
      mode: "standard" | "fast" | "ultrafast";
      inputTokens: number;
      outputTokens: number;
      latencyMs: number;
    }
    
    function recordBilling(info: BillingInfo) {
      // 可寫入 DB / message queue / metrics 平台
      console.log("billing", info);
    }
    
    async function callWithBilling(args) {
      const start = Date.now();
    
      const resp = await client.chat.completions.create(args);
    
      const latencyMs = Date.now() - start;
    
      const usage = resp.usage || { prompt_tokens: 0, completion_tokens: 0 };
    
      recordBilling({
        userId: args.userId,
        mode: args.mode,
        inputTokens: usage.prompt_tokens,
        outputTokens: usage.completion_tokens,
        latencyMs,
      });
    
      return resp;
    }
    

    觀測建議:
    – 對每個模式分別追 p50 / p95 latency、錯誤率、token 用量。
    – 若發現 Ultrafast 的 p95 還是拉高,代表已超出合理併發,要下調該模式的並發上限或做排隊,避免 UX 反而變差。


    建議與注意事項

    1. Prompt 長度對吞吐的影響

    • Ultrafast 在輸出速度快,但 長 prompt 會把 latency 推回去:
    • context 越長,首 token 延遲比例越高,對體感影響最大。
    • 建議:
    • 在 Ultrafast 路徑上嚴格控制系統提示與歷史訊息長度:做 aggressive truncation / distillation。
    • 可為 Ultrafast 系列單獨設計較短的 system prompt,避免浪費吞吐優勢。

    2. 錯誤重試策略

    • 對 Ultrafast,重試成本比 Standard 高很多,不要無腦 retry: 3。
    • 建議:
    • 對 429 / 503 類錯誤:改降級路由(例如 Ultrafast → Fast),而不是瘋狂重試 Ultrafast。
    • 對連線中斷(網路層):
      • 若是批次任務,可重試同模式。
      • 若是互動任務且前端已有部分輸出,改成「請用戶重試」會更安全。

    3. 在現有後端封裝 client SDK,避免未來爆炸

    為了讓未來更換硬體(例如新版 Cerebras 叢集、其他加速方案)或模型版本時不產生 breaking changes:

    • 永遠不要在業務碼裡直接調 OpenAI SDK:
    • 統一用 llmClient.call({ mode, messages, ... }) 這種內部介面。
    • 在 SDK 層:
    • 隱藏實際 model 名稱,只暴露 mode、capability(例如 "chat" | "tool" | "embed")。
    • 一個簡單 pattern:
    // llmClient.ts
    
    export type SpeedMode = "standard" | "fast" | "ultrafast";
    
    export interface LLMRequest {
      mode: SpeedMode;
      messages: any[];
      stream?: boolean;
      // 未來可新增:capability, modelFamily 等
    }
    
    export async function callLLM(req: LLMRequest) {
      const modelMap = {
        standard: "gpt-5.6-sol",
        fast: "gpt-5.6-sol-fast",
        ultrafast: "gpt-5.6-sol-ultrafast",
      } as const;
    
      return client.chat.completions.create({
        model: modelMap[req.mode],
        messages: req.messages,
        stream: req.stream,
      });
    }
    

    未來即使 OpenAI 把 Ultrafast 換到別的硬體、別的 model 名稱,你只需要改 modelMap,上層延遲感知路由與業務邏輯完全不用動。


    總結: Ultrafast GPT-5.6 Sol 把「速度」變成了一個產品級別的一等公民。對工程團隊來說,真正的價值不是 750 tok/s 這個數字,而是:你可以用架構與路由策略,把速度、成本、SLO 解耦開來。只要在 SDK 層做好抽象、在 gateway 層做好觀測和限流,就能相對平滑地把 Ultrafast 接進現有系統,讓高價值的互動場景先享受到 14× 加速帶來的體驗提升。

    🚀 你現在可以做的事

    • 在現有後端加一層 llmClient.call({ mode, ... }) 抽象,統一管理 Standard/Fast/Ultrafast 路由
    • 為現有產品標註 request_type、user_tier,實作一個簡單的延遲感知決策函式來選擇模式
    • 在 gateway 或 API 層加上 token 與 mode 的計費與 p95 latency 監控儀表板,觀察 Ultrafast 對成本與體驗的影響
  • 程式碼化工具呼叫:Agent 下一步

    程式碼化工具呼叫:Agent 下一步

    📌 本文重點

    • Mistral 讓模型一次寫出完整可執行工具計畫
    • Plan 作為結構化 AST,提升可觀測性與可重放
    • 適合多步 workflow、高可靠與多代理協作場景

    Mistral 的程式碼化工具呼叫(code implemented tool calls)瞄準的痛點很直接:現有 Agent/工具呼叫機制太「鬆」——模型吐出一坨 JSON,外層 orchestrator 再硬湊成多步流程,結果是:

    • 提示工程很重(得教模型怎麼排程、怎麼分步)
    • 工具呼叫不可觀測、不易重放(難做 debug/retry)
    • 多代理協作下狀態跟交易邊界很容易打結

    Mistral 想做的是:讓模型在生成過程中「寫出一個可執行的工具呼叫計畫」,再由執行器直接跑這段計畫,介於「純自然語言」與「完整程式語言」之間,變成一種半結構化、可解釋的 agent 程式碼。


    重點說明

    1. 什麼是「程式碼化工具呼叫」?

    用工程語言講,就是讓模型輸出類似這樣的東西:

    plan = [
      call(tool="search_user", args={"email": "foo@bar.com"}),
      if_("result.found", then=[
        call(tool="update_user_status", args={"id": "result.id", "status": "active"})
      ], else=[
        call(tool="create_user", args={"email": "foo@bar.com"})
      ])
    ]
    

    重點不在語法,而在語意:

    • 這不是單一 function_call,而是一段可執行的呼叫腳本
    • 包含控制流程(if/loop)、工具序列、甚至回滾/補償邏輯
    • 可以被引擎解析、記錄、觀測、部分重試

    💡 關鍵: 透過「一次生成完整腳本」,LLM 從逐步決策器變成計畫產生器,大幅降低 orchestrator 與提示工程負擔

    跟 OpenAI function calling 或 MCP 比較:

    • function calling:一次「我要叫哪個工具 + 參數」,多步需要多輪迭代
    • MCP:標準化「工具服務」與「資源」,但仍偏一次一個呼叫
    • 程式碼化工具呼叫:一次生成完整 workflow blueprint,執行器負責跑與監控

    2. 與現有框架(OpenAI / MCP / LangChain)的差異

    心智模型差異:

    • LangChain / 大部分 Agent:
    • LLM 每回合決定「下一步做什麼」,像 ReAct / Planning+Execution
    • 多步推理由 orchestrator 負責追蹤與 loop
    • Mistral 這種設計:
    • 一次吐出一個計畫(plan as code),再由 runtime 執行
    • LLM 變成「計畫生成器」,而不是「每一步都要決策的狀態機」

    具體好處:

    1. 更少提示工程:
    2. 不用在 system prompt 裡教一大堆「遇到 X 就呼叫 Y、記得先查再算」
    3. 而是用工具 DSL + 執行規則約束模型:你只能用這些原語寫流程

    4. 可觀測性:

    5. 計畫本身就是一種可序列化的 execution graph
    6. 可以記錄在 DB,提供 UI 看「每次 Agent 做了哪些步驟、用了哪些工具」

    7. 重試與補償更簡單:

    8. Plan 是結構化的,你可以:
      • 只重跑失敗的 node
      • 將已成功步驟標記為 committed,失敗則跑補償工具

    💡 關鍵: Plan 作為結構化 execution graph,天然支援觀測、重試與補償,比傳統一輪一呼叫模式更適合長鏈路任務


    3. 多步推理、長任務與多代理的實際價值

    多步推理 / 長任務:

    • 對需要多輪工具呼叫(查資料 → 計算 → 寫回 DB → 發通知)的任務,
      一次生成計畫比每步都叫 LLM 決策更穩定也更便宜
    • 可以做:
    • 長任務分段執行:每段是獨立的計畫
    • 中途中斷再恢復:計畫 + 執行游標即可恢復

    多代理協作:

    • 一個 agent 產生計畫,別的 agent 只負責執行子計畫
    • 或一個高階「戰略 agent」產生 plan,交給「執行 agent」執行
    • Plan 本身扮演協作協議:各 agent 只對自己負責的子樹負責

    實作範例

    以下用 Python 與 TypeScript 模擬一個「程式碼化工具呼叫」風格的框架。重點在介面設計與工程落地,不依賴特定廠商 API。


    1. Python:計畫表示 & 執行器介面

    from typing import Any, Dict, List, Literal, Callable
    
    class ToolCall:
        def __init__(self, name: str, args: Dict[str, Any], retry: int = 0):
            self.type: Literal["tool_call"] = "tool_call"
            self.name = name
            self.args = args
            self.retry = retry
    
    class IfNode:
        def __init__(self, condition: str, then: List[Any], otherwise: List[Any] | None = None):
            self.type: Literal["if"] = "if"
            self.condition = condition  # e.g. "ctx['user']['exists'] == True"
            self.then = then
            self.otherwise = otherwise or []
    
    PlanNode = ToolCall | IfNode
    
    class Plan:
        def __init__(self, steps: List[PlanNode]):
            self.steps = steps
    
    
    class ToolRegistry:
        def __init__(self):
            self._tools: Dict[str, Callable[[Dict[str, Any]], Any]] = {}
    
        def register(self, name: str):
            def decorator(fn):
                self._tools[name] = fn
                return fn
            return decorator
    
        def get(self, name: str) -> Callable[[Dict[str, Any]], Any]:
            return self._tools[name]
    
    
    tools = ToolRegistry()
    
    @tools.register("search_user")
    def search_user(args: Dict[str, Any]):
        # 呼叫現有微服務 / DB
        ...
    
    @tools.register("create_user")
    def create_user(args: Dict[str, Any]):
        ...
    
    
    class PlanExecutor:
        def __init__(self, tools: ToolRegistry):
            self.tools = tools
    
        def run(self, plan: Plan, ctx: Dict[str, Any]):
            for step in plan.steps:
                self._run_node(step, ctx)
            return ctx
    
        def _run_node(self, node: PlanNode, ctx: Dict[str, Any]):
            if isinstance(node, ToolCall):
                self._run_tool(node, ctx)
            elif isinstance(node, IfNode):
                branch = node.then if eval(node.condition, {}, {"ctx": ctx}) else node.otherwise
                for sub in branch:
                    self._run_node(sub, ctx)
    
        def _run_tool(self, node: ToolCall, ctx: Dict[str, Any]):
            fn = self.tools.get(node.name)
            attempt = 0
            while True:
                try:
                    result = fn(node.args)
                    ctx[node.name] = result
                    return
                except Exception as e:
                    attempt += 1
                    if attempt > node.retry:
                        # 這裡可以記錄觀測資料,觸發補償
                        raise e
    

    要點:

    • Plan 是一個結構化 AST,可以序列化/儲存
    • PlanExecutor 是純程式碼,模型只產生 Plan 描述,不負責執行
    • 工具描述透過 ToolRegistry 管理,未來可以輸出 JSON schema 給 LLM 看

    2. 模型輸出格式(給 LLM 的 contract)

    你可以用 system prompt 明確要求模型輸出這種 JSON:

    {
      "steps": [
        {
          "type": "tool_call",
          "name": "search_user",
          "args": { "email": "{{user_email}}" },
          "retry": 1
        },
        {
          "type": "if",
          "condition": "ctx['search_user']['found'] == True",
          "then": [
            {
              "type": "tool_call",
              "name": "update_user_status",
              "args": {"id": "{{ctx.search_user.id}}", "status": "active"}
            }
          ],
          "otherwise": [
            {
              "type": "tool_call",
              "name": "create_user",
              "args": {"email": "{{user_email}}"}
            }
          ]
        }
      ]
    }
    

    重點:

    • Plan schema 是固定的,模型只在這個 schema 內填充內容
    • 執行器可以根據 retry 實作內建重試策略
    • condition 可以限制為簡單表達式(避免讓模型寫任意 Python)

    3. TypeScript:與微服務整合 & 錯誤恢復

    type ToolCall = {
      type: 'tool_call';
      name: string;
      args: Record<string, any>;
      retry?: number;
    };
    
    type IfNode = {
      type: 'if';
      condition: string; // 例如 "ctx.order.status === 'PAID'"
      then: PlanNode[];
      otherwise?: PlanNode[];
    };
    
    export type PlanNode = ToolCall | IfNode;
    
    export interface Tool {
      name: string;
      // 可以包 HTTP call / gRPC / queue message
      invoke: (args: any, ctx: any) => Promise<any>;
    }
    
    export class PlanRunner {
      constructor(private tools: Map<string, Tool>) {}
    
      async run(plan: PlanNode[], ctx: any, options?: { txnId?: string }) {
        for (const step of plan) {
          await this.runNode(step, ctx, options);
        }
        return ctx;
      }
    
      private async runNode(node: PlanNode, ctx: any, options?: { txnId?: string }) {
        if (node.type === 'tool_call') {
          await this.runTool(node, ctx, options);
        } else if (node.type === 'if') {
          const cond = this.evalCondition(node.condition, ctx);
          const branch = cond ? node.then : node.otherwise ?? [];
          for (const s of branch) await this.runNode(s, ctx, options);
        }
      }
    
      private async runTool(node: ToolCall, ctx: any, options?: { txnId?: string }) {
        const tool = this.tools.get(node.name);
        if (!tool) throw new Error(`Tool not found: ${node.name}`);
    
        const maxRetry = node.retry ?? 0;
        let attempt = 0;
        while (true) {
          try {
            const result = await tool.invoke(node.args, ctx);
            ctx[node.name] = result;
            // 可在這裡寫入 observability / event log
            return;
          } catch (err) {
            attempt++;
            if (attempt > maxRetry) {
              // 此處可以觸發補償工具,例如 compensate_${node.name}
              throw err;
            }
          }
        }
      }
    
      private evalCondition(expr: string, ctx: any): boolean {
        // 建議使用安全 expression evaluator,而不是直接 eval
        return Function('ctx', `return (${expr});`)(ctx);
      }
    }
    

    與現有微服務整合建議:

    • 每個工具是對應一個微服務或某個 bounded context 的 use case
    • 工具輸出應明確標示是否已提交 side effect(方便補償)
    • PlanRunner 可以在每個 tool call 前後寫入 event log,方便追蹤

    建議與注意事項

    1. 模型自由度過高 → 亂呼工具

    風險:

    • 模型可能:
    • 亂寫 condition 表達式
    • 緊密 loop 呼叫昂貴工具
    • 混用不應該同時出現的工具(跨 bounded context)

    建議:

    • 提供有限 DSL:例如只允許 if、不允許任意 while
    • 在 runtime 做 plan validation:
    • 最大深度、最大工具呼叫數
    • 禁用某些工具組合
    • 聯合 靜態規則 + LLM 自檢:生成後再請同一模型對 plan 做 sanity check

    2. 狀態與交易邊界混亂

    痛點:

    • Plan 容易跨越多個系統邊界:DB、支付、通知系統
    • 一旦中途失敗,很難知道哪一步已真正「提交」

    建議:

    • 把 工具當作 transactional boundary:
    • Tool 內部自行處理 local transaction
    • Tool 對外暴露「已提交 / 可補償」資訊
    • 在 Plan schema 中加入:
    • idempotency_key
    • compensate_tool(可選)

    範例:

    {
      "type": "tool_call",
      "name": "charge_payment",
      "args": { "order_id": "123" },
      "idempotency_key": "order-123-charge",
      "compensate_tool": "refund_payment"
    }
    

    3. 專利風險與開源 / 自建 Agent 平台

    Mistral 申請專利的關鍵關注點在於:

    • 「在生成過程中嵌入可執行工具呼叫計畫」這種整體 workflow
    • 若你建立的框架:
    • 讓 LLM 生成一段帶控制流程的工具呼叫「程式」
    • 再由執行器直接執行

    可能與專利有重疊風險。

    對開源 / 自建平台的實務建議:

    1. 盡量採用分步決策(step-wise)方式(每步 function calling),避免明確 branding 成「plan as code」
    2. 若要實作類似能力,注意:
    3. 檢查專利條款與地域適用範圍
    4. 避免與專利文本中的特定 claim 結構一模一樣
    5. 企業內部自用系統較少被追訴,但商用 SaaS/開源框架就要謹慎,特別是標榜「code implemented tool calls」之類的功能時。

    💡 關鍵: 若將「LLM 產生可執行計畫 + 執行器直接跑」打包成商用產品,需特別留意與既有專利 claim 的重疊風險


    總結:何時值得導入程式碼化工具呼叫?

    適用場景:

    • 任務天然就是多步 workflow(CRM、自動化運維、財務流程)
    • 需要強觀測性、可重放、可審計的 Agent
    • 多代理/多服務協同,想要一個「共通語言」描述任務

    不適用場景:

    • 單步問答或簡單工具呼叫(RAG 查一次資料就結束)
    • 對專利/法務非常敏感且需求不強時

    對有 AI 開發經驗的你,可以先:

    1. 在現有 Agent 系統上加一層簡單的 Plan schema(如本文示範)
    2. 讓模型輸出 plan,再由你自己的執行器跑
    3. 逐步增加:retry、compensation、observability

    這就是「程式碼化工具呼叫」在工程上的落地版本:不是只靠提示工程,而是用一個可執行、可觀測、可管控的計畫語言,把 LLM 變成真正的 workflow generator。


    🚀 你現在可以做的事

    • 在現有 Agent 專案中,加上一個最小可行的 Plan schema,讓 LLM 先輸出計畫再執行
    • 把現有工具封裝進 ToolRegistry 或類似結構,開始收集執行 log 以觀測計畫執行情況
    • 實作簡單的 plan validation 規則(最大深度/最大步數),並用一兩個實際業務 workflow 試跑驗證
  • 用 Docker 做一次性安全 AI Agent 沙盒

    用 Docker 做一次性安全 AI Agent 沙盒

    📌 本文重點

    • Agent 執行程式必須強制進 Docker 沙盒
    • 工具層限制不夠,需從執行環境畫界
    • 以最小權限、隔離與審計集中管理程式執行

    第一個痛點很直接:讓 Agent 能跑程式,又不讓它毀你主機、偷你資料或亂打外部服務。從 Rovo 被 PDF 隱藏指令牽著走,到 OpenClaw 為了搶健身房名額去「半駭半腳本」打 API,核心問題都是:你給了 Agent 工具權限,它就有能力放大任何輸入或目標的風險。本文的結論是實作面很務實:把「執行程式」這件事強制丟進一次性的 Docker 容器沙盒,搭配最小權限、網路/檔案隔離與審計,讓 Agent 成為受控服務,而不是在主機上為所欲為的黑盒。

    💡 關鍵: 把所有「程式執行」集中到一次性沙盒裡,是把 Agent 從高風險黑盒變成可控服務的核心做法


    重點說明

    1. 為什麼需要「一次性沙盒」而不是單純 API 限制

    • Rovo 的案例:攻擊者把指令藏在 PDF,Agent 幫忙從 Jira/Confluence 撈敏感資料,自動送到外部伺服器且不留操作痕跡。你就算限制 tool schema,還是擋不住「正當 API 被惡意使用」。
    • Gym hack 案例:OpenClaw 收到「幫我排到前面」這種模糊目標,就會自然探索網站邊界。只要你給它 HTTP client 或瀏覽器能力,沒有技術上的「這裡不能做」的牆。
    • 結論:工具層級的限制不夠,你必須在「執行環境」上畫界:這段 code 只能在隔離的容器裡跑;這個容器只能存取限定資料;超時就殺掉;所有輸出都被記錄與審核。

    💡 關鍵: 單靠限制工具參數無法阻止「正當 API 被惡用」,必須從執行環境切斷風險擴散路徑


    Docker 沙盒設計的核心原則

    1. 最小權限 + 只讀檔案系統

    • 使用非 root user、關掉不必要的 capabilities(CAP_NET_ADMIN 等)。
    • 根檔案系統 readonly,只有特定目錄(例如 /tmp/work)可寫,避免 Agent在容器內長期累積垃圾或做持久化攻擊。

    2. 網路與檔案系統隔離

    • 默認 無外網,只有明確允許的出口(例如企業 MCP / API gateway)。
    • 不掛宿主機目錄,尤其不要掛 /var/run/docker.sock,這是最常見的逃逸坑。
    • 針對多 Agent 系統,容器間一律不互通,避免 Agent 彼此側通道傳遞資料。

    3. 資源限制 + 超時

    • 用 --cpus、--memory、--pids-limit 等參數防止無限 fork/吃爆 RAM。
    • 在工具層加 硬超時(例如 5–30 秒), timeout 就 docker kill。

    4. 日誌與審計

    • 把 Agent 的 code、stdin、stdout、stderr 全部打包成事件,寫到集中式 log / SIEM。
    • 在多 Agent 架構中,透過 MCP / 工具網關,把「誰在什麼上下文下開了沙盒、跑了什麼」都留下 audit trail。

    多 Agent 系統裡的「動態沙盒策略」

    多 Agent coding 常見失敗點之一是:角色設計清楚,但行為邊界沒明確技術約束。建議是把沙盒視為一個「策略開關」:

    • 規劃幾種沙盒 profile:
    • analysis:只允許在容器內跑靜態分析工具,完全沒網路。
    • integration-test:允許打 staging 環境,有限 CPU/MEM。
    • prod-readonly:只能打只讀 API(例如查詢服務),禁止寫操作。

    • Controller Agent 不直接執行程式,而是呼叫一個 run_in_sandbox(profile, code) 工具,由工具決定 spawn 哪種 Docker 容器。

    • 所有 Agent 的「可執行能力」集中到這一個工具上,便於與企業現有 CI/CD、MCP、監控系統整合,把安全策略集中管理。

    💡 關鍵: 用多種沙盒 profile 對應不同 Agent 角色與任務,才能在安全與靈活之間做細緻權衡


    實作範例

    以下用 Python/Node 示範如何把 LLM 工具調用綁定到「在新容器內執行 code」。

    1. Docker 镜像設計

    這是一個極簡、偏安全的 Python 執行沙盒:

    # Dockerfile.sandbox
    FROM python:3.11-slim
    
    # 建立非 root 使用者
    RUN useradd -m sandbox && mkdir -p /app && chown -R sandbox:sandbox /app
    USER sandbox
    
    WORKDIR /app
    
    # 只安裝必要套件
    RUN pip install --no-cache-dir pytest requests
    
    # 預設為只讀根檔案系統;允許 /tmp/work 可寫(由 run script 控制)
    ENV PYTHONUNBUFFERED=1
    CMD ["python", "-u", "main.py"]
    

    注意:

    • 不要在這個鏡像裡放企業敏感設定檔或憑證。需要時改用 API gateway + 短期 token。
    • main.py 可以是一個固定的 runner,從環境變數或掛載目錄讀入待執行的 user code。

    2. Python:在新容器內執行 Agent 產生的程式碼

    假設你在後端定義了一個工具 run_code_in_sandbox 給 LLM 使用:

    import subprocess
    import tempfile
    import uuid
    from pathlib import Path
    
    SANDBOX_IMAGE = "my-org/agent-sandbox:latest"
    
    def run_code_in_sandbox(code: str, timeout_sec: int = 10) -> dict:
        # 為這次執行建立一次性工作目錄
        workdir = Path(tempfile.mkdtemp(prefix="agent-sandbox-"))
        script_path = workdir / "main.py"
        script_path.write_text(code, encoding="utf-8")
    
        container_name = f"agent-sandbox-{uuid.uuid4()}"
    
        cmd = [
            "docker", "run", "--rm",
            "--name", container_name,
            # 資源限制
            "--cpus", "0.5",          # 最多半顆 CPU
            "--memory", "512m",       # 限制記憶體
            "--pids-limit", "128",    # 限制子行程
            # 禁用網路:完全隔離
            "--network", "none",
            # 根檔案系統掛為 readonly
            "--read-only",
            # 掛載工作目錄到 /app,並提供 /tmp/work 可寫
            "-v", f"{workdir}:/app:ro",
            "-v", f"{workdir}/tmp:/tmp/work:rw",
            SANDBOX_IMAGE,
        ]
    
        try:
            proc = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=timeout_sec,
            )
        except subprocess.TimeoutExpired:
            # 超時直接 kill 容器
            subprocess.run(["docker", "kill", container_name], capture_output=True)
            return {"ok": False, "error": "timeout"}
    
        return {
            "ok": proc.returncode == 0,
            "stdout": proc.stdout,
            "stderr": proc.stderr,
            "exit_code": proc.returncode,
        }
    

    幾個關鍵點:

    • 沒有掛宿主機敏感目錄,也沒有掛 /var/run/docker.sock,避免 Agent 直接控制 Docker daemon。
    • --network none:這個 profile 完全不能出網。若要出網,請另外設計受控 network profile,例如只允許打企業 MCP gateway。
    • --read-only 搭配小範圍可寫目錄,避免 Agent 於容器內持久化惡意腳本。

    在你的 LLM tool schema 中,可以這樣暴露給模型:

    {
      "name": "run_code_in_sandbox",
      "description": "在隔離的 Docker 容器內執行短程式碼(無網路、有限資源)",
      "parameters": {
        "type": "object",
        "properties": {
          "language": {
            "type": "string",
            "enum": ["python"],
            "description": "目前只支援 Python"
          },
          "code": {
            "type": "string",
            "description": "要執行的程式碼,必須是單檔腳本"
          }
        },
        "required": ["language", "code"]
      }
    }
    

    3. Node.js:以 MCP / 工具網關方式整合

    如果你採用 MCP 或類似工具網關,建議把沙盒能力封裝成一個 service tool,而不是每個 Agent 都直接呼叫 Docker。

    // sandboxTool.ts
    import { execFile } from "child_process";
    import { promisify } from "util";
    const execFileAsync = promisify(execFile);
    
    export async function runInSandbox(params: {
      code: string;
      profile?: "analysis" | "integration-test";
    }) {
      const profile = params.profile ?? "analysis";
    
      const dockerArgs =
        profile === "integration-test"
          ? ["--cpus", "1", "--memory", "1g", "--network", "sandbox-staging"]
          : ["--cpus", "0.5", "--memory", "512m", "--network", "none"];
    
      const { stdout, stderr } = await execFileAsync("python", [
        "run_sandbox.py",
        JSON.stringify({ code: params.code, dockerArgs }),
      ], { timeout: 15000 });
    
      // 在這裡寫 audit log 到你的集中式監控
      // logSandboxEvent({ profile, codeSnippet: params.code.slice(0, 500), stdout, stderr })
    
      return { stdout, stderr };
    }
    

    在 MCP server 端,你只要把這個 runInSandbox 暴露為工具,並在工具 metadata 中標註:

    • scope:只允許特定角色的 Agent 使用(例如 CodeExecutor)。
    • audit:所有呼叫自動記錄到審計管線。

    這樣一來,企業的 Agent 系統就有一致的安全邊界:所有程式執行都要經過同一層沙盒服務,方便治理與合規。


    建議與注意事項

    1. 千萬不要掛宿主 Docker socket

    最常見也最危險的做法,是為了讓測試方便,直接在容器內掛:

    -v /var/run/docker.sock:/var/run/docker.sock
    

    這等同給了容器(也就是 Agent)對宿主機 Docker daemon 的完全控制權,能:

    • 啟動任意 privileged 容器;
    • 掛載宿主機任意路徑;
    • 讀取其他服務的環境變數與機密。

    結論:在 Agent 沙盒場景,禁止掛 docker.sock 是硬規則。 需要 orchestrate 容器時,請在宿主或受控 sidecar 上做,而不是讓 Agent 直接控制。

    2. 網路出口要明確設計,而不是「先開再說」

    • Rovo 被利用,就是因為 Agent默許能打外部網路,把敏感資料送走。
    • 建議預設 no egress,再逐步開放:
    • 只允許打企業 API gateway。
    • Gateway 再依使用者、任務、Agent role 做細粒度授權。

    避免一開始就給 Agent 完整的 requests / fetch 能力,卻沒有 outbound policy。

    3. 限制 fork、磁碟寫入與長時間運行

    • 使用 --pids-limit 防止 fork bomb。
    • 用 --read-only + 小範圍可寫目錄限制磁碟寫入,並定期清理一次性目錄。
    • 工具實作上務必加 硬超時,不可只依賴 Agent 自己判斷何時該結束。

    4. 在企業場景下與現有系統整合

    把 Agent 沙盒當成一個可以被納入現有治理架構的「服務」:

    • CI/CD:
    • 把沙盒鏡像視為一個版本化的 artifact,透過 pipeline 發布。
    • 變更權限、套件時要走同樣的審核流程。

    • MCP / 工具網關:

    • 透過 MCP 將沙盒工具集中管理,設定 scope(哪些 Agent 角色可以呼叫)、quota(每日執行次數)、審計策略。
    • 在多模型、多 Agent 環境中,統一用一套沙盒服務取代每個團隊自己寫的「隨便跑程式 API」。

    • 監控與審計:

    • 將每次沙盒執行事件(who / which agent / what code / which profile / result)送到 Log system / SIEM。
    • 在發現異常 pattern(例如大量嘗試未授權 API 操作)時,能快速追溯與調整策略。

    總結:Docker 沙盒的價值不只是「比較安全」,更是讓 Agent 行為可以被管理、可觀測、可審計。只要你把「執行程式」這件事全部強制走沙盒,並搭配最小權限與網路出口策略,你就能在保持開發效率的前提下,大幅降低 Rovo 類型的資料外洩與 OpenClaw 類型的「創意駭客」事件死角。


    🚀 你現在可以做的事

    • 在本機或測試環境建一個最小權限的 agent-sandbox Docker 鏡像,實驗一次性容器執行程式碼
    • 把現有 Agent 的「程式執行工具」改成呼叫 run_code_in_sandbox 類型服務,強制走沙盒
    • 檢查你的系統是否有容器掛載 /var/run/docker.sock 或無網路出口限制,列出並規劃修補清單
  • MCP 無狀態化後的企業 Agent 新架構

    MCP 無狀態化後的企業 Agent 新架構

    📌 本文重點

    • MCP 無狀態後,會話管理回到 Agent / Orchestrator
    • ACL、租戶隔離集中在 MCP Gateway 控制
    • Loop / graph / memory 拉升到框架層,工具端保持純 stateless

    MCP 改成無狀態(stateless)之後,一個很直接的好處是:你不必再在工具層維護「會話」。會話管理、記憶、ACL、租戶隔離,全部拉回到 Agent 平台或中間層控管。

    好處是:

    • 工具 server 更單純、可橫向擴展、可獨立部署與審核
    • 可觀察性與安全治理集中在一層,易於做 ADR 類型的觀察與基準測試
    • 任何「狀態」錯誤不會被藏在 MCP server 裡,而是明確暴露在 orchestration 層

    以下用一個典型架構:前端 Orchestrator + MCP Gateway + 多個工具 server,拆解你需要怎麼重構。


    重點說明:MCP 無狀態後的三個核心變更

    1. 會話被砍掉:改用 request-scoped metadata

    新版 MCP 刪除 session / conversation state,協議只管:

    • 一次呼叫的 工具名稱(tool)、參數(args)
    • 一組可選的 metadata / headers

    上下文與記憶不再由 MCP 持有,而是由:

    • Client / Orchestrator:維護任務 graph、loop、記憶
    • 中間層(MCP Gateway):附加租戶、風險標籤、追蹤 id
    • 工具端:只在必要時讀取 metadata,自己不產生「隱藏狀態」

    你要把原本塞在 MCP session 內的資訊,改寫成 每個 request 都帶的 metadata(如:tenant_id, agent_task_id, risk_level)。

    💡 關鍵: MCP 不再維護任何會話狀態,所有上下文都改成「每次請求顯式帶 metadata」,讓狀態集中在可治理的 Orchestrator 層。

    2. Stateless MCP 上做 ACL 與租戶隔離

    沒有 session,不代表不能做授權。反而更乾淨:

    • 每個 MCP request 都帶 caller identity(user / agent / tenant)
    • Gateway 根據 metadata 做 per-tool ACL,決定能不能 call 該工具
    • 工具 server 自身只 trust Gateway 轉給它的 identity,不自己管理 session

    典型做法是在 MCP 層定義:

    • x-tenant-id:租戶隔離
    • x-agent-id:是哪個 agent runtime/workflow
    • x-permissions:如 read:db,write:file,由 Gateway 檢查是否符合 policy

    3. Loop / Graph / Memory 抬升到框架層

    之前很多人把:

    • 迴圈控制(while loop)
    • 任務 graph / sub-agent 派工
    • 記憶(conversation history / RAG context)

    塞在 MCP server 裡,以為「工具端順便幫我記住上下文」。無狀態之後,你必須把這些搬到 Agent runtime / orchestration framework:

    • Loop:由 Orchestrator 控制是否繼續呼叫 MCP 工具
    • Graph:用 DAG / state machine(例如:Prefect、Temporal、自建 FSM)管理多步驟流程
    • Memory:由獨立記憶服務(Vector DB / KV store)持有,MCP 工具只接收明確的 context(如 docs_chunk_ids)

    這對專案的實際好處是:所有高風險邏輯集中在可治理的層,可以對 loop 數量、工具調用頻率、記憶寫入/讀取做 FinOps、審計與基準測試,而不是到處分散在工具端。

    💡 關鍵: Loop、graph 與 memory 全部拉回 Orchestrator,才能精準控管成本、風險與審計路徑。


    實作範例:用 request metadata 重建「會話」與治理

    以下用一個簡化範例示範:

    • 前端 Orchestrator(可能是自家 Agent runtime)
    • MCP Gateway(實際對接工具 server)
    • 多個 stateless MCP 工具

    1. Orchestrator:每次工具呼叫都帶 context_id / tenant

    # orchestrator.py
    import uuid
    from mcp_client import MCPClient  # 假設有這個 client SDK
    
    client = MCPClient(base_url="https://mcp-gateway.internal")
    
    async def run_agent_task(user_id: str, tenant_id: str, task_input: str):
        task_id = str(uuid.uuid4())
    
        # 這裡的 memory / graph 在 orchestrator 層
        memory_context = load_memory_for_user(user_id)
        workflow_state = init_workflow_state(task_input, memory_context)
    
        while not workflow_state.done:
            tool_request = workflow_state.next_tool_call()
    
            response = await client.call_tool(
                tool_name=tool_request.name,
                args=tool_request.args,
                headers={
                    "x-tenant-id": tenant_id,
                    "x-user-id": user_id,
                    "x-agent-task-id": task_id,
                    "x-workflow-step": str(workflow_state.step),
                    "x-risk-level": "normal",
                },
            )
    
            workflow_state = workflow_state.apply_tool_result(response)
            persist_step_log(task_id, workflow_state, response)
    
        save_memory_for_user(user_id, workflow_state.memory_delta)
        return workflow_state.final_output
    

    重點:

    • 沒有 session id,只有 task_id + step,所有狀態在 orchestrator 層
    • 每個 request 都有完整的治理 metadata,可給 ADR 類工具做觀察與威脅檢測

    💡 關鍵: 用 task_id + step 取代 session,把整個任務完整還原成可審計的步驟序列。

    2. MCP Gateway:per-tool ACL + 租戶隔離

    // mcp-gateway.ts (Node/TypeScript pseudo-code)
    import { verifyToken, checkAclPolicy } from "./auth";
    import { routeToToolServer } from "./router";
    
    async function handleMcpRequest(req, res) {
      const { toolName, args } = req.body;
      const headers = req.headers;
    
      const token = headers["authorization"];
      const identity = await verifyToken(token); // 解析 user/agent/tenant
    
      const tenantId = headers["x-tenant-id"] ?? identity.tenantId;
      const agentTaskId = headers["x-agent-task-id"];
    
      // ACL 檢查:哪些工具可被哪個 identity 使用
      const allowed = await checkAclPolicy({
        identity,
        tenantId,
        toolName,
      });
      if (!allowed) {
        return res.status(403).json({ error: "tool_not_allowed" });
      }
    
      // 建立統一的 observability context
      const observabilityContext = {
        tenantId,
        userId: identity.userId,
        agentId: identity.agentId,
        agentTaskId,
        toolName,
        timestamp: Date.now(),
        requestId: generateRequestId(),
      };
    
      logRequest(observabilityContext, args); // 提供給 ADR / SIEM
    
      const toolResponse = await routeToToolServer(toolName, args, {
        "x-tenant-id": tenantId,
        "x-agent-task-id": agentTaskId,
        "x-request-id": observabilityContext.requestId,
      });
    
      logResponse(observabilityContext, toolResponse);
      return res.json(toolResponse);
    }
    

    重點:

    • ACL 與租戶隔離在 Gateway 層,而不是工具 server 內部
    • observability context(requestId, agentTaskId, toolName)可以直接餵給 ADR 類似的威脅檢測/基準測試

    3. MCP 工具 server:純 stateless,禁止偷塞 state

    # tools/db_query_server.py
    from fastapi import FastAPI, Header
    
    app = FastAPI()
    
    @app.post("/tools/db_query")
    async def db_query(query: str, x_tenant_id: str = Header(...), x_agent_task_id: str = Header(None)):
        # 僅用 tenant_id 做資料邊界控制,不維護會話
        if not is_tenant_allowed_to_query(x_tenant_id, query):
            return {"error": "tenant_query_not_allowed"}
    
        result = execute_query_for_tenant(query, x_tenant_id)
    
        # 嚴禁在 server 內部開 global session dict
        return {
            "rows": result,
            "meta": {
                "agent_task_id": x_agent_task_id,
            },
        }
    

    重點:

    • 工具 server 僅依賴 headers 決定授權與資料範圍
    • 不建議在工具 server 開 global cache 作為「會話記憶」,避免破壞 stateless 模型與可觀察性

    建議與注意事項:遷移時常見坑與最佳實踐

    1. Server 假設有持久連線 / session

    舊版 MCP 或自家協定常常:

    • 用 WebSocket / 長連線維護 session
    • 把 user state 放在 in-memory dict(例如 sessions[user_id])

    在改成 stateless MCP 時:

    • 所有 state 都要變成可持久化的 store(DB / Redis / Vector DB),由 Orchestrator 控制
    • 工具端只讀取 request headers,不再假設「同一連線就是同一會話」

    實務建議:

    • 先畫出「哪些資料被當作 session state 使用」
    • 將它們搬到 明確的 Memory API 上(例如:load_memory(user_id) / save_memory(user_id, delta))

    2. 工具端偷塞 state,導致 log 無法重建任務路徑

    常見情境:

    • 工具 server 依賴 local cache / global dict 記住「上一次呼叫結果」
    • log 只有 request/response,但沒記錄「為什麼 agent 會做出這個決策」

    在 MCP 無狀態模型下,要做到可觀察性與安全基準測試(ADR 類工具),你需要:

    • 每個 step 的決策都由 Orchestrator log(含 prompt, context, tool call)
    • 工具 response 只是一個 pure function 輸出,不混入隱藏狀態

    實務建議:

    • 強制所有工具 server 通過 MCP Gateway,Gateway 追加 x-request-id,並集中 log
    • 在 Orchestrator 保存 完整任務 graph,例如:
    {
      "agent_task_id": "task-123",
      "steps": [
        {"step": 1, "tool": "search_docs", "request_id": "r-1"},
        {"step": 2, "tool": "db_query", "request_id": "r-2"}
      ]
    }
    

    有了這些資料,像 Uber 的 ADR 就可以做:

    • 單一任務的威脅路徑分析(哪一步嘗試讀敏感資料)
    • 持續的安全基準測試(某種輸入是否總是觸發高風險工具)

    3. Loop / Graph / Memory 不要塞在 MCP 裡

    受「while loop 就是 agent」的影響,很多人以前:

    • 在 MCP server 裡面直接實作 while loop 反覆 call LLM
    • 把 graph / workflow 寫在工具端 code 裡

    這在無狀態更新後會變成技術負債:

    • 難以在平台層做 FinOps(每個 loop 要花多少 token / 工具成本)
    • 難以對 loop 數量與深度做 安全限制(避免無限迴圈或風險疊加)

    最佳實踐:

    • Loop:在 Orchestrator(或專門的 agent runtime)層,用明確的限制,如 max_steps, max_cost
    • Graph:用可視化、可配置的 workflow 定義(YAML / JSON / DSL),不寫死在 MCP 工具程式碼裡
    • Memory:獨立成一個工具或服務(如 memory.read, memory.write),由 ACL 管控誰能讀寫

    4. 安全與治理:利用 stateless 做更強的控制平面

    MCP 無狀態其實大幅簡化了 企業級治理:

    • 所有工具呼叫都經過同一層 Gateway,可以:
    • 限制每個 agent / tenant 的 工具配額與成本
    • 對特定工具設高風險標籤,必須經過額外審核或輸入過濾
    • Observability context(tenantId, agentTaskId, toolName, requestId)可以直接餵給:
    • ADR / SIEM / 自家監控平台,做威脅檢測與基準測試

    實務上,你可以定義一個簡單的 治理 schema:

    # governance.yaml
    agents:
      finance_report_agent:
        max_tool_calls: 50
        allowed_tools:
          - db_read_only
          - email_notify
        risk_budget: medium
    
    tools:
      db_read_only:
        risk_level: high
        requires_human_review: true
    

    然後在 Orchestrator + MCP Gateway 共同實作:

    • 在 loop 前檢查 max_tool_calls
    • Gateway 看到 risk_level: high 時,會把這些 call 寫入特別的安全 log,供 ADR 分析

    結論

    MCP 無狀態化的關鍵結論:

    • 會話狀態不再是協議責任,而是 Agent 平台責任
    • 上下文與記憶要由 client / orchestration / memory 工具明確持有
    • stateless 帶來更乾淨的可觀察性與安全治理模型,能與 ADR 這類工具自然整合

    如果你的企業 Agent 平台還在仰賴 MCP session 或工具端隱藏 state,現在是重構的好時機:把 loop、graph、memory、ACL、租戶隔離全部拉回到框架層,留給 MCP 的只有乾淨、可審核、可擴展的工具呼叫。

    🚀 你現在可以做的事

    • 審視現有 MCP / 工具 server,列出所有依賴 session 或 in-memory state 的地方,規劃改成 metadata + Orchestrator state
    • 為 MCP Gateway 增加 x-tenant-id、x-agent-task-id、x-request-id 等 headers,並接入現有的 ADR / SIEM / 監控系統
    • 設計一份 governance.yaml 或類似設定檔,為主要 agent 定義 max_tool_calls、allowed_tools 與 risk_budget,並在 Orchestrator 中強制執行
  • Qwen3.8-Max 長任務實戰與部署攻略

    Qwen3.8-Max 長任務實戰與部署攻略

    📌 本文重點

    • 長任務要以任務級 state 而非單次 context 設計
    • 混合雲端超大模型與本地 27B 降本增效
    • 透過分層記憶、快照與觀察性實現可恢復長任務

    阿里把 Qwen3.8-Max (2.4T) 拉到開源權重,加上 Qwen3.8-27B 這種「中杯」模型,對工程團隊最大意義是:第一次可以在自己掌控的環境裡,穩定做「跑幾天、不斷線、能恢復」的長任務——重現論文、長鏈路 refactor、甚至晶片設計流程,而不是被單次 context 長度與單次 API call 綁死。

    💡 關鍵: 開源 2.4T 級模型 + 27B 中杯,使「跨天長任務」首次在自控環境中變得可行且可恢復。

    這篇從工程視角拆三件事:長任務架構設計、部署選型與多模型搭配、企業級觀察性與成本防護欄,最後用一個「重現論文實驗 / 多階段 refactor」的管線當具體範例。


    重點說明:長任務超大模型的工程切面

    1. 長任務/長上下文的核心設計要點

    Qwen3.8-Max、Kimi K3、DeepSeek V4 Flash 這類模型都強調能處理「多階段、跨天」任務。工程上關鍵不是單次 context 有多長,而是:

    1. Checkpoint 分段:
    2. 對長任務維持 任務級 state,而非完全依賴模型的上下文。
    3. 每一階段輸出轉成 結構化快照(JSON、向量庫),用來恢復 /續跑,而不是重餵全部原始對話。

    4. 任務分解 + 多階段推理管線:

    5. 超大模型負責:規劃 + 難推理 + 棘手 code review。
    6. 中型模型(例如 Qwen3.8-27B)負責:常規 summarization、log 解讀、重複模式生成。
    7. Pipeline 以「任務節點 (TaskNode)」為最小單位,每個節點可重跑,可掛 cache。

    8. 長鏈路記憶:分層設計:

    9. 熱記憶:當前階段所需的 1–2k 關鍵 tokens,直接進 context。
    10. 冷記憶:向量索引(RAG)、中間結果快照。
    11. 超大模型變成「查詢 +推理」引擎,而不是所有歷史都塞進它的 prompt。

    2. 部署選項:雲 API vs 自架 GPU / 集群

    現在 Qwen3.8-Max 提供 付費 API,權重預計開源;Qwen3.8-27B 已確認可在 約 17GB VRAM 上跑(Daniel Han 的實測)。工程決策可以粗略這樣切:

    💡 關鍵: 約 17GB VRAM 即可跑 27B,使中小團隊能低成本自架中型模型配合雲端超大模型。

    雲 API(Max / Kimi / DeepSeek)適合:

    • 須最高模型能力、推理品質優先。
    • 任務量不大但單次任務超長、需要穩定長鏈路追蹤。
    • 不想維護推理集群、只做應用層(產品團隊常見)。

    自架 GPU / 集群(27B / 7B)適合:

    • 高吞吐、預算敏感,能接受稍弱能力換大量併發。
    • 有 On-prem 合規需求(金融、醫療資料不能出域)。
    • 想做定制微調、系統 prompt、工具集成深度控制。

    典型架構是 混合多模型:

    • 雲端 Qwen3.8-Max / DeepSeek V4 Flash:只用在「規劃 /關鍵判斷 /難度最高的 code reasoning」節點。
    • 本地 Qwen3.8-27B:跑 routine summarization、日常 RAG query、內部工具代理。

    3. 與 DeepSeek / Kimi 的差異與 Trade-off

    從現有 benchmark 和社群評價來看:

    • 能力:Qwen3.8-Max 在 coding /軟體任務上略優,整體接近 Kimi K3 / DeepSeek V4 Flash。
    • 推理延遲:超大模型延遲本來就高,雲 API 通常會有 長任務模式(寬限 timeout + 狀態追蹤)。自架時則要自己處理超長推理的 timeout /重試。
    • 記憶管理:Kimi / DeepSeek 已內建較成熟的長記憶機制(server 端 RAG + 任務追蹤);Qwen 開源權重的優勢是:你可以自己決定 記憶層級與格式,不被封閉系統限制。
    • 成本模式:
    • Qwen3.8-Max API 標價示例:Input \$2 /M tokens、Output \$6 /M tokens(依 Reddit 貼文),長任務要小心爆成本。
    • 自架 27B:一次性硬體 + 電費,長期大量任務通常更便宜,但需要 DevOps 能力。

    💡 關鍵: Input \$2 /M、Output \$6 /M tokens 的定價,逼迫架構上用「Max 做關鍵推理 + 27B 處理日常」來控制長任務成本。


    實作範例:以「重現一篇論文實驗 / 多階段 codebase refactor」為例

    以下是一個簡化的長任務管線,支援:

    • 論文解析 → 實驗設計 → Code 生成 → 結果分析
    • 隨時 resume,有 觀察性 (logging) 與 成本防護欄。

    1. 任務管線與分層記憶設計

    先定義任務節點與狀態儲存:

    # pseudo-code: 任務節點 / pipeline 定義
    class TaskNode:
        def __init__(self, name, model_role, handler):
            self.name = name              # e.g. "paper_analysis"
            self.model_role = model_role  # "max" or "27b"
            self.handler = handler        # 可重跑的邏輯
    
    class LongTaskState:
        def __init__(self, task_id):
            self.task_id = task_id
            self.snapshots = {}   # {node_name: snapshot_json}
            self.logs = []        # for observability
    
        def save_snapshot(self, node_name, data):
            self.snapshots[node_name] = data
            # 實際上會寫入 DB 或 object storage
    
        def load_snapshot(self, node_name):
            return self.snapshots.get(node_name)
    
        def log(self, level, msg, meta=None):
            self.logs.append({"level": level, "msg": msg, "meta": meta})
    

    接著建立「分層記憶」:把論文內容、codebase 索引到向量庫,用 RAG 控制熱 /冷記憶:

    # pseudo-code: 分層記憶 (RAG + 快照)
    class MemoryLayer:
        def __init__(self, vector_store, snapshot_store):
            self.vector_store = vector_store
            self.snapshot_store = snapshot_store
    
        def query_paper(self, question, top_k=5):
            return self.vector_store.search("paper", question, top_k)
    
        def query_codebase(self, question, top_k=10):
            return self.vector_store.search("code", question, top_k)
    
        def load_intermediate(self, task_id, node_name):
            return self.snapshot_store.get(task_id, node_name)
    
        def save_intermediate(self, task_id, node_name, data):
            self.snapshot_store.put(task_id, node_name, data)
    

    2. 多模型推理管線:Qwen3.8-Max + Qwen3.8-27B

    假設你有:

    • max_client:雲端 Qwen3.8-Max API 客戶端。
    • local27_client:本地部署 Qwen3.8-27B 客戶端(例如 vLLM / llama.cpp)。
    # pseudo-code: 選模型 + 成本防護欄
    MAX_INPUT_LIMIT = 200_000   # 依你的預算與 API 限制調整
    
    def call_model(model_role, prompt, max_output_tokens=4096):
        if model_role == "max":
            if len(prompt) > MAX_INPUT_LIMIT:
                raise ValueError("input tokens exceed MAX_INPUT_LIMIT")
            return max_client.generate(
                model="qwen-3.8-max",
                input=prompt,
                max_output_tokens=max_output_tokens,
                # 可以加上 temperature, top_p 等
            )
        else:
            return local27_client.generate(
                model="qwen-3.8-27b",
                input=prompt,
                max_output_tokens=max_output_tokens,
            )
    

    以下是三個節點示意:

    1. 論文解析(Max):任務規劃 + 實驗拆解。
    2. codebase 分析(27B + RAG):找出需 refactor 的模組。
    3. 關鍵 refactor 方案(Max):高階設計 + 風險分析。
    # 節點 1:論文解析
    
    def handle_paper_analysis(state: LongTaskState, memory: MemoryLayer, paper_text: str):
        ctx = state.load_snapshot("paper_analysis")
        if ctx:  # 支援 resume
            return ctx
    
        prompt = f"""
    你是一名資深研究工程師,請從以下論文中抽取:
    1. 實驗設定 (dataset, metrics, hyperparams)
    2. 關鍵貢獻
    3. 可能的工程落地風險
    
    輸出 JSON,schema:
    {{
      "experiments": [{{"name": str, "config": dict}}],
      "contributions": [str],
      "risks": [str]
    }}
    
    論文內容:
    {paper_text}
    """
        resp = call_model("max", prompt, max_output_tokens=8192)
        result = json.loads(resp["output_text"])  # 需加 error handling
        state.save_snapshot("paper_analysis", result)
        return result
    
    # 節點 2:codebase 分析 (RAG + 27B)
    
    def handle_code_analysis(state: LongTaskState, memory: MemoryLayer, question: str):
        cached = state.load_snapshot("code_analysis")
        if cached:
            return cached
    
        docs = memory.query_codebase(question, top_k=20)
        context = "\n\n".join(d["content"] for d in docs)
    
        prompt = f"""
    你是資深後端工程師,根據以下 code 片段,找出需要改動的模組與檔案路徑,並說明原因。
    
    [相關程式碼摘錄]
    {context}
    
    問題:{question}
    
    請輸出為 JSON,schema:
    {{
      "modules": [{{"path": str, "reason": str}}]
    }}
    """
        resp = call_model("27b", prompt, max_output_tokens=4096)
        result = json.loads(resp["output_text"])  # 需加 error handling
        state.save_snapshot("code_analysis", result)
        return result
    
    # 節點 3:關鍵 refactor 方案 (Max)
    
    def handle_refactor_plan(state: LongTaskState, memory: MemoryLayer):
        plan = state.load_snapshot("refactor_plan")
        if plan:
            return plan
    
        paper_info = state.load_snapshot("paper_analysis")
        code_info = state.load_snapshot("code_analysis")
    
        prompt = f"""
    你是一名首席軟體架構師。根據以下資訊設計一個分階段 refactor 方案,要求:
    - 每階段可獨立部署
    - 每階段都有 rollback 計畫
    - 清楚列出對實驗結果重現的影響
    
    [論文實驗摘要]
    {json.dumps(paper_info, ensure_ascii=False)}
    
    [需要改動的模組]
    {json.dumps(code_info, ensure_ascii=False)}
    
    請輸出為 JSON,schema:
    {{
      "phases": [{{"id": int, "description": str, "files": [str], "risk": [str], "rollback": [str]}}]
    }}
    """
        resp = call_model("max", prompt, max_output_tokens=8192)
        result = json.loads(resp["output_text"])  # 需加 error handling
        state.save_snapshot("refactor_plan", result)
        return result
    

    3. 企業環境下的觀察性、任務恢復與成本控制

    在企業環境跑長任務,以下三件事必做:

    1. 觀察性 (logging & metrics):

    2. 每個 TaskNode 紀錄:開始 /結束時間、使用模型、input_tokens / output_tokens 數、錯誤碼。

    3. 放進集中式 log (如 ELK、ClickHouse),可追蹤某次 pipeline 的 token 成本。

    4. 任務恢復 (resume):

    5. 每個節點輸出都用 state.save_snapshot,並寫入 DB / object storage。

    6. entrypoint 支援 resume_from=node_name,方便從中間節點重跑,而不是從頭吞全部 context。

    7. 成本防護欄:

    8. 全局 quota:一天內對 Qwen3.8-Max 的 input /output tokens 上限,超過就 fallback 到 27B 或排隊。

    9. batch 策略:大量相似小任務(例如 log summary)批次送給 27B,本地處理;只把「需要人決策」的部分交給 Max。

    建議與注意事項:常見坑與最佳實踐

    1. context 爆炸:

    2. 不要把整篇論文 + 整個 codebase 都塞進 prompt,即使模型支持 1M tokens。

    3. 用 RAG +中間結果快照:先用 27B 做初步萃取,再用 Max 做重推理,context 控制在幾千 tokens 內。

    4. token 成本失控:

    5. 對 Qwen3.8-Max 這種 \$2/\$6 per M tokens 的模型:

      • 所有節點都要紀錄 input_tokens / output_tokens,計算 pipeline 單價。
      • 高頻任務優先跑本地 27B,只在需要「deep reasoning」時升級到 Max。
    6. 長鏈路 state 不一致:

    7. 不要把「模型上次說的東西」當唯一真相。所有關鍵中間結果都要轉成 明確 schema (JSON)。

    8. 上游節點輸出要做 schema validation(可以用 Pydantic)與版本控管,避免後續節點因格式變動直接崩壞。

    9. 多模型切換的延遲 /錯誤處理:

    10. 對雲端 Max:加 重試與退避機制,長任務時避免整個 pipeline 因一個 500 error 失敗。

    11. 設計好 fallback 策略:Max 超時時,先用 27B 生成較粗的答案,讓任務不中斷,事後再補。

    12. 自架 27B 的 VRAM 預算:

    13. Qwen3.8-27B 在約 17GB VRAM 可跑是利好,但那是經過量化 /優化後的情境。

    14. 生產環境請預留額外 VRAM 給:並發請求、KV cache、RAG embedding 模型;單張 24GB 以上 GPU 會比較安全。

    結論:如果你手上有需要「跨天、跨多階段、多次嘗試」的長任務(論文重現、超大型 refactor、風控模組設計),現在可以用 Qwen3.8-Max + Qwen3.8-27B + 分層記憶 (RAG+快照) 搭起一個真正工程化可恢復的管線,把超大模型從「一次性助手」變成「可監控、可控成本的長程代理」。

    🚀 你現在可以做的事

    • 在自家 codebase 中實作文中的 TaskNode 與 LongTaskState,搭起最小可用長任務管線
    • 部署一個本地 Qwen3.8-27B(如用 vLLM 或 llama.cpp),並接上雲端 Qwen3.8-Max 做多模型實驗
    • 為現有 LLM 任務加入 token 計費 log 與 MAX_INPUT_LIMIT 檢查,建立成本防護欄與 resume_from 能力
  • Cloudflare 邊緣 TCP/gRPC 打造 AI 推理網路

    Cloudflare 邊緣 TCP/gRPC 打造 AI 推理網路

    📌 本文重點

    • Cloudflare 邊緣現支援原生入站 TCP/gRPC
    • 可在邊緣直接跑 gRPC 推理、RAG 與微服務
    • 相較 K8s,全球互動式 AI 延遲與成本更優
    • 可作為既有 gRPC/K8s 架構的前置 Edge layer

    Cloudflare 開放 Workers/Containers 入站 TCP + gRPC,直接解掉幾個很實際的痛點:AI 推理只能走 HTTP/JSON、無法沿用現有 gRPC 微服務、邊緣節點難以做長連線與串流推理。現在你可以在 Cloudflare 邊緣跑 原生 gRPC 推理服務、RAG 檢索節點與微服務 Mesh,延遲更低、成本更像 CDN,而不是像傳統 K8s 叢集那樣重。


    重點說明

    1. 邊緣執行模型:Workers vs Containers

    Cloudflare 現在有兩種主要邊緣運行模式:

    • Workers:類似 serverless JavaScript/TypeScript 執行環境
    • 適合:輕量 API 轉接、認證、路由、簡單推理管線
    • 過去:只支援 HTTP/S,你必須把 gRPC 轉成 HTTP JSON,再打到後端
    • 現在:支援 入站 TCP/gRPC,可直接在邊緣 terminate gRPC

    • Containers(Cloudflare Workers AI / Cloudflare Containers):執行打包好的映像

    • 適合:完整推理服務、向量檢索、Python/Go gRPC server
    • 行為更接近傳統 Pod,但由 Cloudflare 在全球 PoP 管理

    這次更新的核心是:兩者都能收 TCP/gRPC 流量,你可以在邊緣直接跑 grpc-go 或 grpc-python 的 server,不必再把邊緣當純 HTTP 反向代理。

    💡 關鍵: Workers 與 Containers 現在都能直接處理入站 TCP/gRPC,等於在 Cloudflare 邊緣即可部署完整 gRPC 微服務與 AI 推理節點。

    2. 入站 TCP/gRPC 開放後能做什麼?

    從開發者視角,幾個立刻有用的場景:

    • AI 推理服務在邊緣:
    • 在每個區域部署輕量 gRPC 推理節點,靠 Anycast + 邊緣路由 讓使用者打到最近 PoP
    • 使用串流 gRPC (Server streaming) 傳回 token,體感延遲大幅下降

    • RAG 檢索節點:

    • 在邊緣跑向量查詢服務(Faiss/pgvector/自家引擎),gRPC 介面
    • 利用 Cloudflare 的 colo 分布,把索引按地理或租戶分片

    • 沿用既有 gRPC 微服務:

    • 以前你要在前面加 API Gateway/Ingress,把 gRPC 轉 HTTP 或 terminate TLS
    • 現在可以在邊緣直接 terminate gRPC,後面用 TCP/gRPC 轉發到 origin 或在邊緣直接處理

    實際好處:

    • 延遲:請求在使用者附近的 PoP 就完成認證、路由,甚至整個推理
    • 成本模型:按使用計費、無需維護多個 K8s cluster,只維護映像與 Workers code
    • 協議相容性:不用再為「Cloudflare 只懂 HTTP」寫一層轉換邏輯

    💡 關鍵: 對互動式 AI 與全球用戶服務,邊緣 gRPC 推理能在不改協議的前提下顯著降低體感延遲與基礎設施維運成本。

    3. 與 Kubernetes + Ingress 的比較

    假設你現在有一套標準架構:Cloud LB → Ingress (Envoy/NGINX) → gRPC microservices / AI inference pods。

    用 Cloudflare 邊緣替代部分角色時,可以這樣看:

    • 延遲:
    • K8s:流量通常集中在幾個 region(us-east, ap-southeast),跨區通訊不可避免
    • Cloudflare:請求先到最近 PoP,如果推理在邊緣完成,RTT 直接以「使用者到 PoP」為主

    • 成本:

    • K8s:你要付 cluster 固定成本(control plane + node),再加 LB、egress 等
    • Cloudflare:付 Workers/Containers 執行 + 帶寬,沒有 cluster idle 成本,對 burst 型推理更友善

    • 可觀測性:

    • K8s:你掌握 Pod metric、sidecar tracing,能非常細緻
    • Cloudflare:以 request-level logs/metrics 為主,容器內的應用層 metric 要自己上報(例如推到 Prometheus/Grafana Cloud)

    結論:

    • 對 全球服務、互動式 AI(chat、copilot),邊緣方案延遲優勢很明顯
    • 對 重度內網 microservices mesh,K8s 還是比較好管理細粒度通訊與觀測

    💡 關鍵: 邊緣 gRPC 更適合面向外部用戶的全球互動式 AI;內部複雜微服務 Mesh 則仍以 K8s 為主,再用 Cloudflare 當前置入口。


    實作範例

    以下用簡化的範例示意:在 Cloudflare 邊緣部署一個 gRPC AI 推理 API,前面 Workers 做零信任與路由,後面 Container 跑真正的 gRPC server。

    1. Terraform:宣告 TCP/gRPC 入口與 Workers Route

    # cloudflare.tf
    provider "cloudflare" {
      api_token = var.cloudflare_api_token
    }
    
    resource "cloudflare_account" "this" {
      name = "my-ai-edge-account"
    }
    
    # 建立域名與 gRPC 入口
    resource "cloudflare_zone" "ai_api" {
      name = "ai.example.com"
    }
    
    # 開啟 gRPC(HTTP/2 + TCP)入口
    resource "cloudflare_worker_route" "grpc_route" {
      zone_id     = cloudflare_zone.ai_api.id
      pattern     = "ai.example.com/*"
      script      = cloudflare_worker_script.grpc_edge.name
    }
    
    resource "cloudflare_worker_script" "grpc_edge" {
      name    = "grpc-edge-worker"
      content = file("./dist/grpc-edge-worker.js")
    }
    

    這裡的關鍵是 route 指到 Workers script,Cloudflare 會依協議分流;當客戶端用 gRPC over TLS 打 ai.example.com,會由對應的 PoP 進入 Worker。

    2. Workers:在邊緣處理 TCP/gRPC 連線與零信任

    Cloudflare 新增了類似 TCP sockets API 的能力(以下為概念化程式碼,實際 API 名稱以官方為準):

    // grpc-edge-worker.js
    export default {
      async fetch(request, env, ctx) {
        // HTTP 入口:可用來做健康檢查、簡單 REST 包裝
        return new Response("OK", { status: 200 });
      },
    
      async tcp(conn, env, ctx) {
        // conn: 表示入站 TCP 連線(含 TLS 已由 Cloudflare terminate)
    
        // 1. 零信任:檢查 mTLS 或 JWT(假設從 Cloudflare Access 傳入 header)
        const identity = await verifyZeroTrust(conn, env);
        if (!identity.valid) {
          await conn.close();
          return;
        }
    
        // 2. 將 TCP 流量轉發到邊緣 Container 中的 gRPC server
        const backendConn = await env.AI_GRPC_ORIGIN.connect({
          host: "ai-grpc.internal", // 邊緣 Container 的內部 host
          port: 50051,
        });
    
        // 3. 做簡單的連線鏡射(pipe)
        conn.pipeTo(backendConn.writable, { preventClose: false });
        backendConn.readable.pipeTo(conn.writable, { preventClose: false });
      },
    };
    
    async function verifyZeroTrust(conn, env) {
      // 範例:檢查 Cloudflare Access JWT 或 mTLS client cert
      // 回傳 { valid: true/false }
      return { valid: true };
    }
    

    重點:

    • tcp(conn, env, ctx) 代表邊緣上對入站 TCP 的 handler
    • 你可以在這裡做 零信任驗證、租戶路由、限流,再把流量交給後面的 gRPC server
    • 後面的 server 可以跑在 Cloudflare Containers,也可以指到你自己的 origin(例如 GCP/K8s)

    3. Container:gRPC AI 推理服務

    假設你在 Cloudflare Container 裡跑一個簡化的 Python gRPC server:

    # inference_server.py
    import grpc
    from concurrent import futures
    from proto import inference_pb2, inference_pb2_grpc
    
    class InferenceServicer(inference_pb2_grpc.InferenceServiceServicer):
        def StreamChat(self, request, context):
            # request.prompt, request.user_id
            for token in run_llm_stream(request.prompt):
                yield inference_pb2.ChatToken(token=token)
    
        def Embed(self, request, context):
            vec = embed_text(request.text)
            return inference_pb2.Embedding(vector=vec)
    
    def serve():
        server = grpc.server(futures.ThreadPoolExecutor(max_workers=8))
        inference_pb2_grpc.add_InferenceServiceServicer_to_server(
            InferenceServicer(), server
        )
        server.add_insecure_port("[::]:50051")
        server.start()
        server.wait_for_termination()
    
    if __name__ == "__main__":
        serve()
    

    你只需要把這個映像部署在 Cloudflare Containers,並與前面的 Worker 透過內部 TCP 相接。


    建議與注意事項

    1. TLS / 零信任設計

    • 對外:讓 Cloudflare 統一做 TLS 終結(TLS termination),使用 Cloudflare Access/ZT 控制身份
    • 對內:邊緣 Worker 與 Container/Origin 之間建議使用 mTLS,防止橫向移動
    • 若你已有 API Gateway(Kong/Envoy),可把 Cloudflare 當 前置 Edge layer,只保留 Gateway 在核心 region 接收來自邊緣的 gRPC

    2. 連線壽命與狀態管理

    • gRPC streaming 本質是長連線,要注意:
    • Worker 執行時間限制:確保 Cloudflare 的執行模型允許長時間 pipe,必要時把邏輯放到 Container
    • 斷線重試:在客戶端設計 token-based resume(例如 cursor 或 offset),避免整個對話因 PoP 切換中斷

    • 状態建議:

    • 把 對話狀態 / session 放在外部儲存(KV、Durable Objects、Redis),不要綁在單一 PoP

    3. Origin 的擴展策略

    • 若推理服務仍在自家 K8s:
    • 用 Cloudflare 邊緣做 全局入口與零信任,後面依租戶或地域分流到不同 cluster
    • 注意 跨 region 帶寬成本:邊緣到 origin 的流量可能集中在幾個 region

    • 若全面搬到 Cloudflare Containers:

    • 建議 按地區部署多個副本,配合 Workers 做地理路由
    • 用外部 observability stack(例如 OpenTelemetry + Tempo/Loki)把邊緣 metric 拉回統一平台

    4. 常見坑與避雷

    • 連線數上限:
    • 高併發 gRPC streaming 會吃 connection slot,要確認 Cloudflare 帳號/方案的限制
    • 建議在 Workers 做 client-side throttling 或 queue,避免瞬間打爆 Container

    • 冷啟動:

    • Workers 冷啟動通常比 Lambda 類服務快,但 Container 啟動仍有延遲
    • 對延遲敏感的推理 API:

      • 預熱核心路由(例如每 PoP 保留最常用模型的 warm instance)
      • 使用簡單的 health probe 定期打模型,維持容器活躍
    • 與現有 API Gateway 串接:

    • 遷移策略:先把 外部客戶端改打 Cloudflare 邊緣 gRPC 入口,由 Workers 轉發到原 API Gateway
    • 等確定運作穩定,再逐步把認證、限流移到邊緣

    5. 一個可落地的「邊緣 AI API」樣板

    總結一個可直接採用的樣板架構:

    1. Domain & TLS:ai.example.com 由 Cloudflare 管理,開啟 TLS、gRPC 支援
    2. Edge Worker:
    3. 實作 tcp() handler
    4. 做零信任/租戶識別/流量路由(ex: enterprise vs free)
    5. Edge Containers:
    6. 多區部署 AI 推理 gRPC server + RAG 檢索
    7. 共享外部向量庫或依地區分片
    8. Observability:
    9. 邊緣 logs/metrics 推到集中式平台
    10. gRPC tracing 使用 OpenTelemetry,將 trace context 從 Worker 轉接到 Container
    11. 漸進式遷移:
    12. 先讓 5–10% 流量經邊緣 gRPC 入口,觀察延遲與錯誤率
    13. 穩定後再逐步將 region LB 退場,讓 Cloudflare 成為主要入口

    關鍵結論:

    如果你的 AI 產品面向全球使用者、有既有 gRPC 微服務、又不想維護多套 K8s 邊緣集群,Cloudflare 邊緣 入站 TCP/gRPC 是一個能在短期內帶來 延遲優化 + 架構簡化 的實際選項,值得在下一版推理網路設計中優先評估。


    🚀 你現在可以做的事

    • 到 Cloudflare Docs 搜尋「Workers TCP gRPC」了解最新 API 與限制
    • 把現有一個 gRPC 推理服務包成 Container,嘗試部署到 Cloudflare Containers 並透過 Worker 轉發
    • 在現有架構前加一個試驗性 ai-edge.example.com 網域,先導入 5–10% gRPC 流量觀察延遲與穩定性
  • Claude Prompt Caching 長對話成本優化實戰

    Claude Prompt Caching 長對話成本優化實戰

    📌 本文重點

    • 只為「新算的 token」付錢才省成本
    • 穩定前綴(system、tools、長期記憶)要被快取
    • cache key 必須含模型、租戶與權限版本
    • 長對話可降 30–60% 推理成本

    長對話裡真正燒錢的不是「整個 context 有幾個 token」,而是每一輪新算了多少 token。Claude 的 prompt caching 就是把那些每輪都重複出現的前綴(system prompt、tool schema、長期記憶等)標記成可重用的計算結果,只為「新出現的部分」付錢。實務上,你可以在 RAG、Agent、Code Assistant 裡大幅壓低長會話成本,同時讓模型保持完整上下文。

    💡 關鍵: 掌握「每輪新算 token」才是控成本的核心,而不是只看整個 context 長度。


    重點說明:為什麼「重用前綴」省錢

    1. 計費與計算模型:只對「新算過的 token」付錢

    現代 LLM(包含 Claude)推理時,會對輸入序列做一次注意力與前向計算。若供應商支援 前綴快取,就能把已算過的 embedding / KV cache 重新使用,對於已快取的部分:

    • 計費:只收少量或不收額外費用(依供應商設計)
    • 計算:避免重跑 attention / matmul

    • 快取什麼:可穩定重複的 prefix

    典型可快取區塊:

    • System prompt:角色、風格、平台規範
    • Tool schema:function calling / tools JSON schema
    • 長期記憶 / 專案上下文:repo 結構、domain handbook、RAG 長期摘要
    • 長期對話前半段:幾十輪以上的歷史,只要不會被頻繁重寫

    • API 層設計:cache key + 失效策略是核心

    要讓 prompt caching 在多模型、多供應商環境可維護,必須明確管理:

    • cache key 組成:模型版本 + system prompt hash + tools hash + tenant + 權限版本
    • 失效策略:
      • 模型升級(model_version 變更)
      • 工具清單或 schema 改動
      • tenant 權限/角色變更

    💡 關鍵: 把模型 ID、租戶與權限版本都放進 cacheKey,才能避免快取錯配與資料洩漏。


    實作範例:切分前綴與多層快取

    1. Prompt 結構切分策略

    我們先定義一個標準化的 prompt 結構:

    // TypeScript / Node pseudo-code
    interface PromptSegments {
      system: string;          // 系統指令
      toolsSchema: object[];   // 工具定義 (OpenAI / Claude tools 格式)
      longTermContext: string; // 長期記憶 / 專案說明 / RAG summary
      shortHistory: string;    // 近期對話 (例如最後 10 輪)
      userInput: string;       // 本輪使用者輸入
    }
    
    function buildMessages(segments: PromptSegments) {
      return [
        { role: "system", content: segments.system },
        { role: "system", content: `TOOLS_SCHEMA:\n${JSON.stringify(segments.toolsSchema)}` },
        { role: "system", content: `LONG_TERM_CONTEXT:\n${segments.longTermContext}` },
        { role: "assistant", content: segments.shortHistory },
        { role: "user", content: segments.userInput },
      ];
    }
    

    前 3 段(system / toolsSchema / longTermContext)就是要被 prompt caching 鎖定的 prefix;shortHistory 則保留可替換的空間,避免整個對話歷史變成難以管理的快取。

    2. Node 版 cache middleware(多模型、多供應商共用)

    假設你有一個抽象的 LLMClient,在呼叫前注入 cache metadata:

    // Node.js pseudo-code
    import crypto from "crypto";
    
    interface CacheMetadata {
      cacheKey: string;
      cacheTtlSec: number;
    }
    
    function buildCacheKey({
      provider,
      model,
      system,
      toolsSchema,
      tenantId,
      permissionsVersion,
    }: {
      provider: string;
      model: string;
      system: string;
      toolsSchema: object[];
      tenantId: string;
      permissionsVersion: string;
    }): string {
      const hashInput = JSON.stringify({ system, toolsSchema, tenantId, permissionsVersion });
      const hash = crypto.createHash("sha256").update(hashInput).digest("hex");
      return `${provider}:${model}:${hash}`; // 核心:模型 + 前綴內容 + 租戶/權限
    }
    
    async function withPromptCache(llmClient, segments: PromptSegments, ctx) {
      const cacheKey = buildCacheKey({
        provider: ctx.provider,          // "anthropic" / "openai" / "azure-openai" ...
        model: ctx.model,                // 例如 "claude-3.7-sonnet"
        system: segments.system,
        toolsSchema: segments.toolsSchema,
        tenantId: ctx.tenantId,
        permissionsVersion: ctx.permissionsVersion,
      });
    
      const messages = buildMessages(segments);
    
      const response = await llmClient.chat({
        messages,
        // 自定義或供應商原生字段
        metadata: {
          // 自家 caching layer 用
          promptCache: {
            cacheKey,
            segmentsCached: ["system", "toolsSchema", "longTermContext"],
            ttlSec: 3600, // 長期 context 一小時有效
          },
        },
      });
    
      return response;
    }
    

    在多供應商環境下的重點:

    • cacheKey 必須在抽象層就固定格式,不要依賴各家私有的 cache token。
    • 將「哪些 segment 可快取」「TTL 設定」寫在自家 metadata.promptCache,再由底層 adapter 映射到各家 API(例如 Anthropic 的 prompt caching 參數、OpenAI 未來的前綴重用機制等)。

    3. Python 版:RAG + Agent + Code Assistant 共用策略

    示範一個 Python middleware,處理三種場景:

    # Python pseudo-code
    import hashlib
    from typing import List, Dict, Any
    
    class PromptCacheMiddleware:
        def __init__(self, backend):
            self.backend = backend  # Redis / in-memory / provider-native
    
        def _hash(self, payload: Dict[str, Any]) -> str:
            raw = repr(payload).encode("utf-8")
            return hashlib.sha256(raw).hexdigest()
    
        def build_cache_key(self, provider: str, model: str, tenant: str,
                             system: str, tools: List[Dict[str, Any]],
                             perm_version: str) -> str:
            hash_part = self._hash({"system": system, "tools": tools,
                                    "tenant": tenant, "perm": perm_version})
            return f"{provider}:{model}:{hash_part}"
    
        def call(self, llm_client, segments, ctx, scenario: str):
            # scenario: "rag" | "agent" | "code"
            cache_key = self.build_cache_key(
                ctx["provider"], ctx["model"], ctx["tenant"],
                segments.system, segments.tools_schema,
                ctx["permissions_version"],
            )
    
            ttl = 3600 if scenario in ("rag", "code") else 600
    
            messages = build_messages(segments)
    
            # 自家 cache backend,也可以是 provider 的 prompt cache
            cached_prefix = self.backend.get(cache_key)
            if cached_prefix:
                # 若使用 provider-native KV cache,可在這裡直接標記使用
                pass
    
            resp = llm_client.chat(
                messages=messages,
                metadata={
                    "prompt_cache": {
                        "cache_key": cache_key,
                        "ttl_sec": ttl,
                        "segments_cached": ["system", "toolsSchema", "longTermContext"],
                    }
                },
            )
            self.backend.set(cache_key, "USED", ttl)
            return resp
    

    這樣你就可以在同一層 middleware 裡:

    • 為 RAG 保留穩定的 long-term summary 前綴
    • 為 Agent 快取工具清單與角色設定
    • 為 Code Assistant 快取 repo / 專案上下文

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

    1. 工具清單動態變化 → cache 大量失效

    2. 問題:Agent 工具列表若每次請求都依據情境動態調整,toolsSchema hash 會頻繁變動,導致 cacheKey 不可重用。

    3. 建議:

      • 將工具分成「核心必備工具」與「情境工具」,只對核心工具段做 caching。
      • 使用 mid-conversation tool changes(例如 Claude Opus 5 的 beta 功能)讓工具權限在對話中調整,而不觸發整個 prefix 重算。
    4. 模型升級 → 快取錯配

    5. 問題:模型從 claude-3.6 換到 claude-3.7,如果 cacheKey 沒把 model / version 納入,就會拿舊模型算出的 prefix 餵給新模型,出現行為差異。

    6. 建議:

      • 必須把 model id / version 放在 cacheKey 開頭(前面範例已包含)。
      • 建多供應商兼容測試(對齊「同一前綴 + 不同模型」在工具呼叫、輸出格式上的差異),參考 LLM Provider Quirks 文章的思路。
    7. 多租戶 SaaS:tenant 隔離

    8. 問題:如果 cacheKey 沒包含 tenant / 權限版本,在多租戶環境下可能把 A 客戶的長期記憶前綴給 B 客戶用,造成嚴重資料洩漏。

    9. 建議:

      • tenantId 與 permissionsVersion 必須是 cacheKey 的一部分。
      • 權限變更(role / scope 調整)時,明確 bump permissionsVersion,強制快取失效。
      • 對於敏感工具(例如能查詢客戶資料庫的 tool),可直接標記為 不參與 prompt caching,或使用獨立 cache 空間。
    10. 避免 cache 污染:RAG + Agent + Code Assistant

    11. 污染範例:

      • RAG summary 中意外混入使用者敏感資料,然後被快取成長期前綴。
      • Agent 工具列表在某次測試中加入 dev-only 工具,被 cache,之後所有 production 對話都看得到。
    12. 建議:
      • 在產生長期 context / tools schema 前,做一次 安全與敏感資料清洗。
      • 長期前綴內容最好由後端控制(例如固定的 RAG summary service),而不是讓使用者直接寫入。
      • 為 dev / staging / prod 分別建立獨立 cache namespace。

    實際好處:你專案會得到什麼

    引入 Claude Code Prompt Caching 這類前綴快取機制後,對典型專案有幾個直接好處:

    • 長對話成本顯著下降:像 code assistant 或長期顧問型 Agent,一整天會話的 token 數看起來驚人,但前 70–90% 的前綴計算可以重用。實務上常見是 30–60% 推理成本下降。
    • 維持完整上下文又不必瘋狂裁剪:有快取後,你可以保留更多長期記憶與工具說明,而不是每次都為了成本把 context 切到只剩最近幾輪。
    • 跨供應商、多模型快速試錯:抽象層設計好 cacheKey 與前綴切分後,就能在不同模型間切換時,維持一致的成本控制策略,不怕「換模型就打回重算」。

    💡 關鍵: 只要一開始就設計好前綴切分與快取策略,prompt caching 就能變成穩定的基礎設施,而不是事後補救。

    只要在專案一開始就規劃好 prompt 結構、cache key、失效策略,你就能把 prompt caching 當成基礎設施來使用,而不是事後補上去的微調。

    🚀 你現在可以做的事

    • 在現有專案裡明確切分 system、toolsSchema、longTermContext 等前綴區塊
    • 為你的 LLM 抽象層加入統一的 cacheKey 與 metadata.promptCache 設計
    • 在開發環境先測試 RAG、Agent、Code Assistant 三種場景的快取策略與失效機制