標籤: Python

  • 把 ERP 變成 AI 的執行單元:iDempiere OData × MCP Server 整合策略

    重點摘要(TL;DR)

    • iDempiere(開源 ERP)的 REST/OData API 包裝成 MCP server,任何支援 MCP 的 AI 工具(Claude / ChatGPT / Cursor / Claude Code / VS Code)都能直接呼叫 ERP。
    • Microsoft 已經做了 Dynamics 365 ERP MCP server(2026/4 文件更新),三類工具設計:Data tools(OData CRUD)、Form tools(模擬使用者操作)、Action tools(直接呼叫 business class)。這個設計可以直接借鏡到 iDempiere
    • iDempiere REST 已經提供 api/v1/auth(JWT)、api/v1/models(OData CRUD)、api/v1/windowsapi/v1/processes 四類 endpoint — 剛好對應 Microsoft 的三類工具
    • 整合到腦子系統:LLM Gateway + MCP Server 雙軌設計,Gateway 管 LLM 流量,MCP 管 ERP tool calls,各自有 audit log,iDempiere 內建的 AD_Role 直接當權限層,不用自寫 ABAC。
    • 本文是腦子系統四部曲的第五篇延伸(ERP 整合層)。前四篇:Why / How / Scale / Tools

    一、為什麼 iDempiere OData 是腦子系統缺的拼圖

    前四篇文章把 AI 治理系統蓋好了:LLM Gateway、雙引擎、Harness、Chat-native Agent。但有個關鍵問題沒解決 — AI 怎麼安全地讀寫公司的真實業務資料?

    大部分公司現況:業務資料躺在 ERP 裡,AI 透過 prompt 拿不到;或者員工自己貼資料給 AI(踩到 A 級資料禁令)。第一篇的核心哲學「AI 時代不做 UI,做給 AI 安全取資料的入口」需要一個具體載體。

    iDempiere(開源 ERP)的 REST/OData API 剛好就是這個載體 — 它本來就是「給機器讀的標準介面」,而且整套權限、Audit、租戶隔離都已經 30 年累積在 iDempiere 的 AD(Application Dictionary)裡。不用重新造輪子,直接接到 AI

    二、事實核對:iDempiere REST API 真實狀態(2026/5)

    本文涉及的 iDempiere 技術細節都來自官方來源,以下是 2026/5 撰文時的驗證結果:

    事實 驗證來源
    REST plugin 由 BX Service GmbH 維護,GPLv2,used in production iDempiere Wiki
    支援 iDempiere release 12 及 master,plugin 已運作於 v9/v10 GitHub Repo
    官方文件站 idempiere-rest-docs
    Swagger UI 互動式 API 探索 hengsin/idempiere-rest-swagger-ui

    2.1 四個主要 API 端點

    • POST/PUT api/v1/auth/tokens — JWT 認證(token 1 小時有效)
    • api/v1/models/{tableName} — PO(Persistent Object)CRUD,支援 OData filter
    • api/v1/windows/{windowSlug}/tabs/{tabSlug} — Window/Tab 互動(對應 ERP UI 的視窗結構)
    • api/v1/processes/{processSlug} — Process 呼叫(DocAction、報表、自動化流程)

    以及附加端點:檔案存取、Reference 資料、Cache 管理、Workflow、Scheduler 資訊。[來源]

    三、業界典範:Microsoft Dynamics 365 ERP MCP Server

    Microsoft 在 2026/4/27 發表了 Dynamics 365 Finance & Operations 的 MCP server 完整文件 — Microsoft Learn這是目前業界最完整的 ERP × AI 整合範例,值得借鏡。

    3.1 Microsoft 的三類工具設計

    類別 用途 代表工具(Microsoft)
    Data tools OData CRUD operations data_find_entities、data_create_entities、data_update_entities、data_delete_entities
    Form tools 模擬使用者在 UI 上的操作(點按鈕、填表、開分頁) form_click_control、form_set_control_values、form_save_form
    Action tools 直接呼叫 ERP 內部 business logic class api_find_actions、api_invoke_action

    3.2 三個關鍵設計原則(直接可借鏡)

    1. 動態 context:MCP server 每次 tool call 都根據 agent 安全角色和環境配置「動態」回傳 context — 「the security role of the authenticated user for the agent determines which objects are returned in the view model」(原文)
    2. 角色限制 = scope 限制:Agent 只看到自己角色能存取的 menu / entities / API,既是安全也是 prompt 效率(context 不會塞太多無關資訊)
    3. Allowed MCP Clients:Microsoft 預設只允許 Copilot Studio 和 VS Code 兩個 client ID 存取 MCP,其他 agent platform 必須在 Microsoft Entra ID 註冊後加入白名單 — 不是「誰來都能接」

    四、把這個設計搬到 iDempiere

    關鍵 insight:iDempiere REST 的四個 endpoint,剛好對應 Microsoft 的三類工具設計,直接 mapping:

    Microsoft 三分類 iDempiere REST 對應 endpoint 說明
    Data tools api/v1/models/{table} PO CRUD + OData filter,直接套
    Form tools api/v1/windows/{slug}/tabs/{slug} Window/Tab 結構,可模擬「打開視窗、切分頁、設欄位」
    Action tools api/v1/processes/{slug} Process 呼叫(DocAction、報表、自動化)

    結論:你不用設計 MCP server 的工具分類,直接複製 Microsoft 的三分類,把 iDempiere REST 端點包裝進去即可。

    五、MCP 是什麼,為什麼是關鍵

    Model Context Protocol 是 Anthropic 2024/11 發布的開源協議,定義 AI 應用怎麼跟外部資料來源、工具、工作流溝通。官方比喻:「USB-C port for AI applications」。[來源]

    5.1 為什麼是 ERP × AI 的關鍵

    • 標準協議,一次寫多處用:同一個 MCP server 可以同時被 Claude Desktop / Claude Code / Cursor / VS Code / ChatGPT 接([來源])
    • 不是 prompt engineering 的小聰明,是基礎建設層
    • 已成 industry standard:Anthropic / OpenAI / Microsoft 都採納

    5.2 寫 MCP server 的工具(2026/5 驗證)

    • Python SDK:modelcontextprotocol/python-sdk v1.x stable(v2 pre-alpha 開發中)
    • 安裝:uv add "mcp[cli]"pip install "mcp[cli]"
    • Transport:stdio、SSE、Streamable HTTP 三種
    • 認證:OAuth 2.1 resource server 標準

    六、實作範例:iDempiere MCP server v0

    下面是用 FastMCP + httpx 實作的最小可行版本,展示三類工具的骨架。注意:這是教學範例,production 版需要加上錯誤處理、重試、token refresh、審計 log 等。

    # idempiere_mcp_server.py
    from mcp.server.fastmcp import FastMCP
    import httpx
    from typing import Optional
    
    mcp = FastMCP("iDempiere-MCP")
    IDEMPIERE_BASE = "https://idempiere.example.com/api/v1"
    
    # ───────── Auth ─────────
    @mcp.tool()
    async def authenticate(
        username: str,
        password: str,
        client_id: int,
        role_id: int,
        organization_id: int = 0,
        warehouse_id: int = 0,
        language: str = "en_US"
    ) -> dict:
        """One-shot authentication with all parameters.
        Returns session token valid for 1 hour."""
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{IDEMPIERE_BASE}/auth/tokens",
                json={
                    "userName": username,
                    "password": password,
                    "parameters": {
                        "clientId": client_id,
                        "roleId": role_id,
                        "organizationId": organization_id,
                        "warehouseId": warehouse_id,
                        "language": language,
                    }
                }
            )
            resp.raise_for_status()
        return resp.json()
    
    # ───────── Data Tools (OData CRUD) ─────────
    @mcp.tool()
    async def query_records(
        token: str,
        table_name: str,
        filter_expr: Optional[str] = None,
        top: int = 50
    ) -> dict:
        """Query iDempiere PO records via OData.
    
        Filter examples (note iDempiere uses 'neq' not 'ne'):
          - "IsCustomer eq true and contains(Name, 'Acme')"
          - "Created gt 2026-04-01T00:00:00Z"
        """
        params = {"$top": top}
        if filter_expr:
            params["$filter"] = filter_expr
        async with httpx.AsyncClient() as client:
            resp = await client.get(
                f"{IDEMPIERE_BASE}/models/{table_name}",
                params=params,
                headers={"Authorization": f"Bearer {token}"}
            )
            resp.raise_for_status()
        return resp.json()
    
    @mcp.tool()
    async def create_record(token: str, table_name: str, data: dict) -> dict:
        """Create a PO record. Caller must include all mandatory fields.
        Tip: query AD_Column WHERE IsMandatory='Y' to discover them first."""
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{IDEMPIERE_BASE}/models/{table_name}",
                json=data,
                headers={"Authorization": f"Bearer {token}"}
            )
            resp.raise_for_status()
        return resp.json()
    
    # ───────── Action Tools (Process call) ─────────
    @mcp.tool()
    async def run_process(
        token: str,
        process_slug: str,
        parameters: dict
    ) -> dict:
        """Execute an iDempiere Process (e.g. DocAction, scheduled job, report).
    
        Parameters must be FLAT top-level keys, NOT a 'parameters' array:
          Correct:  {"StatementYear": 2026, "StatementPeriod": "2"}
          Wrong:    {"parameters": [{"parameterName": ..., "value": ...}]}
        """
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{IDEMPIERE_BASE}/processes/{process_slug}",
                json=parameters,
                headers={"Authorization": f"Bearer {token}"}
            )
            resp.raise_for_status()
        return resp.json()
    
    if __name__ == "__main__":
        mcp.run(transport="streamable-http")

    這支 script 跑起來後,任何支援 MCP 的 client(Claude Desktop / Claude Code / Cursor / VS Code)都可以連到 http://localhost:8000 並使用上述工具。

    6.1 範例對話(架構驗證)

    員工(在 chat 工具中):
      「幫我查最近 10 筆訂單金額大於 100 萬的客戶」
    
    AI agent(透過 MCP 自動執行):
      1. authenticate(...) → 拿到 session token
      2. query_records(
           token=...,
           table_name="C_Order",
           filter_expr="GrandTotal gt 1000000",
           top=10
         )
      3. 解析結果,回給員工
    
    員工看到:
      「最近 10 筆大於 100 萬的訂單列表如下:...」

    注意:第 1 步的 authenticate 只執行一次,session token 1 小時有效,後續 query 都用同一個 token。

    七、整合進腦子系統:雙軌架構

    員工 chat app (LINE / Mattermost / Telegram / Slack)
        ↓
    Chat-native Agent (QwenPaw) 或 Coding Agent (Claude Code)
        │
        ├─ LLM 流量 ───→ 公司 LLM Gateway (LiteLLM + Portkey)
        │                ├─ 分級/脫敏/路由
        │                └─ → 雲端 frontier 或本地 Ollama
        │
        └─ Tool calls ─→ iDempiere MCP Server (自製)
                          ├─ OAuth 2.1 / Allowed Clients 白名單
                          ├─ Data tools (OData CRUD)
                          ├─ Form tools (Window/Tab 互動)
                          ├─ Action tools (Process call)
                          └─ Audit log → SIEM
                          ↓
                      iDempiere REST API (api/v1/*)
                          ↓ (內建 AD_Role 過濾)
                      iDempiere PostgreSQL

    關鍵設計:

    • LLM Gateway 跟 MCP Server 是兩條平行軌道:Gateway 管 prompt,MCP 管 tool calls。兩者都要 audit log,可獨立縱深防禦
    • 權限不重複設計:iDempiere 內建 AD_Role 直接當權限層,MCP server 帶 user 的 token 進去,iDempiere 自動套 role 過濾資料 — 不用自寫 ABAC 規則
    • Allowed MCP Clients 白名單:借鏡 Microsoft 設計,只允許特定 agent platform 接 MCP server,不是「誰來都能接」

    八、權限層的對應(這是最大紅利)

    員工從 chat app 問問題時的完整權限路徑:

    1. 員工 LINE/Slack ID → Agent 認 ALLOWED_USERS 白名單
    2. Agent → MCP Server,帶員工的 iDempiere session token
    3. MCP Server → iDempiere REST,帶 token
    4. iDempiere 自動套員工的 AD_Role 過濾資料:
       - 業務員角色 → 只看自己的 SalesRep 訂單
       - CFO 角色 → 看全公司
       - RD 角色 → 完全看不到業務資料
    5. 回應只含「員工角色應該看到」的資料

    iDempiere 30 年累積的 AD_Role / AD_Window_Access / AD_Column 權限設計直接拿來用。這比自己在 Gateway 寫 ABAC 簡單一個量級

    九、為什麼這比 Dynamics 365 / NetSuite MCP server 適合中小規模製造業

    特性 Dynamics 365 ERP MCP iDempiere + 自製 MCP
    License Microsoft 訂閱 + Copilot 點數 GPLv2 開源
    Hosting Cloud only(Tier 2+) self-host / air-gapped 可
    Tool 計費 0.1 Copilot Credits per tool call(非 Copilot Studio 環境) 0(自架)
    A 級資料 需透過 Cloud,法規場景受限 完全本地處理
    客製化 透過 ICustomAPI + AI tool framework 直接改 plugin / 加 process

    對製造業中小集團、要 air-gapped 的法規場景、預算有限的公司:iDempiere + 自製 MCP server 是唯一既可離線又能整合 AI 的開源路徑

    十、工程藍圖:漸進式 v0 → v1 → v2

    v0:Read-only Data Tools(2-4 週,1 RD)

    • 實作 authenticate + query_records(本文範例)
    • 支援 5-10 個常用 table:C_BPartner、C_Order、M_Product、M_InOut、M_Movement、AD_User、R_Request 等
    • OData filter 支援 eq / neq / contains / gt / lt
    • 串接 Claude Desktop 或 Claude Code 測試

    v1:加入 Action Tools(2-4 週)

    • 實作 run_process(DocAction、報表、自動化)
    • 實作 create_record / update_record(POST/PUT)
    • 處理 mandatory field 偵測(自動查 AD_Column WHERE IsMandatory=’Y’)
    • token 自動 refresh(1 小時過期)
    • 串接公司 LLM Gateway,流量都過 audit

    v2:Form Tools + 進階 Window 互動(4-8 週)

    • 包裝 api/v1/windows/{slug}/tabs/{slug}
    • 讓 AI 模擬「打開視窗、切分頁、設欄位、按按鈕」
    • 處理複雜流程(發票核銷、應收沖帳等)
    • 整合 Allowed MCP Clients 白名單機制

    對中小企業:v0 可能就夠用 80%。對中大型集團:v0 → v1 → v2 漸進式投資,12-16 週完整版可上線。

    十一、結語:把 AI 變成 ERP 的執行單元

    前四篇腦子系統的 AI 仍然是「跟業務資料分開的工具」 — 員工問問題,AI 回答。

    加上 iDempiere MCP Server,AI 變成能直接動 ERP 的執行單元:查訂單、開請款單、跑 process、生報表。員工從 chat app 一句話完成原本要打開 ERP 點 5 個選單的工作。

    這才是「AI 時代不做 UI,做給 AI 安全取資料的入口」的真實落地。RD 不再被 UI 工單吃掉,而 80% 不寫 code 的員工終於能用一句中文操作 ERP。

    對企業 IT 主管的具體下一步:

    1. 裝 bxservice/idempiere-rest plugin 到既有 iDempiere(若還沒)
    2. 用 Postman 測 4 個主要 endpoint(repo 內有 collection)
    3. 用本文 v0 範例寫 MCP server,跑在開發機 localhost
    4. 掛到 Claude Desktop / Claude Code 試用,驗證權限層運作
    5. 確認可用後,搬上公司內網,接入 LLM Gateway

    延伸閱讀:腦子系統四部曲 + 本篇

    可運作的 Reference Links(2026/5 撰文時驗證)

    iDempiere 官方資源

    MCP 官方資源

    業界 ERP MCP server 參考

    OData 標準

  • 後端老兵的工具箱:C# 非同步、Python 逆向工程、架構選型實戰

    後端老兵的工具箱:C# 非同步、Python 逆向工程、架構選型實戰

    寫了十幾年後端,從 SQL Server DBA 一路走到架構師,我發現後端開發實戰能力的核心從來不只是會寫 SQL。真正拉開差距的,是你工具箱裡有多少把不同的扳手。這篇文章整理了我在 dotblogs 上累積的幾個關鍵主題:C# 非同步模式、Python 逆向工程實戰、資料庫架構選型、壓力測試,以及從 DBA 到架構師的技術演進。每一個主題都不是教科書式的介紹,而是實際踩過坑之後的心得。後端開發實戰最重要的一課:不要只讀,要動手測。

    TL;DR 重點摘要

    • C# 非同步控制:SemaphoreSlim 比 lock 更適合 async 場景,BlockingCollection 是 producer-consumer 的標準解法,別再自己造輪子。
    • Python 逆向工程:當 API 太貴,直接逆向 Web 介面的 JSON 回應是可行路線,但要做好延遲控制與路徑文件化。
    • 架構選型:資料庫不是選最潮的,而是看存取模式決定 — 高頻讀用 cache、搜尋用倒排索引、日誌用列式儲存。
    • 壓測不是選配:沒壓測過的系統就是紙老虎,JMeter 的 Thread Group + Listener 是最基本的品質門檻。

    1. C# 非同步模式 — Semaphore 與 BlockingCollection

    在 .NET 後端開發中,非同步處理是繞不開的主題。當你有 100 個 HTTP 請求要同時發出去,但目標伺服器只能承受 10 個並發時,你需要的不是 lock,而是 SemaphoreSlim

    為什麼 Semaphore 比 lock 更適合 async?

    很多人習慣用 lock 來控制並發,但 lock 有一個致命問題:它不支援 async/await。你不能在 lock 區塊裡面 await,否則會收到編譯錯誤。即使你繞過去了(用 Monitor),async 的 continuation 可能在不同執行緒上執行,導致 unlock 失敗。

    SemaphoreSlim 則原生支援 WaitAsync(),專為 async 場景設計。它的心智模型是「停車場」:車位(permit)有限,滿了就在外面等,有車出來才放行。

    // SemaphoreSlim throttling concurrent HTTP requests
    public async Task<List<string>> FetchAllAsync(List<string> urls)
    {
        // Only allow 10 concurrent requests
        var semaphore = new SemaphoreSlim(10);
        var httpClient = new HttpClient();
        var tasks = urls.Select(async url =>
        {
            await semaphore.WaitAsync();
            try
            {
                var response = await httpClient.GetStringAsync(url);
                return response;
            }
            finally
            {
                semaphore.Release();
            }
        });
    
        var results = await Task.WhenAll(tasks);
        return results.ToList();
    }
    

    注意 finally 裡的 Release() — 不管成功或失敗都要釋放,否則 permit 會洩漏,最終所有請求都會卡住。

    BlockingCollection:Producer-Consumer 的標準解法

    BlockingCollection<T> 是 .NET 內建的執行緒安全佇列,底層預設使用 ConcurrentQueue。它最強大的特性是 GetConsumingEnumerable() — consumer 端可以用 foreach 持續等待新資料,直到 producer 呼叫 CompleteAdding()

    // Producer-Consumer pattern with BlockingCollection
    var queue = new BlockingCollection<WorkItem>(boundedCapacity: 100);
    
    // Producer thread
    Task.Run(() =>
    {
        foreach (var item in GetWorkItems())
        {
            // Blocks if queue is full (back-pressure!)
            queue.Add(item);
            Console.WriteLine($"Produced: {item.Id}");
        }
        queue.CompleteAdding(); // Signal no more items
    });
    
    // Consumer thread
    Task.Run(() =>
    {
        // Blocks automatically when queue is empty
        // Exits when CompleteAdding() is called and queue is drained
        foreach (var item in queue.GetConsumingEnumerable())
        {
            ProcessItem(item);
            Console.WriteLine($"Consumed: {item.Id}");
        }
    });
    

    這裡的 boundedCapacity: 100 是關鍵 — 它提供了背壓(back-pressure)機制。當 consumer 處理速度跟不上 producer 時,佇列滿了 producer 就會被阻塞,而不是無限制地吃記憶體。這跟傳統 ThreadPool 的固定 worker 模型不同:ThreadPool 維護一組固定的執行緒,而 Semaphore + BlockingCollection 用的是「等待」機制,更彈性也更省資源。


    2. Python 逆向工程 — 當 API 太貴,我就自己拆

    2021 年我碰到一個需求:要把 2600 多筆地址轉成經緯度座標。Google Maps Geocoding API 當時每 1000 次要收 5 美元,算一算要十幾美元。對一個一次性的專案來說,這太不划算了。

    於是我打開 Chrome DevTools,觀察 Google Maps 搜尋框的網路請求。發現它回傳的不是標準 JSON API,而是一個巢狀極深的陣列結構。座標藏在類似 d[16][0][0][7][1][3] 這種路徑裡。

    import requests
    import json
    import time
    import random
    
    def geocode_address(address):
        """Reverse-engineered Google Maps search to extract coordinates."""
        url = "https://www.google.com/maps/search/"
        params = {"q": address}
        headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                           "AppleWebKit/537.36 Chrome/91.0"
        }
    
        resp = requests.get(url, params=params, headers=headers)
        text = resp.text
    
        # The response contains a deeply nested JSON-like structure
        # Find the coordinate data block
        try:
            # Extract the nested array from response
            start = text.find("window.APP_INITIALIZATION_STATE")
            if start == -1:
                return None
    
            # Parse the nested structure
            data_start = text.find("[", start)
            data_end = text.find("];", data_start) + 1
            raw = text[data_start:data_end]
    
            # Navigate the deeply nested path for coordinates
            parsed = json.loads(raw)
            # Path varies by response type, document it!
            lat = parsed[16][0][0][7][1][3][0]
            lng = parsed[16][0][0][7][1][3][1]
            return (lat, lng)
        except (IndexError, KeyError, json.JSONDecodeError):
            return None
    
    
    def batch_geocode(addresses):
        """Process addresses with random delays to avoid detection."""
        results = {}
        for i, addr in enumerate(addresses):
            coords = geocode_address(addr)
            results[addr] = coords
            print(f"[{i+1}/{len(addresses)}] {addr} -> {coords}")
    
            # Random delay: 3~15 seconds to mimic human behavior
            delay = random.uniform(3, 15)
            time.sleep(delay)
    
        return results
    

    幾個實戰經驗:

    • 隨機延遲是必須的:固定間隔很容易被偵測,3 到 15 秒的隨機延遲更像人類行為。
    • 文件化提取路徑d[16][0][0][7][1][3] 這種路徑完全沒有語意,Google 隨時可能改結構。你必須在程式碼旁邊寫清楚這條路徑代表什麼,否則三個月後你自己也看不懂。
    • 錯誤處理要寬容:有些地址搜不到、有些回傳結構不同,用 try/except 包住並記錄失敗的地址,事後手動補。
    • 這不是長期方案:逆向工程的結果隨時會因為前端改版而失效,只適合一次性或低頻的資料收集。

    最終我用這個方法在一個週末處理完 2600 多筆地址,成本是零。但我也很清楚:如果這是一個需要長期維護的服務,老老實實付 API 費用才是正道。


    3. 架構選型 — 不同系統該用什麼資料庫?

    2021 年我在部落格上寫了一系列「架構師慢慢學」的文章,其中最受歡迎的是資料庫選型。核心觀點只有一句話:資料庫的選擇應該由存取模式驅動,而不是技術潮流。

    太多團隊因為「大家都在用 MongoDB」就把所有東西塞進去,結果需要 JOIN 時痛不欲生。也有人因為「Redis 很快」就把所有資料都放 cache,然後面對一致性問題束手無策。

    系統類型 vs 資料庫選型對照表

    系統類型 建議資料庫 原因
    後台管理系統(Admin) RDBMS(PostgreSQL / SQL Server) 低併發、需要複雜查詢和 JOIN、資料完整性優先
    高流量前台系統 RDBMS 後端 + Redis 前端 RDBMS 保證資料正確,Redis 用 Key/Value 加速讀取
    日誌 / Log 系統 列式儲存(ClickHouse)+ 倒排索引(Elasticsearch) 寫入量大、需要聚合分析和全文搜尋
    交易系統 RDBMS + Cache + 一致性協議 ACID 不可妥協,cache 用於讀加速但需要失效策略
    即時監控系統 時序資料庫(InfluxDB / TimescaleDB) 時間序列寫入優化、自動聚合降精度
    搜尋系統 RDBMS 後端 + Elasticsearch 前端 RDBMS 為資料源,ES 提供倒排索引加速模糊搜尋

    幾個決策原則:

    1. 先問讀寫比例:讀多寫少 → 考慮加 cache;寫多讀少 → 考慮列式儲存或訊息佇列緩衝。
    2. 再問一致性需求:金融交易不能最終一致,社群按讚可以。
    3. 最後問查詢模式:需要 JOIN → RDBMS;需要全文搜尋 → 倒排索引;需要時間範圍聚合 → 時序 DB。

    這不是什麼高深的理論,但我看過太多團隊在第一步就跳過去,直接被「這個技術很紅」帶著走。


    4. JMeter 壓測 — 不壓測的系統都是紙老虎

    你寫的 API 在開發機上跑得飛快,但上線後 100 人同時用就當機了。這種事我見過不止一次。壓力測試不是「有空再做」的事情,它是品質門檻。

    JMeter 基本設定

    Apache JMeter 是免費的壓測工具,核心概念只有三個:

    • Thread Group(執行緒群組):模擬多少使用者同時操作。設定 Number of Threads = 100 就是 100 個並發使用者。
    • Sampler(取樣器):每個使用者要做什麼動作。最常用的是 HTTP Request Sampler,填入 URL、Method、Body 就行。
    • Listener(監聽器):收集結果的報表。Summary Report 給你吞吐量和錯誤率,Aggregate Report 給你百分位數響應時間。

    關鍵指標怎麼看

    指標 意義 健康標準(參考)
    Throughput 每秒處理的請求數 依業務而定,但應隨並發數線性增長直到瓶頸
    P90 Response Time 90% 的請求在此時間內完成 一般 API < 500ms
    P99 Response Time 99% 的請求在此時間內完成 應 < P90 的 3 倍,否則有長尾問題
    Error Rate 失敗請求的百分比 < 0.1% 為優秀,> 1% 要警覺

    常見錯誤

    1. 在同一台機器上跑 JMeter 和被測服務:JMeter 本身也吃 CPU 和記憶體,會互相干擾。壓測機和被測機必須分開。
    2. 沒有暖機(Warm-up):JVM 或 .NET 的 JIT 編譯在前幾次請求時會比較慢,應該先跑一輪不計入結果的請求。
    3. 只看平均值:平均響應時間 200ms 看起來很好,但如果 P99 是 5 秒,代表每 100 個使用者就有 1 個等 5 秒。看百分位數才有意義。
    4. 不模擬真實場景:所有人都打同一個 API endpoint 不代表真實負載。應該混合不同操作的比例。

    我的習慣是:在專案中期就開始跑基準壓測,而不是上線前才慌張地補。早發現瓶頸,修復成本低十倍。


    5. 從 DBA 到架構師 — 我的技術演進路線

    我的技術路線不是一開始就規劃好的,而是一步步堆疊出來的:

    • 2020:SQL Server DBA — 每天看執行計劃、調索引、處理 deadlock。這個階段讓我理解了資料庫內部的儲存引擎、鎖機制、B-Tree 索引結構。
    • 2021:.NET 後端開發 — 開始寫 C# Web API,發現 DBA 背景讓我寫出的 SQL 比大多數工程師都好。但也發現自己在非同步、設計模式上的不足。
    • 2022-2023:全端開發 — 接觸前端、Python、爬蟲、自動化。工具箱從一把螺絲起子變成一整個工具箱。
    • 2024-2026:架構設計 — 開始做系統設計、技術選型、效能規劃。發現以前每個階段的經驗都在這裡匯聚。

    DBA 背景帶給我的不公平優勢

    理解資料庫內部運作,會從根本上改變你寫應用程式的方式:

    • 你知道 SELECT * 在有 covering index 時多浪費多少 I/O,所以你會主動只選需要的欄位。
    • 你知道 NVARCHARVARCHARDATALENGTH() 下的差異,所以你會根據實際資料選擇正確的型別。
    • 你看過太多 table scan 的慘案,所以你設計 API 時會強制分頁,而不是讓使用者一次撈全部。
    • 你理解 transaction isolation level 的差異,所以你知道什麼時候用 READ COMMITTED SNAPSHOT 可以大幅降低鎖爭用。

    給後端工程師的建議

    如果你想往架構方向發展,我的建議是:學任何東西都要動手測試

    • 想搞懂索引?建一個百萬筆的測試表,比較有索引和沒索引的執行計劃。
    • 想搞懂 SemaphoreSlim?寫一個 console app,開 1000 個 task,觀察不同 permit 數量的效果。
    • 想搞懂資料庫選型?不要只讀比較文章,自己用 Docker 裝一個 ClickHouse、一個 PostgreSQL,塞相同的資料,跑相同的查詢,比較速度。

    2021 年我在部落格上寫「架構師慢慢學」系列時,最大的收穫不是寫出來的文章,而是為了寫文章去做的那些實驗。讀十篇文章不如自己跑一次 DATALENGTH() 比較 CHARVARCHAR 的儲存差異。那個數字會刻在你腦子裡,比任何文章都深。


    結語

    後端工程師的價值不在於精通某一個框架或語言,而在於工具箱的廣度和深度。C# 的非同步模式讓你處理高並發,Python 的靈活性讓你快速解決一次性問題,架構選型的思維讓你做出正確的技術決策,壓力測試讓你對系統有信心。

    這些東西沒辦法在一天內學會,但每一個都值得你花時間去實驗。畢竟,不壓測的系統是紙老虎,不動手的學習是紙上談兵

    如果你也是從 DBA 或其他專精領域起步的工程師,不要覺得自己起步晚。每一個階段的深度經驗,都會在你走向架構師的路上成為別人沒有的武器。

  • LangGraph 多模型實戰:從零到 Production 的完整教學

    重點摘要

    • LangGraph 讓你把不同 AI 模型串成自動化流水線:Claude 負責「想」,Groq Llama 負責「寫」,各司其職
    • 本文從零開始,帶你走完四個階段:基本流水線 → 智慧路由 → Streaming + 容錯 → FastAPI 部署
    • 每個階段都有完整可執行的程式碼,跟著做就能跑
    • 核心區別:Claude Code / Cursor 是「你的工具」,LangGraph 是「你造工具的材料」——當你要造產品給別人用時才需要它
    • 實測結果:比全用 Claude Sonnet 省 75% 成本,同時保留深度分析的品質

    這篇文章是給誰看的?

    如果你符合以下任一情況,這篇文章就是為你寫的:

    • 你想做一個 AI chatbot(客服、內部助手、產品功能),但不知道怎麼開始
    • 你已經在用 Claude / GPT,但覺得全部用最貴的模型太浪費錢
    • 你聽過「多模型協作」但不確定具體怎麼實作
    • 你想知道 LangGraph 跟你平常用的 Claude Code / Cursor / Copilot 到底差在哪

    本文從零開始,帶你走完四個階段,每個階段都有完整可執行的程式碼。你可以在任何一個階段停下來——不是每個人都需要走到 production。

    先搞清楚:這跟 Claude Code / Cursor / Copilot 完全不同

    在往下讀之前,先釐清最容易搞混的一件事:

    面向 Claude Code / Cursor / Copilot LangGraph
    本質 開發者工具 — 你用它寫 code 開發框架 — 你用它造產品
    使用者是誰 你自己(開發者) 你的客戶 / 你的系統 / 你的團隊
    使用場景 日常寫程式、debug、重構 建 chatbot、自動化流程、API 服務
    互動方式 人 ↔ AI 即時對話 程式碼自動跑,可以無人值守
    模型選擇 工具幫你選好(通常固定一個) 你自己決定哪個步驟用哪個模型
    計費方式 月費訂閱(工具包了) 純 API 按量計費,你自己控制成本

    用一個比喻:Claude Code 是你請了一個很強的工程師坐在旁邊幫你寫 code;LangGraph 是你在蓋一條自動化產線,產線上有不同的機器人各司其職

    如果你只是日常寫程式,用 Claude Code 就好,不需要 LangGraph。往下讀之前,確認你的需求是「造一個東西給別人用」,而不是「讓自己寫 code 更快」。

    為什麼不同的 AI 模型要分工合作?

    沒有一個模型什麼都最好。每個模型有不同的強項和定價:

    能力 Claude Sonnet Groq Llama 70B Groq Llama 8B
    架構設計 / Code Review ⭐ 最強 普通
    程式碼生成 ⭐ 強且快 普通
    簡單問答 大材小用 大材小用 ⭐ 夠用且極便宜
    生成速度 ~50 tok/s ~300 tok/s ~800 tok/s
    費用 (Output/1M tokens) $15.00 $0.79 $0.08

    核心邏輯:讓擅長「想」的模型去想,讓擅長「做」的模型去做,讓便宜的模型處理瑣事。就像軟體團隊裡,架構師出規格、工程師寫程式、實習生回答簡單問題一樣。

    LangGraph 是什麼?一分鐘看懂

    LangGraph 是 LangChain 團隊開發的有狀態流程編排框架。三個核心概念:

    1. Node(節點)— 一個步驟,比如「用 Sonnet 設計」或「用 Llama 實作」
    2. Edge(邊)— 步驟之間的連線,決定「做完 A 接著做 B」
    3. State(狀態)— 所有節點共享的資料,比如設計規格、程式碼、審查結果

    把這三個組合起來,就是一個可以自動跑的流水線。你定義「什麼步驟做什麼、什麼條件走什麼路」,框架負責執行。

    環境準備(所有階段共用)

    開始之前,你需要準備兩個 API key 和安裝套件。這一步做完,後面四個階段都不用再設定。

    1. 取得 API Key(兩個都有免費額度)

    ⚠️ 注意:這是 API key,跟 Claude.ai 的月費訂閱、GitHub Copilot 的訂閱完全無關。API 是按用量計費的。

    2. 建立專案

    mkdir langgraph-duo && cd langgraph-duo
    
    # 安裝套件
    pip install langgraph langchain-anthropic langchain-groq python-dotenv
    
    # 建立 .env 檔案(填入你的 key)
    cat > .env << 'EOF'
    ANTHROPIC_API_KEY=sk-ant-xxxx
    GROQ_API_KEY=gsk_xxxx
    EOF

    3. 驗證連線

    python3 -c "
    from dotenv import load_dotenv; load_dotenv()
    from langchain_anthropic import ChatAnthropic
    from langchain_groq import ChatGroq
    
    sonnet = ChatAnthropic(model='claude-sonnet-4-6', max_tokens=50)
    llama = ChatGroq(model='llama-3.3-70b-versatile', max_tokens=50)
    
    print('Sonnet:', sonnet.invoke('Say OK').content)
    print('Llama:', llama.invoke('Say OK').content)
    "

    兩行都印出 OK,就可以開始了。

    第一階段:Duo 流水線(設計 → 實作 → 審查)

    目標:讓 Claude Sonnet 設計規格,Groq Llama 寫程式碼,Sonnet 再審查品質。審查不通過自動重做,最多 3 次。

    適合場景:批量生成程式碼、自動化 code review、需要品質把關的程式碼生成。

    Task → [Sonnet 設計] → [Llama 實作] → [Sonnet 審查]
                                ↑               |
                                └── 未通過 ──────┘  (最多 3 次)
                                      ↓ 通過
                                     END

    完整程式碼:duo.py

    from typing import TypedDict
    from dotenv import load_dotenv
    from langgraph.graph import StateGraph, END
    from langchain_anthropic import ChatAnthropic
    from langchain_groq import ChatGroq
    from langchain_core.messages import HumanMessage, SystemMessage
    
    load_dotenv()
    
    # 模型分工:Sonnet 想,Llama 做
    sonnet = ChatAnthropic(model="claude-sonnet-4-6", max_tokens=4096)
    llama = ChatGroq(model="llama-3.3-70b-versatile", max_tokens=4096)
    
    # 所有節點共享的狀態
    class AgentState(TypedDict):
        task: str             # 原始需求
        design: str           # Sonnet 的設計規格
        code: str             # Llama 的實作
        review: str           # Sonnet 的審查結果
        revision_notes: str   # 修改指示(給重試用)
        approved: bool        # 是否通過
        attempt: int          # 重試次數
    
    MAX_ATTEMPTS = 3
    
    # Node 1: Sonnet 設計
    def design_node(state):
        response = sonnet.invoke([
            SystemMessage(content="You are a senior architect. Produce a precise technical spec with function signatures, edge cases, and pseudocode."),
            HumanMessage(content=f"Request: {state['task']}")
        ])
        return {"design": response.content}
    
    # Node 2: Llama 實作
    def implement_node(state):
        prompt = f"Spec:\n{state['design']}"
        if state.get("revision_notes"):
            prompt += f"\n\nFix these issues:\n{state['revision_notes']}"
        response = llama.invoke([
            SystemMessage(content="Implement per spec. Output only Python code."),
            HumanMessage(content=prompt)
        ])
        return {"code": response.content, "attempt": state.get("attempt", 0) + 1}
    
    # Node 3: Sonnet 審查
    def review_node(state):
        response = sonnet.invoke([
            SystemMessage(content="Review code vs spec. Reply VERDICT: APPROVED or REJECTED with details."),
            HumanMessage(content=f"Spec:\n{state['design']}\n\nCode:\n{state['code']}")
        ])
        review = response.content
        approved = "APPROVED" in review.upper()
        return {"review": review, "approved": approved,
                "revision_notes": "" if approved else review}
    
    # 條件路由:通過 → 結束,未通過 → 回去重做
    def should_continue(state):
        if state["approved"] or state["attempt"] >= MAX_ATTEMPTS:
            return "end"
        return "revise"
    
    # 組裝 Graph
    workflow = StateGraph(AgentState)
    workflow.add_node("design", design_node)
    workflow.add_node("implement", implement_node)
    workflow.add_node("review", review_node)
    workflow.set_entry_point("design")
    workflow.add_edge("design", "implement")
    workflow.add_edge("implement", "review")
    workflow.add_conditional_edges("review", should_continue,
                                   {"end": END, "revise": "implement"})
    app = workflow.compile()
    
    # 跑!
    result = app.invoke({
        "task": "Write a Python function that reads a CSV and returns column averages as a dict",
        "design": "", "code": "", "review": "",
        "revision_notes": "", "approved": False, "attempt": 0,
    })
    
    print("=== CODE ===")
    print(result["code"])
    print(f"\n{'✅ APPROVED' if result['approved'] else '⚠️ BEST EFFORT'} after {result['attempt']} attempt(s)")

    實測結果

    階段 模型 結果
    🧠 Design Claude Sonnet 4.6 產出 4 個參數、10 個 edge case、9 步 pseudocode 的完整規格
    ⚡ Implement Groq Llama 70B 完整 Python 函數,含 type hints、docstring、error handling
    🔍 Review Claude Sonnet 4.6 VERDICT: APPROVED — 第一次就通過,沒有重試

    到這裡你已經有一個能跑的多模型流水線了。如果你的需求是「批量生成程式碼 + 自動品質把關」,可以停在這個階段。

    第二階段:Smart Router(自動選模型的聊天機器人)

    目標:做一個像 ChatGPT 一樣的對話介面,但底下不是固定一個模型,而是自動根據問題類型選最適合的模型。

    適合場景:客服 chatbot、團隊內部 AI 助手、需要控制 API 成本的聊天服務。

    你的問題 → [Llama 8B 分類器] → 判斷類型
                                      |
                      ┌────────────────┼────────────────┐
                      ↓                ↓                ↓
                [Claude Sonnet]  [Llama 70B]      [Llama 8B]
                 深度分析          寫程式            簡單問答
                      ↓                ↓                ↓
                      └────────────────┼────────────────┘
                                       ↓
                                    回答你

    分類器怎麼運作?

    分類器用最便宜的 Llama 8B(每次呼叫不到 $0.0001)讀取問題,輸出一個 JSON 判斷結果。分類規則:

    分類 路由到 觸發條件 費用 (Output/1M)
    🧠 深度思考 Claude Sonnet 架構分析、比較權衡、code review、規劃 $15.00
    ⚡ 寫程式 Llama 70B 實作函數、生成腳本、重構、修 bug $0.79
    💨 快速回答 Llama 8B 打招呼、簡單問答、定義、基礎算數 $0.08

    核心程式碼

    import json
    from langgraph.graph import StateGraph, END
    
    # 分類器:Llama 8B 讀問題,判斷該走哪條路
    def router_node(state):
        response = llama_8b.invoke([
            SystemMessage(content="""Classify into one category:
    - "sonnet": complex reasoning, analysis, architecture
    - "llama_70b": write code, implement, fix bugs
    - "llama_8b": greetings, simple facts, casual chat
    Reply ONLY JSON: {"route": "...", "reason": "..."}"""),
            HumanMessage(content=state["question"])
        ])
        parsed = json.loads(response.content)
        return {"route": parsed["route"]}
    
    # 條件路由:根據分類結果,走不同的回答節點
    workflow = StateGraph(RouterState)
    workflow.add_node("router", router_node)
    workflow.add_node("sonnet", answer_sonnet)
    workflow.add_node("llama_70b", answer_llama_70b)
    workflow.add_node("llama_8b", answer_llama_8b)
    
    workflow.set_entry_point("router")
    workflow.add_conditional_edges("router", lambda s: s["route"], {
        "sonnet": "sonnet",
        "llama_70b": "llama_70b",
        "llama_8b": "llama_8b",
    })
    app = workflow.compile()

    實測分類準確度:7/7

    問題 路由結果 正確?
    Hi! 💨 Llama 8B
    1+1=? 💨 Llama 8B
    Write a Python quicksort ⚡ Llama 70B
    Compare microservices vs monolith 🧠 Sonnet
    What is Docker? 💨 Llama 8B
    幫我寫一個 REST API ⚡ Llama 70B
    分析 Redis cache vs CDN 優缺點 🧠 Sonnet

    到這裡你有一個能自動選模型的聊天機器人了。但它還缺兩個東西:回答會一次全部吐出(不是逐字顯示),而且 Groq 掛了就整個壞掉。第三階段解決這兩個問題。

    第三階段:Streaming + Fallback(讓它不會掛)

    目標:回答逐字出現(像 ChatGPT 一樣流暢),而且模型掛了自動切換備用模型。

    為什麼這一步很重要:沒有 Streaming 的 chatbot,使用者體驗像 2010 年的網頁——按下送出,等 5 秒,突然一大段文字出現。沒有 Fallback 的服務,Groq 一限速(免費版每分鐘 30 次),你的整個服務就掛了。

    Streaming:體感延遲降 25 倍

    方式 使用者體驗 程式碼差別
    invoke()(非串流) 等 5 秒 → 突然出現整段文字 response = model.invoke(messages)
    stream()(串流) 0.2 秒開始出字 → 像打字一樣流暢 for chunk in model.stream(messages)
    # 只需要把 .invoke() 改成 .stream()
    # 然後迭代每個 chunk 即時輸出
    
    for chunk in model.stream(messages):
        print(chunk.content, end="", flush=True)  # flush=True 強制即時顯示

    Fallback:模型掛了自動切換

    主要模型 Fallback 模型 切換代價
    🧠 Sonnet ⚡ Llama 70B 分析品質略降,速度更快
    ⚡ Llama 70B 🧠 Sonnet 速度略慢,品質更高
    💨 Llama 8B ⚡ Llama 70B 稍慢稍貴,但一定能回答
    # Fallback 模式:try 主要模型,失敗自動切備用
    try:
        for chunk in primary_model.stream(messages):
            print(chunk.content, end="", flush=True)
    except Exception:
        print("⚠️ Primary failed, switching to fallback...")
        for chunk in fallback_model.stream(messages):
            print(chunk.content, end="", flush=True)

    完整的 router_stream.py 把 Streaming + Fallback + 對話歷史 + 使用統計整合在一起,不到 200 行。跑起來就是一個帶有自動模型切換的 terminal 聊天機器人。

    到這裡你有一個穩定的、體驗流暢的聊天機器人了。但它還是跑在你的 terminal 裡,只有你能用。第四階段讓它變成任何人都能呼叫的 API 服務。

    第四階段:FastAPI 部署(讓別人能用)

    目標:把 chatbot 包成 HTTP API,讓網頁、App、Line bot、Slack bot 都能呼叫。

    為什麼是 FastAPI:Python 生態最主流的 API 框架,原生支援 async、自動生成 API 文件、社群龐大。

    API 端點設計

    端點 方式 適合場景
    POST /chat 非串流,回傳完整 JSON Slack/Line bot、後端呼叫、批次處理
    POST /chat/stream SSE 串流,逐 token 推送 網頁聊天窗、需要即時體感的 UI
    GET /health 健康檢查 負載平衡器、監控系統
    GET /sessions/{id} 取得對話歷史 Debug、對話紀錄查詢
    DELETE /sessions/{id} 清除對話 使用者開始新對話

    啟動與測試

    # 安裝額外套件
    pip install fastapi uvicorn sse-starlette
    
    # 啟動 server
    python server.py
    # → 🤖 Smart Router API running on http://localhost:8900
    
    # 測試非串流
    curl -X POST http://localhost:8900/chat \
      -H "Content-Type: application/json" \
      -d '{"message": "什麼是 Docker?", "session_id": "test-user"}'
    
    # 回應範例:
    # {
    #   "answer": "Docker 是一個容器化平台...",
    #   "model_used": "llama_8b",
    #   "route_reason": "simple definition question",
    #   "session_id": "test-user",
    #   "fallback_used": false,
    #   "elapsed_seconds": 0.85
    # }

    前端怎麼接 SSE 串流?

    如果你要做網頁聊天窗,前端只需要幾行 JavaScript:

    // 瀏覽器原生 EventSource API
    const response = await fetch('/chat/stream', {
      method: 'POST',
      headers: {'Content-Type': 'application/json'},
      body: JSON.stringify({message: '寫一個排序函數', session_id: 'user-1'})
    });
    
    const reader = response.body.getReader();
    const decoder = new TextDecoder();
    
    while (true) {
      const {done, value} = await reader.read();
      if (done) break;
      const text = decoder.decode(value);
      // 每個 chunk 到了就即時顯示在畫面上
      document.getElementById('chat').innerHTML += text;
    }

    為什麼用 SSE 而不是 WebSocket?

    聊天場景是「使用者送一次訊息、AI 回一次」的單向推送。SSE 比 WebSocket 更適合:

    • 更簡單 — 單向傳輸,不需要管雙向連線
    • 瀏覽器原生 — EventSource API 內建自動重連
    • 穿透力好 — 走標準 HTTP,能通過代理和 CDN
    • 夠用 — 使用者的訊息用 POST 送,AI 的回應用 SSE 推

    四個階段的完整對照

    階段 檔案 新增能力 適合誰 可以停在這嗎?
    1. Duo 流水線 duo.py 設計→實作→審查 批量生成程式碼
    2. Smart Router router.py 自動選模型 個人實驗、學習
    3. Streaming + Fallback router_stream.py 逐字輸出 + 自動容錯 團隊內部使用
    4. FastAPI 部署 server.py HTTP API + SSE + Session 對外服務、產品整合

    費用比較:到底能省多少?

    假設一天 100 個問題,其中 20% 深度分析、30% 寫程式、50% 簡單問答:

    方案 月成本估算 品質
    全部用 Claude Sonnet ~$54 最高,但簡單題大材小用
    Smart Router(自動切換) ~$13.50 深度題用 Sonnet,其餘用 Llama
    全部用 Groq Llama 70B ~$4.20 最便宜,但分析品質弱

    Smart Router 方案比全用 Sonnet 省 75%,同時保留了深度分析任務的 Sonnet 品質。規模越大差距越明顯——1000 個使用者的 SaaS 產品,月省 $3000+。

    什麼場景適合?什麼場景不適合?

    ✅ 適合的場景

    1. 客服 Chatbot — 70% 簡單題用 Llama 8B 秒回,複雜題自動升級到 Sonnet
    2. 團隊 AI 助手 — 接 Slack,PM 問策略用 Sonnet,工程師要 code 用 Llama 70B
    3. 自動化 Pipeline — CI/CD 中的 AI code review,PR 提交自動跑
    4. SaaS 產品 — 加 AI 功能但要控成本,簡單摘要用 Llama,深度分析用 Sonnet
    5. 批量內容生成 — 50 篇產品描述:Sonnet 定規範 → Llama 批量寫 → Sonnet 抽檢

    ❌ 不適合的場景

    • 日常寫程式 — 用 Claude Code 或 Cursor 就好,不需要 LangGraph
    • 一次性分析 — 直接貼給 Claude 問就好,不需要搭 pipeline
    • 不需要控成本 — 個人使用月花不到 $10,Smart Router 省下的錢不值得建置成本

    三個問題判斷法

    在決定要不要用之前,問自己:

    1. 使用者是誰? — 你自己 → 不需要。別人(客戶/團隊/系統)→ 繼續看
    2. 會跑多少次? — 幾次 → 直接呼叫 API。幾百幾千次 → LangGraph 有意義
    3. 需要品質分級嗎? — 所有問題都要最高品質 → 用最強模型。不同問題可以不同品質 → Smart Router

    三個都答「後者」才值得用 LangGraph。

    走完四個階段之後,還有什麼?

    如果你的服務要上正式商業環境,還有幾個面向需要處理:

    面向 做什麼 不做的後果
    RAG(檢索增強) 接向量資料庫,讓 AI 查你的文件回答 AI 只能回答通用知識,不懂你的業務
    評估(LangSmith) 追蹤每次呼叫的路由、延遲、成本 不知道 Router 分類準不準,成本失控
    Session 持久化 用 Redis 存對話歷史(目前在記憶體) Server 重啟,所有對話消失
    認證 API key 或 JWT 驗證 任何人都能呼叫你的 API,幫你花錢
    Prompt Injection 防護 驗證使用者輸入,防止惡意 prompt 使用者讓 AI 做不該做的事
    Docker 容器化 打包成 Docker image 部署 環境不一致,部署困難
    Subgraph 嵌套 Router 判斷「寫程式」→ 啟動整個 Duo 流水線 只能單步回答,沒有品質把關

    完整專案結構

    langgraph-duo/
    ├── .env                  # API keys(不要 commit)
    ├── .gitignore            # 排除 .env
    ├── requirements.txt      # 所有套件
    ├── duo.py                # 階段 1:設計→實作→審查 流水線
    ├── duo_notebook.ipynb    # 階段 1 的 Jupyter 版
    ├── router.py             # 階段 2:Smart Router 基本版
    ├── router_stream.py      # 階段 3:+ Streaming + Fallback
    └── server.py             # 階段 4:FastAPI HTTP API

    所有程式碼都有詳細的中英文註解,說明每個模型的選用原因和適用場景。從哪個階段開始都可以,每個檔案都是獨立可執行的。

    總結

    LangGraph 多模型流水線的核心價值不是「用 AI 寫程式更快」——如果只是速度,直接用一個模型最快。它的價值在於把 AI 當成團隊來管理:讓擅長設計的去設計、擅長實作的去實作、擅長審查的去審查。

    從 Duo 流水線到 Smart Router,再到帶有 Streaming、Fallback、API 部署的生產版本,每一步都是從「能跑」走向「能上線」的必經之路。你不需要一次走完四個階段——先跑通第一階段,確認這個架構對你有用,再往下走。

    相關閱讀:LangChain/LangGraph 深度分析:架構師、顧問、個人公司的實戰指南 | Claude Code Agent Teams:從穩定執行到自動化代碼審查的完整指南 | 舊系統整合場景下,會用 vs 不會用 Claude Code 的差距