目標先講清楚:
從LangChain 的技術文件和範例,了解Agent Middleware怎麼解決「無法精細控制 context engineering」的問題
解決方法:
不再只靠提示詞(prompt)「拜託模型要遵守」,而是在模型呼叫的前、中、後放上可組合、可測試、可審計的攔截器(middleware),確定性地裁訊息、裁工具、換模型、驗證輸出。
LangChain 在 v1 引入 Agent Middleware:
before_model
:模型呼叫前可以補資料/改 state/甚至跳過模型;modify_model_request
:僅對本次呼叫動態調整「可見的工具清單(白名單/黑名單)」「訊息/提示詞」「模型與參數」「tool choice」;after_model
:模型輸出後做JSON/Schema 驗證、合規審核、HITL 人審、fallback/重試。這些都是執行期的硬性控制,而不是把規則寫在一段脆弱的大 prompt 裡。
使用者問:「幫我改期 11/20 → 11/23」。
before_model
:先從 CRM 查到「企業客戶 12% 優惠」寫入 state(模型第一次就知道商規)。modify_model_request
:這回合任務很簡單→把模型降級/降溫;同時只暴露 xxxx_
開頭的兩個查價工具(看不到就不會亂用)。after_model
:產出訂單草稿→驗證 JSON schema,不合格自動重試一次,合格才放行。等於LLM 的「每一回合」拆成 before_model → modify_model_request → (LLM) → after_model 這三個攔截點
from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware
from langchain.tools import tool
from langchain_openai import ChatOpenAI
# --- 兩個示意工具(皆以 xxxx_ 開頭,方便白名單) ---
@tool
def xxxx_search_fares(route: str, date: str) -> str:
"""查詢票價(唯讀)"""
return f"FARES for {route} on {date}: ..."
@tool
def xxxx_search_inventory(hotel: str, date: str) -> str:
"""查詢庫存(唯讀)"""
return f"INVENTORY for {hotel} on {date}: ..."
# --- 1) before_model:補事實(查 CRM) ---
class CRMEnricher(AgentMiddleware):
def before_model(self, state, *, node_name=None):
# 假裝查到企業金級 12% 折扣
crm = {"corpTier": "Gold", "discount": 0.12}
# 把事實寫進 state,後續提示/推理都能看到
state["biz_rules"] = crm
return state # 交給下一層
# --- 2) modify_model_request:切模型 + 白名單工具 + tool choice ---
class RequestShaper(AgentMiddleware):
def modify_model_request(self, request, *, node_name=None):
"""
request 內通常含有:
- messages(系統/人/代理)
- model / model_config(溫度、max_tokens…)
- tools(本回合可見的工具清單)
- tool_choice(讓/強制模型用工具)
"""
# 判斷此回合是「簡單改期」=> 降級模型 + 降溫
request.model = ChatOpenAI(model="gpt-4o-mini") # 僅影響這次呼叫
request.model_config = {"temperature": 0.0, "max_tokens": 512}
# 只保留以 xxxx_ 開頭的工具(白名單)
request.tools = [t for t in request.tools if t.name.startswith("xxxx_")]
# 至少要用一個工具(供支援的 provider 使用)
request.tool_choice = "any" # 或指定某工具 id/name
# 視需要也可在這裡修剪 messages(控長/補系統指示)
# request.messages = trim_history(request.messages, keep=30)
return request
# --- 3) after_model:驗證 JSON;必要時 HITL ---
class OutputGuard(AgentMiddleware):
def after_model(self, state, output, *, node_name=None):
"""
檢查 LLM 的「計畫/工具呼叫/結構」是否合規。
- 不合格:回傳一個「要求重試」的信號(或改寫 output 再丟回)
- 高風險:發出 interrupt(HITL),暫停等待人工
"""
# 假裝檢查 output["plan"] 是否為合格 JSON 結構
if not is_valid_json_like(output):
# 要求 agent 以「同一回合」重試一次(具體 API 依版本)
output["retry"] = True
return output
# 假裝當前動作是 "db_write" => 觸發人審(HITL)
if output.get("next_action") == "db_write":
# 具體的 interrupt 觸發方式依你用的 checkpointer/平台
output["interrupt"] = {
"reason": "High risk action: db_write",
"require_human_approval": True,
}
return output
return output
def is_valid_json_like(output) -> bool:
# 這裡放 JSON/Schema 驗證(示意)
return True
# --- 裝配 Agent:三個 middleware 疊起來 + 兩個工具 ---
base_model = ChatOpenAI(model="gpt-4o") # 預設主模型(可被 RequestShaper 單回合切換)
agent = create_agent(
model=base_model,
tools=[xxxx_search_fares, xxxx_search_inventory],
middleware=[CRMEnricher(), RequestShaper(), OutputGuard()],
)
# --- 呼叫(示意) ---
user_msg = "請把 11/20 的行程改到 11/23,並確認票價與飯店是否可改。"
result = agent.invoke({"messages": [{"role": "user", "content": user_msg}]})
print(result)
心得:
langchain的agent middleware可以把原先混在agent_node中的功能抽出來,在多個node(langGraph)都需要調用同一個邏輯或規則時,把它包含middleware比較方便;但是考慮到改成middleware造成現有的程式維護複雜度增加,應該會再觀望看看。
使用Drew Breunig的How Long Contexts Fail,為這個系列文做總結