AgentMiddleware is All You Need

业内对 Agent 的抽象已经存在三年了，现在社区中有众多 Agent 框架采用了相似的核心概念——LangChain, Pydantic-AI, Agno, Google ADK, claude-agent-sdk。 但它们大部分都有相似的处境：在开发者需要精细控制 Agent 运行时，无法给开发者足够的掌控力，导致开发者遇到非简单场景时不得不舍弃框架这层抽象。

这篇文章是我对「如何搭好一个 Agent」的一份理解。

一、AI Workflow 与 ReAct Agent

生产环境里常见的有两种形态：

形态一：AI Workflows

LLM 被嵌入预设的固定流程，模型本身没有充足的自主性：

用户问题
  → [程序 + LLM] 识别意图，命中预设场景
  → [程序] 执行该场景的固定流程
           （查员工信息 → 并行查业务单据 → 数据整合 → ...）
  → [LLM] 总结 + 话术引导

形态二：ReAct Agent

LLM 自主决策，工具按需调用：

用户问题 → [LLM：决定调哪个工具] → [工具执行] → [LLM：决定下一步] → ...

两者的根本区别在于谁拥有控制流。

AI Workflows 有它存在的充分理由：

• 公司内部模型智力有限——业务数据敏感，通常无法调用外部闭源高阶模型
• ReAct 一次回答需要 5+ 分钟（涉及 2+ 次 LLM 调用），延迟无法接受
• 流程可控、可审计，符合合规要求

它不是糟粕，而是对现阶段模型能力的务实应对。但它的扩展性问题本质上是传统软件工程问题，不需要新的范式。

今天的生产环境大多是混合架构（Hybrid Agentic Workflow）： 用 Workflow 把控全局框架和关键节点，在模糊决策处嵌入 Agent 做局部推理。 这既是当下的工程务实，也是通往纯 Agent 时代的必经之路——随着模型能力持续提升，Agent 的可靠性会逼近"确定性"，固定编排的价值自然边际递减。

下文聚焦于 ReAct Agent。

二、ReAct Agent 的扩展性困境

一个 ReAct Agent 仅由几个简单的部分组成：

• Model（大语言模型）
• Prompt（模型的提示词）
• Tools（一组供模型调用的工具）

其核心算法同样简单： 用户首先用某条输入消息调用 Agent，Agent 随即进入循环： 调用工具、将 Assistant 消息和 Tool 消息追加到其状态中，直到决定不再调用任何工具并最终结束。

graph TD
    A([request]) --> B(model)
    B -.->|action| C(tools)
    C -->|observation| B
    B -.-> D([result])

用 LangChain 写出来大概是这样：

from langchain.agents import create_agent

# ① Model
model = ChatOpenAI(model="gpt-5")

# ② Prompt
system_prompt = "You are a helpful assistant."

# ③ Tools
@tool
def get_weather(location: str):
    """Call to get the weather from a specific location."""
    if any([city in location.lower() for city in ["sf", "san francisco"]]):
        return "It's sunny in San Francisco, but you better look out if you're a Gemini ."
    else:
        return f"I am not sure what the weather is in {location}"


backend_tools = [get_weather]

agent = create_agent(
    model=model,
    tools=backend_tools,
    system_prompt=system_prompt
)

与此同时，虽然搭建一个基础 agent 抽象并不难（几乎所有框架都做到了这一点），但要让这个抽象足够灵活以支撑生产环境，却并非易事。 形如 create_agent 的用法极简且优雅，但它对开发者几乎不透明——你难以干预它的内部运行机制。

以「让 Agent 调用前端工具」为例——前端可以注册一批在浏览器端执行的工具，Agent 调用它们时，逻辑运行在用户的浏览器里，可以读写 React 组件状态、触发 UI 动画、驱动 Generative UI 渲染。整个交互流程如下：

sequenceDiagram
    participant 前端
    participant 后端
    participant 模型
    前端 ->> 后端: 上传前端工具定义（名称、参数 Schema）
    后端 ->> 模型: 绑定后端工具 + 前端工具，发起对话

    loop 推理循环
        模型 ->> 后端: 决定调用后端工具 A 和前端工具 B
        后端 ->> 后端: 执行后端工具 A
        后端 -->> 前端: 中断执行，下发前端工具 B 的调用请求（含参数）
        前端 ->> 前端: 在浏览器中执行前端工具 B
        前端 ->> 后端: 返回前端工具 B 的执行结果
        后端 ->> 模型: 汇总结果，继续推理
    end

    模型 ->> 后端: 生成最终回答
    后端 -->> 前端: 返回结果

要让这张图跑起来，两件事缺一不可：调用 LLM 前，需要把前端注册的工具动态注入工具列表，让模型感知到它们的存在；工具调用阶段，需要识别哪些是前端工具，拦截并中断后端执行，把调用参数发回客户端，等浏览器执行完毕再继续。 两件事分属不同层次——模型输入处理、工具调用拦截——必须同时生效才能形成完整功能。

这不是 create_agent 能直接支持的——它的内部运行机制对开发者不透明，你无法在"调用 LLM 时"或"执行工具时"插入自定义逻辑。

这也不是调用前端工具独有的问题，其根本原因在于：context engineering——模型输入的上下文决定了其输出结果。

为了让模型表现可靠，开发者需要对模型输入做精细的控制：

1. 希望调整 agent 的"状态"，使其不仅仅包含消息
2. 希望更精细地控制究竟向模型发送什么内容
3. 希望更精细地控制执行步骤的顺序

这些控制逻辑会蔓延并散落在 Agent 的各个层次，而那个简单的 agent 循环虽然适合入门，当你不断突破 agent 能力的边界时，就不得不修改它的各个部分。

于是不少开发者纷纷将 LangChain 框架改用更底层的 LangGraph，以框架抽象为代价换取更强的上下文掌控能力。

用 LangGraph 写出来的标准 Agent 大概是这样：

from typing import Annotated
from typing_extensions import TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_core.messages import ToolMessage


# ① State
class State(TypedDict):
    messages: Annotated[list, add_messages]

# ② Model node
def call_model(state: State):
    model_with_tools = model.bind_tools(backend_tools)
    return {"messages": [model_with_tools.invoke(state["messages"])]}

# ③ Tool node
def tool_node(state: State):
    result = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"]))
    return {"messages": result}

# ④ 条件边
def should_continue(state: State):
    return "tools" if state["messages"][-1].tool_calls else END

# ⑤ Graph 装配
graph = StateGraph(State)
graph.add_node("model", call_model)
graph.add_node("tools", tool_node)
graph.add_conditional_edges("model", should_continue)
graph.add_edge("tools", "model")
app = graph.compile()

这个骨架本身没问题。问题出在往上加功能的时候。

同样以「让 Agent 调用前端工具」为例，在 LangGraph 里实现它，逻辑会散落在两处：

• Model node：调用 LLM 前，把前端注册的工具动态注入工具列表（model.bind_tools(backend_tools + client_tools)）
• Tool node：识别前端工具，通过 interrupt() 暂停后端执行，把调用参数发回客户端，等浏览器执行完毕再继续

一个功能，散落在两个节点。每加一个这样的功能，model node 和 tool node 就各自膨胀一点，最终变成两坨谁都不敢动的代码。

类似的问题反复出现：

「Agent 自主任务规划」：Agent 自主维护一份 todo list，追踪复杂任务的执行进度。 要实现它，需要在 State 里新增 todos 字段、在 model node 里注入 write_todos 工具和任务管理提示词、在 tool node 里处理 write_todos 的调用并把结果写回 State——三处都要改。

「高危操作人工确认」：Agent 在执行删除文件、执行 shell 命令等敏感操作前，需要暂停并等待人工审批。 要实现它，需要在 model node 之后、tool node 之前插入一段判断逻辑——识别本次 LLM 返回中是否有高危工具调用，若有则触发中断等待人工决策（通过、编辑参数或拒绝），再把结果交给 tool node 执行。这段逻辑同样游离在 Agent 骨架的各个节点之间。

三、答案：Middleware

LangChain 引入了 AgentMiddleware，解决的正是这个问题。

核心思想：一个功能涉及的所有层次（state、工具定义、模型输入处理、工具调用），全部封装在同一个 Middleware 类里，而不是散落在 graph 的各个角落。Agent 的骨架固定不变，功能通过 Middleware 进行模块化的插拔和组合，互不干扰。

六个钩子

AgentMiddleware 提供了六个钩子，覆盖 Agent 执行的全生命周期：

graph TD
    Req([request])
    BA(before_agent)
    BM(before_model)
    WMC(wrap_model_call)
    AM(after_model)
    WTC(wrap_tool_call)
    AA(after_agent)
    Res([result])
    Req --> BA
    BA --> BM
    BM --> WMC
    WMC --> AM
    AM -.-> WTC
    WTC --> BM
    AM -.-> AA
    AA --> Res

回到前端工具的例子

上一节里，「让 Agent 调用前端工具」在 LangGraph 里的逻辑散落在两处。用 Middleware 的思路，把它们全部收进一个类——ClientToolMiddleware：

import json
from contextvars import ContextVar
from dataclasses import dataclass
from typing import Any, Callable, Awaitable, TypedDict

from langchain.agents.middleware import AgentMiddleware, ModelRequest, ModelResponse
from langchain_core.tools import BaseTool, StructuredTool
from langgraph.prebuilt.tool_node import ToolCallRequest
from langgraph.types import Command, interrupt

_CLIENT_TOOL_CALL_INTERRUPT_KEY = "client_tool_call"
_client_tools_var: ContextVar[tuple[BaseTool, ...]] = ContextVar("client_tools", default=())


@dataclass
class Tool:
    name: str
    description: str
    inputSchema: dict


def set_client_tools(tools: list[Tool]) -> None:
    """前端调用，注册本次会话的客户端工具"""
    _client_tools_var.set(tuple(_to_langchain_tool(t) for t in tools))


def _to_langchain_tool(tool: Tool) -> BaseTool:
    """将 MCP Tool 定义转换为 LangChain StructuredTool"""

    def invoke_tool(**kwargs: Any) -> str:
        # 触发 interrupt，暂停后端执行，把调用请求下发给前端
        result = interrupt({_CLIENT_TOOL_CALL_INTERRUPT_KEY: {"name": tool.name, "args": kwargs}})
        return json.dumps(result, ensure_ascii=False) if not isinstance(result, str) else result

    async def ainvoke_tool(**kwargs: Any) -> str:
        result = interrupt({_CLIENT_TOOL_CALL_INTERRUPT_KEY: {"name": tool.name, "args": kwargs}})
        return json.dumps(result, ensure_ascii=False) if not isinstance(result, str) else result

    return StructuredTool.from_function(
        name=tool.name,
        description=tool.description or "",
        args_schema=tool.inputSchema,
        func=invoke_tool,
        coroutine=ainvoke_tool,
    )


class ClientToolMiddleware(AgentMiddleware):

    def wrap_model_call(self, request: ModelRequest, handler) -> ModelResponse:
        # ① 把前端工具合并进工具列表，让模型感知到它们的存在
        client_tools = _client_tools_var.get()
        if client_tools:
            request = request.override(tools=[*request.tools, *client_tools])
        return handler(request)

    async def awrap_model_call(self, request: ModelRequest, handler) -> ModelResponse:
        ...  # 此处省略 async 样板代码

    def wrap_tool_call(self, request: ToolCallRequest, handler) -> ToolMessage | Command:
        # ② 识别前端工具，调用时内部触发 interrupt 暂停，等待前端响应
        tool_name = request.tool_call.get("name", "")
        client_tool = next((t for t in _client_tools_var.get() if t.name == tool_name), None)
        if client_tool:
            return client_tool.invoke(request.tool_call, request.runtime.config)
        return handler(request)

    async def awrap_tool_call(self, request: ToolCallRequest, handler) -> ToolMessage | Command:
        ...  # 此处省略 async 样板代码

原本散落在 model node、tool node 两处的逻辑，现在全收进了一个类：

• wrap_model_call：每次调用 LLM 前，把前端工具动态合并进工具列表
• wrap_tool_call：每次工具调用时，识别前端工具并触发 interrupt() 暂停，等待前端响应

使用时，只需一行：

agent = create_agent(model, tools=backend_tools, middleware=[ClientToolMiddleware()])

开箱即用的 Middleware

LangChain 官方提供了不少开箱即用的 Middleware：

DeepAgents 是 LangChain 团队做的「Agent 开箱即用套件」，它的 create_deep_agent 本质上就是在 create_agent 之上预装了一套 Middleware 栈：

Middleware 的执行顺序

当你挂载多个 Middleware 时，它们的执行顺序遵循一套直觉上很自然的规则——和 Express/Koa 的洋葱模型一致：

agent = create_agent(
    model="gpt-4.1",
    tools=[...],
    middleware=[middleware1, middleware2, middleware3],
)

• before_* 钩子：按注册顺序执行（1 → 2 → 3）
• after_* 钩子：按注册逆序执行（3 → 2 → 1）
• wrap_* 钩子：嵌套执行，像函数调用栈一样层层包裹

middleware1.wrap_model_call()
  → middleware2.wrap_model_call()
    → middleware3.wrap_model_call()
      → 实际调用模型
    ← middleware3
  ← middleware2
← middleware1

这意味着排在前面的 Middleware 拥有最外层的控制权——它最先看到请求，最后看到响应。 如果你需要某个 Middleware（比如 PIIMiddleware）作为最终的安全兜底，把它放在列表第一位就好。

实际使用中，大多数 Middleware 之间是正交的，顺序无所谓。 这也是 LangChain 官方推荐的最佳实践「Keep middleware focused - each should do one thing well」。

ReAct Agent 的骨架很简单，复杂的是骨架上长出来的那些能力。

Middleware 的价值在于：把「复杂」模块化，让每一块功能都可以独立开发、独立测试、自由组合。

Middleware 的思想不是 LangChain 专有的，你可以在任何框架的基础上参考这个设计自己实现一套。