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 處理暫時性錯誤。

<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>