LangChain Middleware 深度解析：函數呼叫時序與執行邏輯全攻略

0. 前言：為什麼要深入理解 Middleware？

LangChain v1 引入了一套完整的 AgentMiddleware 系統，讓開發者能夠在不修改核心 Agent 邏輯的前提下，插入各種橫切關注點（cross-cutting concerns）：

  • 重試與回退機制
  • 請求/回應日誌
  • 人在迴圈審核
  • Token 計數與對話摘要
  • PII 檢測與過濾

但要正確使用這套系統，你必須理解：

  1. 每個 Hook 何時被呼叫？
  2. Hook 之間的執行順序是什麼？
  3. 哪些 Hook 可以修改狀態？哪些可以控制流程？
  4. 多個 Middleware 如何組合？

本文將從源碼層級，完整剖析這些問題。

1. Middleware 架構總覽

1.1 模組結構

LangChain v1 的 Middleware 系統位於 langchain/agents/middleware/：

langchain/agents/middleware/
├── __init__.py          # 主要匯出點
├── types.py             # 核心型別定義
├── decorators.py        # Hook 裝飾器
├── summarization.py     # 摘要 Middleware
├── human_in_the_loop.py # 人工審核 Middleware
├── retry.py             # 重試 Middleware
└── ...                  # 其他內建 Middleware

1.2 核心匯出

from langchain.agents.middleware import (
    # 基礎類別
    AgentMiddleware,              # Middleware 基類
    AgentState,                   # Agent 狀態容器
    ModelRequest,                 # 模型請求
    ModelResponse,                # 模型回應

    # 生命週期鉤子裝飾器
    before_agent,                 # Agent 啟動前
    before_model,                 # 模型呼叫前
    after_model,                  # 模型呼叫後
    after_agent,                  # Agent 結束後

    # 攔截器裝飾器
    wrap_model_call,              # 包裝模型呼叫
    wrap_tool_call,               # 包裝工具呼叫

    # 配置
    hook_config,                  # Hook 行為配置
)

2. Hook 類型分類

LangChain Middleware 提供兩大類 Hook：

2.1 Node-Style Hooks（節點式鉤子）

這類 Hook 可以觀察狀態、修改狀態、以及控制執行流程：

Hook執行時機可修改狀態可跳轉
before_agentAgent 啟動時（僅一次）YesNo
before_model每次模型呼叫前YesYes
after_model每次模型呼叫後YesYes
after_agentAgent 結束時（僅一次）YesNo

2.2 Wrap-Style Hooks（包裝式鉤子）

這類 Hook 可以完全控制被包裝函數的執行：

Hook執行時機特性
wrap_model_call包裹模型呼叫可呼叫 0/1/N 次、可修改請求、可轉換回應
wrap_tool_call包裹工具呼叫可呼叫 0/1/N 次、可修改請求、可轉換回應

2.3 Request Modification Hook（請求修改鉤子）

Hook執行時機特性
modify_model_request模型呼叫前只能修改單次請求，不能修改永久狀態

3. 函數簽名詳解

3.1 before_agent()

def before_agent(
    self,
    state: AgentState,
    runtime: Runtime
) -> dict[str, Any] | None:
    """
    Agent 啟動時呼叫一次。

    Args:
        state: 當前 Agent 狀態
        runtime: 執行時期上下文（包含 config、context 等）

    Returns:
        None: 不做任何修改
        dict: 要合併到 AgentState 的更新
    """

使用場景：

  • 初始化 Middleware 專用的狀態欄位
  • 記錄 Agent 啟動時間
  • 預處理輸入訊息

3.2 before_model()

def before_model(
    self,
    state: AgentState,
    runtime: Runtime
) -> dict[str, Any] | None:
    """
    每次模型呼叫前執行。

    Args:
        state: 當前 Agent 狀態
        runtime: 執行時期上下文

    Returns:
        None: 不做任何修改
        dict: 可包含：
            - 狀態更新（messages、自訂欄位等）
            - jump_to: "model" | "tools" | "__end__"

    Constraints:
        - 在 before_model 內不能 jump_to "model"（會造成無限迴圈）
    """

使用場景：

  • Token 計數與對話摘要
  • 動態調整 system prompt
  • 條件性終止 Agent

3.3 after_model()

def after_model(
    self,
    state: AgentState,
    runtime: Runtime
) -> dict[str, Any] | None:
    """
    每次模型呼叫後執行。

    Args:
        state: 包含模型回應的更新狀態
        runtime: 執行時期上下文

    Returns:
        None: 不做任何修改
        dict: 可包含狀態更新與 jump_to

    Note:
        執行順序是**反向**的（後定義的先執行）
    """

使用場景：

  • 人在迴圈審核（在工具執行前攔截）
  • 回應後處理
  • 錯誤檢測與處理

3.4 after_agent()

def after_agent(
    self,
    state: AgentState,
    runtime: Runtime
) -> dict[str, Any] | None:
    """
    Agent 結束時呼叫一次。

    Args:
        state: 最終 Agent 狀態
        runtime: 執行時期上下文

    Returns:
        None 或最終狀態更新
    """

使用場景：

  • 記錄執行統計
  • 清理資源
  • 最終輸出格式化

3.5 wrap_model_call()

def wrap_model_call(
    self,
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
    """
    完全包裝模型呼叫。

    Args:
        request: ModelRequest 物件，包含：
            - model: BaseChatModel
            - messages: list[AnyMessage]
            - system_message: SystemMessage | None
            - tools: list[BaseTool | dict]
            - response_format: ResponseFormat | None
            - state: AgentState
            - runtime: Runtime
            - model_settings: dict
        handler: 呼叫下一層的函數

    Returns:
        ModelResponse，包含：
            - result: list[BaseMessage]
            - structured_response: Any

    Capabilities:
        - 呼叫 handler 0 次（短路返回）
        - 呼叫 handler 1 次（正常流程）
        - 呼叫 handler N 次（重試邏輯）
        - 修改 request 後再呼叫
        - 轉換 response 後再返回
    """

使用場景：

  • 重試機制
  • 請求/回應日誌
  • 快取層
  • 回退到備援模型

3.6 wrap_tool_call()

def wrap_tool_call(
    self,
    request: ToolCallRequest,
    handler: Callable[[ToolCallRequest], ToolCallResponse]
) -> ToolCallResponse:
    """
    完全包裝工具呼叫。

    與 wrap_model_call 類似，但針對工具執行。
    """

3.7 modify_model_request()

def modify_model_request(
    self,
    request: ModelRequest,
    state: AgentState
) -> ModelRequest:
    """
    修改模型請求（僅限該次呼叫）。

    Args:
        request: 原始 ModelRequest
        state: 當前狀態（唯讀參考）

    Returns:
        修改後的 ModelRequest

    可修改欄位：
        - tools
        - prompt (system message)
        - messages
        - model
        - model_settings
        - response_format
        - tool_choice

    Constraints:
        - 不能修改永久狀態
        - 不能控制執行流程
    """

4. 執行時序：完整流程圖

4.1 Agent 完整生命週期

╔══════════════════════════════════════════════════════════════════╗
║                     AGENT INVOCATION START                       ║
╚══════════════════════════════════════════════════════════════════╝
                              ↓
┌────────────────────────────────────────────────────────────────────┐
│  [1] before_agent hooks                                            │
│      執行順序: Middleware[0] → [1] → [2] → ...                     │
│      每個 Middleware 僅執行一次                                     │
└────────────────────────────────────────────────────────────────────┘
                              ↓
            ╔════════════════════════════════════════╗
            ║     AGENT LOOP (重複直到結束)          ║
            ╠════════════════════════════════════════╣
            ║                                        ║
            ║  ┌──────────────────────────────────┐  ║
            ║  │ [2a] before_model hooks          │  ║
            ║  │     順序: [0] → [1] → [2]        │  ║
            ║  │     可修改狀態                    │  ║
            ║  │     可 jump_to: tools/__end__    │  ║
            ║  └──────────────────────────────────┘  ║
            ║                  ↓                     ║
            ║  ┌──────────────────────────────────┐  ║
            ║  │ [2b] modify_model_request hooks  │  ║
            ║  │     順序: [0] → [1] → [2]        │  ║
            ║  │     request → request' → req''   │  ║
            ║  │     只改請求，不改狀態            │  ║
            ║  └──────────────────────────────────┘  ║
            ║                  ↓                     ║
            ║  ┌──────────────────────────────────┐  ║
            ║  │ [2c] wrap_model_call             │  ║
            ║  │     獨立攔截器                    │  ║
            ║  │     可呼叫 handler 0/1/N 次      │  ║
            ║  └──────────────────────────────────┘  ║
            ║                  ↓                     ║
            ║  ┌──────────────────────────────────┐  ║
            ║  │ [2d] MODEL INVOCATION            │  ║
            ║  │     LLM 處理請求                  │  ║
            ║  └──────────────────────────────────┘  ║
            ║                  ↓                     ║
            ║  ┌──────────────────────────────────┐  ║
            ║  │ [2e] after_model hooks           │  ║
            ║  │     ⚠️ 反向順序: [2] → [1] → [0] │  ║
            ║  │     可修改狀態                    │  ║
            ║  │     可 jump_to: model/tools/end  │  ║
            ║  └──────────────────────────────────┘  ║
            ║                  ↓                     ║
            ║  ┌──────────────────────────────────┐  ║
            ║  │ [2f] TOOL EXECUTION (if needed)  │  ║
            ║  │     wrap_tool_call 攔截每個工具   │  ║
            ║  └──────────────────────────────────┘  ║
            ║                  ↓                     ║
            ║       有更多工具呼叫？                  ║
            ║         ↓是        ↓否                 ║
            ║    回到 [2a]    離開迴圈               ║
            ║                                        ║
            ╚════════════════════════════════════════╝
                              ↓
┌────────────────────────────────────────────────────────────────────┐
│  [3] after_agent hooks                                             │
│      執行順序: Middleware[0] → [1] → [2] → ...                     │
│      每個 Middleware 僅執行一次                                     │
└────────────────────────────────────────────────────────────────────┘
                              ↓
╔══════════════════════════════════════════════════════════════════╗
║                     AGENT INVOCATION END                         ║
╚══════════════════════════════════════════════════════════════════╝

4.2 詳細時序圖（三個 Middleware）

假設我們有三個 Middleware：A、B、C

sequenceDiagram
    participant User
    participant Agent
    participant A as Middleware A
    participant B as Middleware B
    participant C as Middleware C
    participant LLM

    User->>Agent: invoke()

    Note over Agent: before_agent (順向)
    Agent->>A: before_agent()
    A-->>Agent: state update
    Agent->>B: before_agent()
    B-->>Agent: state update
    Agent->>C: before_agent()
    C-->>Agent: state update

    loop Agent Loop
        Note over Agent: before_model (順向)
        Agent->>A: before_model()
        A-->>Agent: state/jump
        Agent->>B: before_model()
        B-->>Agent: state/jump
        Agent->>C: before_model()
        C-->>Agent: state/jump

        Note over Agent: modify_model_request (順向)
        Agent->>A: modify_model_request(req)
        A-->>Agent: req'
        Agent->>B: modify_model_request(req')
        B-->>Agent: req''
        Agent->>C: modify_model_request(req'')
        C-->>Agent: req'''

        Agent->>LLM: invoke(req''')
        LLM-->>Agent: response

        Note over Agent: after_model (逆向!)
        Agent->>C: after_model()
        C-->>Agent: state/jump
        Agent->>B: after_model()
        B-->>Agent: state/jump
        Agent->>A: after_model()
        A-->>Agent: state/jump
    end

    Note over Agent: after_agent (順向)
    Agent->>A: after_agent()
    Agent->>B: after_agent()
    Agent->>C: after_agent()

    Agent-->>User: final response

5. 關鍵設計決策解析

5.1 為什麼 after_model 是反向執行？

這遵循了經典的 Middleware Onion Model（洋蔥模型）：

Request Flow (順向):
    A.before → B.before → C.before → [Core]

Response Flow (逆向):
    [Core] → C.after → B.after → A.after

這樣設計的好處：

  1. 對稱性：第一個處理請求的，最後處理回應
  2. 封裝完整性：每個 Middleware 可以在進入時設置上下文，在離開時清理
  3. 巢狀攔截：外層 Middleware 可以完整包裹內層的行為

5.2 modify_model_request vs before_model

特性modify_model_requestbefore_model
修改請求YesNo（需透過狀態間接影響）
修改永久狀態NoYes
控制流程NoYes（可 jump_to）
影響範圍單次呼叫整個 Agent 會話

使用指南：

  • 只需要調整這次 LLM 呼叫的參數 → modify_model_request
  • 需要修改會話狀態或控制流程 → before_model

5.3 wrap_model_call 的強大之處

wrap_model_call 是最靈活的 Hook，因為它完全控制執行：

# 模式 1: 短路返回（不呼叫 LLM）
@wrap_model_call
def cached_response(request, handler):
    if cache_hit:
        return ModelResponse(result=[AIMessage(content="cached")])
    return handler(request)

# 模式 2: 重試邏輯
@wrap_model_call
def retry_on_error(request, handler):
    for attempt in range(3):
        try:
            return handler(request)
        except RateLimitError:
            time.sleep(2 ** attempt)
    raise

# 模式 3: 回退模型
@wrap_model_call
def fallback_model(request, handler):
    try:
        return handler(request)
    except Exception:
        fallback_request = request.override(model=backup_model)
        return handler(fallback_request)

# 模式 4: 請求/回應轉換
@wrap_model_call
def transform_io(request, handler):
    modified_request = request.override(
        messages=[sanitize(m) for m in request.messages]
    )
    response = handler(modified_request)
    return ModelResponse(
        result=[post_process(m) for m in response.result]
    )

6. ModelRequest 與 ModelResponse 詳解

6.1 ModelRequest 結構

@dataclass
class ModelRequest:
    model: BaseChatModel                        # LLM 實例
    messages: list[AnyMessage]                  # 對話歷史
    system_message: SystemMessage | None        # 系統訊息
    system_prompt: str | None                   # 系統提示字串
    tool_choice: Any | None                     # 工具選擇策略
    tools: list[BaseTool | dict] | None         # 可用工具列表
    response_format: ResponseFormat | None      # 結構化輸出格式
    state: AgentState | None                    # 當前狀態
    runtime: Runtime[ContextT] | None           # 執行時期上下文
    model_settings: dict[str, Any] | None       # 模型特定設定

    def override(self, **kwargs) -> ModelRequest:
        """
        返回新的 ModelRequest，指定欄位被覆蓋。
        遵循不可變模式 - 原始物件不變。
        """

6.2 ModelResponse 結構

@dataclass
class ModelResponse:
    result: list[BaseMessage]              # 模型輸出的訊息
    structured_response: Any = None        # 結構化輸出（如果有指定 response_format）

6.3 不可變模式

ModelRequest.override() 確保 Middleware 鏈中每個環節都能安全修改請求：

def modify_model_request(self, request, state):
    # 原始 request 不受影響
    return request.override(
        model=self.custom_model,
        system_message=SystemMessage(content="Enhanced prompt")
    )

7. Middleware 組合與鏈式執行

7.1 定義多個 Middleware

agent = create_agent(
    model="gpt-4o",
    tools=[...],
    middleware=[
        LoggingMiddleware(),     # Index 0
        RetryMiddleware(),       # Index 1
        SummarizationMiddleware() # Index 2
    ]
)

7.2 執行順序總結

輸入階段（順向）:
    before_agent:         [0] → [1] → [2]
    before_model:         [0] → [1] → [2]
    modify_model_request: [0] → [1] → [2]

                    ↓ [MODEL CALL] ↓

輸出階段（逆向）:
    after_model:          [2] → [1] → [0]  ← 注意反向！
    after_agent:          [0] → [1] → [2]

7.3 狀態累積

每個 Middleware 的狀態更新會影響後續 Middleware：

# Middleware A
def before_model(self, state, runtime):
    return {"call_count": state.get("call_count", 0) + 1}

# Middleware B - 可以看到 A 的更新
def before_model(self, state, runtime):
    # state["call_count"] 已經被 A 更新
    if state["call_count"] > 10:
        return {"jump_to": "__end__"}

8. 內建 Middleware 實作分析

8.1 SummarizationMiddleware

目的：當對話歷史過長時自動摘要

class SummarizationMiddleware(AgentMiddleware):
    def __init__(
        self,
        model: str,
        trigger: dict  # {"tokens": 4000, "messages": 10}
    ):
        self.model = model
        self.trigger = trigger

    def before_model(self, state: AgentState, runtime: Runtime):
        messages = state["messages"]

        # 檢查是否需要摘要
        token_count = count_tokens(messages)
        if token_count > self.trigger["tokens"]:
            # 保留最近的訊息
            recent = messages[-self.trigger.get("keep", 5):]
            older = messages[:-self.trigger.get("keep", 5)]

            # 摘要舊訊息
            summary = self._summarize(older)

            return {
                "messages": [
                    SystemMessage(content=f"Previous conversation summary: {summary}"),
                    *recent
                ]
            }
        return None

Hook 使用：before_model（觀察性 + 狀態修改）

8.2 HumanInTheLoopMiddleware

目的：在特定工具執行前暫停等待人工審核

class HumanInTheLoopMiddleware(AgentMiddleware):
    def __init__(self, interrupt_on: dict):
        # {"send_email": {"allowed_decisions": ["approve", "reject"]}}
        self.interrupt_on = interrupt_on

    def after_model(self, state: AgentState, runtime: Runtime):
        last_msg = state["messages"][-1]

        if hasattr(last_msg, "tool_calls"):
            for tool_call in last_msg.tool_calls:
                tool_name = tool_call["name"]
                if tool_name in self.interrupt_on:
                    # 中斷執行，等待人工介入
                    return {
                        "pending_approval": tool_call,
                        "jump_to": "__end__"  # 暫停 Agent
                    }
        return None

Hook 使用：after_model（觀察性 + 流程控制）

8.3 RetryMiddleware

目的：自動重試失敗的模型呼叫

class RetryMiddleware(AgentMiddleware):
    def __init__(self, max_retries: int = 3, backoff: float = 1.0):
        self.max_retries = max_retries
        self.backoff = backoff

    def wrap_model_call(self, request, handler):
        last_error = None

        for attempt in range(self.max_retries):
            try:
                return handler(request)
            except (RateLimitError, APIError) as e:
                last_error = e
                wait_time = self.backoff * (2 ** attempt)
                time.sleep(wait_time)

        raise last_error

Hook 使用：wrap_model_call（攔截器）

9. 使用裝飾器定義 Middleware

除了類別方式，也可以用裝飾器快速定義：

9.1 生命週期鉤子

from langchain.agents.middleware import before_model, after_model

@before_model
def log_request(state: AgentState, runtime: Runtime):
    print(f"Model call with {len(state['messages'])} messages")
    return None

@after_model
def log_response(state: AgentState, runtime: Runtime):
    last_msg = state["messages"][-1]
    print(f"Model response: {last_msg.content[:100]}...")
    return None

9.2 帶配置的鉤子

@before_model(can_jump_to=["__end__"])
def limit_turns(state: AgentState, runtime: Runtime):
    if len(state["messages"]) > 50:
        return {"jump_to": "__end__"}
    return None

9.3 包裝鉤子

@wrap_model_call
def measure_latency(request: ModelRequest, handler):
    start = time.time()
    response = handler(request)
    latency = time.time() - start
    print(f"Model latency: {latency:.2f}s")
    return response

10. 實戰範例：完整 Middleware 組合

from langchain.agents import create_agent
from langchain.agents.middleware import (
    AgentMiddleware,
    before_agent,
    before_model,
    after_model,
    wrap_model_call,
)

# 自訂 Middleware：追蹤呼叫統計
class CallTracker(AgentMiddleware):
    def before_agent(self, state, runtime):
        return {"model_calls": 0, "tool_calls": 0}

    def before_model(self, state, runtime):
        return {"model_calls": state["model_calls"] + 1}

    def after_agent(self, state, runtime):
        print(f"Stats: {state['model_calls']} model calls")
        return None

# 裝飾器式 Middleware：日誌
@before_model
def log_input(state, runtime):
    print(f"[LOG] Input messages: {len(state['messages'])}")
    return None

@after_model
def log_output(state, runtime):
    print(f"[LOG] Output received")
    return None

# 裝飾器式 Middleware：重試
@wrap_model_call
def retry_handler(request, handler):
    for i in range(3):
        try:
            return handler(request)
        except Exception as e:
            if i == 2:
                raise
            print(f"Retry {i+1}/3...")

# 組合所有 Middleware
agent = create_agent(
    model="gpt-4o",
    tools=[search_tool, calculator_tool],
    system_prompt="You are a helpful assistant.",
    middleware=[
        CallTracker(),      # 追蹤統計
        log_input,          # 輸入日誌
        log_output,         # 輸出日誌
        retry_handler,      # 重試機制
    ],
    checkpointer=InMemorySaver()
)

# 執行
response = agent.invoke({
    "messages": [HumanMessage(content="What is 25 * 48?")]
})

11. 已知限制與注意事項

11.1 v1.0.0a14 已知問題

  1. ModelResponse 匯入問題

    # 這個可能失敗
from langchain.agents.middleware import ModelResponse

# 改用這個
from langchain.agents.middleware.types import ModelResponse

  2. middleware 與 state_schema 互斥

    # 這會出錯
agent = create_agent(
    middleware=[...],
    state_schema=CustomState  # 不能同時使用
)

  3. 缺少工具層級鉤子

    • 沒有 before_tool / after_tool
    • 只能用 wrap_tool_call 攔截

11.2 效能考量

  • 每個 Middleware 都會增加執行開銷
  • wrap_model_call 中避免不必要的 handler 呼叫
  • 大量 Middleware 會影響響應延遲

12. Hook 特性總結表

Hook類型範圍時機可改狀態可跳轉執行順序
before_agentNodeAgent啟動一次YesNo順向
before_modelNode每次模型呼叫LLM 前YesYes順向
modify_model_requestRequest每次模型呼叫LLM 前NoNo順向
wrap_model_callWrap每次模型呼叫包裹 LLM透過回傳No獨立
after_modelNode每次模型呼叫LLM 後YesYes逆向
wrap_tool_callWrap每次工具呼叫包裹工具透過回傳No獨立
after_agentNodeAgent結束一次YesNo順向

13. 結語

LangChain v1 的 Middleware 系統提供了一個強大且靈活的擴展機制：

  1. Node-style hooks 讓你觀察和修改 Agent 狀態
  2. Wrap-style hooks 讓你完全控制函數執行
  3. 洋蔥模型 確保請求和回應的對稱處理
  4. 裝飾器語法 讓簡單場景變得更簡潔

掌握這套系統，你就能：

  • 實現可靠的重試與回退機制
  • 加入精細的監控與日誌
  • 建立人在迴圈的審核流程
  • 優化 Token 使用與對話管理

Middleware 不只是「插件」，而是生產級 Agent 的必要組成部分。