標籤: 模型路由策略

  • Fugu 式多模型協作實戰拆解

    Fugu 式多模型協作實戰拆解

    📌 本文重點

    • 單一 LLM 容易遇到成本與供應商風險問題
    • Fugu 用任務類型與路由實現多模型協作
    • 多模型需要良好編排、仲裁與可觀測性
    • 建議從抽象 Task 與 adapter 漸進導入

    單一 LLM 做所有事情的痛點很明確:成本不可控、供應商風險高、效能無法對應不同任務類型。Sakana 的 Fugu 路線給了一個很務實的答案:用一層編排(orchestration)把多個模型(雲端 + 本地 + 專用 code model)協作起來,把「選模型、聚合結果、錯誤控制」變成一套可維護的工程結構,而不是散落在業務程式碼裡的 if-else。


    重點說明

    1. Fugu 式類型系統:先定「任務類型」,再談選哪個模型

    Fugu 的關鍵不是多模型本身,而是任務類型(task types)+ 類型安全的輸入輸出

    • 每個任務明確定義:
    • input schema(如 QueryTask, CodeGenTask, LongFormTask
    • output schema(如 Answer, CodePatch, SearchPlan
    • 路由層只依賴任務類型與 metadata(長度、成本上限、延遲 SLA),不直接寫死「如果是寫程式就用 XXX」。

    範例(TypeScript 風格的 pseudo code):

    // 1. 任務類型定義
    interface BaseTaskMeta {
      maxLatencyMs: number;
      maxCostUSD: number;
      priority: 'low' | 'normal' | 'high';
    }
    
    interface QueryTask {
      type: 'query';
      input: { question: string; context?: string };
      meta: BaseTaskMeta & { allowWebSearch: boolean };
    }
    
    interface CodeGenTask {
      type: 'codegen';
      input: { spec: string; language: string };
      meta: BaseTaskMeta & { needTests: boolean };
    }
    
    interface LongFormTask {
      type: 'longform';
      input: { topic: string; minWords: number };
      meta: BaseTaskMeta & { allowStreaming: boolean };
    }
    
    type Task = QueryTask | CodeGenTask | LongFormTask;
    

    好處:

    • 模型路由只看 Task,不看業務細節,方便之後替換模型 / 供應商。
    • 不同模型可以有不同的 prompt / tool schema,但在編排層都被包成同一個 Task 抽象。

    💡 關鍵: 先用類型把任務抽象好,之後換模型或換供應商就變成「改路由」而不是「重寫業務程式碼」。


    2. 模型路由:任務類型 × 長度 × 成本

    一個實用的路由策略通常只靠幾個欄位就夠了:

    • 任務類型
    • codegen → 專用 code model(如 o3-miniDeepSeek-Coder、本地 Qwen code)
    • query → 一般對話模型(OpenAI / Anthropic / 本地)
    • longform → 長 context 模型(如 200k+ context),或拆段 + 聚合
    • 長度估計:預估輸入 token + 預計輸出 token,超過本地模型 context 就路由到雲端長上下文模型。
    • 成本與延遲
    • maxCostUSD 控制是否可以打貴模型
    • maxLatencyMs 決定是否啟用並行查詢 + 快速仲裁

    簡化版路由器:

    function routeModel(task: Task): 'openai:gpt-4.1-mini' | 'local:qwen' | 'openai:o3-mini' {
      const estTokens = estimateTokens(task.input);
    
      if (task.type === 'codegen') {
        // code 任務預設走專用 code model
        return task.meta.maxCostUSD < 0.05 ? 'local:qwen' : 'openai:o3-mini';
      }
    
      if (task.type === 'longform') {
        if (estTokens > 120_000) return 'openai:gpt-4.1-mini';
        return 'local:qwen';
      }
    
      // query 一般問答
      if (task.meta.maxLatencyMs < 3000) {
        // 低延遲預算 → 本地或較小雲端模型
        return 'local:qwen';
      }
    
      return 'openai:gpt-4.1-mini';
    }
    

    這類路由就是 Fugu 類型系統在工程上的落地:先把任務分型,路由邏輯就自然長出來

    💡 關鍵:maxLatencyMsmaxCostUSD 這類 metadata 控制路由,可以在同一套架構裡同時優化成本與延遲。


    3. 回覆聚合與仲裁:多模型輸出怎麼合成一個答案

    多模型協作的價值在於:

    • 一部分模型擅長查(search / recall),一部分擅長寫(rewrite / explain)
    • 或同一任務交給兩個模型,透過仲裁降低幻覺

    典型做法:

    1. 並行呼叫 2–3 個模型:如本地 Qwen + 雲端 GPT
    2. 用一個「仲裁模型」來閱讀所有候選答案,輸出最終回覆與信心分數

    仲裁 prompt 示意:

    const arbiterPrompt = `你是仲裁模型。你會看到多個模型的回答,請:
    1. 比較其一致性與是否自相矛盾。
    2. 檢查是否有推理錯誤或明顯幻覺。
    3. 選出最可信的一個,並在有疑慮時標記「不確定」。
    
    輸出 JSON:
    {
      "winner": "model_a" | "model_b",
      "confidence": 0-1,
      "final_answer": "...",
      "notes": "..."
    }`;
    

    好處:

    • 提高可靠性(特別是檢索或工具調用密集場景)
    • 可以把仲裁結果記錄下來,用於後續離線分析各模型表現

    成本上升是必然,常見做法是:

    • 只在 高價值任務 / 有風險的 domain(法律、醫療) 開啟仲裁
    • 其他場景靠單模型 + tool verification 解決

    4. 單一 LLM vs 多 LLM 編排:實際取捨

    單一 LLM

    • 優點:實作簡單、debug 容易、觀測鏈短
    • 缺點:
    • 價格彈性差:所有任務都用貴模型
    • 供應商風險:價格調整、限額、區域封鎖都直接影響產品
    • 難以對應極端需求(超長上下文、線下敏感數據)

    多 LLM 編排(Fugu 路線):

    • 優點:
    • 不同任務用最合適的模型 → 可靠性與成本可同時優化
    • 可以把敏感任務 route 到本地模型,降低隱私風險
    • 雲端服務掛了可以 fallback 到次佳方案
    • 缺點:
    • 觀測與 debug 變複雜(誰的錯?哪一層出問題?)
    • 延遲可能放大(串聯多步、多模型仲裁)
    • 各家 API、tool schema、系統提示格式不一致,需要一層 adapter

    對多數專案來說:

    • MVP 階段 → 單一 LLM + 清楚的 abstraction
    • 成本 / 隱私壓力出現後 → 漸進式導入多模型編排,而不是一次重寫

    💡 關鍵: 多模型不是為了「酷」,而是為了在成本、可靠性、隱私之間取得更穩定的折衷。


    實作範例:OpenAI + 本地 Qwen + 專用 code model

    以下給出一個可自建的最小多模型編排骨架(Node/TypeScript 風格,但概念可套任何語言)。

    1. API 介面設計

    對前端/上游只暴露一個 API:POST /v1/ai/execute,輸入統一的 Task 結構。

    // express / fastify handler
    app.post('/v1/ai/execute', async (req, res) => {
      const task: Task = req.body;
    
      const modelId = routeModel(task);
    
      const controller = new AbortController();
      const timeout = setTimeout(() => controller.abort(), task.meta.maxLatencyMs);
    
      try {
        const rawResponse = await callModel(modelId, task, { signal: controller.signal });
        const parsed = normalizeOutput(task, rawResponse);
    
        await logTask({ task, modelId, rawResponse: parsed });
    
        res.json(parsed);
      } catch (e) {
        const fallback = await tryFallback(task, modelId);
        res.json(fallback);
      } finally {
        clearTimeout(timeout);
      }
    });
    

    2. 模型 adapter:解決不同 API / token 格式

    常見坑是:

    • 系統訊令格式不同(OpenAI messages vs 本地單純 prompt
    • tool / function call schema 不同

    用 adapter 隔離差異:

    async function callModel(modelId: string, task: Task, opts: { signal: AbortSignal }) {
      switch (modelId) {
        case 'openai:gpt-4.1-mini':
          return callOpenAI(task, opts);
        case 'openai:o3-mini':
          return callOpenAICode(task, opts);
        case 'local:qwen':
          return callLocalQwen(task, opts);
        default:
          throw new Error(`Unknown model ${modelId}`);
      }
    }
    
    async function callOpenAI(task: Task, { signal }: { signal: AbortSignal }) {
      const messages = buildMessagesFromTask(task);
      const resp = await openai.chat.completions.create({
        model: 'gpt-4.1-mini',
        messages,
        temperature: 0.2,
        response_format: { type: 'json_object' },
        signal,
      });
      return resp.choices[0].message.content;
    }
    
    async function callLocalQwen(task: Task, { signal }: { signal: AbortSignal }) {
      const prompt = buildPromptFromTask(task); // 單一 string
      const resp = await fetch('http://localhost:8000/v1/completions', {
        method: 'POST',
        body: JSON.stringify({
          model: 'qwen-32b-instruct',
          prompt,
          max_tokens: 2048,
          temperature: 0.1,
        }),
        signal,
      }).then(r => r.json());
      return resp.choices[0].text;
    }
    

    只要嚴格把「怎麼跟模型講話」鎖在 adapter 裡,上層就可以只面對 Task


    3. 超時與 fallback 策略

    簡單可行的策略:

    1. 以延遲為主的 fallback

    2. 本地模型超時 → fallback 到雲端小模型

    3. 雲端模型錯誤/超時 → fallback 到本地(或退化版回答)
    async function tryFallback(task: Task, failedModelId: string) {
      const fallbackId = pickFallbackModel(task, failedModelId);
      if (!fallbackId) throw new Error('No fallback model');
    
      const raw = await callModel(fallbackId, task, { signal: AbortSignal.timeout(2000) });
      return normalizeOutput(task, raw, { degraded: true, usedFallback: true, fallbackId });
    }
    
    1. 回應標註退化狀態:在 normalizeOutput 中加入:
    {
      "answer": "...",
      "meta": {
        "modelId": "local:qwen",
        "usedFallback": true,
        "fallbackFrom": "openai:gpt-4.1-mini",
        "degraded": true
      }
    }
    

    讓前端可以決定是否顯示「此回答為備援模型生成」。


    4. 觀測與日誌結構:解決「昨晚 agent 到底做了什麼」

    多模型編排很容易變成黑箱。建議最少做到:

    • 每個 Task 一個 traceId
    • log 中至少包含:
    • traceId, task.type, task.meta
    • 選擇的 modelId、fallback 情況
    • 每步 latency、token 使用量
    • 仲裁結果(如果有)

    示意:

    interface TaskLog {
      traceId: string;
      taskType: Task['type'];
      modelId: string;
      fallbackFrom?: string;
      latencyMs: number;
      inputTokens: number;
      outputTokens: number;
      success: boolean;
      error?: string;
    }
    
    async function logTask(log: TaskLog) {
      // 可寫入 ClickHouse / BigQuery / Elastic
      console.log(JSON.stringify({ kind: 'taskLog', ...log }));
    }
    

    這類結構化 log 是後續做「loop engineering」(自動 self-correct / regression test)與成本優化的基石。


    建議與注意事項

    1. 延遲放大:多模型 ≠ 多倍延遲

    • 儘量並行呼叫可獨立的模型,再用仲裁合併。
    • 嚴格設定 per-model timeout,避免某個模型拖垮整個請求。
    • 對長任務(如自動寫測試、長時間 agent loop)要分段 log,避免只看到「跑了一小時,掛了」。

    2. 工具 / 系統 prompt 不一致

    • 不同家模型對 system / tools / function calling 的支援度不同。
    • 最好的做法:
    • 定義自己的 工具層 schema(如 JSON Tool 定義)
    • 在 adapter 把它映射成各模型需要的格式
    • 千萬避免在業務邏輯裡到處寫 if (model === 'gpt-4.1-mini') 這種分支。

    3. 責任歸屬與 debug 困難

    多模型編排容易出現:

    • prompt 沒設好 → 模型亂回答
    • 路由策略不合理 → 小模型被丟去做艱難任務
    • 仲裁錯誤 → 明明較好的答案被丟掉

    實務上建議:

    • 為每一層定義清楚的「契約」
    • 路由層:輸入 Task,輸出 modelId,不關心內容
    • adapter:保證把 Task 翻譯成該模型最佳格式
    • 仲裁層:對模型輸出負責,不對業務邏輯負責
    • 做回溯時先問:錯在路由、adapter、模型本身、還是仲裁?

    4. 不要一開始就 over-engineer

    • 若你現在是:一個雲端 LLM + 少數工具,建議只先做:
    • 抽出 Task 類型
    • 寫好 model adapter(即使目前只有一個模型)
    • 之後要引入本地 Qwen、專用 code model、Fugu 式仲裁機制,就只是替換實作,而不是重寫整個系統。

    核心結論: 多模型協作不是把更多模型硬塞進系統,而是用 清晰的任務類型 + 路由 + 仲裁 + 可觀測性,把「哪個模型做什麼」變成一個可以演進的工程決策。從這個角度看,你可以在自己的專案裡做一個「迷你 Fugu」,用極少的代碼換來更好的成本控制、可靠性與供應商彈性。


    🚀 你現在可以做的事

    • 在現有專案中抽出一層 Task 類型與 routeModel(),把模型選擇邏輯從業務程式碼移出來
    • 寫一個簡單的 model adapter(例如包一層 callModel()),即使目前只支援單一 gpt-4.1-mini
    • 為每次模型呼叫加上 traceId 與結構化 log,開始累積日後做成本與可靠性優化所需的資料