Agent 循环与 Middleware

一句话：create_agent 是一个预置好的“模型—工具”循环图，middleware 是这套装配里插入横切逻辑的主路径——摘要、重试、限流、脱敏、人工审批都在这里做。

适用版本：LangChain v1.x。Middleware 是 v1 的核心扩展机制，0.x 的 callback/自定义 Agent 写法已不是主路径。

为什么需要

一个能跑的 Agent 和能上线的 Agent 之间，差的全是横切关注点：对话太长导致上下文溢出；模型或工具偶发失败，需要重试与降级；模型陷入循环、反复调同一个工具烧钱；发邮件、退款、删数据之前必须有人确认；日志里不能出现手机号、身份证、信用卡；不同租户要用不同的 system prompt 或工具集。

把这些写进业务代码，会让每个 Agent 都重复一遍；写进继承类，会和框架版本绑死。Middleware 提供的是一组稳定的生命周期钩子。

需要说清边界：固定 hook 位是图运行时（LangGraph、Spring AI 这类）的扩展方式——你只能在给定位置插逻辑；插件化运行时把扩展粒度下沉到可替换组件，连组件本身都能换。两者解决的不是同一层问题，不要把 hook 数量当成扩展性指标。

核心机制

Agent 循环与钩子位置

这张图回答 middleware 能插在哪、不能插在哪：钩子只出现在主路径的固定位置，回边就是循环本身。

钩子只能插在循环的固定位置，所以 middleware 改的是某一阶段的输入输出，改不了图的拓扑；需要新拓扑就得下到 LangGraph。包裹式钩子按列表顺序嵌套，外层先拿到请求、最后拿到结果，限流和鉴权放外层才能在真实调用前生效。恢复时节点从头重跑，before_model 和工具钩子可能执行两次，所以钩子里的副作用必须幂等。

• 节点式钩子（before_model、after_model）：拿到当前 state，返回要合并的更新，或返回 None 表示不改。适合消息裁剪、注入上下文、审计。
• 包裹式钩子（wrap_model_call、wrap_tool_call）：像装饰器一样包住一次调用，可以改请求、重试、捕获异常、短路返回。适合超时、重试、权限检查。

执行顺序

middleware=[a, b, c] 时，节点式钩子按列表顺序执行；包裹式钩子形成嵌套，a 在最外层，c 最靠近真实调用。

所以顺序有意义：限流和鉴权放在外层，摘要和格式化放在内层。

内置 middleware

Middleware	作用	典型参数
SummarizationMiddleware	历史过长时自动摘要	model、trigger、keep
HumanInTheLoopMiddleware	指定工具调用前暂停等人审	interrupt_on={工具名: 配置}
ModelCallLimitMiddleware	限制模型调用次数	run_limit、thread_limit、exit_behavior
ToolCallLimitMiddleware	限制工具调用次数（可按住 thread/run/工具）	tool_name、run_limit、thread_limit
ToolRetryMiddleware	工具失败重试	max_retries、on_failure
ModelRetryMiddleware	模型调用失败重试	max_retries、退避策略
ModelFallbackMiddleware	主模型失败时切备用模型	备用模型列表
PIIMiddleware	检测/脱敏/掩码	类型、strategy、作用方向
ToolErrorMiddleware	把工具异常转成可读结果	错误处理函数
LLMToolSelectorMiddleware	工具太多时先筛工具	候选工具、选择模型

HumanInTheLoopMiddleware 需要 checkpointer 才能暂停和恢复；thread_limit 也需要 checkpointer 记录跨轮次数。

自定义 middleware

from langchain.agents import AgentState, create_agent
from langchain.agents.middleware import before_model, wrap_tool_call

@before_model
def inject_tenant(state: AgentState, runtime) -> dict | None:
    # runtime.context 来自 invoke(..., context={...})
    return {"messages": [{"role": "system", "content": f"当前租户：{runtime.context.get('tenant', 'unknown')}"}]}

@wrap_tool_call
def audit(request, handler):
    print("调用工具:", request.tool_call["name"], request.tool_call["args"])
    return handler(request)

返回值会被合并进 state；不要在这里做不可回滚的副作用，因为节点可能因恢复而重跑。

实战场景

场景：客服 Agent 的六层 middleware，顺序是设计出来的

痛点：横切逻辑各写各的早晚出事——审计包在脱敏内层，日志里留下身份证原文；限流放摘要外层，被裁掉的 token 照样计费；审批钩子在最外层，人确认的参数后来又被内层改写。middleware 列表存在的意义就是把"谁先谁后"从口头约定变成装配代码，而顺序本身要按失效代价排。

middleware=[
    audit_log,                      # 1 外层：审计要看到"原始请求 + 最终结果"，且被拒绝的调用也要留痕
    PIIMiddleware(strategy="mask"), # 2 安全类短路：脱敏失败必须拒绝，不能带原文进模型
    ModelCallLimitMiddleware(run_limit=8, exit_behavior="end"),
    ModelFallbackMiddleware(backup_model),
    SummarizationMiddleware(model=cheap, trigger=("tokens", 6000), keep=("messages", 20)),
    HumanInTheLoopMiddleware(interrupt_on={"create_refund": True, "send_email": True}),
]

排法背后的三条规则：

• 短路类在最外层：审计和 PII 被内层短路后外层根本看不到这次调用——限流放摘要外层会先计数，脱敏放审计内层则日志里已经泄漏；
• 改请求的在内层：摘要靠近真实调用，它裁剪的结果就是模型实际收到的内容，放在外层会被后面的逻辑再改动，trace 对不上；
• 审批必须在所有"改参数"的钩子之后：HumanInTheLoop 让人确认的必须是最终参数——若限流/重试包在它外层，恢复重跑时参数可能已被改写，审批就失去了意义。

事故复盘：审批通过后备注消息发了两遍

现象：退款审批通过，用户在企微收到两条"已为您提交退款"。链路还原：notify_middleware 挂在 after_model 里，审批触发 interrupt 挂起；Command(resume=True) 恢复时整个节点从头重跑，after_model 第二次执行，通知再发一次。

三种修法按侵入度排序：

1. 副作用出钩子——middleware 只做标记，通知移到图末尾的独立节点（见 05 篇），恢复重跑不再经过它；
2. 幂等键——notify(user_id, thread_id, resume_generation) 带去重键，重发直接被缓存挡掉；
3. 恢复感知——钩子里检查 runtime 的执行次数，第二次进入跳过副作用；最脆，新钩子容易忘加，只当临时止血。

这个事故的一般形式是：任何"钩子里发生的写操作"都要按"至少一次执行"设计。middleware 的无状态建议（生产动作最后一条）和恢复重跑语义是同一条约束的两面：正因为钩子会重跑，才不许它持有跨轮状态；正因为钩子幂等做不到，才要把副作用挪出钩子。

最小上手

一个带摘要、限流和审计的最小 Agent：

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware, ModelCallLimitMiddleware, before_model
from langgraph.checkpoint.memory import InMemorySaver

@before_model
def log_call(state, runtime):
    print("模型调用，当前消息数:", len(state["messages"]))

agent = create_agent(
    model="openai:gpt-5.5",
    tools=[get_weather],
    system_prompt="你是客服助手，回答要简洁。",
    middleware=[
        log_call,
        SummarizationMiddleware(model="openai:gpt-5.5-mini", trigger=("tokens", 4000), keep=("messages", 20)),
        ModelCallLimitMiddleware(run_limit=8, exit_behavior="end"),
    ],
    checkpointer=InMemorySaver(),
)

cfg = {"configurable": {"thread_id": "user-42"}}
print(agent.invoke({"messages": [{"role": "user", "content": "你好"}]}, cfg)["messages"][-1].text)

生产环境把 InMemorySaver 换成 PostgresSaver，否则进程重启后 thread 状态全丢。

生产动作

• 必须设上限：ModelCallLimitMiddleware + ToolCallLimitMiddleware，再配整体超时；只设一个容易被另一种循环绕过。
• 上下文管理前置：先摘要再截断；摘要用便宜的小模型，保留最近 N 条原文。
• 重试要分类：429、超时、5xx 可重试；参数错误、权限拒绝不要重试。重试必须配退避和上限。
• 降级要有兜底：主模型故障切备用模型，备用也失败就返回可解释的错误，不要让请求挂住。
• 人工审批只覆盖高风险工具：转账、退款、发邮件、删数据；读操作不要也拦，否则没人用。
• 脱敏在入口做：PII 在进入模型前处理，日志和追踪里也不能留原文。
• middleware 保持无状态：需要跨轮状态就放 checkpointer/store，放进程内变量会在多副本下失效。

边界

• middleware 不是安全沙箱：它能拦截工具调用，但真正的权限校验必须在工具或后端服务里，模型和框架都可能被绕过。
• 钩子会因恢复而重跑：中断恢复时节点从头执行，钩子里的副作用可能重复，必须幂等。
• 摘要会丢信息：摘要模型看不懂的细节就永久丢了；关键事实应写入 store 或结构化状态。
• 顺序错会互相抵消：限流放在摘要外层会先被计数，放在内层可能已经烧了 token。
• 过度使用会难排查：每加一个 middleware 就多一层隐式行为，链路追踪必须能看到每层耗时和输入输出。

讲清楚

Agent 的本质是一个循环：调模型、执行工具、再调模型。create_agent 把这个循环和状态管理预置好，middleware 则在循环的固定位置插钩子。

节点式钩子拿 state、返回更新，适合改消息、注入上下文、审计；包裹式钩子包住一次模型或工具调用，适合重试、超时、降级、权限检查。列表顺序决定嵌套顺序，限流和鉴权放外层，摘要和格式化放内层。

上线必备的是三类：上下文管理、成本与循环上限、高风险操作的人工审批。它们都依赖 checkpointer，所以持久化不是可选项。最后记住：middleware 不是安全边界，真正的权限必须在工具和后端做。

适用：LangChain v1.x | 初稿 2026-10