作者: tm731531

  • PDF 生成最佳實踐:Builder 模式的優雅 API

    商業價值:統一的 PDF 生成讓「出貨單、撿貨單一鍵產生」,直接支撐 導讀篇提到「處理速度提升 10 倍」——從每單 3 分鐘縮短到批次 1 秒。

    前言:電商系統的 PDF 需求

    OMS 系統需要產生各種 PDF 文件:

    文件類型 用途 特殊需求
    出貨標籤 貼在包裹上 條碼、收件人資訊
    訂單明細 放在包裹內 商品清單、金額
    撿貨單 倉庫作業 商品位置、數量
    對帳單 商戶結算 表格、統計
    挑戰:傳統 PDF 產生程式碼冗長、難以維護

    解決方案:Builder 模式

    使用範例(先看效果)

    // 產生出貨單 PDF
    byte[] pdf = PdfBuilder.create()
    .title(“出貨單”)
    .separator()
    .text(“訂單編號:ORD-2024-001234”)
    .text(“出貨日期:2024-03-18”)
    .newLine()
    .subtitle(“商品明細”)
    .table(orderItems)
    .newLine()
    .subtitle(“收件資訊”)
    .text(“收件人:王小明”)
    .text(“電話:0912-345-678”)
    .text(“地址:台北市信義區信義路五段7號”)
    .newPage()
    .image(barcodeBytes)
    .build();
    效果:程式碼像文件結構一樣直覺,易讀易維護

    PdfBuilder 實作

    核心結構

    public class PdfBuilder {

    private List<Element> elements = new ArrayList<>();
    private FontConfig fontConfig;

    // 工廠方法
    public static PdfBuilder create() {
    return new PdfBuilder(FontConfig.defaultConfig());
    }

    public static PdfBuilder create(FontConfig config) {
    return new PdfBuilder(config);
    }

    private PdfBuilder(FontConfig config) {
    this.fontConfig = config;
    }
    }

    文字方法

    /**
    * 標題(最大字體)
    */

    public PdfBuilder title(String text) {
    elements.add(new TextElement(text, fontConfig.getTitleFont()));
    return this; // 回傳 this 支援鏈式呼叫
    }

    /**
    * 副標題
    */

    public PdfBuilder subtitle(String text) {
    elements.add(new TextElement(text, fontConfig.getSubtitleFont()));
    return this;
    }

    /**
    * 一般文字
    */

    public PdfBuilder text(String text) {
    elements.add(new TextElement(text, fontConfig.getBodyFont()));
    return this;
    }

    /**
    * 粗體文字
    */

    public PdfBuilder boldText(String text) {
    elements.add(new TextElement(text, fontConfig.getBoldFont()));
    return this;
    }

    版面控制

    /**
    * 換行
    */

    public PdfBuilder newLine() {
    elements.add(new NewLineElement());
    return this;
    }

    /**
    * 分隔線
    */

    public PdfBuilder separator() {
    elements.add(new SeparatorElement());
    return this;
    }

    /**
    * 換頁
    */

    public PdfBuilder newPage() {
    elements.add(new NewPageElement());
    return this;
    }

    表格支援

    /**
    * 新增表格
    * @param data 二維資料,第一列為標題
    */

    public PdfBuilder table(List<List<String>> data) {
    elements.add(new TableElement(data, TableStyle.BORDERED));
    return this;
    }

    /**
    * 新增表格(指定樣式)
    */

    public PdfBuilder table(List<List<String>> data, TableStyle style) {
    elements.add(new TableElement(data, style));
    return this;
    }

    圖片與條碼

    /**
    * 新增圖片
    */

    public PdfBuilder image(byte[] imageBytes) {
    elements.add(new ImageElement(imageBytes));
    return this;
    }

    /**
    * 新增條碼(自動產生)
    */

    public PdfBuilder barcode(String content) {
    byte[] barcodeImage = BarcodeGenerator.generate(content);
    elements.add(new ImageElement(barcodeImage));
    return this;
    }


    輸出 PDF

    /**
    * 產生 PDF byte 陣列
    */

    public byte[] build() {
    ByteArrayOutputStream output = new ByteArrayOutputStream();
    Document document = new Document(PageSize.A4);

    try {
    PdfWriter.getInstance(document, output);
    document.open();

    for (Element element : elements) {
    if (element instanceof NewPageElement) {
    document.newPage();
    } else {
    document.add(element.render());
    }
    }

    document.close();

    } catch (DocumentException e) {
    throw new PdfGenerationException(“PDF 產生失敗”, e);
    }

    return output.toByteArray();
    }

    /**
    * 直接寫入檔案
    */

    public void toFile(String filename) {
    byte[] pdf = build();
    try (FileOutputStream fos = new FileOutputStream(filename)) {
    fos.write(pdf);
    } catch (IOException e) {
    throw new PdfGenerationException(“檔案寫入失敗”, e);
    }
    }


    中文字型支援

    public class FontConfig {

    private static final String FONT_PATH = “/fonts/NotoSansTC-Regular.ttf”;
    private BaseFont baseFont;

    public static FontConfig defaultConfig() {
    return new FontConfig();
    }

    private FontConfig() {
    try {
    // 載入支援中文的字型
    baseFont = BaseFont.createFont(
    FONT_PATH,
    BaseFont.IDENTITY_H, // Unicode 支援
    BaseFont.EMBEDDED // 嵌入字型
    );
    } catch (Exception e) {
    throw new RuntimeException(“字型載入失敗”, e);
    }
    }

    public Font getTitleFont() {
    return new Font(baseFont, 24, Font.BOLD);
    }

    public Font getSubtitleFont() {
    return new Font(baseFont, 18, Font.BOLD);
    }

    public Font getBodyFont() {
    return new Font(baseFont, 12, Font.NORMAL);
    }

    public Font getBoldFont() {
    return new Font(baseFont, 12, Font.BOLD);
    }
    }


    完整範例:出貨單

    public byte[] generateShippingLabel(Order order) {
    // 準備商品明細表格
    List<List<String>> items = new ArrayList<>();
    items.add(List.of(“商品”, “數量”, “單價”, “小計”));

    for (OrderItem item : order.getItems()) {
    items.add(List.of(
    item.getName(),
    String.valueOf(item.getQuantity()),
    formatPrice(item.getPrice()),
    formatPrice(item.getSubtotal())
    ));
    }

    items.add(List.of(“合計”, “”, “”, formatPrice(order.getTotal())));

    // 產生 PDF
    return PdfBuilder.create()
    // 標題區
    .title(“出貨單”)
    .separator()
    .text(“訂單編號:” + order.getOrderId())
    .text(“出貨日期:” + LocalDate.now())
    .newLine()

    // 條碼
    .barcode(order.getOrderId())
    .newLine()

    // 商品明細
    .subtitle(“商品明細”)
    .table(items)
    .newLine()

    // 收件資訊
    .subtitle(“收件資訊”)
    .text(“收件人:” + order.getReceiverName())
    .text(“電話:” + order.getReceiverPhone())
    .text(“地址:” + order.getReceiverAddress())

    .build();
    }


    總結

    設計 效果
    Builder 模式 鏈式呼叫,程式碼簡潔
    流式 API 像寫 HTML 一樣直覺
    字型抽象 換字型只改一處
    元件化 表格、圖片、條碼都是元件

    為什麼不用其他方案?

    方案 優點 缺點 結論
    HTML 轉 PDF 會 HTML 就會用 排版難控制、分頁問題 簡單報表可用
    Word 範本 業務人員能改 需要額外軟體、格式問題 不推薦
    JasperReports 功能強大 學習曲線陡、設計器難用 複雜報表可考慮
    iText + Builder 程式碼控制、可測試 要寫程式碼 工程師友好

    實戰踩坑

    坑 1:中文字型問題

    預設字型不支援中文,產出的 PDF 全是方框。要嵌入支援中文的字型(如 Noto Sans TC)。第一次載入字型要 2-3 秒,後來改成應用程式啟動時預載。

    坑 2:出貨單格式每平台不同

    蝦皮、Momo、Yahoo 的出貨單長得不一樣。最初想做成一模一樣,後來發現不可能(平台會檢查格式)。解法:只統一「我們自己的出貨單」,平台的標籤用平台 API 下載。

    坑 3:大量 PDF 記憶體爆炸

    一次產生 500 張出貨單,記憶體直接爆。解法:改成串流處理,產一張輸出一張,不要全部放在記憶體。


    系列導航

    ◀ 上一篇
    HTTP 客戶端
    📚 返回目錄 下一篇 ▶
    JSON 序列化
  • HTTP 客戶端設計:OkHttp 連接池與多場景應用

    商業價值:穩定的 HTTP 客戶端讓系統「能可靠地跟 17 個平台通訊」,這是 導讀篇提到「99% 庫存準確率」的基礎——API 不穩定,庫存同步就會失敗。

    前言:呼叫外部 API 的挑戰

    多通路系統需要呼叫大量外部 API:

    API 類型 特性 挑戰
    蝦皮 API 流量限制嚴格 需要控制請求頻率
    物流 API 回應慢 需要較長逾時
    支付 API 高可靠性要求 需要重試機制
    問題:每次都建立新連線 → 效能差、資源浪費

    解決方案:OkHttp 連線池

    連線池設定

    @Configuration
    public class HttpClientConfig {

    @Bean
    public OkHttpClient okHttpClient() {
    return new OkHttpClient.Builder()
    // 連線池設定
    .connectionPool(new ConnectionPool(
    100, // 最大閒置連線數
    5, TimeUnit.MINUTES // 閒置時間
    ))

    // 逾時設定
    .connectTimeout(10, TimeUnit.SECONDS)
    .readTimeout(30, TimeUnit.SECONDS)
    .writeTimeout(30, TimeUnit.SECONDS)

    // 重試
    .retryOnConnectionFailure(true)

    .build();
    }
    }

    設定 說明
    maxIdleConnections 100 最多保持 100 條閒置連線
    keepAliveDuration 5 分鐘 閒置連線保持時間
    connectTimeout 10 秒 建立連線逾時
    readTimeout 30 秒 讀取回應逾時

    HTTP 客戶端封裝

    @Component
    public class HttpClientService {

    @Autowired
    private OkHttpClient okHttpClient;

    /**
    * GET 請求
    */

    public HttpResult get(String url, Map<String, String> headers) {
    Request request = new Request.Builder()
    .url(url)
    .headers(Headers.of(headers))
    .get()
    .build();

    return execute(request);
    }

    /**
    * POST 請求(JSON)
    */

    public HttpResult postJson(String url, Object body, Map<String, String> headers) {
    String json = JsonUtil.toJson(body);

    Request request = new Request.Builder()
    .url(url)
    .headers(Headers.of(headers))
    .post(RequestBody.create(json, MediaType.parse(“application/json”)))
    .build();

    return execute(request);
    }

    /**
    * POST 請求(Form)
    */

    public HttpResult postForm(String url, Map<String, String> params, Map<String, String> headers) {
    FormBody.Builder formBuilder = new FormBody.Builder();
    params.forEach(formBuilder::add);

    Request request = new Request.Builder()
    .url(url)
    .headers(Headers.of(headers))
    .post(formBuilder.build())
    .build();

    return execute(request);
    }

    private HttpResult execute(Request request) {
    try (Response response = okHttpClient.newCall(request).execute()) {
    return HttpResult.builder()
    .statusCode(response.code())
    .body(response.body() != null ? response.body().string() : null)
    .headers(response.headers().toMultimap())
    .success(response.isSuccessful())
    .build();

    } catch (IOException e) {
    return HttpResult.builder()
    .success(false)
    .errorMessage(e.getMessage())
    .build();
    }
    }
    }


    回應結果封裝

    @Data
    @Builder
    public class HttpResult {
    private boolean success;
    private int statusCode;
    private String body;
    private Map<String, List<String>> headers;
    private String errorMessage;

    /**
    * 解析 JSON 回應
    */

    public <T> T parseJson(Class<T> clazz) {
    if (!success || body == null) {
    return null;
    }
    return JsonUtil.fromJson(body, clazz);
    }

    /**
    * 解析 JSON 陣列回應
    */

    public <T> List<T> parseJsonList(Class<T> clazz) {
    if (!success || body == null) {
    return Collections.emptyList();
    }
    return JsonUtil.fromJsonList(body, clazz);
    }
    }


    追蹤 Header 傳遞

    支援分散式追蹤,自動傳遞追蹤 Header:

    @Component
    public class TracingHttpClient {

    private static final List<String> TRACING_HEADERS = List.of(
    “x-request-id”,
    “x-b3-traceid”,
    “x-b3-spanid”,
    “x-b3-parentspanid”,
    “x-b3-sampled”
    );

    @Autowired
    private HttpClientService httpClient;

    /**
    * 從當前請求提取追蹤 Header
    */

    public Map<String, String> extractTracingHeaders(HttpServletRequest request) {
    Map<String, String> headers = new HashMap<>();

    for (String name : TRACING_HEADERS) {
    String value = request.getHeader(name);
    if (value != null) {
    headers.put(name, value);
    }
    }

    return headers;
    }

    /**
    * 發送請求,自動帶入追蹤 Header
    */

    public HttpResult getWithTracing(String url, HttpServletRequest currentRequest) {
    Map<String, String> headers = extractTracingHeaders(currentRequest);
    return httpClient.get(url, headers);
    }
    }


    重試機制

    @Component
    public class RetryableHttpClient {

    @Autowired
    private HttpClientService httpClient;

    /**
    * 帶重試的請求
    */

    public HttpResult getWithRetry(String url, Map<String, String> headers, int maxRetries) {
    int attempt = 0;
    HttpResult result = null;

    while (attempt < maxRetries) {
    result = httpClient.get(url, headers);

    if (result.isSuccess()) {
    return result;
    }

    // 只對可重試的錯誤重試
    if (!isRetryable(result.getStatusCode())) {
    break;
    }

    attempt++;
    sleep(calculateBackoff(attempt));
    }

    return result;
    }

    private boolean isRetryable(int statusCode) {
    // 5xx 錯誤和 429 (Too Many Requests) 可重試
    return statusCode >= 500 || statusCode == 429;
    }

    private long calculateBackoff(int attempt) {
    // 指數退避:1秒, 2秒, 4秒…
    return (long) Math.pow(2, attempt – 1) * 1000;
    }
    }

    HTTP 狀態碼 是否重試 原因
    2xx 不需要 成功
    4xx (非 429) 不重試 客戶端錯誤,重試也沒用
    429 重試 流量限制,稍後重試
    5xx 重試 伺服器暫時錯誤

    使用範例

    @Service
    public class ShopeeApiClient {

    @Autowired
    private RetryableHttpClient httpClient;

    public List<ShopeeOrder> getOrders(String accessToken) {
    String url = “https://partner.shopeemobile.com/api/v2/order/get_order_list”;

    Map<String, String> headers = Map.of(
    “Authorization”, “Bearer “ + accessToken,
    “Content-Type”, “application/json”
    );

    HttpResult result = httpClient.getWithRetry(url, headers, 3);

    if (!result.isSuccess()) {
    throw new ApiException(“Shopee API 錯誤: “ + result.getErrorMessage());
    }

    return result.parseJsonList(ShopeeOrder.class);
    }
    }


    總結

    設計 效果
    連線池 復用連線,減少建立成本
    逾時設定 避免請求無限等待
    結果封裝 統一處理成功/失敗
    追蹤 Header 支援分散式追蹤
    重試機制 自動處理暫時性錯誤

    為什麼不用其他方案?

    方案 優點 缺點 結論
    HttpURLConnection JDK 內建 API 難用、功能少 不推薦
    Apache HttpClient 功能完整 API 複雜、依賴多 可用但重
    Spring RestTemplate Spring 整合好 已被標記為 maintenance 舊專案可用
    Spring WebClient 非同步、Reactive 學習曲線、除錯困難 Reactive 專案用
    OkHttp 輕量、效能好、API 簡潔 非 Spring 原生 同步 HTTP 首選

    實戰踩坑

    坑 1:連線池用完了

    早期沒設連線池,每次請求都建新連線。流量一大,系統噴 Connection refused。加上連線池後,效能提升 10 倍,問題消失。

    坑 2:逾時設太長

    最初 readTimeout 設 60 秒。某平台 API 掛了,執行緒都在等待,整個服務卡死。改成 30 秒 + 重試機制後,就算 API 慢也能處理。

    坑 3:沒處理 429 Too Many Requests

    蝦皮有流量限制,瘋狂打 API 會收到 429。最初沒處理,一直重試反而更慢。加上指數退避(1秒、2秒、4秒…)後,流量限制問題大幅改善。


    系列導航

    ◀ 上一篇
    DTO 設計
    📚 返回目錄 下一篇 ▶
    PDF 生成
  • DTO 地獄求生指南:管理數百個資料傳輸物件

    商業價值:良好的 DTO 設計讓「平台 API 變更只影響一個類別」,這是 導讀篇提到「新增平台 2-3 週上線」的技術基礎。

    前言:DTO 地獄

    在多通路系統中,每個平台的資料格式都不同:

    平台 訂單編號欄位 金額欄位 時間格式
    蝦皮 order_sn total_amount Unix timestamp
    Momo orderNo orderAmount yyyy/MM/dd HH:mm
    Yahoo OrderId TotalPrice ISO 8601
    PChome order_id amount yyyy-MM-dd
    問題:如果每個平台都用不同的 DTO,會有數百個類別,維護困難。

    解決方案:三層 DTO 架構

    設計原則:

    1. 外部 DTO:對應平台 API 的原始格式
    2. 內部 DTO:系統統一的資料格式
    3. 轉換器:負責格式轉換

    層次說明

    層次 命名規則 範例 用途
    外部 DTO {Platform}{Action}Request/Response ShopeeGetOrderResponse 對應平台 API
    內部 DTO {Entity}DTO OrderDTO 系統內部傳輸
    轉換器 {Platform}Converter ShopeeConverter 格式轉換

    外部 DTO:對應平台 API

    /**
    * 蝦皮訂單 API 回應(對應蝦皮 API 文件)
    */

    @Data
    public class ShopeeGetOrderResponse {

    // 蝦皮欄位名稱(底線命名)
    @JsonProperty(“order_sn”)
    private String orderSn;

    @JsonProperty(“order_status”)
    private String orderStatus;

    @JsonProperty(“total_amount”)
    private BigDecimal totalAmount;

    @JsonProperty(“create_time”)
    private Long createTime; // Unix timestamp

    @JsonProperty(“buyer_username”)
    private String buyerUsername;

    @JsonProperty(“item_list”)
    private List<ShopeeOrderItem> itemList;
    }

    /**
    * Momo 訂單 API 回應(對應 Momo API 文件)
    */

    @Data
    public class MomoGetOrderResponse {

    // Momo 欄位名稱(駝峰命名)
    private String orderNo;
    private String orderStatus;
    private BigDecimal orderAmount;
    private String orderDate; // yyyy/MM/dd HH:mm
    private String customerName;
    private List<MomoOrderItem> products;
    }


    內部 DTO:統一格式

    /**
    * 系統內部訂單 DTO(統一格式)
    */

    @Data
    @Builder
    public class OrderDTO {

    // 統一的欄位命名
    private String orderId;
    private String platformOrderId;
    private ChannelType channel;
    private OrderStatus status;
    private BigDecimal totalAmount;
    private ZonedDateTime createdAt; // 統一用 ZonedDateTime
    private String buyerName;
    private List<OrderItemDTO> items;
    }


    轉換器:格式轉換

    @Component
    public class ShopeeConverter {

    /**
    * 蝦皮格式 → 內部格式
    */

    public OrderDTO toOrderDTO(ShopeeGetOrderResponse response) {
    return OrderDTO.builder()
    .platformOrderId(response.getOrderSn())
    .channel(ChannelType.SHOPEE)
    .status(mapStatus(response.getOrderStatus()))
    .totalAmount(response.getTotalAmount())
    .createdAt(convertTimestamp(response.getCreateTime()))
    .buyerName(response.getBuyerUsername())
    .items(convertItems(response.getItemList()))
    .build();
    }

    // Unix timestamp → ZonedDateTime
    private ZonedDateTime convertTimestamp(Long timestamp) {
    return Instant.ofEpochSecond(timestamp)
    .atZone(ZoneId.of(“Asia/Taipei”));
    }

    // 蝦皮狀態 → 系統狀態
    private OrderStatus mapStatus(String shopeeStatus) {
    return switch (shopeeStatus) {
    case “UNPAID” -> OrderStatus.PENDING_PAYMENT;
    case “READY_TO_SHIP” -> OrderStatus.PENDING_SHIPMENT;
    case “SHIPPED” -> OrderStatus.SHIPPED;
    case “COMPLETED” -> OrderStatus.COMPLETED;
    case “CANCELLED” -> OrderStatus.CANCELLED;
    default -> OrderStatus.UNKNOWN;
    };
    }
    }

    @Component
    public class MomoConverter {

    private static final DateTimeFormatter MOMO_DATE_FORMAT =
    DateTimeFormatter.ofPattern(“yyyy/MM/dd HH:mm”);

    /**
    * Momo 格式 → 內部格式
    */

    public OrderDTO toOrderDTO(MomoGetOrderResponse response) {
    return OrderDTO.builder()
    .platformOrderId(response.getOrderNo())
    .channel(ChannelType.MOMO)
    .status(mapStatus(response.getOrderStatus()))
    .totalAmount(response.getOrderAmount())
    .createdAt(convertDate(response.getOrderDate()))
    .buyerName(response.getCustomerName())
    .items(convertItems(response.getProducts()))
    .build();
    }

    // Momo 日期格式 → ZonedDateTime
    private ZonedDateTime convertDate(String dateStr) {
    LocalDateTime ldt = LocalDateTime.parse(dateStr, MOMO_DATE_FORMAT);
    return ldt.atZone(ZoneId.of(“Asia/Taipei”));
    }
    }


    狀態對照表

    不同平台的訂單狀態對照:

    系統狀態 蝦皮 Momo Yahoo
    PENDING_PAYMENT UNPAID 01 Unpaid
    PENDING_SHIPMENT READY_TO_SHIP 02 Processing
    SHIPPED SHIPPED 03 Shipped
    COMPLETED COMPLETED 04 Completed
    CANCELLED CANCELLED 99 Cancelled

    使用範例

    @Service
    public class OrderSyncService {

    @Autowired private ShopeeConverter shopeeConverter;
    @Autowired private MomoConverter momoConverter;

    public List<OrderDTO> syncOrders(ChannelType channel, Merchant merchant) {

    if (channel == ChannelType.SHOPEE) {
    // 呼叫蝦皮 API,取得蝦皮格式
    List<ShopeeGetOrderResponse> shopeeOrders = shopeeApi.getOrders();

    // 轉換成內部格式
    return shopeeOrders.stream()
    .map(shopeeConverter::toOrderDTO)
    .toList();
    }

    if (channel == ChannelType.MOMO) {
    // 呼叫 Momo API,取得 Momo 格式
    List<MomoGetOrderResponse> momoOrders = momoApi.getOrders();

    // 轉換成內部格式
    return momoOrders.stream()
    .map(momoConverter::toOrderDTO)
    .toList();
    }

    // … 其他平台
    }
    }

    效果:業務邏輯層只處理統一的 OrderDTO,不用關心各平台的差異。

    總結

    設計 效果
    外部 DTO 對應 API API 變更只影響一個類別
    內部 DTO 統一格式 業務邏輯不受平台影響
    獨立轉換器 轉換邏輯集中管理
    狀態對照表 統一的訂單狀態

    為什麼不用其他方案?

    方案 優點 缺點 結論
    直接用 Map 不用定義類別 無型別安全、IDE 無法幫忙 除錯困難
    一個 DTO 打天下 類別數量少 欄位爆炸、不知道哪些是哪個平台 維護噩夢
    MapStruct 自動轉換 減少手寫程式碼 複雜轉換還是要手寫 可搭配使用
    三層 DTO + 轉換器 清晰、可測試 類別數量多 大型系統首選

    實戰踩坑

    坑 1:平台欄位名稱一直變

    蝦皮某次升級把 item_list 改成 items,只有外部 DTO 需要改,業務邏輯完全不受影響。如果沒有分層,全系統都要搜尋取代。

    坑 2:狀態對照表不完整

    PChome 新增了一個「部分出貨」狀態,我們的對照表沒有,結果 mapping 成 UNKNOWN,訂單卡住不處理。教訓:每個平台的狀態值要定期 review

    坑 3:Converter 邏輯越來越肥

    最初 Converter 只做欄位 mapping,後來塞進去驗證、預設值、業務邏輯…變成 God Class。後來拆成 Converter(純 mapping)+ Validator + Enricher,各司其職。


    系列導航

    ◀ 上一篇
    健康檢查
    📚 返回目錄 下一篇 ▶
    HTTP 客戶端
  • 分佈式健康檢查:自定義 Spring Boot Actuator

    商業價值:健康檢查讓系統「自動發現問題、自動恢復」,直接支撐 導讀篇提到的 99% 庫存準確率——系統不穩定就不可能有準確的庫存。

    前言:為什麼需要健康檢查?

    在微服務架構中,一個服務可能依賴多個外部元件:

    元件 用途 掛掉的影響
    PostgreSQL 主資料庫 無法讀寫訂單
    Redis 快取 效能下降
    Kafka 訊息佇列 無法非同步處理
    Solr 搜尋引擎 無法搜尋訂單
    問題:Kubernetes 預設只檢查 HTTP 回應,無法知道資料庫是否正常。

    Spring Boot Actuator 健康檢查

    基本設定

    # application.yml
    management:
    endpoints:
    web:
    base-path: /
    exposure:
    include: health, info, metrics

    endpoint:
    health:
    show-details: always
    show-components: always

    health:
    # 啟用各元件的健康檢查
    db:
    enabled: true
    redis:
    enabled: true

    健康檢查端點

    端點 用途 使用場景
    /health 完整健康狀態 監控系統
    /health/liveness 存活檢查 K8s liveness probe
    /health/readiness 就緒檢查 K8s readiness probe

    自定義健康檢查指標

    Kafka 健康檢查

    @Component
    public class KafkaHealthIndicator implements HealthIndicator {

    @Value(“${kafka.bootstrap-servers}”)
    private String bootstrapServers;

    private AtomicReference<Health> cachedHealth =
    new AtomicReference<>(Health.unknown().build());

    @Override
    public Health health() {
    return cachedHealth.get();
    }

    /**
    * 背景執行緒定期檢查,避免阻塞健康檢查端點
    */

    @Scheduled(fixedRate = 30000) // 每 30 秒檢查一次
    public void checkHealth() {
    try {
    Properties props = new Properties();
    props.put(“bootstrap.servers”, bootstrapServers);
    props.put(“request.timeout.ms”, “5000”);

    try (AdminClient admin = AdminClient.create(props)) {
    admin.listTopics().names().get(5, TimeUnit.SECONDS);
    }

    cachedHealth.set(Health.up()
    .withDetail(“servers”, bootstrapServers)
    .build());

    } catch (Exception e) {
    cachedHealth.set(Health.down()
    .withDetail(“error”, e.getMessage())
    .build());
    }
    }
    }

    Solr 健康檢查

    @Component
    public class SolrHealthIndicator implements HealthIndicator {

    @Autowired
    private SolrClient solrClient;

    private AtomicReference<Health> cachedHealth =
    new AtomicReference<>(Health.unknown().build());

    @Override
    public Health health() {
    return cachedHealth.get();
    }

    @Scheduled(fixedRate = 30000)
    public void checkHealth() {
    try {
    SolrPingResponse response = solrClient.ping();
    int status = response.getStatus();

    if (status == 0) {
    cachedHealth.set(Health.up()
    .withDetail(“responseTime”, response.getQTime())
    .build());
    } else {
    cachedHealth.set(Health.down()
    .withDetail(“status”, status)
    .build());
    }

    } catch (Exception e) {
    cachedHealth.set(Health.down()
    .withDetail(“error”, e.getMessage())
    .build());
    }
    }
    }


    健康檢查回應範例

    {
    “status”: “UP”,
    “components”: {
    “db”: {
    “status”: “UP”,
    “details”: {
    “database”: “PostgreSQL”,
    “validationQuery”: “isValid()”
    }
    },
    “kafka”: {
    “status”: “UP”,
    “details”: {
    “servers”: “kafka:9092”
    }
    },
    “redis”: {
    “status”: “UP”,
    “details”: {
    “version”: “7.0.0”
    }
    },
    “solr”: {
    “status”: “UP”,
    “details”: {
    “responseTime”: 5
    }
    }
    }
    }

    Kubernetes 整合

    # deployment.yaml
    spec:
    containers:
    – name: oms-service
    # 存活檢查:程式是否還活著
    livenessProbe:
    httpGet:
    path: /health/liveness
    port: 8080
    initialDelaySeconds: 30
    periodSeconds: 10
    timeoutSeconds: 5
    failureThreshold: 3

    # 就緒檢查:是否可以接受流量
    readinessProbe:
    httpGet:
    path: /health/readiness
    port: 8080
    initialDelaySeconds: 20
    periodSeconds: 5
    timeoutSeconds: 3
    failureThreshold: 3

    Probe 類型 失敗後行為 使用場景
    liveness 重啟 Pod 程式死當、無回應
    readiness 從 Service 移除 暫時無法服務(如 DB 斷線)

    設計考量

    為什麼用背景執行緒 + 快取?

    • 健康檢查端點需要快速回應(< 1秒)
    • 外部元件檢查可能很慢(網路延遲)
    • Kubernetes 頻繁呼叫(每 5-10 秒)
    設計 說明
    背景檢查 每 30 秒執行一次,不阻塞端點
    結果快取 AtomicReference 儲存最新狀態
    逾時設定 檢查逾時 5 秒,避免卡住
    狀態詳情 包含時間、錯誤訊息等資訊

    監控整合

    將健康狀態匯出到 Prometheus:

    # 健康狀態指標
    health_check_status{component=”kafka”} 1
    health_check_status{component=”solr”} 1
    health_check_status{component=”redis”} 1
    health_check_status{component=”db”} 1

    # 檢查執行時間
    health_check_duration_seconds{component=”kafka”} 0.023
    health_check_duration_seconds{component=”solr”} 0.005


    總結

    設計 效果
    自定義 HealthIndicator 檢查所有依賴元件
    背景執行 + 快取 端點回應快速
    K8s Probe 整合 自動重啟/移除故障 Pod
    Prometheus 匯出 歷史趨勢監控

    為什麼不用其他方案?

    方案 優點 缺點 結論
    只靠 K8s 預設檢查 零設定 只檢查 HTTP 回應,不知道 DB 狀態 不夠
    外部監控工具打 API 不侵入程式碼 只知道 API 回應,不知道內部狀態 補充用
    自己寫健康檢查 API 完全控制 要自己處理快取、超時 重複造輪子
    Actuator + 自訂 整合好、可擴展 要學 Spring 生態 Spring 專案首選

    實戰踩坑

    坑 1:健康檢查太慢導致 Pod 被殺

    最初健康檢查直接連 Kafka,網路慢時要 10 秒才回應。K8s 以為 Pod 死了,不斷重啟。解法:改成背景執行緒定期檢查,健康端點只回傳快取結果。

    坑 2:Liveness 和 Readiness 混用

    最初兩個 Probe 用同一個端點。結果 Kafka 斷線時,所有 Pod 都被重啟(Liveness 失敗)。正確做法:Liveness 只檢查「程式還活著」,Readiness 檢查「能不能接流量」。Kafka 斷線應該是 Readiness 失敗(從 Service 移除),不是 Liveness 失敗(重啟)。

    坑 3:忘記設定 initialDelaySeconds

    應用程式啟動要 30 秒,但健康檢查 10 秒就開始。結果 Pod 永遠起不來,一直被重啟。


    系列導航

    ◀ 上一篇
    多租戶認證
    📚 返回目錄 下一篇 ▶
    DTO 設計
  • 企業級多租戶認證:Token 驗證實戰

    商業價值:多租戶認證讓「一套系統服務多個商戶」,是 SaaS 模式的技術基礎。這直接支撐 導讀篇提到的 70% 人力成本降低(多商戶共用系統,不用每家都建一套)。

    前言:SaaS 系統的認證挑戰

    在多租戶 SaaS 系統中,我們面臨兩種使用者:

    角色 說明 存取範圍
    商戶 (Merchant) 使用系統的客戶 只能看到自己的資料
    平台 (Platform) 系統管理者 可以看到所有商戶資料
    安全需求:

    • 商戶 A 絕對不能看到商戶 B 的訂單
    • Token 被盜用時要能快速失效
    • 支援多裝置同時登入

    Token 設計

    Token 結構

    {
    “tokenId”: “uuid-xxxx-xxxx”,
    “userType”: “MERCHANT”,
    “userId”: “U001”,
    “merchantId”: “M001”,
    “permissions”: [“ORDER_READ”, “ORDER_WRITE”],
    “issuedAt”: “2024-03-18T10:00:00Z”,
    “expiresAt”: “2024-03-18T12:00:00Z”
    }

    Token 驗證流程

    請求進來


    ┌─────────────────────┐
    │ 1. 檢查 Token 存在 │
    └──────────┬──────────┘


    ┌─────────────────────┐
    │ 2. 檢查 Token 有效 │ ──── 過期/無效 ──→ 401 Unauthorized
    └──────────┬──────────┘


    ┌─────────────────────┐
    │ 3. 載入使用者資訊 │
    └──────────┬──────────┘


    ┌─────────────────────┐
    │ 4. 設定安全上下文 │
    └──────────┬──────────┘


    繼續處理請求

    Filter 實作

    @Component
    public class TokenAuthFilter extends OncePerRequestFilter {

    @Autowired
    private TokenService tokenService;

    @Autowired
    private TokenCache tokenCache;

    @Override
    protected void doFilterInternal(
    HttpServletRequest request,
    HttpServletResponse response,
    FilterChain chain) throws Exception {

    // 1. 從 Header 取得 Token
    String token = extractToken(request);

    if (token == null) {
    sendError(response, “Missing token”);
    return;
    }

    // 2. 驗證 Token(先查快取)
    TokenInfo tokenInfo = tokenCache.get(token);
    if (tokenInfo == null) {
    tokenInfo = tokenService.validate(token);
    if (tokenInfo != null) {
    tokenCache.put(token, tokenInfo);
    }
    }

    if (tokenInfo == null || tokenInfo.isExpired()) {
    sendError(response, “Invalid or expired token”);
    return;
    }

    // 3. 設定安全上下文
    SecurityContext.set(tokenInfo);

    try {
    chain.doFilter(request, response);
    } finally {
    SecurityContext.clear();
    }
    }

    private String extractToken(HttpServletRequest request) {
    String header = request.getHeader(“Authorization”);
    if (header != null && header.startsWith(“Bearer “)) {
    return header.substring(7);
    }
    return null;
    }
    }


    商戶隔離

    所有資料存取都要加上商戶過濾:

    @Repository
    public class OrderRepository {

    public List<Order> findOrders(OrderQuery query) {
    // 從安全上下文取得當前商戶
    TokenInfo token = SecurityContext.get();

    if (token.isMerchant()) {
    // 商戶只能查自己的資料
    query.setMerchantId(token.getMerchantId());
    }
    // 平台角色可以查所有商戶(不加過濾條件)

    return jdbcTemplate.query(
    “SELECT * FROM orders WHERE merchant_id = ? AND …”,
    query.getMerchantId(),

    );
    }
    }

    安全保證:即使前端傳入其他商戶 ID,後端也會強制覆蓋為當前登入商戶的 ID。

    Token 快取策略

    策略 設定 原因
    快取時間 5 分鐘 減少 DB 查詢,但不會太舊
    最大數量 10,000 避免記憶體爆炸
    淘汰策略 LRU 不常用的 Token 先移除
    @Bean
    public Cache<String, TokenInfo> tokenCache() {
    return Caffeine.newBuilder()
    .maximumSize(10_000)
    .expireAfterWrite(Duration.ofMinutes(5))
    .recordStats()
    .build();
    }

    Token 失效機制

    當需要強制登出時:

    @Service
    public class TokenService {

    /**
    * 強制失效單一 Token
    */

    public void revokeToken(String tokenId) {
    // 1. 從資料庫標記失效
    tokenRepository.markRevoked(tokenId);

    // 2. 從快取移除
    tokenCache.invalidate(tokenId);

    // 3. 發布失效事件(通知其他節點)
    eventPublisher.publish(new TokenRevokedEvent(tokenId));
    }

    /**
    * 強制登出商戶所有 Token
    */

    public void revokeAllTokens(String merchantId) {
    List<String> tokens = tokenRepository.findByMerchant(merchantId);
    tokens.forEach(this::revokeToken);
    }
    }


    權限檢查

    @RestController
    public class OrderController {

    // 需要 ORDER_READ 權限
    @RequirePermission(“ORDER_READ”)
    @GetMapping(“/api/orders”)
    public List<Order> listOrders() {
    return orderService.findOrders();
    }

    // 需要 ORDER_WRITE 權限
    @RequirePermission(“ORDER_WRITE”)
    @PostMapping(“/api/orders/{id}/ship”)
    public Order shipOrder(@PathVariable String id) {
    return orderService.ship(id);
    }
    }


    安全總結

    安全機制 實作方式 防護目標
    Token 驗證 Filter 攔截 未授權存取
    商戶隔離 強制覆蓋 merchantId 資料外洩
    Token 快取 Caffeine + 分散式同步 效能
    強制失效 事件發布 + 快取清除 帳號被盜
    權限檢查 註解 + AOP 越權操作

    為什麼不用其他方案?

    方案 優點 缺點 結論
    Session-based 簡單、狀態伺服器端管理 不適合分散式、需要 Session 複製 單機可用
    JWT(無狀態) 不需查 DB、自包含 無法主動失效、Token 可能很大 短期 Token 可用
    OAuth 2.0 標準協議、支援第三方 複雜、需要額外服務 需要 SSO 時用
    Token + 快取 可主動失效、效能好 需要 Redis SaaS 系統首選

    實戰踩坑

    坑 1:商戶資料外洩

    早期版本有個 API 忘記加商戶過濾,某商戶發現可以透過改 URL 的 orderId 看到別人的訂單。好險被內部測試發現,立刻修復並全面檢查所有 API。教訓:所有 Repository 方法預設都要加 merchantId 過濾

    坑 2:Token 快取同步問題

    三台 Server,使用者在 A 機登出,但 B、C 機的快取還有 Token。結果登出後還能繼續操作 30 秒。解法:登出時發布事件到所有節點,同步清除快取。

    坑 3:權限設計太細

    最初設計了 50+ 種權限,結果商戶搞不懂、客服也解釋不清。後來簡化成 5 個角色(管理員、主管、操作員、財務、唯讀),大幅降低維護成本。


    系列導航

    ◀ 上一篇
    Kafka 事件驅動
    📚 返回目錄 下一篇 ▶
    健康檢查
  • Kafka 事件驅動架構:打造高可用訂單處理系統

    商業價值:事件驅動架構讓系統能「處理速度提升 10 倍」,從 4-8 小時縮短到 25-35 分鐘。詳見 導讀篇的 ROI 計算

    前言:為什麼需要事件驅動?

    想像一個場景:使用者在後台點擊「同步蝦皮訂單」。

    同步處理的問題:

    • 蝦皮 API 回應慢 → 使用者等待 30 秒以上
    • API 超時 → 整個請求失敗
    • 大量請求 → 伺服器資源耗盡

    解決方案:非同步事件驅動

    使用者請求 背景處理
    │ │
    ▼ │
    ┌─────────┐ 發送訊息 ┌─────────┐
    │ Web API │ ──────────────► │ Kafka │
    └─────────┘ └────┬────┘
    │ │
    ▼ ▼
    回應成功 ┌───────────────┐
    (立即返回) │ Consumer Job │
    │ 慢慢處理… │
    └───────────────┘
    效果:使用者立即收到「已排程」回應,實際同步在背景執行。

    架構設計

    Topic 設計:每個通路獨立

    Topic 名稱 用途 Consumer
    oms-action-shopee 蝦皮相關動作 Shopee Consumer
    oms-action-momo Momo 相關動作 Momo Consumer
    oms-action-yahoo Yahoo 相關動作 Yahoo Consumer
    oms-action-pchome PChome 相關動作 PChome Consumer
    為什麼要分開?

    • 蝦皮 API 壞了,不影響 Momo 訂單處理
    • 可以針對不同平台調整 Consumer 數量
    • 方便監控各平台的處理狀況

    Producer:發送訊息

    @Service
    public class ActionProducer {

    private final KafkaTemplate<String, String> kafkaTemplate;

    /**
    * 發送動作到對應的通路 Topic
    */

    public void sendAction(ChannelType channel, ActionMessage message) {
    String topic = “oms-action-“ + channel.getCode();
    String payload = JsonUtil.toJson(message);

    kafkaTemplate.send(topic, message.getMerchantId(), payload)
    .addCallback(
    result -> log.info(“發送成功: {}”, topic),
    error -> log.error(“發送失敗: {}”, error.getMessage())
    );
    }
    }

    訊息格式設計

    {
    “header”: {
    “messageId”: “uuid-xxxx-xxxx”,
    “timestamp”: “2024-03-18T10:30:00Z”,
    “traceId”: “trace-xxxx”
    },
    “body”: {
    “merchantId”: “M001”,
    “actionType”: “SYNC_ORDERS”,
    “parameters”: {
    “startDate”: “2024-03-17”,
    “endDate”: “2024-03-18”
    }
    }
    }

    Consumer:處理訊息

    @Component
    public class ActionConsumer {

    @Autowired
    private ChannelFactory channelFactory;

    @KafkaListener(topics = “oms-action-shopee”)
    public void consumeShopee(String message) {
    processAction(ChannelType.SHOPEE, message);
    }

    @KafkaListener(topics = “oms-action-momo”)
    public void consumeMomo(String message) {
    processAction(ChannelType.MOMO, message);
    }

    private void processAction(ChannelType channel, String message) {
    try {
    // 1. 解析訊息
    ActionMessage action = JsonUtil.fromJson(message);

    // 2. 取得對應的通路處理器
    ChannelAction handler = channelFactory.getAction(channel);

    // 3. 執行動作
    ActionResult result = handler.execute(action);

    // 4. 回寫結果
    saveResult(action, result);

    } catch (Exception e) {
    // 5. 錯誤處理
    handleError(message, e);
    }
    }
    }


    錯誤處理策略

    錯誤類型 處理方式 範例
    暫時性錯誤 重試 3 次 API 超時、網路問題
    永久性錯誤 記錄並跳過 資料格式錯誤
    未知錯誤 進入 Dead Letter Queue 系統異常
    @Bean
    public DefaultErrorHandler errorHandler() {
    // 設定重試策略
    BackOff backOff = new ExponentialBackOff(1000L, 2.0);
    backOff.setMaxElapsedTime(30000L); // 最多重試 30 秒

    return new DefaultErrorHandler(
    (record, exception) -> {
    // 重試失敗後,送到 Dead Letter Queue
    sendToDeadLetterQueue(record, exception);
    },
    backOff
    );
    }


    監控與告警

    監控指標 正常值 告警條件
    Consumer Lag < 1000 > 5000 持續 5 分鐘
    處理時間 < 5 秒 > 30 秒
    錯誤率 < 1% > 5%
    Dead Letter 數量 0 > 10

    效能調校

    # application.yml
    spring:
    kafka:
    consumer:
    # 每次拉取的最大筆數
    max-poll-records: 100

    # 拉取間隔
    fetch-min-size: 1
    fetch-max-wait: 500ms

    producer:
    # 批次發送設定
    batch-size: 16384
    buffer-memory: 33554432

    # 壓縮
    compression-type: lz4


    總結

    設計 效果
    非同步處理 使用者不用等待 API 回應
    Topic 分離 通路故障隔離
    重試機制 暫時性錯誤自動恢復
    Dead Letter Queue 問題訊息不遺失
    監控告警 問題即時發現

    為什麼不用其他方案?

    方案 優點 缺點 結論
    同步處理 簡單、好除錯 使用者要等、效能差 小流量可用
    Redis Queue 輕量、快速 持久化弱、無法分區 簡單場景可用
    RabbitMQ 功能豐富、可靠 吞吐量不如 Kafka 適合複雜路由
    Kafka 高吞吐、持久化、分區 學習曲線、維運成本 大流量首選

    實戰踩坑

    坑 1:Consumer Lag 暴增

    雙 11 當天 Consumer Lag 飆到 50,000+,訂單處理延遲 2 小時。原因:單一 Consumer 處理太慢。解法:增加 Consumer 數量到 Partition 數量,同時優化處理邏輯(批次處理)。

    坑 2:訊息重複消費

    Consumer 處理到一半掛掉,重啟後同一筆訂單被處理兩次,導致重複出貨。解法:加入冪等性檢查(用訂單 ID 去重)。

    坑 3:Topic 沒分開

    最初所有平台共用一個 Topic,蝦皮 API 壞了堵住整條 Queue,Momo 訂單也跟著延遲。後來拆成每個平台獨立 Topic,故障隔離。


    系列導航

    ◀ 上一篇
    工廠模式
    📚 返回目錄 下一篇 ▶
    多租戶認證
  • 多通路電商系統架構:用工廠模式整合 17 個平台

    商業價值:這篇介紹的工廠模式讓「新增平台從 2-3 個月縮短到 2-3 週」,直接影響 導讀篇提到的 80% 擴展成本降低

    前言:當你要同時對接 17 個電商平台

    在多通路電商系統中,我們需要整合多個平台:

    平台類型 範例 特性
    綜合電商 蝦皮、Momo、Yahoo、PChome 訂單量大、API 複雜
    開店平台 Shopify、Shopline、91APP 彈性高、客製化多
    國際平台 樂天、Coupang、Amazon 多語系、跨境物流
    問題:每個平台的 API 格式、認證方式、資料結構都不同。如果用 if-else 判斷,程式碼會變成災難。

    解決方案:工廠模式 + 策略模式

    核心思想:定義統一介面,每個平台各自實作。新增平台時,只需要新增一個實作類別。

    Step 1:定義統一介面

    所有平台都必須實作這個介面:

    /**
    * 電商平台動作介面
    * 所有平台整合都必須實作這個介面
    */

    public interface ChannelAction {

    // 取得平台設定(API URL、版本等)
    ChannelSetting getSetting();

    // 取得平台授權 Token
    TokenResult getAccessToken(Merchant merchant);

    // 驗證必要資料是否齊全
    boolean validateRequired(ActionRequest request);

    // 執行實際動作(同步訂單、更新庫存等)
    ActionResult execute(ActionRequest request);
    }

    Step 2:每個平台各自實作

    蝦皮實作範例:

    @Component
    public class ShopeeAction implements ChannelAction {

    @Override
    public ChannelSetting getSetting() {
    return ChannelSetting.builder()
    .apiUrl(“https://partner.shopeemobile.com”)
    .version(“v2”)
    .authType(AuthType.OAUTH2)
    .build();
    }

    @Override
    public TokenResult getAccessToken(Merchant merchant) {
    // 蝦皮使用 OAuth2 + 簽章驗證
    String signature = generateSignature(merchant);
    return callShopeeAuthAPI(merchant, signature);
    }

    @Override
    public ActionResult execute(ActionRequest request) {
    // 呼叫蝦皮 API 執行動作
    return callShopeeAPI(request);
    }
    }

    Momo 實作範例:

    @Component
    public class MomoAction implements ChannelAction {

    @Override
    public ChannelSetting getSetting() {
    return ChannelSetting.builder()
    .apiUrl(“https://api.momo.com.tw”)
    .version(“v1”)
    .authType(AuthType.API_KEY)
    .build();
    }

    @Override
    public TokenResult getAccessToken(Merchant merchant) {
    // Momo 使用 API Key 驗證
    return TokenResult.of(merchant.getMomoApiKey());
    }

    @Override
    public ActionResult execute(ActionRequest request) {
    // 呼叫 Momo API 執行動作
    return callMomoAPI(request);
    }
    }

    Step 3:工廠類別統一管理

    @Component
    public class ChannelFactory {

    private final Map<ChannelType, ChannelAction> actionMap;

    // Spring 自動注入所有 ChannelAction 實作
    public ChannelFactory(List<ChannelAction> actions) {
    this.actionMap = actions.stream()
    .collect(Collectors.toMap(
    action -> action.getSetting().getChannelType(),
    action -> action
    ));
    }

    /**
    * 根據通路類型取得對應的實作
    */

    public ChannelAction getAction(ChannelType channelType) {
    ChannelAction action = actionMap.get(channelType);
    if (action == null) {
    throw new UnsupportedChannelException(
    “不支援的通路: “ + channelType
    );
    }
    return action;
    }
    }


    使用方式

    業務邏輯層只需要這樣呼叫:

    @Service
    public class OrderSyncService {

    @Autowired
    private ChannelFactory channelFactory;

    public SyncResult syncOrders(ChannelType channel, Merchant merchant) {
    // 1. 取得對應的通路實作
    ChannelAction action = channelFactory.getAction(channel);

    // 2. 取得授權 Token
    TokenResult token = action.getAccessToken(merchant);

    // 3. 執行同步
    ActionRequest request = ActionRequest.builder()
    .merchant(merchant)
    .token(token)
    .actionType(ActionType.SYNC_ORDERS)
    .build();

    return action.execute(request);
    }
    }

    優點:不管是蝦皮、Momo、Yahoo 還是其他平台,呼叫方式完全一樣。新增平台時,業務邏輯層完全不用改。

    新增平台有多簡單?

    假設要新增 Coupang 韓國平台:

    @Component
    public class CoupangAction implements ChannelAction {

    @Override
    public ChannelSetting getSetting() {
    return ChannelSetting.builder()
    .apiUrl(“https://api-gateway.coupang.com”)
    .version(“v2”)
    .authType(AuthType.HMAC)
    .build();
    }

    // … 實作其他方法
    }

    步驟 工作內容 影響範圍
    1 建立 CoupangAction 類別 只有新檔案
    2 實作 ChannelAction 介面 只有新檔案
    3 加上 @Component 註解 只有新檔案
    4 完成!Spring 自動註冊 零修改現有程式

    設計模式總結

    模式 用途 在這裡的應用
    策略模式 定義演算法家族,讓它們可互換 每個平台是一個策略
    工廠模式 封裝物件建立邏輯 根據通路類型取得實作
    依賴注入 解耦合 Spring 自動管理
    效益:

    • 新增通路:從 2-3 個月縮短到 2-3 週
    • 維護成本:改一個平台不影響其他平台
    • 測試:每個平台可以獨立單元測試

    為什麼不用其他方案?

    方案 優點 缺點 結論
    if-else 判斷 簡單直接 每次加平台要改核心程式碼 小規模可用,超過 3 個平台就很痛苦
    Switch Case 比 if-else 清楚 還是要改核心程式碼 同上
    反射 + 設定檔 完全不改程式碼 除錯困難、IDE 無法追蹤 過度設計,維護成本高
    工廠 + 策略 新增只加檔案、Spring 自動註冊 需要理解設計模式 中大型系統的最佳平衡

    實戰踩坑

    坑 1:平台 API 變更沒通知

    蝦皮某次 API 升級,回傳欄位名稱從 order_sn 改成 ordersn。因為每個平台有獨立的 Action 類別,我們只需要改 ShopeeAction,其他 16 個平台完全不受影響。如果用 if-else,改錯一行就全部爆炸。

    坑 2:忘記加 @Component

    新人寫好 CoupangAction 卻沒加 @Component,Spring 沒註冊到 Factory。呼叫時直接噴 UnsupportedChannelException。後來在程式碼審查加入檢查項:「確認新 Action 有 @Component」。

    坑 3:介面設計太死

    最初 ChannelAction 只有 execute() 一個方法。後來發現有些平台需要 OAuth 刷新 Token、有些需要 Webhook 處理。介面改了三次才穩定。教訓:先做 3-5 個平台再抽象,別一開始就過度設計


    系列導航

    ◀ 上一篇
    導讀篇
    📚 返回目錄 下一篇 ▶
    Kafka 事件驅動
  • 🏗️ LangChain/LangGraph 深度分析:架構師、顧問、個人公司的實戰指南

    **作者的話**:本文從架構設計、商業決策、工程化、生態演進、個人公司戰略等 8 個維度,深度分析 LangChain/LangGraph 在 AI 開發中的真實定位。本文附帶完整的 Jupyter Notebook,讓你能親身體驗多 LLM 集成的實際效果。

    **閱讀時間**:20-30 分鐘 | **難度**:中等偏高 | **實用度**:⭐⭐⭐⭐⭐

    📖 目錄

    • [現狀速覽](#現狀速覽)
    • [1️⃣ 架構設計維度](#1️⃣-架構設計維度)
    • [2️⃣ 工程化可維護性](#2️⃣-工程化可維護性)
    • [3️⃣ 性能特徵](#3️⃣-性能特徵)
    • [4️⃣ 生態成熟度](#4️⃣-生態成熟度)
    • [5️⃣ 供應商風險與多 LLM 價值](#5️⃣-供應商風險與多-llm-價值)
    • [6️⃣ 生產化成本](#6️⃣-生產化成本)
    • [7️⃣ 組織影響](#7️⃣-組織影響維度)
    • [8️⃣ 長期戰略](#8️⃣-長期戰略與演進)
    • [個人公司實戰指南](#個人公司的實戰指南)
    • [完整 Jupyter Notebook](#完整-jupyter-notebook)
    • [快速開始指南](#快速開始指南)
    • [最終決策框架](#最終決策框架)

    現狀速覽

    2026 年 3 月的 AI 框架生態:
    
                        複雜度
                         ↑
                ┌────────┴────────┐
                │   LangGraph    │ ← 複雜工作流
                │  (新興但專業)  │   多 Agent
                └────────┬────────┘
                ┌────────┴────────┐
                │   LangChain    │ ← 快速原型
                │   (成熟生態)   │   簡單應用
                └────────┬────────┘
        ┌───────┬────────┴────────┬───────┐
        │       │                 │       │
    Claude    GPT-4            Gemini  Llama
     SDK       SDK              API     API
    (最優)    (最優)          (最優)  (最優)
    
    規則:
    • 簡單 + 單 LLM     → 官方 SDK(最快)
    • 簡單 + 多 LLM     → LangChain(靈活)
    • 複雜 + 多 LLM     → LangGraph(專業)
    • 複雜 + 單 LLM     → 官方 SDK(官方優化)

    1️⃣ 架構設計維度

    LangChain:管道式思想(Pipeline Pattern)

    # LangChain 的核心抽象:Chain(順序執行)
    # 概念:A → B → C(線性管道)
    
    from langchain_anthropic import ChatAnthropic
    from langchain.chains import LLMChain
    from langchain.prompts import PromptTemplate
    
    # 簡單的 Chain
    llm = ChatAnthropic(model="claude-opus-4-6")
    
    prompt = PromptTemplate(
        input_variables=["task"],
        template="驗證:{task}"
    )
    
    chain = LLMChain(llm=llm, prompt=prompt)
    result = chain.run(task="檢查 API Schema")

    **架構評價**:

    LangGraph:狀態機思想(State Machine Pattern)

    # LangGraph 的核心抽象:Graph(有向無環圖)
    # 概念:在不同「狀態」間轉移
    
    from langgraph.graph import StateGraph, END
    from langchain_anthropic import ChatAnthropic
    from typing import TypedDict
    
    # 定義狀態
    class AuditState(TypedDict):
        api_findings: str
        db_findings: str
        ui_findings: str
        final_report: str
    
    # 定義節點(每個節點是狀態轉移)
    def verify_api(state: AuditState) -> dict:
        llm = ChatAnthropic(model="claude-opus-4-6")
        result = llm.invoke("驗證 API Schema...")
        return {"api_findings": result.content}
    
    def verify_db(state: AuditState) -> dict:
        # state 中自動帶了前一步的結果
        llm = ChatAnthropic(model="claude-opus-4-6")
        result = llm.invoke(f"根據 API 驗證結果:{state['api_findings']}... 驗證資料層...")
        return {"db_findings": result.content}
    
    def verify_ui(state: AuditState) -> dict:
        llm = ChatAnthropic(model="claude-opus-4-6")
        result = llm.invoke(f"根據 API 驗證結果:{state['api_findings']}... 驗證 UI...")
        return {"ui_findings": result.content}
    
    def final_report(state: AuditState) -> dict:
        # 可以訪問所有之前的結果
        llm = ChatAnthropic(model="claude-opus-4-6")
        report = llm.invoke(f"""
        綜合以下發現生成報告:
        API: {state['api_findings']}
        DB: {state['db_findings']}
        UI: {state['ui_findings']}
        """)
        return {"final_report": report.content}
    
    # 構建圖
    graph = StateGraph(AuditState)
    graph.add_node("api", verify_api)
    graph.add_node("db", verify_db)
    graph.add_node("ui", verify_ui)
    graph.add_node("report", final_report)
    
    # 定義流向(這就是架構)
    graph.add_edge("api", "db")
    graph.add_edge("api", "ui")  # 並行執行
    graph.add_edge("db", "report")
    graph.add_edge("ui", "report")
    graph.set_entry_point("api")
    graph.add_edge("report", END)
    
    # 編譯並執行
    workflow = graph.compile()
    result = workflow.invoke({})
    print(result["final_report"])

    **架構評價**:

    架構對比圖

    LangChain 的流程(線性):
    ═════════════════════
    
    Task 1: API 驗證
       ↓
       你需要手動管理 task1_result
       ↓
    Task 2: DB 驗證
       ↓
       你需要手動管理 task2_result
       ↓
    Task 3: UI 驗證
       ↓
    問題:狀態在各地傳遞,容易出錯
    
    
    LangGraph 的流程(DAG):
    ═════════════════════
    
            ┌─→ Task 2: DB 驗證 ─┐
            │                    ├─→ Task 4: 最終報告
    Task 1: API 驗證              │
            │                    ├─→ END
            └─→ Task 3: UI 驗證 ─┘
    
    好處:
    ✓ 並行執行(Task 2 和 3 同時跑)
    ✓ State 自動流轉(無需手動管理)
    ✓ 流程結構清晰(add_edge 就是架構)

    2️⃣ 工程化可維護性

    代碼複雜度對比:簡單 vs 複雜場景

    場景 A:簡單線性流程(3 個 Task)

    **用 LangChain 實現**:

    class SimpleAudit:
        def __init__(self, llm):
            self.llm = llm
            self.results = {}
    
        def task1_api_check(self):
            result = self.llm.invoke("檢查 API...")
            self.results['task1'] = result
            return result
    
        def task2_db_check(self):
            # 手動管理 task1 的結果
            prev = self.results['task1']
            result = self.llm.invoke(f"根據 {prev} 檢查資料層...")
            self.results['task2'] = result
            return result
    
        def task3_ui_check(self):
            prev = self.results['task2']
            result = self.llm.invoke(f"根據 {prev} 檢查 UI...")
            self.results['task3'] = result
            return result
    
        def run(self):
            self.task1_api_check()
            self.task2_db_check()
            self.task3_ui_check()
            return self.results['task3']
    
    # 問題:狀態散亂,難以追蹤
    # 程式碼行數:40 行

    **用 LangGraph 實現**:

    from langgraph.graph import StateGraph, END
    from typing import TypedDict
    
    class AuditState(TypedDict):
        task1_result: str
        task2_result: str
        task3_result: str
    
    graph = StateGraph(AuditState)
    
    def task1(state):
        result = llm.invoke("檢查 API...")
        return {"task1_result": result.content}
    
    def task2(state):
        result = llm.invoke(f"根據 {state['task1_result']} 檢查資料層...")
        return {"task2_result": result.content}
    
    def task3(state):
        result = llm.invoke(f"根據 {state['task2_result']} 檢查 UI...")
        return {"task3_result": result.content}
    
    graph.add_node("task1", task1)
    graph.add_node("task2", task2)
    graph.add_node("task3", task3)
    graph.add_edge("task1", "task2")
    graph.add_edge("task2", "task3")
    graph.add_edge("task3", END)
    
    # 好處:State 清晰,流程一目瞭然
    # 程式碼行數:25 行(簡潔 40%)

    場景 B:複雜分支流程(並行 + 分支)

    需求:
    Task 1 → [Task 2 並行 Task 3] → Task 4

    **用 LangChain 實現**(噩夢):

    import concurrent.futures
    import threading
    
    class ComplexAudit:
        def __init__(self, llm):
            self.llm = llm
            self.results = {}
            self.lock = threading.Lock()
    
        def task1(self):
            result = self.llm.invoke("Task 1...")
            with self.lock:
                self.results['task1'] = result
    
        def task2(self):
            # 需要等待 task1 完成
            while 'task1' not in self.results:
                time.sleep(0.1)
            result = self.llm.invoke(f"Task 2 based on {self.results['task1']}...")
            with self.lock:
                self.results['task2'] = result
    
        def task3(self):
            while 'task1' not in self.results:
                time.sleep(0.1)
            result = self.llm.invoke(f"Task 3 based on {self.results['task1']}...")
            with self.lock:
                self.results['task3'] = result
    
        def task4(self):
            while 'task2' not in self.results or 'task3' not in self.results:
                time.sleep(0.1)
            result = self.llm.invoke(f"Task 4 based on {self.results['task2']} and {self.results['task3']}...")
            with self.lock:
                self.results['task4'] = result
    
        def run(self):
            # 手動管理並行
            executor = concurrent.futures.ThreadPoolExecutor(max_workers=4)
    
            f1 = executor.submit(self.task1)
            f1.result()  # 等 task1 完成
    
            f2 = executor.submit(self.task2)
            f3 = executor.submit(self.task3)
            f2.result()
            f3.result()
    
            f4 = executor.submit(self.task4)
            f4.result()
    
            return self.results['task4']
    
    # 問題:手動管理並行,容易出 deadlock
    # 手動管理依賴,難以維護
    # 程式碼行數:70+ 行(複雜且容易出錯)

    **用 LangGraph 實現**(優雅):

    from langgraph.graph import StateGraph, END
    
    graph = StateGraph(AuditState)
    
    graph.add_node("task1", task1)
    graph.add_node("task2", task2)
    graph.add_node("task3", task3)
    graph.add_node("task4", task4)
    
    # 定義並行和依賴關係(自動處理)
    graph.add_edge("task1", "task2")
    graph.add_edge("task1", "task3")
    graph.add_edge("task2", "task4")
    graph.add_edge("task3", "task4")
    graph.add_edge("task4", END)
    
    # LangGraph 自動:
    # ✓ 並行執行 task2 和 task3
    # ✓ 等待兩個都完成後再執行 task4
    # ✓ 管理所有依賴
    
    # 程式碼行數:18 行(簡潔 75%,且完全無 bug)

    複雜度對比表

    ╔════════════════════╦══════════════╦═════════════╦══════════════╗
    ║     指標           ║ LangChain    ║ LangGraph   ║ Claude SDK   ║
    ╠════════════════════╬══════════════╬═════════════╬══════════════╣
    ║ 簡單線性           ║ 40 行        ║ 25 行       ║ 20 行        ║
    ║ 複雜並行           ║ 80+ 行       ║ 22 行       ║ 120+ 行      ║
    ║ 易於測試           ║ ⭐⭐         ║ ⭐⭐⭐⭐⭐ ║ ⭐⭐⭐       ║
    ║ 易於除錯           ║ ⭐⭐         ║ ⭐⭐⭐⭐   ║ ⭐⭐⭐       ║
    ║ 新人上手時間       ║ 1-2 天       ║ 3-5 天      ║ 1 天         ║
    ║ 複雜流程上手時間   ║ 2-3 周       ║ 1-2 周      ║ 3+ 周        ║
    ╚════════════════════╩══════════════╩═════════════╩══════════════╝

    3️⃣ 性能特徵

    執行效率測試

    測試場景:質量檢查系統(4 個 Task,各 1 次 LLM 調用)
    環境:16GB RAM,單用戶,冷啟動
    
    ╔════════════════════╦══════════════╦═════════╦═══════════╗
    ║ 方案               ║ 總執行時間   ║ 開銷    ║ 記憶體    ║
    ╠════════════════════╬══════════════╬═════════╬═══════════╣
    ║ Claude API 直調    ║ 48 秒        ║ 基準    ║ 120 MB   ║
    ║ LangChain          ║ 52 秒        ║ +8%     ║ 180 MB   ║
    ║ LangGraph          ║ 54 秒        ║ +12%    ║ 220 MB   ║
    ╚════════════════════╩══════════════╩═════════╩═══════════╝
    
    分析:
    ✓ 在 I/O 密集型(LLM 調用)中,開銷可忽略
    ✓ 實際瓶頸是 LLM API 延遲(30-40 秒),不是框架
    ✓ 高併發時(100+ 並行)資源差異才明顯

    高並發測試

    100 個同時請求,每個 4 個 Task:
    
    ╔════════════════════╦═══════════╦═════════════╗
    ║ 方案               ║ 總記憶體  ║ CPU 使用率  ║
    ╠════════════════════╬═══════════╬═════════════╣
    ║ Claude API 直調    ║ 2.5 GB    ║ 45%         ║
    ║ LangChain          ║ 3.8 GB    ║ 52%         ║
    ║ LangGraph          ║ 4.5 GB    ║ 58%         ║
    ╚════════════════════╩═══════════╩═════════════╝
    
    高並發時的差異較明顯,但:
    • 多數公司不會有 100 個同時 AI 請求
    • 可以用隊列和批處理解決
    • 不是技術選型的主要考慮

    4️⃣ 生態成熟度

    框架成熟度對比

    ╔════════════════════╦═══════════════╦══════════════╦═══════════════╗
    ║ 維度               ║ LangChain     ║ LangGraph    ║ Claude SDK    ║
    ╠════════════════════╬═══════════════╬══════════════╬═══════════════╣
    ║ GitHub Star        ║ 90k+          ║ 新興         ║ 新興          ║
    ║ 文檔品質           ║ ⭐⭐⭐ 豐富 ║ ⭐⭐ 完善中 ║ ⭐⭐⭐⭐⭐ ║
    ║ Stack Overflow     ║ ⭐⭐⭐⭐      ║ ⭐          ║ ⭐⭐⭐⭐⭐   ║
    ║ 企業採用           ║ ⭐⭐⭐⭐      ║ 新興         ║ 新興          ║
    ║ 更新頻率           ║ 每週          ║ 每月         ║ 每週          ║
    ║ 向後相容性         ║ ⚠️ 經常破壞  ║ ✅ 穩定      ║ ✅ 穩定       ║
    ║ 第三方集成         ║ ⭐⭐⭐⭐⭐   ║ ⭐⭐        ║ ⭐⭐⭐⭐     ║
    ╚════════════════════╩═══════════════╩══════════════╩═══════════════╝

    生態演進預測

    時間軸:2024-2029
    
    2024:分化和專業化開始
    ├─ LangChain:從「全能」變成「簡單應用」
    ├─ LangGraph:從「升級」變成「複雜工作流標準」
    └─ 官方 SDK:從「簡單」變成「優化和專業」
    
    2025:明確分層
    ├─ 官方 SDK(Claude/OpenAI/Gemini)⭐⭐⭐⭐⭐
    │  用戶:想要最優化的,願意被鎖定
    │
    ├─ LangGraph ⭐⭐⭐⭐
    │  用戶:複雜工作流,需要多 LLM
    │
    └─ LangChain ⭐⭐⭐
       用戶:快速原型,簡單應用
    
    2026-2029:優勝劣汰
    ├─ LangGraph 成為業界標準(類似 Docker)
    ├─ LangChain 邊緣化為「輕量級」工具
    └─ 官方 SDK 深度優化(強者恆強)

    5️⃣ 供應商風險與多 LLM 價值

    多 LLM 成本-收益分析

    場景 1:純粹的成本優化

    現狀:用 Claude Opus,月成本 $5,000
    
    優化後:
    - 複雜任務用 Opus($3,000)
    - 簡單任務用 Sonnet($500)
    - 標準任務用 GPT-4($1,000)
    
    結果:月成本 $4,500(-10%)
    
    但需要投入:
    ✓ 學習 LangGraph:2-3 周
    ✓ 測試不同 LLM:1-2 周
    ✓ 维护多模型邏輯:+20% 維護成本
    
    淨收益(年):
    成本節省:$6,000
    維護成本:$3,000
    實際節省:$3,000(不值得)

    場景 2:供應商備份(企業級需求)

    風險:
    ├─ Claude 故障 → 全系統掛
    ├─ Claude 漲價 100% → 成本翻倍
    └─ Claude 停止服務(小概率)
    
    對策:支持 Claude + GPT-4 自動備份
    
    實現成本:
    ✓ 用 LangGraph:+50% 代碼(已在複雜系統中攤銷)
    ✓ 測試備份邏輯:1-2 周
    ✓ 維護兩個模型:+30% 維護
    
    收益:
    ✓ 可用性 99.95% → 99.99%(0.04% 提升)
    ✓ 如果 Claude 故障,無損切換
    ✓ 談判籌碼增加(買方權力提升)
    
    適用場景:
    ✓ 企業級應用(可用性要求高)
    ✓ 金融/醫療(SLA 要求)
    ✓ 長期服務(>3 年)
    
    不適用場景:
    ❌ 初創應用
    ❌ 成本敏感(成本優先)
    ❌ 短期項目

    多 LLM 決策樹

    你需要多 LLM 嗎?
    
    ├─ Q1: 月 LLM 成本 > $3,000?
    │  ├─ YES → 可能值得優化成本
    │  └─ NO → 跳過
    │
    ├─ Q2: 有企業級客戶(SLA 要求)?
    │  ├─ YES → 供應商備份很重要
    │  └─ NO → 降低優先級
    │
    ├─ Q3: 預計 3 年內有 5+ 複雜項目?
    │  ├─ YES → 框架複用價值高
    │  └─ NO → 單項目的框架學習成本太高
    │
    └─ 結論:
       3 個 YES → 學 LangGraph + 多 LLM(ROI > 1.5x)
       2 個 YES → 考慮學,但不急
       1 個 YES → 暫不需要
       0 個 YES → 保持官方 SDK(省事)

    6️⃣ 生產化成本

    6 個月全生命週期成本對比

    ╔═════════════════════╦══════════════╦═════════════╦══════════════╗
    ║ 環節                ║ LangChain    ║ LangGraph   ║ Claude SDK   ║
    ╠═════════════════════╬══════════════╬═════════════╬══════════════╣
    ║ 1. 開發成本         ║              ║             ║              ║
    ║   ├─ 學習          ║ 3 天 ($600)  ║ 5 天 ($1k)  ║ 2 天 ($400)  ║
    ║   ├─ 寫代碼        ║ 4 天         ║ 3 天        ║ 5 天         ║
    ║   └─ 測試          ║ 2 天         ║ 1 天        ║ 1 天         ║
    ║   小計:           ║ $3,400       ║ $3,200      ║ $3,000       ║
    ╠═════════════════════╬══════════════╬═════════════╬══════════════╣
    ║ 2. 部署成本         ║              ║             ║              ║
    ║   └─ 監控/日誌      ║ $2,000       ║ $2,000      ║ $3,000       ║
    ╠═════════════════════╬══════════════╬═════════════╬══════════════╣
    ║ 3. 運維成本(6月)  ║              ║             ║              ║
    ║   ├─ LLM API       ║ $18,000      ║ $18,000     ║ $18,000      ║
    ║   ├─ 監控工具      ║ $600         ║ $600        ║ $3,000       ║
    ║   └─ 人力維護      ║ $3,600       ║ $1,800      ║ $3,600       ║
    ║   小計:           ║ $22,200      ║ $20,400     ║ $24,600      ║
    ╠═════════════════════╬══════════════╬═════════════╬══════════════╣
    ║ 總成本              ║ $27,600      ║ $25,600     ║ $30,600      ║
    ║ 初期貴 vs LangGraph ║ +$2,000      ║ 基準        ║ +$5,000      ║
    ║ 長期便宜度(/年)   ║ -$1,200      ║ -$2,400     ║ +$1,200      ║
    ╚═════════════════════╩══════════════╩═════════════╩══════════════╝
    
    結論:
    • LangGraph 初期略貴,但長期最便宜
    • Claude SDK 初期便宜,但運維成本高(缺監控)
    • 複雜項目超過 1 年,LangGraph ROI 最高

    監控工具成本對比

    ╔════════════════════╦═══════════════╦═════════════════╦═══════════╗
    ║ 功能               ║ LangSmith*    ║ 自建監控        ║ Claude無  ║
    ║                    ║ (LangChain)   ║ (Claude SDK)    ║ 原生工具  ║
    ╠════════════════════╬═══════════════╬═════════════════╬═══════════╣
    ║ Agent 追蹤         ║ ✅ 內置       ║ 手動寫 logging  ║ ❌        ║
    ║ 成本               ║ $100-500/月   ║ $1,000 setup    ║ $0        ║
    ║ 可視化             ║ ✅ Web UI     ║ ❌ CLI only     ║ ❌        ║
    ║ 版本管理           ║ ✅            ║ ❌              ║ ❌        ║
    ║ 易用性             ║ ⭐⭐⭐⭐      ║ ⭐              ║ -         ║
    ║ 6 月成本           ║ $600-3,000    ║ $1,000+人力     ║ $0        ║
    ╚════════════════════╩═══════════════╩═════════════════╩═══════════╝
    
    *LangSmith 支持 LangChain 和 LangGraph

    7️⃣ 組織影響維度

    團隊規模與框架選擇

    團隊規模:1 人
    ├─ LangChain: ✅ 簡單快速
    ├─ LangGraph: ⚠️ 一人維護複雜項目困難
    └─ Claude SDK: ✅ 官方最快
    
    團隊規模:2-3 人
    ├─ LangChain: ⭐⭐⭐ 合適,相對簡單
    ├─ LangGraph: ⭐⭐⭐ 複雜項目用它更清晰
    └─ Claude SDK: ⭐⭐⭐ 也可以,複雜項目時新人難上手
    
    團隊規模:4-5 人
    ├─ LangChain: ⭐⭐ 開始混亂,每個人 chain 寫法不同
    ├─ LangGraph: ⭐⭐⭐⭐⭐ 最優(標準化架構)
    └─ Claude SDK: ⭐⭐⭐ 可以,但複雜項目時溝通成本高
    
    團隊規模:10+ 人
    ├─ LangChain: ❌ 災難(無標準)
    ├─ LangGraph: ⭐⭐⭐⭐⭐ 完美(可複用框架)
    └─ Claude SDK: ⭐⭐⭐ 需要層層抽象才能用

    知識遷移成本

    在同一團隊中,第 2 個複雜項目的成本:
    
    LangChain:
    ├─ 第 1 個項目:7 天開發
    ├─ 第 2 個項目:6 天開發(相似度 +1 天)
    ├─ 第 3 個項目:5.5 天開發
    └─ 問題:每個項目的 chain 組織方式不同,無法直接複用
    
    LangGraph:
    ├─ 第 1 個項目:8 天開發
    ├─ 第 2 個項目:4 天開發(框架直接複用 -50%)
    ├─ 第 3 個項目:3 天開發(框架 + patterns 複用)
    ├─ 第 4 個項目:2.5 天開發
    └─ 優勢:框架標準化,新項目變成填空題
    
    Claude SDK:
    ├─ 第 1 個項目:7 天開發
    ├─ 第 2 個項目:7 天開發(複雜項目要重新設計)
    ├─ 第 3 個項目:7 天開發
    └─ 問題:每個複雜項目都要「重新輪子」
    
    轉折點:
    5 個項目後,LangGraph 團隊節省 > 20 天
    = 節省 $10,000 人力成本

    8️⃣ 長期戰略與演進

    3 年技術景觀預測

    2026 年 3 月(現在)
    ═══════════════════
    
    成熟度曲線:
      功能性
        ↑
        │        官方 SDK
        │     ╱────────╲
        │   ╱          ╲
        │  ╱  LangChain  ╲
        │ ╱                ╲    預期
        ├──────────────────────→ 時間
        │              LangGraph
        │         ╱──────
        │      ╱
        └────╱
    
    LangChain:成熟期(市場份額 40-50%)
    LangGraph:成長期(市場份額 15-20%)
    官方 SDK:優化期(市場份額 30-40%)
    
    
    2026 年 12 月(1 年後)
    ═════════════════════
    
    預測:
    ├─ LangGraph 市場份額 ↑ 25-30%
    ├─ LangChain 邊緣化到「簡單應用」
    ├─ 官方 SDK 加強 Agent 功能
    └─ 出現新的垂直框架(垂直行業專用)
    
    
    2028 年(2 年後)
    ═════════════════
    
    穩定格局:
    ├─ 官方 SDK:40%(強者恆強)
    ├─ LangGraph:35%(業界標準)
    ├─ LangChain:15%(輕量級遺產)
    └─ 其他新框架:10%
    
    LangGraph 地位:
    類似 Docker(容器標準)或 Terraform(IaC 標準)
    = 學好 LangGraph 的人會很值錢

    對不同角色的影響

    對開發者:
    ├─ 現在學 LangGraph → 3 年後稀缺技能
    ├─ 工資溢價:+20-30%
    └─ 就業機會:多(所有複雜 AI 項目都用)
    
    對企業主(個人公司):
    ├─ 現在投 LangGraph → 成為專家
    ├─ 能做 $10k+ 的大項目
    └─ 競爭對手少(優勢期 2-3 年)
    
    對顧問:
    ├─ 現在掌握 → 成為「多 LLM 架構顧問」
    ├─ 諮詢費:$200-500/小時
    └─ 稀缺度:高(很少人懂跨 LLM 架構)
    
    對企業 CTO:
    ├─ 現在採用 LangGraph → 技術領先
    ├─ 吸引人才(用最新技術)
    └─ 降低風險(不被 LLM 廠商鎖定)

    個人公司的實戰指南

    3 年商業規劃

    Year 1:「活著」 (現金流 $3k-5k/月)
    ════════════════════════════════
    
    目標:
    ✓ 月收 $3k-5k,活著
    ✓ 積累 5+ 個項目案例
    ✓ 建立初步品牌
    
    技術策略:
    ✅ 用 Claude SDK(最快)
    ✅ 多做簡單項目(量的積累)
    ✅ 不投 LangGraph(現金流太緊)
    
    客戶定位:
    - 預算 $2k-5k 的小項目
    - 中小 SaaS,簡單 AI 需求
    
    利潤模式:
    - 快速交付 = 高周轉率
    - 靠項目量賺錢
    
    風險:
    ⚠️ 技術債累積
    ⚠️ 被有團隊的公司碾壓複雜項目
    ⚠️ 月收入天花板(個人規模)
    
    
    Year 2:「競爭」 (現金流 $5k-10k/月)
    ═══════════════════════════════════
    
    目標:
    ✓ 月收 $5k-10k,穩定
    ✓ 建立技術品牌
    ✓ 完成 2-3 個複雜項目
    
    技術策略:
    ✅ 評估多 LLM 需求(看客戶反饋)
    ✅ 如果有複雜項目 → 投 2-3 周學 LangGraph
    ✅ 開始有意識地設計可複用框架
    ✅ 寫技術博客(建立影響力)
    
    客戶定位:
    - 升級到 $5k-15k 的項目
    - 中型企業開始看重品質
    - 複雜 AI 需求的客戶
    
    利潤模式:
    - 品質溢價(能做複雜項目,競爭少)
    - 技術諮詢(不只寫代碼)
    - 開始有長期維護合約
    
    技術積累:
    - 沉澱「複雜 Multi-Agent 系統設計」
    - 建立 LangGraph 的內部框架
    
    
    Year 3:「擴展」 (現金流 $10k-20k/月+)
    ═══════════════════════════════════════
    
    目標:
    ✓ 月收 $10k-20k,或融資,或招人
    ✓ 成為某領域的專家
    ✓ 建立品牌和 IP
    
    技術策略:
    ✅ LangGraph 完全掌握(成為專家)
    ✅ 多 LLM 支持變成標配
    ✅ 考慮開源項目(建立 IP)
    ✅ 建立「標準交付流程」(為招人做準備)
    
    客戶定位:
    - 大型企業的 POC/試驗
    - 或連續的長期合約
    - $15k-50k 以上
    
    利潤模式:
    - 顧問 + 開發(高價值)
    - 長期維護 + 迭代(穩定現金流)
    - 可能融資(有 IP 有客戶)
    
    技術 IP:
    - 「複雜 Agent 系統」的業界聲譽
    - LangGraph 專家(稀缺性)
    - 可能有開源項目

    POC 階段正確的行動清單

    **你現在在 POC + 找 TA 階段,應該這樣做:**

    優先級排序:
    
    1️⃣ 找到 TA(目標客戶)⭐⭐⭐⭐⭐
       時間:現在 - 4 周
       行動:
       □ 列出 5-10 個潛在客戶類型
       □ 分析他們的痛點
       □ 評估付費意願
       □ 深入訪談 2-3 個最有潛力的
    
       為什麼:沒有 TA,框架選擇沒意義
    
    2️⃣ 快速 POC Demo⭐⭐⭐⭐
       時間:第 3-4 周
       行動:
       □ 用 Claude API 直接寫腳本(2-3 天)
       □ 不用框架(省時間)
       □ 快速迭代(客戶反饋驅動)
    
       為什麼:驗證想法,不需要完美代碼
    
    3️⃣ Beta 測試和反饋⭐⭐⭐⭐
       時間:第 5-8 周
       行動:
       □ 找 3-5 個 beta 用戶
       □ 快速迭代(週期 1-2 周)
       □ 記錄反饋
    
       為什麼:市場信號最重要
    
    4️⃣ 評估和決策
       時間:第 9-12 周
       決定:
       ✓ 有付費意向?→ 準備 Productize
       ✓ 需求清晰?→ 投 LangGraph
       ✓ 方向不對?→ 及時止損或轉向
    
    5️⃣ 技術選型(這時才考慮)⭐⭐
       時間:12 周以後
       決策:
       • 簡單應用 → Claude SDK
       • 複雜工作流 → LangGraph
       • 多 LLM 需求 → LangGraph + 多 LLM
    
    千萬別:
    ❌ 現在投 2-3 周學 LangGraph
    ❌ POC 代碼寫得很漂亮
    ❌ 等完美再給客戶看

    完整 Jupyter Notebook

    下面是完整的、可運行的 Jupyter Notebook 代碼。你可以複製到 `.ipynb` 文件中運行。

    📌 準備工作

    # 1. 安裝依賴
    pip install langchain langchain-anthropic langchain-openai langchain-google-genai langgraph pandas matplotlib python-dotenv
    
    # 2. 創建 .env 文件
    cat > .env << EOF
    ANTHROPIC_API_KEY=your_claude_api_key
    OPENAI_API_KEY=your_openai_api_key
    GOOGLE_API_KEY=your_google_api_key
    EOF
    
    # 3. 運行 Jupyter
    jupyter notebook

    🔧 完整 Notebook 代碼

    Cell 1:安裝依賴

    import subprocess
    import sys
    
    packages = [
        'langchain',
        'langchain-anthropic',
        'langchain-openai',
        'langchain-google-genai',
        'langgraph',
        'pandas',
        'matplotlib',
        'python-dotenv'
    ]
    
    print("📦 安裝必要的包...")
    for package in packages:
        subprocess.check_call([sys.executable, "-m", "pip", "install", "-q", package])
        print(f"✅ {package}")
    
    print("\n🎉 安裝完成!")

    Cell 2:配置 API Keys

    import os
    from dotenv import load_dotenv
    
    # 加載 .env 文件中的 API Keys
    load_dotenv()
    
    # 檢查 API Keys
    apis = {
        'ANTHROPIC_API_KEY': '🔑 Claude (Anthropic)',
        'OPENAI_API_KEY': '🔑 GPT-4 (OpenAI)',
        'GOOGLE_API_KEY': '🔑 Gemini (Google)'
    }
    
    print("\n📍 檢查 API Keys 狀態:\n")
    available_apis = []
    
    for env_var, name in apis.items():
        if os.getenv(env_var):
            status = f"✅ {name}: 已配置"
            available_apis.append(env_var)
        else:
            status = f"❌ {name}: 未配置"
        print(status)
    
    if not available_apis:
        print("\n⚠️ 警告:沒有設置任何 API Key!")
        print("請按照上面的說明設置 .env 文件或環境變量。")
    else:
        print(f"\n✅ 可用的 LLM:{len(available_apis)} 個")

    Cell 3:初始化 LLM

    from langchain_anthropic import ChatAnthropic
    from langchain_openai import ChatOpenAI
    from langchain_google_genai import ChatGoogleGenerativeAI
    import time
    import json
    
    print("\n🚀 初始化多個 LLM...\n")
    
    # 初始化 LLM(都初始化,但只會使用可用的)
    llms = {}
    llm_configs = {
        'Claude': {
            'class': ChatAnthropic,
            'params': {'model': 'claude-opus-4-6'},
            'env': 'ANTHROPIC_API_KEY'
        },
        'GPT-4': {
            'class': ChatOpenAI,
            'params': {'model': 'gpt-4-turbo', 'temperature': 0.7},
            'env': 'OPENAI_API_KEY'
        },
        'Gemini': {
            'class': ChatGoogleGenerativeAI,
            'params': {'model': 'gemini-pro'},
            'env': 'GOOGLE_API_KEY'
        }
    }
    
    # 嘗試初始化每個 LLM
    for name, config in llm_configs.items():
        try:
            if os.getenv(config['env']):
                llm = config['class'](**config['params'])
                llms[name] = llm
                print(f"✅ {name}: 初始化成功")
            else:
                print(f"⏭️  {name}: 跳過(未設置 {config['env']})")
        except Exception as e:
            print(f"❌ {name}: 初始化失敗 - {str(e)[:50]}")
    
    if llms:
        print(f"\n✅ 成功初始化 {len(llms)} 個 LLM")
    else:
        print(f"\n⚠️ 沒有成功初始化任何 LLM,請檢查 API Keys")
    
    available_llms = list(llms.keys())
    print(f"\n📋 可用的 LLM:{', '.join(available_llms)}")

    Cell 4:對比測試

    # 定義測試任務
    TEST_PROMPT = """簡短回答(不超過 100 字):
    什麼是 LangChain 和 LangGraph 的主要區別?
    
    用 JSON 格式回答:
    {
      "差異": "...",
      "適用場景": "..."
    }
    """
    
    print("\n📝 測試任務:")
    print(f"提示詞:{TEST_PROMPT[:80]}...\n")
    
    # 存儲結果
    results = {}
    execution_times = {}
    
    print("\n🔄 執行中...\n")
    print("="*80)
    
    # 對每個可用的 LLM 執行
    for llm_name, llm in llms.items():
        print(f"\n▶️  {llm_name} 開始...")
    
        start_time = time.time()
    
        try:
            response = llm.invoke(TEST_PROMPT)
            elapsed = time.time() - start_time
    
            results[llm_name] = response.content
            execution_times[llm_name] = elapsed
    
            print(f"✅ {llm_name} 完成 ({elapsed:.2f}s)")
            print(f"\n📄 回答:")
            print(f"{response.content[:150]}...")
            print("\n" + "-"*80)
    
        except Exception as e:
            print(f"❌ {llm_name} 出錯:{str(e)[:100]}")
            print("\n" + "-"*80)
    
    print("\n" + "="*80)
    print(f"\n✅ 測試完成!共執行 {len(results)} 個 LLM")

    Cell 5:執行時間對比

    import pandas as pd
    import matplotlib.pyplot as plt
    
    # 創建時間對比表
    if execution_times:
        df_times = pd.DataFrame([
            {'LLM': name, '執行時間 (秒)': time}
            for name, time in execution_times.items()
        ]).sort_values('執行時間 (秒)')
    
        print("\n⏱️  執行時間對比")
        print("="*50)
        print(df_times.to_string(index=False))
        print("="*50)
        print(f"\n平均時間:{df_times['執行時間 (秒)'].mean():.2f}s")
        print(f"最快:{df_times.iloc[0]['LLM']} ({df_times.iloc[0]['執行時間 (秒)']:.2f}s)")
    
        # 繪製柱狀圖
        plt.figure(figsize=(10, 5))
        colors = ['#FF6B6B' if t == df_times['執行時間 (秒)'].min() else '#4ECDC4'
                  for t in df_times['執行時間 (秒)']]
    
        plt.bar(df_times['LLM'], df_times['執行時間 (秒)'], color=colors, alpha=0.7, edgecolor='black')
        plt.ylabel('執行時間 (秒)', fontsize=12)
        plt.xlabel('LLM', fontsize=12)
        plt.title('🏃 多 LLM 執行時間對比', fontsize=14, fontweight='bold')
        plt.grid(axis='y', alpha=0.3, linestyle='--')
    
        # 添加數值標籤
        for i, v in enumerate(df_times['執行時間 (秒)']):
            plt.text(i, v + 0.1, f'{v:.2f}s', ha='center', fontweight='bold')
    
        plt.tight_layout()
        plt.show()
    
    else:
        print("⚠️  沒有執行時間數據")

    Cell 6:質量對比

    # 分析回答質量
    if results:
        print("\n📊 回答質量對比")
        print("="*80)
    
        quality_metrics = []
    
        for llm_name, response in results.items():
            # 計算指標
            length = len(response)
            word_count = len(response.split())
            has_json = '{' in response and '}' in response
    
            quality_metrics.append({
                'LLM': llm_name,
                '字數': length,
                '詞數': word_count,
                'JSON 格式': '✅' if has_json else '❌'
            })
    
        df_quality = pd.DataFrame(quality_metrics)
        print(df_quality.to_string(index=False))
        print("="*80)
    
        # 詳細回答
        print("\n📄 詳細回答:\n")
        for llm_name, response in results.items():
            print(f"\n▶️  {llm_name}:")
            print("-" * 70)
            print(response)
            print("-" * 70)

    Cell 7:LangGraph 演示

    from langgraph.graph import StateGraph, END
    from typing import TypedDict
    
    if len(llms) >= 2:
        print("\n🏗️  使用 LangGraph 構建多 LLM 工作流")
        print("="*80)
    
        # 定義狀態
        class ComparisonState(TypedDict):
            task: str
            results: dict  # 存儲多個 LLM 的結果
    
        # 為每個 LLM 定義一個節點
        def create_llm_node(llm_name, llm):
            def node_func(state: ComparisonState) -> dict:
                print(f"\n▶️  {llm_name} 處理中...")
                try:
                    response = llm.invoke(state['task'])
                    result = {
                        'response': response.content[:200] + '...',
                        'status': '✅ 成功',
                        'timestamp': time.time()
                    }
                    state['results'][llm_name] = result
                    print(f"✅ {llm_name} 完成")
                except Exception as e:
                    state['results'][llm_name] = {
                        'error': str(e)[:100],
                        'status': '❌ 失敗'
                    }
                    print(f"❌ {llm_name} 失敗")
    
                return state
    
            return node_func
    
        # 構建 Graph
        graph = StateGraph(ComparisonState)
    
        # 為每個 LLM 添加節點
        for llm_name in llms:
            graph.add_node(llm_name, create_llm_node(llm_name, llms[llm_name]))
    
        # 連接節點(全部並行執行,然後到 END)
        graph.set_entry_point(list(llms.keys())[0])
        for i, llm_name in enumerate(list(llms.keys())[:-1]):
            graph.add_edge(llm_name, list(llms.keys())[i+1])
        graph.add_edge(list(llms.keys())[-1], END)
    
        # 編譯并執行
        workflow = graph.compile()
    
        print("\n📋 執行工作流...\n")
    
        test_task = "用一句話解釋 LangGraph 的核心優勢"
    
        try:
            workflow_result = workflow.invoke({
                'task': test_task,
                'results': {}
            })
    
            print(f"\n✅ 工作流完成!\n")
            print("📊 結果:")
            print("="*80)
    
            for llm_name, result in workflow_result['results'].items():
                print(f"\n{llm_name}:")
                print("-"*70)
                if 'error' in result:
                    print(f"❌ 錯誤:{result['error']}")
                else:
                    print(f"{result['response']}")
    
            print("\n" + "="*80)
    
        except Exception as e:
            print(f"❌ 工作流執行失敗:{str(e)}")
    
    else:
        print("⚠️  需要至少 2 個 LLM 才能演示工作流")

    Cell 8:成本對比

    # 模擬成本數據(基於實際 API 定價 2026 年 3 月)
    cost_data = {
        'Claude': {
            'name': 'Claude Opus 4.6',
            'input_cost': 0.003,  # 每 1K tokens
            'output_cost': 0.015,
            'speed': '中等',
            'quality': '⭐⭐⭐⭐⭐',
            'price_tier': '高端'
        },
        'GPT-4': {
            'name': 'GPT-4 Turbo',
            'input_cost': 0.01,
            'output_cost': 0.03,
            'speed': '快',
            'quality': '⭐⭐⭐⭐⭐',
            'price_tier': '最高'
        },
        'Gemini': {
            'name': 'Gemini Pro',
            'input_cost': 0.0005,
            'output_cost': 0.0015,
            'speed': '最快',
            'quality': '⭐⭐⭐⭐',
            'price_tier': '經濟'
        }
    }
    
    print("\n💰 成本和性能對比")
    print("="*90)
    print(f"{'LLM':<15} {'速度':<10} {'質量':<15} {'輸入成本':<15} {'輸出成本':<15} {'定位':<10}")
    print("="*90)
    
    for llm_name, data in cost_data.items():
        if llm_name in llms:
            status = "✅"
        else:
            status = "⏭️ "
    
        print(f"{status}{llm_name:<13} {data['speed']:<10} {data['quality']:<15} "
              f"${data['input_cost']:<14} ${data['output_cost']:<14} {data['price_tier']:<10}")
    
    print("="*90)
    
    # 估算月成本(假設 1000 萬 tokens 使用)
    print("\n📊 月成本估算(假設 1000 萬 input tokens + 500 萬 output tokens)")
    print("="*60)
    
    cost_estimates = []
    input_tokens = 10_000_000
    output_tokens = 5_000_000
    
    for llm_name, data in cost_data.items():
        monthly_cost = (input_tokens / 1000 * data['input_cost'] +
                       output_tokens / 1000 * data['output_cost'])
        cost_estimates.append({
            'LLM': llm_name,
            '月成本': f"${monthly_cost:.2f}",
            '成本占比': f"{monthly_cost / sum([((input_tokens / 1000 * cost_data[k]['input_cost'] + output_tokens / 1000 * cost_data[k]['output_cost'])) for k in cost_data]) * 100:.1f}%"
        })
    
    df_costs = pd.DataFrame(cost_estimates)
    print(df_costs.to_string(index=False))
    print("="*60)
    
    print("\n💡 建議:")
    print("  • 複雜任務 → 用 Claude(質量最好)")
    print("  • 簡單任務 → 用 Gemini(成本最低)")
    print("  • 要求快速 → 用 GPT-4(速度最快)")
    print("  • 多 LLM 支持 → 用 LangGraph(靈活切換)")

    Cell 9:框架對比

    print("\n🔄 架構對比:LangChain vs LangGraph")
    print("\n" + "="*80)
    
    comparison = {
        "方面": [
            "代碼行數(簡單流程)",
            "代碼行數(複雜流程)",
            "狀態管理",
            "並行執行",
            "易於測試",
            "新人上手",
            "適用場景"
        ],
        "LangChain": [
            "~40 行",
            "80+ 行(複雜)",
            "手動(state 散亂)",
            "極其困難",
            "⭐⭐(中等)",
            "1-2 天",
            "簡單應用、快速原型"
        ],
        "LangGraph": [
            "~25 行",
            "22 行(簡潔)",
            "自動(State TypedDict)",
            "⭐⭐⭐⭐⭐ 天生支持",
            "⭐⭐⭐⭐⭐(優秀)",
            "3-5 天(陡峭但有回報)",
            "複雜工作流、多 Agent"
        ]
    }
    
    df_comparison = pd.DataFrame(comparison)
    print(df_comparison.to_string(index=False))
    print("="*80)
    
    print("\n🎯 何時選擇:")
    print("\n✅ 選 LangChain 當:")
    print("   • 時間緊張(< 2 周)")
    print("   • 應用簡單(< 3 個 Agent)")
    print("   • 快速原型")
    
    print("\n✅ 選 LangGraph 當:")
    print("   • 複雜工作流(3+ Agent)")
    print("   • 有並行邏輯")
    print("   • 需要多 LLM 支持")
    print("   • 長期維護和擴展")
    print("   • 團隊規模 > 3 人")

    Cell 10:最終總結

    print("\n" + "🎯 "*40)
    print("\n📊 你的實驗總結:")
    print("\n" + "="*80)
    
    if execution_times:
        print(f"\n✅ 已測試 {len(llms)} 個 LLM")
        fastest = min(execution_times, key=execution_times.get)
        slowest = max(execution_times, key=execution_times.get)
        print(f"   • 最快:{fastest} ({execution_times[fastest]:.2f}s)")
        print(f"   • 最慢:{slowest} ({execution_times[slowest]:.2f}s)")
        print(f"   • 平均:{sum(execution_times.values())/len(execution_times):.2f}s")
    
    print(f"\n✅ 框架對比結論:")
    print(f"   LangChain: 適合簡單應用、快速原型")
    print(f"   LangGraph: 適合複雜工作流、多 Agent、多 LLM")
    
    print(f"\n✅ 多 LLM 切換的好處:")
    print(f"   • 成本優化:同一套代碼,根據任務選 LLM")
    print(f"   • 供應商備份:一個 LLM 故障,自動切換")
    print(f"   • 市場適應:新 LLM 出現,快速集成")
    
    print(f"\n✅ 給個人公司的建議:")
    print(f"   Year 1: 用 Claude SDK,多做項目(現金流優先)")
    print(f"   Year 2: 評估 LangGraph,學習投入(複雜項目有收益)")
    print(f"   Year 3: 成為專家,多 LLM 支持(競爭優勢)")
    
    print("\n" + "="*80)
    print(f"\n🚀 下一步行動:")
    print(f"   1. 根據你的 POC 找到具體的 TA(目標客戶)")
    print(f"   2. 驗證市場需求(beta 用戶反饋)")
    print(f"   3. 決定是否投資 LangGraph 學習")
    print(f"   4. 考慮多 LLM 支持(如果客戶有需求)")
    
    print(f"\n" + "🎯 "*40)

    快速開始指南

    🎯 5 分鐘快速開始

    # 1. 安裝依賴
    pip install jupyter notebook
    
    # 2. 創建 .env 文件
    cat > .env << EOF
    ANTHROPIC_API_KEY=sk-ant-xxxxxxxxxxxx
    OPENAI_API_KEY=sk-xxxxxxxxxxxx
    GOOGLE_API_KEY=xxxxxxxxxxxx
    EOF
    
    # 3. 創建 Jupyter Notebook
    jupyter notebook
    
    # 4. 在 Notebook 中粘貼上面的代碼
    # 5. 依次執行 Cells

    🔑 API Keys 獲取

    📋 Notebook 執行順序

    1. ✅ Cell 1:安裝依賴 2. ✅ Cell 2:配置 API Keys 3. ✅ Cell 3:初始化 LLM 4. ✅ Cell 4-5:對比測試 5. ✅ Cell 6:質量分析 6. ✅ Cell 7:LangGraph 演示 7. ✅ Cell 8:成本對比 8. ✅ Cell 9:框架對比 9. ✅ Cell 10:總結

    最終決策框架

    三位一體:老闆 × 顧問 × PM

    你同時是三個身份,優先級根據情況變化:
    
    情況 1:現金流緊張(< 1 月)
    優先級:老闆 > PM > 顧問
    決策:快速賺錢最重要
      ├─ 用 Claude SDK(快)
      ├─ 多做小項目
      └─ 不學新框架
    
    情況 2:穩定現金流(3+ 月)
    優先級:顧問 > PM > 老闆
    決策:給客戶好方案,建立品牌
      ├─ 評估 LangGraph(複雜項目)
      ├─ 提高服務質量
      └─ 長期客戶關係
    
    情況 3:複雜項目 + 預算充足
    優先級:PM > 顧問 > 老闆
    決策:交付成功最重要
      ├─ 用 LangGraph(管理複雜度)
      ├─ 投前期設計時間
      └─ 降低風險

    快速決策矩陣

    ╔═══════════════════════════════════════════════════════════════╗
    ║                 你應該選哪個框架?                           ║
    ╠═══════════════════════════════════════════════════════════════╣
    
    1. 項目複雜度?
       ├─ 簡單(1-2 Agent)→ 看 Q3
       ├─ 中等(3-5 Agent)→ 看 Q3
       └─ 複雜(5+ Agent 或並行)→ LangGraph ✓
    
    2. 多 LLM 需求?
       ├─ 不需要 → 用官方 SDK
       └─ 需要 → 加 LangGraph
    
    3. 時間線?
       ├─ < 2 周 → Claude SDK(快)
       ├─ 2-4 周 → LangChain(平衡)
       └─ > 1 月 → LangGraph(品質優先)
    
    4. 團隊規模?
       ├─ 1-2 人 → Claude SDK(簡單)
       ├─ 3-5 人 → LangGraph(標準化)
       └─ 10+ 人 → LangGraph(可複用)
    
    5. 長期規劃?
       ├─ 1 個項目 → Claude SDK(不值得學)
       ├─ 2-3 個 → 考慮 LangGraph
       └─ 5+ 個 → 必須 LangGraph(ROI 高)
    
    ╠═══════════════════════════════════════════════════════════════╣
    
    最終建議:
    
    ✅ 用 Claude SDK 的條件:
       • 簡單應用(< 3 Agent)
       • 時間緊張(< 2 周)
       • 團隊小(1-2 人)
       • 短期項目(< 1 個項目規劃)
    
    ✅ 用 LangChain 的條件:
       • 快速原型
       • 簡單應用
       • 想支持多 LLM(但工作流簡單)
       • 喜歡靈活性
    
    ✅ 用 LangGraph 的條件:
       • 複雜工作流(> 3 Agent)
       • 並行執行或分支邏輯
       • 多 LLM 支持
       • 長期項目規劃(3+ 個)
       • 團隊 > 3 人
    
    ╚═══════════════════════════════════════════════════════════════╝

    核心結論

    1. 框架選擇不是技術問題,是商業問題
       ✓ 有付費客戶?有複雜項目?有長期規劃?
       ✗ 這些都沒有?先驗證市場
    
    2. LangGraph 的價值在於「長期」
       • 第 1 個項目:LangGraph 可能慢
       • 第 3 個項目:LangGraph 快 50%
       • 第 5 個項目:LangGraph 快 70%
    
    3. 多 LLM 不是為了省錢,而是為了自由
       • 供應商備份(可用性)
       • 成本優化(可選)
       • 市場適應(長期)
    
    4. POC 階段最重要的是市場驗證,不是技術選擇
       • 1 周市場反饋 > 2 周完美代碼
       • TA 決定一切
       • 框架是後話
    
    5. 個人公司的出路是「垂直深化」,不是「技術秀肌肉」
       • 某個領域的專家 > 全能技術人
       • $10k-20k 的垂直項目 > $2k-5k 的通用項目
       • 品質溢價 > 技術先進性

    後記

    **給你的最後話**:

    你已經掌握了 Claude Agent SDK,這給了你第一步的優勢。現在的問題不是「選哪個框架」,而是「選哪個市場」。

    LangChain/LangGraph 是在你確定市場和客戶後,為了「長期優化」的選擇。不是「先進」,而是「務實」。

    現在就出去找 TA,快速驗證想法。3 個月後,當你有 3-5 個清晰的客戶需求時,再回過頭來考慮技術選擇。那時候的決策會基於真實數據,而不是假設。

    同時,寫好你的博客。「Claude Agent SDK 深度實踐指南」會成為你的品牌。未來的「LangGraph 多 LLM 架構指南」會是補充。

    競爭力不來自「最新技術」,而來自「解決真實問題的深度」。

    **撰寫時間**:2026-03-17 **版本**:1.0 **包含**:完整分析 + Jupyter Notebook + 決策框架

    **加油!👊**

  • Agent Team 多輪迭代:從失敗到成功的設計演進

    Document Status: Living Document (持續更新)
    Last Updated: 2026-03-17
    Author: Claude Code + Tom
    Purpose: 從 SimpleEC OMS 多次失敗的 Agent Team 經驗中提煉最佳實踐

    🎯 重點摘要

    • 問題:4 次 Agent Team 啟動都在 24+ 小時後卡住,根本原因不在代碼,而在設計與 Prompt
    • 核心原則:編碼依賴(Task.blockedBy/blocks)、明確停止條件、2 小時 timeout、結構化交付物
    • 關鍵對比:Provider Chain (複雜) vs Sequential Phase (簡單) — 後者減少隱式依賴、自動強制約束
    • Prompt 檢查清單:5 大維度、17 項細節檢查,防止 Agent 卡住或無限期等待
    • 成功設計:明確的時間表、具體的交付物格式 (JSON)、禁止行為列表、協調者角色

    序言:為什麼我們一直失敗

    從 2026-03-03 到 2026-03-17,我們嘗試啟動過至少 4 個 Agent Team,每一次都卡住 24+ 小時:

    1. simpleec-oms-audit (3/3) — 任務分配錯亂,Agent 互相等待
    2. simpleec-oms-bidirectional-audit (3/10) — 複雜的雙向驗證設計,Agent 不知道什麼時候應該停止
    3. simpleec-oms-quality-audit (3/16) — Task 被分配給錯誤的 Agent,導致 3 個 Agent 空轉 24 小時

    根本原因不在代碼,而在於

    • ❌ 設計太複雜,隱式依賴沒有編碼
    • ❌ Prompt 寫得不清楚,Agent 不知道什麼時候應該停止或等待
    • ❌ 沒有 Timeout 機制,導致無限期等待
    • ❌ 任務分配邏輯複雜,容易出錯
    • ❌ Agent 沒有明確的”停止條件”,導致亂工作或空轉

    本文將詳細解析這些問題,並提供可複製的最佳實踐。

    🎯 你需要打什麼讓我一步到位

    複製以下內容給 Claude

    我要啟動 Agent Team 做 [你的任務]
    
    系統:[系統名]
    檢查維度:[A], [B], [C]
    時間:[時間限制]
    Agent:[N 個]
    
    創建:TEAM_PLAN.md / AGENT_PROMPTS.md / CHECKLIST.md / settings.json

    📋 成功的 Prompt 範本

    You are ARCHITECT on [TEAM_NAME].
    
    Verify [SYSTEM]:
    - [Check A]
    - [Check B]
    
    DELIVERABLE (JSON):
    { "task_id": 1, "findings": { "score": X, "ready": true } }
    
    BEGIN PHASE 1 NOW.

    第一部分:為什麼 Agent Team 會卡住?

    原因 1:複雜的隱式依賴

    ❌ 壞設計示例:使用 Provider Chain + Consumer Chain 雙向驗證(複雜且失敗)。有 4 個同步點,但都是文字描述,沒有編碼到 Task 系統。Task 之間的依賴關係是隱式的(”Task #1 完成後才能開始 Task #2″,但沒有在 blockedBy 里標記)。Agent 不知道應該等待誰被誰等待

    後果:當任務分配出錯(比如 Task #2 被分給了錯誤的 Agent),整個鏈條斷裂。Architect 和 Backend-QA 無所事事,空轉 24+ 小時,持續消耗內存。

    原因 2:Prompt 寫得不清楚 → Agent 不知道什麼時候該停止

    Prompt 缺少明確的停止條件。完成 Task #1 後,Agent 不知道應該做什麼——是否應該主動聯繫其他 Agent?是否應該重新檢查自己的工作?內存會持續增長(因為 Agent 一直在運行)。

    實際情況:Architect 完成 Task #1 後,發現沒有 Task #2 和 #3 的結果(因為被分配錯了)。開始懷疑,重新讀一遍所有文件。14 小時後,內存從 380MB 漲到 440MB。最後 OOM → 系統當機。

    原因 3:沒有 Timeout 機制

    舊設計 新設計
    Agent Task 超時限制:無
    → 可以無限期等待
    → 可以無限期工作
    → 內存持續增長,最終 OOM
    Agent Task 超時限制:2 小時
    → 2 小時後,自動標記 timeout
    → 解鎖下一個任務或停止
    → 即使出錯,也不會無限期卡住

    原因 4:Agent 之間沒有同步點

    舊設計中,Task #1 在工作,Task #2 也在工作,但 Task #2 應該等待 Task #1。沒人告訴他們應該等待誰。新設計中,用 Task.blockedBy = [1, 2] 明確表示”我要等 Task #1, #2″。

    第二部分:好的 Agent Team 設計原則

    原則 1:Show, Don’t Tell(編碼依賴,不要寫文字)

    為什麼? Task 系統會自動強制依賴順序,不會分配錯誤。不依賴人工判斷。系統可以自動檢測循環依賴或孤立任務。

    原則 2:一個 Agent 做一件事

    ❌ 壞:Task #2 包含 3 個不同的東西(Schema 驗證、Kafka 驗證、Handler 驗證)。Agent 容易分心,難以判斷什麼時候”完成”。

    ✅ 好:Task #2 目標清晰:”驗證數據層能否支持 API 承諾”。包含 3 個驗證點,但都是同一件事的一部分。交付物是一個結構化的 JSON 對象。

    原則 3:明確的交付物結構

    ❌ 壞:”Generate audit report with Executive summary, Critical issues list…” 模糊,無法驗證是否合格。

    ✅ 好:返回結構化 JSON,包含量化指標(alignment score)、布爾標誌(ready_for_next_phase)、具體問題列表。下一個 Agent 可以用 JSON parser 讀取。

    原則 4:Prompt 要明確定義”停止條件”

    Prompt 必須明確說:完成 Task 後,Agent 應該做什麼(答案是:停止工作,等待通知)。沒有”循環工作”的空間。明確的等待機制(team-lead 會告訴你什麼時候開始下一個 Phase)。

    原則 5:Timeout 是必需的

    新設計:2 小時 timeout。時間到了,自動標記 timeout。自動解鎖下一個 Task(即使當前 Task 沒完成)。自動通知 team-lead。防止無限期等待。

    第三部分:Prompt 編寫終極檢查清單

    A. 責任清晰度(Clarity of Responsibility)

    • 一個 Agent,一件事 — 避免重載式任務設計
    • 目標可測 — 不是”驗證 schema 完整性”,而是”生成 JSON with schema_integrity score”
    • 交付物明確 — 結構化 JSON,而不是自由文本報告

    B. 停止條件明確(Clear Stop Conditions)

    • Prompt 明確說什麼時候應該停止 — “After Task #1 is complete, do NOT continue. Wait for team-lead notification.”
    • 沒有”循環工作”的空間 — “After generating findings JSON, stop. Do not re-read files.”
    • 明確的等待機制 — “Wait for team-lead message: ‘Phase 2 begins’”

    C. 依賴關係編碼(Dependency Encoding)

    • Task.blockedBy 明確填寫"blockedBy": [1] 表示”我等 Task #1″
    • Task.blocks 明確填寫"blocks": [3, 4] 表示”我阻擋 Task #3 和 #4″
    • 沒有循環依賴 — Task #1 blocks #2, #2 blocks #3, #3 blocks #1 ← 循環!

    D. 超時保護(Timeout Protection)

    • Prompt 中明確提到超時時間 — “You have 2 hours to complete Task #1”
    • 告訴 Agent 超時時會發生什麼 — “If 2 hours pass, next phase will begin regardless”
    • Timeout 不應該導致錯誤,只是解鎖 — 後自動繼續下一階段

    E. Agent 之間的協調(Inter-Agent Coordination)

    • 明確說是否應該主動聯繫其他 Agent — “Do NOT contact backend-qa or frontend-qa”
    • 明確說如何接收通知 — “You will receive explicit notification from team-lead when Phase 2 begins”
    • 明確說何時應該報告問題 — “Report them in your findings JSON. Do NOT try to fix them yourself.”

    第四部分:Prompt 好與壞的完整對比

    ❌ 壞 Prompt(導致失敗)

    缺少:明確的完成後行為、明確的等待規則、Timeout 時間、停止條件、交付物格式的具體性。

    後果:Architect 完成 Task #1 後,發現沒有 Task #2 和 #3 的結果。不知道應該等待還是繼續工作。開始懷疑自己的分析,重新讀文件。持續讀文件和重新分析 14+ 小時。內存從 380MB 漲到 440MB。最後 OOM,系統當機。

    ✅ 好 Prompt(成功設計)

    特點:明確的時間表(0-2h Phase 1, 2-4h 等待, 4-6h Phase 4)。明確的停止條件(”完成後停止,不要做任何其他事”)。明確的交付物(結構化 JSON)。明確的等待機制(team-lead 會告訴你什麼時候開始 Phase 4)。明確的禁止行為(❌ 列出了不應該做的事)。

    後果:Architect 完成 Task #1。知道自己應該停止(Prompt 明確說了)。進入等待模式(不讀文件,不重新分析)。2 小時後,如果沒有 Task #4 通知,自動 timeout。內存穩定在 380MB(沒有持續增長)。系統正常運行。

    第五部分:常見錯誤與修正

    錯誤 1:Prompt 寫得太長和太複雜

    修正:用清晰的結構(Markdown headers)。用列表而不是段落。用例子而不是抽象描述。用”禁止行為”而不是”允許行為”。目標:500 字清晰 Prompt,而不是 2000 字複雜 Prompt。

    錯誤 2:隱式假設 Agent 會自己推斷

    ❌ 壞:”After Task #1, you will know when to start Task #2″(Agent 怎麼知道?沒人告訴他!)

    ✅ 好:”After Task #1, wait for team-lead message: ‘Task #2 begins’. Only then should you proceed.”

    錯誤 3:交付物格式模糊

    ❌ 壞:”Generate a report with findings”

    ✅ 好:生成 JSON,具有明確的字段結構 (status, findings, critical_issues 等)

    錯誤 4:沒有明確的超時時間

    ✅ 修正Timeout: 2 hours — If 2 hours pass, your task is automatically marked timeout. Next phase will proceed regardless. You will be notified when your timeout occurs.

    第六部分:Agent Team 設計檢查清單

    Pre-Launch Checklist

    系統狀態:內存充足 (>8GB available)。沒有舊的 Agent Team 在運行。Task 目錄是空的。

    Task 設計:每個 Task 有明確的 id, subject, description。每個 Task 的 blockedBy 和 blocks 已填寫(沒有循環依賴)。沒有孤立的 Task。每個 Task 有 2 小時的隱式 timeout。

    Prompt 設計:每個 Agent 的 Prompt 有明確的責任(一件事)。每個 Prompt 有明確的停止條件。每個 Prompt 有具體的交付物格式(JSON)。每個 Prompt 有明確的等待機制。每個 Prompt 禁止了不應該做的事情(❌ 列表)。

    交付物定義:所有交付物都是結構化的(JSON)。每個交付物都有 status 字段。下一個 Agent 能解析上一個 Agent 的交付物。交付物不應該是自由文本,應該是結構化數據。

    同步機制:有明確的”team-lead”角色負責協調。有明確的通知機制。有明確的失敗恢復流程。沒有 Agent 之間的直接通信(都通過 team-lead)。

    Launch Checklist

    監控:每 30 分鐘檢查一次內存使用。監控是否有 Agent 卡住。準備好 2 小時後手動檢查 timeout 機制。

    信號:知道什麼樣的行為表示”卡住”(讀同一個文件超過 1 小時,內存持續增長 10% 以上,沒有進展 30 分鐘以上)。準備好應急措施(2 分鐘內關掉所有 Agent,清理內存,回滾到之前的狀態)。

    第七部分:迭代和改進

    本文將根據實驗結果進行迭代。每當我們啟動新的 Agent Team 時:

    1. 記錄過程 — 啟動時間、Agent 行為、完成時間、內存使用、任何卡住或異常情況
    2. 在本文中更新 — 如果發現新的問題、更好的 Prompt 寫法、或 Timeout 時間不合適的情況
    3. 版本控制 — 每次大的更新,提升版本號。在文件頂部記錄更新日期和內容

    總結

    從失敗中學到的核心真理

    1. 設計要簡單,依賴要明確 — 不要用複雜的 Provider Chain,用簡單的 Sequential Phase
    2. 編碼依賴,不要寫文字blockedByblocks 必須填寫,隱式依賴是毒藥
    3. Prompt 要明確停止條件 — Agent 必須知道什麼時候應該停止
    4. Timeout 是必需的保險 — 沒有 timeout,Agent 可以無限期卡住
    5. 交付物要結構化 — JSON 而不是自由文本,下一個 Agent 能解析
    6. 有一個協調者(team-lead) — 不要讓 Agent 之間直接通信

    下一步:用 v2 設計啟動 Quality Audit Team。記錄過程和結果。根據結果更新本文。建立標準範本(Task 和 Prompt 的 template)。

    實施深度:v3 計畫的四層架構

    前面的原則是設計哲學。但實際部署 v3 計畫時,我們加入了四項關鍵實施細節,把抽象的原則轉換為可監控、可恢復、可自動化的運作

    ①分層監控架構(Team-lead 的心跳)

    v3 計畫引入了 Team-lead 監控層,專職於此,间隔嚴格:30秒 Session 健康檢查、30分鐘超時檢查、60分鐘進度報告、5分鐘內存監控。

    ②事件驅動協調(4 個事件)

    v3 計畫改為事件驅動模型,完全由 Team-lead 控制狀態轉遷。4個事件:Task#1完成→解鎖Task#2&3、Task#2&3都完成→通知Architect進行Task#4、Task#4完成→生成報告。

    ③應急與恢復(5 個故障模式)

    v3 計畫為5種情況預先定義恢復策略:Orphaned Agents(中止審計)、Multiple Timeouts(部分報告)、Invalid JSON(記錄錯誤繼續)、OOM Event(強制超時解鎖)、Stalled Task(檢測重啟或強制超時)。

    ④完整實現範例

    完整的Task生命週期:t=0初始化、t=0-30分鐘Architect驗證、t=25觸發Event1解鎖Task#2&3、t=25-50並行執行QA驗證、t=50觸發Event2&3通知Architect、t=50-110Task#4最終驗證、t=115報告生成完成。總耗時115分鐘,4個Agent共消耗1.8GB內存。

    v3 計畫 vs v2:7 個關鍵改進

    依賴編碼從文字敘述改為JSON(blockedBy/blocks)、停止條件從暗示改為明確、超時機制從提到改為實現(120min+30min+自動unlock)、監控從無改為Team-lead 30秒心跳、協調從Agent間通信改為Team-lead主控、交付物從文字改為JSON schema、故障恢復從無預案改為5個策略。

    「後即時」監控與自動恢復

    v3 計畫的終極目標是達到「後即時(Near Real-Time)監控」:不是等Task失敗才發現,而是任何異常2分鐘內自動檢測並響應。解決之前24小時卡住的問題。

  • Hacker News 每日精選 – 2026-03-17

    🚀 今天的科技圈充滿了對「效率」與「結構」的重新思考。從美國 SEC 準備取消季度財報的震撼彈,到 Meta 重新對底層基礎設施 jemalloc 的承諾,我們正處於一個追求長期價值與開發流程簡化(避免過度審查)的轉折點。

    💡 無論你是關注基礎架構的工程師,還是追蹤市場動態的創業家,今日的資訊流都揭示了一個核心訊息:在 AI 與自動化加速發展的時代,繁文縟節與低效溝通正被無情地淘汰。

    🤖 AI / 機器學習

    Leanstral:用於可信編碼與形式證明工程的開源 Agent

    Mistral AI 推出了 Leanstral,這是一個專門為 Lean 4 形式語言設計的開源模型與 Agent。與傳統 LLM 僅給出機率性的代碼不同,Leanstral 專注於「形式化證明」,這意味著生成的代碼可以經過數學上的驗證,確保其正確性與安全性。對於需要高度信任的關鍵系統開發者來說,這是一大突破。

    • 🚀 重點:將 AI 從「猜測代碼」提升到「證明代碼」。
    • 🎯 目標:減少軟體漏洞並推動形式化方法在業界的普及。

    👉 閱讀原文

    Claude 在 3D 工作流中的應用技巧

    這篇文章分享了如何利用 Claude 輔助 3D 內容創作。作者提到 LLM 在處理複雜的 3D 腳本(如 Blender Python API)與生成著色器代碼(Shaders)時表現優異。透過精確的提示詞與迭代,Claude 能顯著縮短傳統 3D 藝術家在技術瓶頸上花費的時間。

    👉 閱讀原文

    🛠️ 開發工具與基礎設施

    Meta 對 jemalloc 的重新承諾與投入

    Meta(Facebook)宣布將持續投資並優化 jemalloc,這是一個廣泛用於 Linux 系統的高效能記憶體分配器。Meta 透過工程實踐證明,對於處理數兆字節數據的大型基礎設施,精細的記憶體管理能直接轉化為成本節省與系統穩定性,這反擊了「基礎設施已死」的論調。

    👉 閱讀原文

    每一層審核都會讓你慢上 10 倍

    這是一篇關於軟體工程效率的深刻反思。作者 apenwarr 指出,組織中每增加一道審核關卡(Review Layer),項目的推進速度就會呈指數級下降。文章主張與其增加審核來防止錯誤,不如建立自動化測試與快速復原機制,賦予開發者更大的自主權。

    「如果你需要三個人簽字才能發布代碼,你不是在保證品質,而是在保證開發進度停滯。」

    👉 閱讀原文

    從零開始為 Commodore 64 重新打造《猴島小英雄》

    這是一項令人驚嘆的逆向工程與復古開發計畫。作者詳細紀錄了如何將經典冒險遊戲《猴島小英雄》移植到 1982 年的硬體 Commodore 64 上。這不僅是情懷,更展示了在極端硬體限制下,如何透過精妙的內存管理與算法優化來達成現代化遊戲體驗。

    👉 閱讀原文

    💼 創業、商業與政策

    美國 SEC 準備取消季度報告要求

    這可能是近年來對美國資本市場影響最大的政策調整。SEC 正在考慮取消上市公司每三個月發布一次財報的強制要求(10-Q),轉向半年報或更靈活的披露機制。支持者認為這能減少企業的「短期主義」壓力,讓管理層專注於長期戰略而非每季度的股價波動。

    👉 閱讀原文

    Kagi Translate 現在支援「LinkedIn 腔」作為輸出語言

    這是一個帶有黑色幽默的功能更新。付費搜尋引擎 Kagi 旗下的翻譯工具新增了 “LinkedIn Speak” 選項,能將普通的日常對話轉換成充滿企業術語、勵志格言與大量 Emoji 的 LinkedIn 貼文風格。這雖然看似玩笑,卻諷刺了當前職業社交平台上的誇大文化。

    👉 閱讀原文

    🌐 其他值得關注

    「小眾網路」(Small Web)比你想像的還要龐大

    當主流網路充斥著 AI 生成的垃圾訊息與 SEO 內容時,回歸個人部落格、手寫 HTML 網頁的「Small Web」運動正在興起。文章討論了這些由人類親手策劃、不為流量服務的網站,如何構成了一個更有活力、更具真實感的數位生態系統。

    👉 閱讀原文

    美國醫療困境:數據分析

    這個開源專案在 GitHub 上引起熱議,它透過大量的數據視覺化探討了美國醫療系統的高昂成本與低效回報。對於關注社會工程與數據科學的讀者來說,這是一個利用數據解構複雜社會體制的極佳範例。

    👉 閱讀原文

    🎯 今日觀點:效率的重新定義

    今天的熱門話題共同指向了一個趨勢:去除無效的中間層

    • 制度面:SEC 試圖減少頻繁財報帶來的行政負擔與短期壓力。
    • 開發面:我們被提醒審核流程(Review Layer)可能是創新的殺手。
    • 技術面:Leanstral 試圖跳過「人為猜測」,直接用數學確保正確性。

    👨‍💻 給讀者的行動建議:
    檢查你目前的開發流程或工作流,是否有哪一個「審核層級」其實可以被自動化測試更強大的底層工具所取代?在 AI 加速生成的時代,保護你的「專注力」與「開發速度」將是維持競爭力的關鍵。同時,別忘了去「Small Web」逛逛,找回那份屬於人類親手創作的靈感。