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,開始累積日後做成本與可靠性優化所需的資料

留言

發佈留言

發佈留言必須填寫的電子郵件地址不會公開。 必填欄位標示為 *