LangChain v1.3.4 笔记 - 04 Agent 中间件

在 Agent 执行过程中的各个环节暴露出来的钩子函数，使用中间件可以控制和定制 Agent 的执行过程、输出以及增加健壮性（添加一些业务相关的逻辑，比如日志）。

此外中间件还能把一些通用的与业务关系不大的业务抽离出来，让 Agent 具有更好的复用性；带有钩子函数的 Agent 执行逻辑

内置中间件

在 Langchain 中也提供了一系列的内置中间件；

from langchain.agents.middleware import SummarizationMiddleware, ...
from deepagents.middleware.subagents import SubAgentMiddleware, ...

Summarization 对历史消息进行摘要总结（避免上下文爆炸）
HumanInTheLoop 暂停执行、等待人工审核
ModelCallLimit | ToolCallLimit 限制模型或工具的执行次数
ModelFalback 当主模型发生故障时，自动回退到备用模型。
PIIDetetion 敏感信息检测转码
TodoList 让 Agent 做事前先列计划
LLMToolSelector 当工具太多时，用子模型筛选最相关的几个工具
ToolRetry | ModelRetry 通过指数退避算法，设置工具或模型失败时的重试策略
LLMToolEmulator 当工具未完善时，模拟工具执行 mock 方案
ContextEditing 上下文编辑，可以控制最终给模型哪些消息
FileSearch 基于 Glob 和 Grep 检索工具给 Agent 赋予本地文件检索和分析的
FileSystem 为代理提供文件系统，用于存储上下文和长期记忆；来源于 DeepAgents
Subagent 创建子 Agent 来源于 DeepAgents

SummarizationMiddleware

对历史消息进行摘要总结、有压缩上下文的效果；

model 摘要模型，字符串或者 init_chat_model 返回值
trigger 触发条件，列表嵌套元组
messages 摘要的消息树
tokens 摘要 token 数量
fraction 摘要的百分比模型 token 上限 * 系数百分比，浮点数
keep 与 trigger 参数一致，但是仅接受一个值，需要保留的内容
summary_prompt 总结消息的 prompt
trim_tokens_to_summarize 总结调用的最大值，超过则会被修剪

model = init_chat_model( "openai:kimi-k2.6", base_url=os.environ.get("OPENAI_BASE_URL"),
```
profile={"max_input_tokens": 262144}
```
)

messages = [ HumanMessage("你好派大星，我是海绵宝宝"), AIMessage("你好海绵宝宝"), HumanMessage("我们一起去抓水母吧~"), AIMessage("等等我，海绵宝宝！我的水母网... "), HumanMessage("今天工作日，我要去蟹老板那里打工了~") ] agent = create_agent( model, middleware=[ SummarizationMiddleware( model, trigger=[ ("messages", 2),
```
        ("fraction", 0.1),
        ("tokens", 100)
    ],
    
    keep=("messages", 2),
    
    summary_prompt="用中文繁体总结消息内容 \n{messages}"
  )
]
```
) result = agent.invoke({"messages": messages})

for message in result["messages"]: print(message.pretty_print())

HumanInTheLoopMiddleware

中断 Agent 调用工具的过程，审核后再执行；

interrupt_on 审核动作，字段类型
- <tool_name>: bool True 表示 [approve, reject, edit] False 表示跳过
- <tool_name>: { allowed_decisions: [approve, reject, edit], description: xxx } 可以直接通过数组指定模型和描述

description_prefix 中断的描述信息，没有内部 description 的优先级高

from langgraph.checkpoint.memory import InMemorySaver from langgraph.types import Command

@tool def get_weather(city: str, is_forcast=False) -> str: """获取指定城市天气预报""" return f"{city} 是晴天 {'明天有雨' if is_forcast else '明天晴天'}"

agent = create_agent( model, tools=[get_weather], middleware=[ HumanInTheLoopMiddleware( interrupt_on={

      "get_weather": {
        "allowed_decisions": ["edit"],
        "description": "优先级更高的中断描述信息"
      }
    },
    description_prefix="工具调用中断~"
  )
],

checkpointer=InMemorySaver()

)

config = {"configurable": {"thread_id": "1"}} result = agent.invoke({"messages": [HumanMessage("上海的天气怎么样？")]}, config=config)

if result["interrupt"]:

resume_result = agent.invoke(
    Command(resume={
        "decisions": [
          
          
          {
              "type": "edit", 
              "edited_action": {
                  "name": "get_weather",
                  "args": {"city": "上海", "is_forcast": False}
              }
          }
          
          {"type": "approve"}
        ]
    }),
    config=config
)

rprint(resume_result)

PIIMiddleware

敏感信息加密

pii_type 需要被加密的数据类型，支持自定义；"email" | "credit_card" | "url" | "ip" | "mac_address"
strategy 加密方案
- redact 默认值，替换为 [REDACTED_EMAIL] 常量
- mask 部分屏蔽，显示最后几个字符
- hash 转换为 hash
- block 直接抛异常
detector 自定义检测函数正则字符串或者函数
apply_to_input 模型调用前检查默认 True
apply_to_output 模型调用后检查默认 False
apply_to_tool_results 工具执行后检查默认 False

agent = create_agent( model, middleware=[ PIIMiddleware("email", strategy="redact"), PIIMiddleware("url", strategy="hash"), PIIMiddleware("credit_card", strategy="mask"), PIIMiddleware("mac_address", strategy="mask"), PIIMiddleware("ip", strategy="mask") ], )

result = agent.invoke({ "messages": HumanMessage("""这是什么邮箱 1234567890@qq.com 这是什么 mac 地址 02:4A:7C:9E:3F:1B 这是什么 ip 192.168.0.1 这是什么银行卡号 6226 1234 5678 9012 这是什么 url http://www.baidu.com""" ) })

rprint(result["messages"][0].content)

TodoListMiddleware

应对比较复杂任务时，可以为模型提供规划能力，减缓降智

agent = create_agent(
    model,
    middleware=[
      
      
      
      TodoListMiddleware()
    ],
)















result = agent.invoke({"messages": HumanMessage("使用 write_todos 帮我规划一个上海一日游")})

ModelCallLimitMiddleware

thread_limit 线程内模型调用上限
run_limit 本次运行模型调用上限
exit_behavior end | error 优雅退出和抛出异常

agent = create_agent( model, middleware=[ ModelCallLimitMiddleware( thread_limit=1,
```
    exit_behavior="error"
  )
],
tools=[get_weather],
checkpointer=InMemorySaver()
```
) config = {"configurable": {"thread_id": "1"}}

result = agent.invoke({"messages": HumanMessage("今天上海天气怎么样")}, config=config)

ToolCallLimitMiddleware

tool_name 不指定就监控所有工具，否则就指定工具
thread_limit 线程内工具调用上限
run_limit 本次运行工具调用上限
exit_behavior 比较模型中间多出 continue 表示继续运行 Agent，模型自主决策后续行为

ModelFallbackMiddleware

后备模型，当主模型失败时，会使用该模型

model = init_chat_model(
    "openai:kimi-k100", 
    base_url=os.environ.get("OPENAI_BASE_URL"),
)
agent = create_agent(
    model,
    middleware=[
      ModelFallbackMiddleware("deepseek:deepseek-v4-flash")
    ],
)
result = agent.invoke({"messages": [HumanMessage("你是什么模型？")]})

LLMToolSelectorMiddleware

工具选择

agent = create_agent(
    model,
    middleware=[
      LLMToolSelectorMiddleware(
        model=model, 
        max_tools=2, 
        
      )
    ],
    tools=[get_news, get_weather]
)
result = agent.invoke({"messages": [HumanMessage("上海天气如何？使用 get_news 看看新闻")]})

ToolRetryMiddleware

基于指数退避算法，设置工具调用失败时的重试策略并且结合 jitter 实现抖动，防止短期迎来并发

class ToolRetryMiddleware(
    *,
    
    max_retries: int = 2, 
    
    tools: list[BaseTool | str] | None = None, 
    
    retry_on: RetryOn = (Exception, ), 
    
    on_failure: OnFailure = "continue", 
    
    backoff_factor: float = 2,
    
    initial_delay: float = 1,
    
    max_delay: float = 60,
    
    jitter: bool = True
)



agent = create_agent(
    model,
    middleware=[
      ToolRetryMiddleware(
          max_retries=5,
          max_delay=10.0,
          retry_on=(TimeoutError,)
      )
    ],
    tools=[get_weather]
)
res = agent.invoke({"messages": [HumanMessage("上海天气怎么样")]})
rprint(res["messages"])

ModelRetryMiddleware

参数与 ToolRetryMiddleware 一致

agent = create_agent(
    'deepseek:deepseek-cat',
    middleware=[
      ModelRetryMiddleware(
          max_retries=5,
          max_delay=10.0,
          on_failure="continue"
      )
    ]
)

res = agent.invoke({"messages": [HumanMessage("上海天气怎么样")]})
rprint(res["messages"])

LLMToolEmulator

当工具不完善的时候，可以使用该中间件达到 mock 的效果

agent = create_agent(
    model,
    middleware=[
      
      LLMToolEmulator(model=model, tools=['get_weather'])
    ],
    tools=[get_weather]
)

res = agent.invoke({"messages": [HumanMessage("上海天气怎么样")]})
rprint(res["messages"])

中间件的执行顺序

遵循先进后出的策略，1 -> 2 -> 3 -> 3 -> 2 -> 1

自定义中间件

中间件提供了六个钩子函数，按照功能可划分为两组

Node-style hooks

before_agent 智能体调用前
after_agent 智能体调用后
before_model 模型调用前
after_model 模型调用后

Wrap-style hooks

wrap_model_call 模型调用前后执行
wrap_tool_call 工具调用前后执行

每个中间件都会用到其中多个或一个钩子函数，针对中间件的书写也提供了函数式和类两种方案（函数式一般仅使用一个钩子时使用）；函数最终也会被内部包装成类去执行

Node Style -> function

@before_agent
def before_agent_middleware(state: AgentState, runtime: Runtime):
    print("=== before_agent ===")
    return None

@after_agent
def after_agent_middleware(state: AgentState, runtime: Runtime):
    print("=== after_agent ===")
    return None

@before_model
def before_model_middleware(state: AgentState, runtime: Runtime):
    print("=== before_model ===")
    return None

@after_model
def after_model_middleware(state: AgentState, runtime: Runtime):
    print("=== after_model ===")
    return None

agent = create_agent(
    model,
    middleware=[before_agent_middleware, after_agent_middleware, before_model_middleware, after_model_middleware],
)
agent.invoke({"messages": [HumanMessage("hello")]})

接收 AgentState | Runtime 两个参数，返回值为 dict[str, Any] | None。

AgentState 是 Agent 运行中的状态，随着 Agent 的运行发行变化，包括消息列表 也在其中。
Runtime 是 Agent 运行中的上下文环境，包括上下文和长期记忆 。
返回值如果是 None 表示不做任何操作，返回字典可以状态；返回特定字段可以控制 Agent 流程；

def before_model(self, state, runtime): if state.get("count", 0) > 10:
```
  return {"jump_to": "__end__"} 
```
@after_agent def after_agent_middleware(state: AgentState, runtime: Runtime): state["messages"][-1].content += "after_agent" return None

state | runtime 的信息如下

{'messages': [HumanMessage(content='hello', additional_kwargs={}, response_metadata={}, id='f577b4bc-722a-4cf7-9891-263d37120559')]}

Runtime(
    context=None,
    store=None,
    stream_writer=<function Pregel.stream.<locals>.stream_writer at 0x10f47c4a0>,
    heartbeat=<function _no_op_heartbeat at 0x1084f5620>,
    previous=None,
    execution_info=ExecutionInfo(
        checkpoint_id='1f184dd4-4b99-6b70-8001-aef8a60a67cd',
        checkpoint_ns='before_model_middleware.before_model:550f9db1-2052-10ef-6835-8c4766468404',
        task_id='550f9db1-2052-10ef-6835-8c4766468404',
        thread_id=None,
        run_id=None,
        node_attempt=1,
        node_first_attempt_time=1784622130.085105
    ),
    server_info=None,
    control=<langgraph.runtime.RunControl object at 0x10f33ebc0>
)

before_** 通常用于消息修剪、数据脱敏、数据输入验证等操作

after_** 通常用于状态统计、输出验证、格式化响应

can_jump_to 参数

中间件装饰器的参数，决定钩子可以直接跳转到流程的哪些地方

tools 跳转至工具节点
end 跳转至 Agent 流程的末尾，或者第一个 after_agent 钩子
model 跳转至模型节点，或者第一个 before_model 钩子

@after_model(can_jump_to=["tools"]) def after_model_middleware(state: AgentState, runtime: Runtime): print("=== after_model ===") return None

Node Style -> class

类实现要继承 AgentMiddleware 这个类

from langchain.agents.middleware from AgentMiddleware, hook_config

class CustomMiddleware(AgentMiddleware):
    def __init__(self):
        super().__init__()
    
    @hook_config(['end', 'tools', 'model'])
    def before_agent(self, state, runtime):
        return None
    
    def before_model(self, state, runtime):
        return None
    
    def after_agent(self, state, runtime):
        return None
    
    def after_model(self, state, runtime):
        return None

Wrap Style -> function

@wrap_model_call
def wrap_model_call_middleware(request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse],) -> ModelResponse:
    rprint(request)
































































    
    response = handler(request)
    return response



@wrap_tool_call
def model_tool_call_middleware(request: ToolCallRequest, handler: Callable[[ToolCallRequest], ToolMessage | Command]) -> ToolMessage | Command:
    rprint(request)

























































































































































    
    request.tool_call["args"]["is_forcast"] = True
    result = handler(request)
    return result

Wrap Style -> class

class WrapMiddleware(AgentMiddleware):
    def __init__(self):
        super().__init__()

    def wrap_model_call(self, request, handler):
        return None
    
    def wrap_tool_call(self, request, handler):
        return None

中间件执行顺序

class CustomMiddleware(AgentMiddleware):
    def __init__(self):
        super().__init__()

    def before_agent(self, state, runtime):
        print("====== before_agent ======")
        return None
    
    def before_model(self, state, runtime):
        print("====== before_model ======")
        return None
    
    def after_agent(self, state, runtime):
        print("====== after_agent ======")
        return None
    
    def after_model(self, state, runtime):
        print("====== after_model ======")
        return None
    
    def wrap_model_call(self, request, handler):
        print("====== wrap_model_call ====== before")
        result =  handler(request)
        print("====== wrap_model_call ====== after")
        return result
    
    def wrap_tool_call(self, request, handler):
        print("====== wrap_tool_call ====== before")
        result = handler(request)
        print("====== wrap_tool_call ====== after")
        return result


agent = create_agent(
    model,
    middleware=[CustomMiddleware()],
    tools=[get_weather]
)