一 概述与设计目标

• 定位：在 LangChain 1.0中，中间件是围绕智能体执行流程的“可插拔钩子系统”，用于在关键节点对输入、调用、输出进行预处理、拦截、修改与验证，从而实现更强的可控性与可观测性。

• 核心能力：

  • 上下文工程：动态提示词、对话摘要、上下文编辑，避免上下文溢出与跑偏。

  • 流程控制：重试、回退、速率限制、提前终止、条件跳转。

  • 安全与合规：PII检测与脱敏、工具调用权限、人机协同（HITL）。

  • 可观测与成本：日志、监控、缓存、调用预算。

• 执行模型：

  • 顺序：before_*钩子按注册顺序执行；after_*钩子按逆序执行。

  • 嵌套：wrap_*钩子呈“洋葱式”嵌套，外层先进入、内层先返回。

  • 跳转：在 before_model/after_model可返回 {"jump_to": "end|tools|model"}实现流程中断与跳转。

• 与 create_agent 的关系：create_agent是 1.0 构建智能体的标准接口，底层基于 LangGraph 1.0 运行时，天然具备持久化、流式、HITL、时间回溯等能力，中间件即插即用。

二 执行顺序与跳转机制

• 执行顺序规则

  • before_agent→ before_model→ wrap_model_call（外→内）→ 模型调用 → wrap_model_call（内→外）→ after_model（逆序）→ wrap_tool_call（外→内）→ 工具执行 → wrap_tool_call（内→外）→ after_agent。

• 跳转控制

  • 在 before_model/after_model返回 {"jump_to": "end|tools|model"}可中断并跳转；若从 before_model/after_model跳转到 "model"，会重新触发该节点之前的 before_model钩子。

  • 使用跳转需在钩子上应用 @hook_config(can_jump_to=[...])明确声明合法目标。

• 典型用途

  • 合规拦截：检测到敏感操作时 jump_to="end"终止并返回提示。

  • 条件路由：根据用户等级/意图跳过部分工具或直接进入特定处理分支。

三 内置中间件速览

以上中间件覆盖了上下文优化、人工介入、隐私合规、调用限制、容错降级、成本控制等主流生产诉求。

四 自定义中间件实战

• 装饰器方式（单钩点、轻量）

  • 适合快速添加“日志、动态提示词、单点重试/校验”等能力。

  • 示例：动态提示词 + 条件跳转 + 安全校验

    from langchain.agents.middleware import before_model, after_model, wrap_model_call, AgentMiddleware, AgentState, ModelRequest, ModelResponse, dynamic_prompt, hook_config
    from langchain_core.messages import AIMessage
    from typing import Any, Callable, Optional
    
    # before_model：记录消息数
    @before_model
    def log_msg_count(state: AgentState, runtime: Any) -> None:
        print(f"[BeforeModel] 消息数: {len(state['messages'])}")
    
    # after_model：合规拦截
    @after_model(can_jump_to=["end"])
    def block_sensitive(state: AgentState, runtime: Any) -> Optional[dict[str, Any]]:
        last = state["messages"][-1].content
        if "DELETE_ALL" in last.upper():
            return {"messages": [AIMessage("拒绝高危操作。")], "jump_to": "end"}
        return None
    
    # wrap_model_call：失败重试
    @wrap_model_call
    def retry_on_error(req: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]) -> ModelResponse:
        for i in range(3):
            try:
                return handler(req)
            except Exception as e:
                if i == 2:
                    raise
                print(f"重试 {i+1}/3: {e}")

• 类方式（多钩点、可配置）

  • 适合“能力分级、多租户策略、复合规则”等复杂场景。

  • 示例：按用户等级动态切换模型与工具

    from dataclasses import dataclass
    from langchain.agents.middleware import AgentMiddleware
    
    @dataclass
    class UserContext:
        level: str  # "beginner" | "expert"
    
    class ExpertiseMiddleware(AgentMiddleware):
        def __init__(self, default_model, default_tools, expert_model, expert_tools):
            self.default_model = default_model
            self.default_tools = default_tools
            self.expert_model = expert_model
            self.expert_tools = expert_tools
    
        def before_model(self, state: AgentState, runtime: Any) -> None:
            ctx: UserContext = runtime.context.get("user") or UserContext(level="beginner")
            if ctx.level == "expert":
                state["model"] = self.expert_model
                state["tools"] = self.expert_tools
            else:
                state["model"] = self.default_model
                state["tools"] = self.default_tools

  • 使用方式：将上述中间件按需加入 create_agent(middleware=[...])即可生效。

五 最佳实践与选型建议

• 保持单一职责：每个中间件只做一件事；组合多个中间件实现复杂能力。

• 优先内置，再做扩展：优先使用 SummarizationMiddleware、PIIMiddleware、HumanInTheLoopMiddleware、ModelFallbackMiddleware、ToolCallLimitMiddleware等，减少自研成本。

• 顺序很重要：将“全局限制/安全护栏”类中间件靠前；“审计/日志”类靠后，避免被提前短路。

• 优雅处理错误：在重试/回退/拦截中避免抛出未捕获异常导致流程中断。

• 可观测优先：配合日志、指标与缓存中间件，确保可追踪、可回放、可优化。

• 与 LangGraph 协同：中间件提供“轻量定制”，当需要复杂编排、持久化、细粒度状态机时，可在 create_agent基础上深入 LangGraph 1.0运行时。