AI Middleware - Kerwin AI Information

📌 本文重點

直接在業務程式碼呼叫 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）：

路由與模型選擇
依 任務類型、租戶、成本上限、延遲預算 選擇具體模型。
範例策略：summary 走便宜模型；寫 SQL 走更準確模型；UGC 敏感類先加安全模型前置審查。
Tool / Agent 協調
統一管理 工具 registry、Tool 調用權限、Agent orchestration。
在多代理系統中，把「誰能叫什麼工具」和「記憶寫入策略」集中配置，而不是散落在各 Agent 類別裡。
觀測與追蹤（logs / traces / metrics）
每個 LLM 請求、工具呼叫、記憶讀寫 都要有 trace id。
對接 OpenTelemetry 或 APM：把 token 數量、延遲、錯誤碼、cache 命中等變成可查詢的 metrics。
成本與速率限制
按 user / org / feature 做 token 配額與 qps 限制。
將成本計算邏輯（model 單價、權重）集中在控制層。
cache 與重試策略
根據 prompt 指紋 + 入參 做 deterministic 回答的 cache。
定義 哪些錯誤可重試、最大重試次數、退避策略，避免在業務層各自實作。
合規與安全策略注入
對接 DLP、敏感詞檢測、角色權限，把 prompt / tool call / output 走同一條審查管線。
企業內部可在這裡插入 審批流（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。
配置 providers：OpenAI / 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_call、tool_call、memory_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_id、user_id、model_name log schema，並串接 OpenTelemetry 或現有 APM

為現有的多代理或 RAG 系統設計一個 MemoryCurator 類別，強制所有記憶寫入都經過同一治理入口

標籤: AI Middleware

AI Middleware 控制層實戰指南