標籤: Qwen

📌 本文重點

• 單一 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-mini、DeepSeek-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 類型系統在工程上的落地：先把任務分型，路由邏輯就自然長出來。

💡 關鍵： 用 maxLatencyMs、maxCostUSD 這類 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，開始累積日後做成本與可靠性優化所需的資料