Middleware 怎麼讓你的 Agent 變得超好用

Agent 框架百家爭鳴，但真正用過的人都知道，框架給你的 default 行為往往不夠用。你想要記錄每次 Agent 的決策過程、想要加上自己的 error handling、或者想要在特定步驟插入自己的邏輯 — 這時候就需要 middleware 了。

LangChain 在這塊做得還不錯，他們的 middleware 模式讓你能在不改動核心邏輯的情況下，客製化整個 Agent 執行流程。今天就來拆解一下這套機制怎麼運作的。

Agent Harness 是什麼？

先講概念。Agent Harness 其實就是 Agent 的執行環境，負責處理 Agent 的生命週期 — 從接收輸入、調用工具、處理回應、到最終輸出。你可以想像成是 Agent 的「容器」，所有的執行邏輯都在這裡面跑。

預設的 Harness 很陽春，就是簡單的 input → process → output。但實際應用時你會發現，這中間需要塞很多東西：

• Request/Response 的 logging
• Error handling 和 retry 機制
• Rate limiting
• Custom validation
• 執行時間監控

這些東西如果都硬塞到 Agent 本體，code 會變得超亂。middleware 就是來解決這個問題的。

Middleware 的核心概念

LangChain 的 middleware 採用洋蔥模型 (onion model)。每個 middleware 就像洋蔥的一層，request 從外層進來，經過每一層處理，到達最內層的 Agent，然後 response 再反向經過每一層出去。

# 簡化的概念圖
Request → Middleware 1 → Middleware 2 → Agent → Middleware 2 → Middleware 1 → Response

每個 middleware 都有機會在 request 進來時做前處理，在 response 出去時做後處理。這樣的設計讓你可以把不同的關注點分離到不同的 middleware 裡。

實作一個 Logging Middleware

來看個實際例子。假設你想記錄每次 Agent 執行的詳細資訊：

class LoggingMiddleware:
    def __init__(self, logger):
        self.logger = logger
    
    def __call__(self, next_handler):
        def middleware(request):
            # 前處理 - 記錄請求
            self.logger.info(f"Agent request: {request.input}")
            start_time = time.time()
            
            try:
                # 執行下一個 middleware 或 Agent
                response = next_handler(request)
                
                # 後處理 - 記錄成功回應
                duration = time.time() - start_time
                self.logger.info(f"Agent response: {response.output}, duration: {duration:.2f}s")
                
                return response
            except Exception as e:
                # 記錄錯誤
                self.logger.error(f"Agent error: {str(e)}")
                raise
        
        return middleware

這個 middleware 很簡單，就是在 Agent 執行前後記錄一些資訊。但你可以看出這個模式的威力 — 完全不用動到 Agent 的 code，就能加上完整的 logging 功能。

組合多個 Middleware

真正強大的地方是組合。你可以串接多個 middleware，每個負責不同的功能：

# 建立 Agent 執行環境
harness = AgentHarness(
    agent=my_agent,
    middleware=[
        RateLimitMiddleware(max_requests=10, window=60),  # 限流
        ValidationMiddleware(input_schema=schema),        # 驗證
        LoggingMiddleware(logger=logger),                 # 記錄
        RetryMiddleware(max_attempts=3),                  # 重試
    ]
)

這樣一來，每個 request 都會依序經過這四個 middleware 的處理。Rate limiting 確保不會被濫用，validation 確保輸入格式正確，logging 記錄所有活動，retry 處理暫時性錯誤。

進階應用：Context Injection

更進階的用法是透過 middleware 注入 context。比如說，你想要在 Agent 執行過程中取得當前用戶的資訊：

class UserContextMiddleware:
    def __init__(self, user_service):
        self.user_service = user_service
    
    def __call__(self, next_handler):
        def middleware(request):
            # 從 request 中取得 user_id，注入用戶資訊
            user_id = request.metadata.get('user_id')
            if user_id:
                user_info = self.user_service.get_user(user_id)
                request.context['user'] = user_info
            
            return next_handler(request)
        
        return middleware

這樣 Agent 就能透過 request.context['user'] 取得用戶資訊，而且完全不用知道這資訊是怎麼來的。

為什麼這樣設計很香？

這個 middleware 模式有幾個好處：

1. 關注點分離 — 每個功能獨立開發測試
2. 可重用 — 同一個 middleware 可以套用到不同的 Agent
3. 可組合 — 像積木一樣組合出需要的功能
4. 非侵入性 — Agent 本體不需要改動

最重要的是，這讓你的 Agent 系統變得超級有彈性。需要新功能？寫個 middleware。不需要某個功能？從串列中拿掉就好。

老實說，如果你在建 Agent 系統，middleware 模式是必須要考慮的架構。它讓你的系統從「能跑」變成「好維護」，這差距比你想的還大。

<h2>延伸閱讀</h2> <ul> <li><a href="/blog/anatomy-of-agent-harness">拆解 Agent Harness — 你以為的 AI Agent 其實 90% 是 harness</a></li> <li><a href="/blog/agent-design-behind-50000-stars-architectural-breakdown-of-bytedance-deerflow-20">五萬顆星背後的 Agent 設計：ByteDance DeerFlow 2.0 的架構拆解</a></li> <li><a href="/blog/kensho-financial-agent-with-langgraph-practical-analysis-of-multi-agent-system">Kensho 用 LangGraph 做金融 Agent — 多 Agent 系統實戰解析</a></li> </ul>