AI Middleware 控制層實戰指南

AI Middleware 控制層實戰指南

📌 本文重點

  • 直接在業務程式碼呼叫 LLM API 是反模式
  • 需要獨立 AI Middleware 控制層集中治理
  • 控制層是從 demo 到 production 的關鍵分水嶺
  • 多代理與記憶治理必須納入審計與安全設計

先講結論:直接在產品程式碼裡 fetch('https://api.llm.com') 是一種反模式。當功能從「問答」長成「多工具、多代理、多租戶」時,你會需要一層專門的 AI Middleware 控制層,把:

  • 模型路由與選型
  • Tool / Agent 協調
  • log / trace / metrics
  • 成本與速率限制
  • cache / 重試策略
  • 合規與安全策略注入

從業務程式碼中抽出來,集中治理。這一層就是 LLM 應用從 demo 到 production 的分水嶺

💡 關鍵: 把所有 LLM / Tool / Agent 呼叫收斂到單一控制層,是從玩具 demo 變成可維運產品的必要條件


重點說明:為什麼要獨立 AI Middleware 控制層?

1. 從「直接 call 模型」到「控制層」

傳統 Web 三層:UI 層 → Service 層 → DB 層

LLM 應用如果是:UI → 直接呼叫模型 API,你會發現:

  • 想加 多模型路由(如 GPT + Claude + 本地模型)時,只能四處找出 openai.chat.completions.create 逐一改。
  • 想統一 重試 / 超時 / 灰階 / rollback,根本沒有共同入口可以掛。
  • 想做 成本統計、用戶配額,只剩 trace log 回頭瞎猜。

控制層的做法是:

Product Code → AI Middleware(控制層)→ 各家模型 / 工具 / Agent

所有模型呼叫先經過這一層,才能實作像 API Gateway + Service Mesh 那種集中治理能力。

2. 控制層的核心職責拆解

實務上,控制層至少要負責這幾件事(建議直接當 checklist):

  1. 路由與模型選擇
  2. 任務類型、租戶、成本上限、延遲預算 選擇具體模型。
  3. 範例策略:summary 走便宜模型;寫 SQL 走更準確模型;UGC 敏感類先加安全模型前置審查。

  4. Tool / Agent 協調

  5. 統一管理 工具 registry、Tool 調用權限、Agent orchestration。
  6. 在多代理系統中,把「誰能叫什麼工具」和「記憶寫入策略」集中配置,而不是散落在各 Agent 類別裡。

  7. 觀測與追蹤(logs / traces / metrics)

  8. 每個 LLM 請求、工具呼叫、記憶讀寫 都要有 trace id。
  9. 對接 OpenTelemetry 或 APM:把 token 數量、延遲、錯誤碼、cache 命中等變成可查詢的 metrics。

  10. 成本與速率限制

  11. user / org / feature 做 token 配額與 qps 限制。
  12. 將成本計算邏輯(model 單價、權重)集中在控制層。

  13. cache 與重試策略

  14. 根據 prompt 指紋 + 入參 做 deterministic 回答的 cache。
  15. 定義 哪些錯誤可重試、最大重試次數、退避策略,避免在業務層各自實作。

  16. 合規與安全策略注入

  17. 對接 DLP、敏感詞檢測、角色權限,把 prompt / tool call / output 走同一條審查管線。
  18. 企業內部可在這裡插入 審批流(human-in-the-loop)

💡 關鍵: 成本、風險與合規控制若不集中在控制層,很難在日後擴展時補強而不「大重構」

3. 審計追蹤與記憶治理:多代理系統必備

多代理系統與企業場景最常見的兩個雷:

  • 沒有審計 trail:agent 到處點擊、下單、寫 ticket,最後出了事誰也說不清「是第幾步決定下單」?
    → 參考 Reddit 討論:AI agents 比起更多自主性,更需要完整 audit trail。控制層就是自然的落點。

  • 記憶污染嚴重:所有 agent 都能隨意寫向量庫,久了之後充滿過期決策、敏感資料。
    → 引入 Memory Curator 概念:worker agent 只能發出 記憶事件,由專門的記憶治理層決定要不要寫入、寫到哪個 scope(個人 / 團隊 / 專案 / 會話)。

這兩件事如果不在控制層規範好,就會像 Reddit 上那位開發者形容的:「前三週一切正常,之後 retrieval 變成噪音地獄」。


實作範例:用 Genkit 與 Vercel AI Gateway 搭 Node / Python 控制層

下面用兩個路線示範:

  • Node:Google Genkit + Vercel AI Gateway
  • Python:簡易自製 Middleware(搭配 OpenTelemetry

範例一:Node + Genkit Middleware(含 model 路由與 observability)

安裝與基礎設定(簡化示意):

npm install @genkit-ai/core @genkit-ai/openai @genkit-ai/firebase
npm install @opentelemetry/api @opentelemetry/sdk-node

建立 aiClient.ts 當控制層入口:

// aiClient.ts
import { genkit, z } from "@genkit-ai/core";
import { openai } from "@genkit-ai/openai";

const ai = genkit({
  plugins: [openai()],
});

// 定義模型路由策略
const pickModel = (task: string) => {
  if (task === "summary") return "gpt-4.1-mini";
  if (task === "sql") return "gpt-4.1";
  return "gpt-4o";
};

// 中央統一的生成函數
export async function generateText(req: {
  task: "summary" | "sql" | "general";
  input: string;
  userId: string;
}) {
  const model = pickModel(req.task);

  // observability: 附上 trace metadata
  const traceMeta = {
    userId: req.userId,
    task: req.task,
    model,
  };

  return ai.generate(
    {
      model,
      prompt: req.input,
    },
    {
      // middleware hooks: 可插 retry / cache / logging
      trace: traceMeta,
    }
  );
}

在 Genkit middleware hooks 中加入 cache / retry / 安全策略(概念程式):

// middleware.ts
import { registerMiddleware } from "@genkit-ai/core";

registerMiddleware(async (ctx, next) => {
  const key = hashPrompt(ctx.request);

  // 1. Cache 命中
  const cached = await cacheGet(key);
  if (cached) {
    ctx.log.info("cache-hit", { key });
    return cached;
  }

  // 2. 成本與速率限制
  await enforceQuota(ctx.trace.userId, ctx.request.model);

  // 3. 安全策略:例如 DLP 檢查
  await checkPromptPolicy(ctx.request.prompt);

  // 4. 重試包一層
  return retry(async () => {
    const res = await next();
    await cacheSet(key, res);
    return res;
  }, { retries: 2, backoffMs: 200 });
});

前端 / 服務層就只呼叫 generateText,完全不碰 openai.chat 類 API。未來換模型、加灰階、接別家供應商,只要改控制層即可。

範例二:Node + Vercel AI Gateway 作為集中入口

Vercel AI Gateway 提供 統一 endpoint + 多模型路由 + 速率限制 + 規則引擎,很適合當外部控制層。

基本設定(vercel.json 或 UI 設定):

  • 建立一個 Gateway route,例如:https://ai.yourdomain.com/v1/chat
  • 配置 providersOpenAI / Anthropic / 自建模型。
  • 寫 routing rule:
  • if (task == 'summary') -> openai:gpt-4.1-mini
  • if (tenant == 'premium') -> anthropic:claude-3.7

前端程式碼(Next.js 伺服器端)只需要呼叫單一 gateway:

// app/api/ai/route.ts
import { NextRequest, NextResponse } from "next/server";

export async function POST(req: NextRequest) {
  const body = await req.json();

  const res = await fetch(process.env.AI_GATEWAY_URL!, {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      "X-Tenant": body.tenantId,
      "X-Task": body.task,
    },
    body: JSON.stringify({
      messages: body.messages,
      stream: body.stream ?? false,
    }),
  });

  // 這裡可以加 audit log / trace id
  return new NextResponse(res.body, { status: res.status });
}

優點:

  • 速率限制、成本統計、提供者 failover 直接用 Vercel 的 console 配。
  • 灰階與回滾:改 routing rule 即可,比如 10% 流量導到新模型。

💡 關鍵: 把流量管理與模型路由交給 gateway,可用設定檔與後台調整,而不必每次動到應用程式碼

範例三:Python 控制層 + 記憶治理(Memory Curator)

假設你在做多代理系統,希望 worker agent 不能直接寫入向量庫

# ai_control_layer.py
from dataclasses import dataclass
from typing import Literal, Dict, Any

Scope = Literal["agent", "team", "project", "session"]

@dataclass
class MemoryEvent:
  scope: Scope
  content: str
  evidence: Dict[str, Any]
  actor: str  # 哪個 agent

class MemoryCurator:
  def __init__(self, store):
    self.store = store

  def decide(self, event: MemoryEvent):
    # 簡單篩選:過長、含敏感資訊則拒絕
    if len(event.content) > 2000:
      return "discard"
    if "password" in event.content.lower():
      return "discard"

    # 不同 scope 寫不同 index
    index_name = {
      "agent": f"agent-{event.actor}",
      "team": "team-shared",
      "project": "project-global",
      "session": None,  # 只放在 runtime context
    }[event.scope]

    if index_name:
      self.store.write(index_name, event.content, event.evidence)
      return f"written:{index_name}"
    else:
      # session-only: 不寫 durable store
      return "session-only"

# worker agent 不直接調 store
curator = MemoryCurator(store=vectordb)

async def worker_store_memory(proposed_content: str, actor: str):
  event = MemoryEvent(
    scope="project",
    content=proposed_content,
    evidence={"source": "task_result"},
    actor=actor,
  )
  result = curator.decide(event)
  return result

在這個架構下:

  • 所有記憶寫入 都經過 MemoryCurator,可審計、可控。
  • 之後要加 審計 log / OpenTelemetry span 也只改這一處。

建議與注意事項:常見踩坑與實戰指引

1. 不要把控制層寫成巨大 God Service

常見錯誤:

把所有邏輯(模型路由、prompt 模板、tool orchestration、記憶管理、審批流)塞進同一個 ai_service.ts

結果:

  • 難以測試、難以灰階、任何小改都要全 redeploy。
  • 無法對特定職責做獨立 scaling(例如工具呼叫爆量時)。

建議:依職責拆模組,例如:

  • ModelRouter
  • ToolRegistry
  • RetryPolicy
  • QuotaManager
  • MemoryCurator

控制層本身只是一個 薄 orchestration 層,組裝這些模組,不要變成 monolith。

2. 一開始就設計可觀測性 schema

很多團隊是到事故發生後才加 log,結果 schema 雜亂無章,完全無法統計。

最低限度,請在控制層統一定義以下欄位

  • trace_id / span_id:串起同一個 user request 下的所有模型與工具呼叫。
  • user_id / tenant_id / feature_name:方便做成本報表與配額控制。
  • model_name / provider / tokens_in / tokens_out / latency_ms:跑出模型性能與成本比較。
  • cache_hit / retry_count / policy_blocked (bool):分析策略的實際效果。

搭配 OpenTelemetry

  • 在 control layer 的入口建立 root span
  • 每個 model_calltool_callmemory_write 建子 span。
  • 把上述欄位當 attributes 寫入。

3. 灰階與回滾機制不要事後補

Anthropic 的觀察顯示:非程式碼型 agent 在 production 常常因為資料與流程難控而失敗。這不是模型問題,而是缺乏 安全試錯機制

在控制層預先設計:

  • 灰階發布:例如新增模型時,只讓 5% 的流量使用;表現不好立即切回。
  • 可以在 Vercel AI Gateway 或內部 router 實作簡單的百分比路由。
  • 策略回滾:策略(例如更嚴格的 DLP、不同記憶寫入規則)要抽成 可配置,而不是寫死在程式裡。
  • 配合 feature flag 平台(LaunchDarkly 等)效果更佳。

4. 把 audit trail 當成產品需求,不是附加功能

對多步 agent 流程,請明確要求:

  • 每一步 「想做什麼 → 實際做了什麼 → 結果如何」 都寫 audit log。
  • 在 UI(或 admin console)內提供「代理執行歷史」頁面,讓使用者能依 trace id 看到整條鏈。

這件事最好放在控制層完成:

  • 所有 tool_call 都經過控制層。
  • 控制層負責序列化成統一格式:

json
{
"trace_id": "...",
"step": 3,
"actor": "billing_agent",
"tool": "update_invoice",
"input": {"invoice_id": 123},
"output": {"status": "ok"},
"started_at": "...",
"duration_ms": 532
}

這直接提升企業客戶的信任度,也方便日後做合規稽核。


總結: 如果你的專案準備從「demo 給老闆看」變成「真的要上線給客戶用」,請先停下來,把所有 LLM / Tool / Agent 呼叫集中收斂到一個 AI Middleware 控制層。路由、觀測、策略、記憶治理都放在這裡,才能控制成本、降低事故、加快迭代速度。

🚀 你現在可以做的事

  • 整理專案中所有 openai.chat / fetch('llm') 呼叫點,設計一個統一的控制層介面(如 generateText()
  • 在控制層導入基本的 trace_iduser_idmodel_name log schema,並串接 OpenTelemetry 或現有 APM
  • 為現有的多代理或 RAG 系統設計一個 MemoryCurator 類別,強制所有記憶寫入都經過同一治理入口

留言

發佈留言

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