linqianqian/lqq-agent-study @ 8b7a2333542a15d554acbc9ef665a10e5be60911

前言

如果你正在用 LangGraph 搭建 Agent，大概率遇到过这些场景：

Agent 陷入循环，疯狂调用工具把 Token 烧光
模型 API 偶尔抽风，整个链路直接崩掉
用户把身份证号、手机号一股脑塞进对话框
某个"删除生产数据"的操作，Agent 二话不说就执行了

这些问题，靠多写几行 if-else 或者往 System Prompt 里塞规则是解决不了的。你需要的是一个能横切 Agent 执行流程的治理层——这就是 Middleware。

本文把 LangGraph Agent Middleware 的完整知识体系梳理成一篇文章，从架构认知到底层原理，从内置中间件实战到自定义进阶。文中的所有示例围绕一个统一场景——智能报表导出平台——展开，模型选用国内开发者更熟悉的 DeepSeek 系列。

读完这篇，你应该能回答下面几个问题：

Middleware 在 Agent 执行循环的哪些位置介入？
怎么给 Agent 装"刹车"（次数限制、人工确认、脱敏）？
怎么让 Agent 长期稳定运行（重试降级、摘要压缩、上下文清理）？
什么时候用内置 Middleware，什么时候自己写？
request / handler / state / runtime 这几个参数到底怎么用？

一、架构认知：Middleware 是什么，怎么运转

1.1 为什么 Agent 需要被"管"

先看一个典型的 Agent 执行流程：

<!-- 这是一个文本绘图，源码为：flowchart LR

A[👤 用户输入] --> B[🤖 调用大模型]
B --> C{模型决定}
C -->|调工具| D[⚙️ 执行工具]
D --> E[📋 观察结果]
E --> B
C -->|直接回复| F[✅ 返回最终答案] -->

这套循环本身没什么问题，但放到真实业务里，麻烦就来了。假设你在做一个智能报表导出 Agent，它有两个工具：

check_export_permission：查用户有没有导出权限
create_export_task：创建导出任务，可能一次导出几十万行数据

Agent 能自己决策、自己调工具。但如果它决定反复查权限（浪费 Token）、在用户没说清楚场景时直接建导出任务（产生脏数据），或者用户消息里混着别人的手机号就送进了模型——这些都不是 Prompt 里多写两句话能拦住的事。

Middleware 的定位就是：不改变 Agent 做业务的方式，但在它做业务的时候盯着、拦着、兜着底。

<!-- 这是一个文本绘图，源码为：flowchart TB

subgraph Agent执行循环
    direction LR
    U[用户输入] --> M1[before_agent]
    M1 --> M2[before_model]
    M2 --> M3[wrap_model_call]
    M3 --> LLM[大模型调用]
    LLM --> M4[after_model]
    M4 --> M5[wrap_tool_call]
    M5 --> Tool[工具执行]
    Tool --> M6[after_agent]
    M6 --> U
end
M1 -.->|"启动前检查"| M1
M2 -.->|"调用前日志"| M2
M3 -.->|"拦截/重试"| M3
M4 -.->|"响应后处理"| M4
M5 -.->|"工具管控"| M5
M6 -.->|"收尾清理"| M6 -->

LangGraph 提供了六大钩子点，分别对应 Agent 执行流程的不同阶段：

钩子	触发时机	典型用途
`before_agent`	Agent 启动前	初始化状态、检查前置条件
`before_model`	每次调模型之前	审计日志、脱敏、状态检查
`wrap_model_call`	包裹模型调用过程	重试、降级、模型切换
`after_model`	模型返回之后	记录响应、检测 tool_calls、更新计数
`wrap_tool_call`	包裹工具调用过程	权限校验、参数拦截、耗时统计
`after_agent`	Agent 执行结束后	清理状态、发送通知

一句话总结：业务逻辑写在工具里，治理逻辑写在 Middleware 里。职责分离是 Middleware 最核心的设计哲学。

1.2 第一段 Middleware：从零到能观察到 Agent 在做什么

先搭一个最基础的 Agent——没有任何 Middleware，只有两个工具。场景是报表导出平台的权限校验：

from langchain.chat_models import init_chat_model
from langchain.tools import tool


# 使用 DeepSeek 模型，通过阿里云百炼平台接入
# init_chat_model 会自动适配 OpenAI 兼容接口
model = init_chat_model(
    model="deepseek-v4-flash",
    model_provider="openai",
    base_url="https://api.deepseek.com",
    api_key="your_api_key",
)

# ---- 模拟数据 ----
# 实际项目中，这些数据来自数据库或 API
EXPORT_RIGHTS = {
    "zhangsan": {"role": "finance", "region": "all"},
    "lisi": {"role": "operation", "region": "east"},
}


@tool
def check_export_permission(username: str) -> dict:
    """查询用户是否有报表导出权限。
    参数 username 为员工账号（英文名）。
    """
    user_info = EXPORT_RIGHTS.get(username)
    if not user_info:
        return {"username": username, "can_export": False, "reason": "用户不存在"}
    return {
        "username": username,
        "role": user_info["role"],
        "region": user_info["region"],
        "can_export": user_info["role"] in ("finance", "ops_manager"),
    }


@tool
def create_export_task(
    report_name: str, file_format: str, estimated_rows: int, reason: str
) -> dict:
    """创建报表导出任务。
    report_name: 报表名称
    file_format: 导出格式 (xlsx / csv)
    estimated_rows: 预计导出行数
    reason: 导出原因
    """
    return {
        "task_id": "EXPORT-20260706-001",
        "report_name": report_name,
        "file_format": file_format,
        "estimated_rows": estimated_rows,
        "reason": reason,
        "status": "queued",
    }


tools = [check_export_permission, create_export_task]

这个 Agent 已经能干活了，但我们对它的内部行为一无所知。加上两个最轻量的 Middleware 来观察：

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


@before_model
def log_before_model(state: AgentState, runtime):
    """模型调用前执行：看一眼当前上下文里有多少条消息。"""
    # state["messages"] 是 Agent 当前累积的全部对话历史
    # runtime 携带运行时环境信息（后面会详细讲）
    print(f"[审计] 准备调用模型，当前上下文消息数：{len(state['messages'])}")
    return None  # 返回 None 表示不做任何修改


@wrap_tool_call
def log_tool_call(request, handler):
    """工具调用前后各打一条日志，并记录耗时。"""
    tool_name = request.tool_call["name"]
    print(f"[审计] 开始执行工具：{tool_name}")

    # handler(request) 是"真正执行工具"的入口
    # 不调用它，工具就不会执行
    result = handler(request)

    print(f"[审计] 工具执行完毕：{tool_name}")
    return result


# 组装 Agent，把 Middleware 列表传进去
agent = create_agent(
    model=model,
    tools=tools,
    middleware=[
        log_before_model,   # 排在前面：先记录状态
        log_tool_call,      # 排在后面：包裹工具调用
    ],
    system_prompt=(
        "你是报表导出平台的智能助手。"
        "用户询问导出相关问题前，先调用 check_export_permission 确认权限。"
        "只有在用户明确提出导出需求时，才调用 create_export_task 创建任务。"
    ),
)

# 跑一次看看
response = agent.invoke({
    "messages": [
        {
            "role": "user",
            "content": "我是 zhangsan，需要导出本月 east 区域的订单报表，大约 5000 行，xlsx 格式。",
        }
    ]
})

print(response["messages"][-1].content)

运行时你会看到类似这样的输出：

[审计] 准备调用模型，当前上下文消息数：1
[审计] 开始执行工具：check_export_permission
[审计] 工具执行完毕：check_export_permission
[审计] 准备调用模型，当前上下文消息数：3
[审计] 开始执行工具：create_export_task
[审计] 工具执行完毕：create_export_task
[审计] 准备调用模型，当前上下文消息数：5

Agent 的业务能力没变，但现在你能清楚地看到：模型被调了几次、每次调了哪些工具、消息在什么时候膨胀。这就是 Middleware 的第一个价值——可观测性，不侵入业务代码。

1.3 深入 `request` 与 `handler`

写 @wrap_tool_call 或 @wrap_model_call 时，你一定会遇到这两个参数。它们的直觉含义是"拦住一次调用，看看参数，再决定要不要继续"。但很多人卡在对这两个对象的具体结构不熟悉上。

1.3.1 工具级：`ToolCallRequest`

当你在 @wrap_tool_call 里拿到 request 时，它是一个 ToolCallRequest 对象，包含以下关键字段：

字段	含义	用法举例
`request.tool_call["name"]`	模型决定调哪个工具	`if tool_name == "create_export_task"`
`request.tool_call["args"]`	模型生成的调用参数	校验 `estimated_rows` 是否超标
`request.tool_call["id"]`	本次调用的唯一 ID	日志追踪
`request.tool`	工具对象本身	读取工具的 docstring
`request.state`	Agent 当前累积状态	查看已有消息数
`request.runtime`	运行时上下文	读取从业务层传入的 `context`

handler** 是一个可调用对象**——handler(request) 执行之后，真正的工具才会运行。这意味着你可以在调用 handler 之前做校验、修改参数，甚至直接跳过它。

下面是一个实用的例子：在工具执行前拦截超标请求，不调用真正的导出工具，而是返回伪造的 ToolMessage 让 Agent 知道被拦了：

from langgraph_core.messages import ToolMessage


@wrap_tool_call
def guard_export_scale(request, handler):
    """超过 10 万行的导出请求直接拦截，不进入执行队列。"""
    tool_name = request.tool_call["name"]
    args = request.tool_call.get("args", {})

    # 只拦截 create_export_task
    if tool_name == "create_export_task" and args.get("estimated_rows", 0) > 100_000:
        # 不调 handler，直接返回 ToolMessage 给模型
        # 模型收到这条消息后，会告知用户"太大了，换个小范围"
        return ToolMessage(
            content=f"导出被拦截：预计行数 {args.get('estimated_rows')} 超过上限 100000，请缩小筛选范围后重试。",
            tool_call_id=request.tool_call["id"],
        )

    # 正常放行
    return handler(request)

四种 handler 用法模式：

<!-- 这是一个文本绘图，源码为：flowchart TD

R[收到 request] --> Q{怎么处理 handler?}
Q -->|"模式1: 透明放行"| A["return handler(request)"]
Q -->|"模式2: 拦截替换"| B["return ToolMessage(...)"]
Q -->|"模式3: 重试覆盖"| C["handler(request.override(model=backup))"]
Q -->|"模式4: 链式修改"| D["修改 request 后调 handler"] -->

⚠️ 重要：不调 handler = 原始操作从未发生；调两次 handler = 操作执行两次——对于发邮件、扣款这类操作，这可能造成严重后果。

1.3.2 模型级：`ModelRequest`

@wrap_model_call 里的 request 是 ModelRequest，结构更"厚"：

字段	含义
`request.messages`	即将发送给模型的消息列表——就是上文说的"上下文"
`request.model`	当前使用的模型对象
`request.tools`	模型可用的工具列表
`request.model_settings`	模型参数（temperature 等）
`request.state` / `request.runtime`	同工具级

这里有一个很关键的用法——request.override(model=...)，可以在不重建 Agent 的情况下临时切换模型：

@wrap_model_call
def local_first_then_cloud(request, handler):
    """先用本地模型，失败 3 次后切云端模型。"""
    # 先用本地 Ollama 模型（省成本）
    try:
        return handler(request)  # 使用 Agent 默认模型
    except Exception:
        # 本地挂了，切到 DeepSeek 云端
        print("[降级] 本地模型不可用，切换到云端 DeepSeek")
        return handler(request.override(model=cloud_model))

# 一个典型的 Agent 运行流程：
用户输入 → 
  [wrap_model_call 拦截] → LLM 思考 → 决定调用工具 get_weather →
  [wrap_tool_call 拦截] → 执行 get_weather 工具 → 
  [wrap_model_call 拦截] → LLM 再次思考 → 生成最终回复 → 返回给用户

**<font style="color:rgb(15, 17, 21);background-color:rgb(235, 238, 242);">wrap_model_call</font>** 在模型每次被调用时触发，包括首次决策和后续的反思轮次。
**<font style="color:rgb(15, 17, 21);background-color:rgb(235, 238, 242);">wrap_tool_call</font>** 只在模型明确要求调用工具时触发。

1.4 深入 `state` 与 `runtime`

如果说 request / handler 关注的是"这一次调用"，那 state / runtime 关注的就是"Agent 当前整体处于什么状态"。

<!-- 这是一个文本绘图，源码为：flowchart LR

subgraph 单次调用视角
    R[request] --> H[handler]
end
subgraph 全局运行视角
    S[state: 累积消息/计数] --- RT[runtime: 外部上下文]
end -->

1.4.1 `state`：Agent 的内部状态

state（类型 AgentState）是 Agent 在运行过程中不断累积的数据。最核心的字段是 state["messages"]——所有对话历史和工具结果都在里面。

你可以扩展 AgentState，添加自定义字段来追踪业务指标：

from typing import TypedDict
from typing_extensions import NotRequired
from langchain.agents.middleware import AgentState, before_model, after_model


class ExportAgentState(AgentState):
    """扩展默认状态，增加模型调用计数和大任务标记。"""
    # AgentState 已经自带 messages 字段，这里只加新字段
    # NotRequired 表示可以不传，Middleware 内部自己维护
    model_call_count: NotRequired[int]
    blocked_requests: NotRequired[int]  # 被拦截的请求次数


@before_model(state_schema=ExportAgentState)
def audit_before_model(state: ExportAgentState, runtime):
    """每次调模型前，看一眼当前统计。"""
    count = state.get("model_call_count", 0)
    blocked = state.get("blocked_requests", 0)
    print(f"[统计] 第 {count + 1} 次调模型 | 已拦截 {blocked} 次")
    return None


@after_model(state_schema=ExportAgentState)
def update_stats(state: ExportAgentState, runtime):
    """模型返回后，把计数器 +1。"""
    # after_model 返回 dict 可以直接更新 AgentState
    return {"model_call_count": state.get("model_call_count", 0) + 1}

关键点：@before_model / @after_model 需要指定 state_schema，告诉 LangGraph 你的自定义状态有哪些字段。@after_model 返回的 dict 会被浅合并进 state。

1.4.2 `runtime`：外部注入的上下文

runtime.context 存放的是不属于对话本身、但从外部传入的运行时信息——比如请求 ID、用户角色、业务标识。

from langgraph.runtime import Runtime


class RunContext(TypedDict):
    """定义 runtime.context 的结构。"""
    request_id: str     # 用于日志追踪
    user_role: str      # 当前用户的角色
    tenant_id: str      # 租户标识（多租户场景）


@before_model(state_schema=ExportAgentState)
def inject_context(state: ExportAgentState, runtime: Runtime[RunContext]):
    """从 runtime.context 读取业务信息，用于日志关联。"""
    ctx = runtime.context or {}
    print(
        f"[上下文] request_id={ctx.get('request_id')} "
        f"user_role={ctx.get('user_role')} "
        f"tenant={ctx.get('tenant_id')}"
    )
    return None


# 创建 Agent 时声明 context 结构
agent = create_agent(
    model=model,
    tools=tools,
    middleware=[audit_before_model, inject_context, update_stats],
    state_schema=ExportAgentState,  # 声明自定义状态
    context_schema=RunContext,      # 声明上下文结构
    system_prompt="你是报表导出平台的智能助手。",
)

# 调用时传入 context
result = agent.invoke(
    {"messages": [{"role": "user", "content": "导出本月的订单报表"}]},
    context={
        "request_id": "req-20260706-001",
        "user_role": "finance",
        "tenant_id": "t-1234",
    },
)

重要：runtime.context 不会自动让模型看到。如果你想让它影响模型行为，需要在 Middleware 里显式地把信息写入 state 或拼入 prompt——这正是 @dynamic_prompt 的用武之地（见第四章）。

二、安全风控：给 Agent 装上刹车和护栏

线上 Agent 有三类典型风险：调用失控（无限循环烧 Token）、操作越权（关键动作未经确认）、隐私泄露（PII 直接喂给模型）。这一章用三个内置 Middleware 分别解决。

2.1 调用次数限制：防住"无穷循环"

假设 Agent 要查一个异步导出任务的进度。任务可能跑好几分钟，Agent 会不停地轮询。但如果没有限制，它能在 30 秒内调用 200 次模型——Token 账单直接爆炸。

ModelCallLimitMiddleware 和 ToolCallLimitMiddleware 就是专门干这个的。

ModelCallLimitMiddleware：限制模型调用次数

from langchain.agents.middleware import ModelCallLimitMiddleware


# 模拟一个"前 4 次返回处理中，第 5 次返回完成"的进度查询工具
export_progress = {"task_001": 0}


@tool
def check_export_progress(task_id: str) -> str:
    """查询导出任务的进度。"""
    export_progress[task_id] = export_progress.get(task_id, 0) + 1
    attempt = export_progress[task_id]
    if attempt < 5:
        return f"第 {attempt} 次查询：任务仍在处理中，请稍候。"
    return f"第 {attempt} 次查询：导出完成，文件已生成。"


# 不加限制：Agent 会一直查到第 5 次
# 加上 ModelCallLimitMiddleware(run_limit=3)：最多调 3 次模型就强制结束
agent = create_agent(
    model=model,
    tools=[check_export_progress],
    middleware=[
        ModelCallLimitMiddleware(
            run_limit=3,          # 单次 invoke 最多调 3 次模型
            exit_behavior="end",  # 到达上限后尝试优雅结束（生成总结）
        ),
    ],
    system_prompt=(
        "你是导出任务进度查询助手。"
        "当用户查询任务进度时，调用 check_export_progress。"
        "如果任务还在处理中，继续查询直到完成。"
    ),
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "帮我查 task_001 的导出进度，持续查到完成为止。"}]
})
# 输出类似："Model call limits exceeded: run limit (3/3)"
# Agent 被强制刹车，不会无限循环
print(result)

ToolCallLimitMiddleware：限制特定工具的调用次数

某些场景下你不想限制整个 Agent 的模型调用次数，只想限制某一个工具——比如进度查询工具被频繁轮询。ToolCallLimitMiddleware 只针对指定工具生效：

from langchain.agents.middleware import ToolCallLimitMiddleware

agent = create_agent(
    model=model,
    tools=[check_export_progress],
    middleware=[
        ToolCallLimitMiddleware(
            tool_name="check_export_progress",  # 只限制这个工具
            run_limit=2,                         # 最多调 2 次
            exit_behavior="continue",            # 达到上限后通知模型继续（不抛异常）
        ),
    ],
    system_prompt=(
        "你是导出任务进度查询助手。"
        "当 check_export_progress 返回限制信息时，告知用户当前进度并建议稍后再查。"
    ),
)
# exit_behavior="continue" 的效果：
# 工具达到上限后，Middleware 返回一条 ToolMessage 告知模型"已达调用上限"，
# 模型可以据此生成友好的回复，而不是直接报错

两个 Middleware 可以混用：同时配置 ModelCallLimitMiddleware（全局上限）和 ToolCallLimitMiddleware（单工具上限），Agent 会在任一限制触发时终止或被拦截。

run_limit vs thread_limit

参数	作用域	适用场景
`run_limit`	单次 `invoke()`	防止一次请求中过度调用
`thread_limit`	整个对话线程（需 checkpointer）	跨多轮对话的累计限制

thread_limit 需要配合 checkpointer 使用，因为跨轮追踪需要持久化状态：

from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
    model=model,
    tools=[check_export_progress],
    middleware=[
        ModelCallLimitMiddleware(
            run_limit=10,
            thread_limit=50,  # 整个对话生命周期最多 50 次模型调用
            exit_behavior="end",
        ),
    ],
    checkpointer=InMemorySaver(),  # 必需：跨轮追踪需要存储
    system_prompt="你是导出任务进度查询助手。",
)

`exit_behavior` 的三种模式

值	行为	适用场景
`"end"`	优雅结束，生成兜底回复	大多数场景
`"error"`	抛出异常，上层捕获处理	需要明确感知限流事件
`"continue"`	（仅工具限制）返回限制信息给模型，让模型决定怎么说	模型可以继续回复用户的场景

2.2 关键操作人工确认：Human-in-the-Loop

有些操作 Agent 不应该自己决定——比如创建一个可能消耗大量资源的导出任务。HumanInTheLoopMiddleware 让 Agent 在执行特定工具前暂停并等待人类审批。

<!-- 这是一个文本绘图，源码为：sequenceDiagram

participant U as 👤 用户
participant A as 🤖 Agent
participant M as 🛑 HITL Middleware
participant T as ⚙️ 工具

U->>A: 帮我导出一份报表
A->>M: 准备调用 create_export_task
M-->>A: ⏸️ 暂停！需要人工审批
M->>U: 请确认：创建导出任务？
U->>M: ✅ approve（或 ✏️ edit / ❌ reject）
M->>T: 放行（或修改参数后放行）
T->>A: 返回执行结果
A->>U: 任务已创建 -->

实战：为危险操作加上确认环节

from typing import Literal
from pydantic import BaseModel, Field
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver


# 第一步：用 Pydantic 定义工具的输入结构
# 结构化的参数让模型生成更准确，也方便人工审批时查看
class ExportTaskInput(BaseModel):
    """导出任务的参数结构。模型会按照这个 schema 生成参数。"""
    report_name: str = Field(description="报表名称，如 'east_region_orders_202607'")
    file_format: Literal["xlsx", "csv"] = Field(description="导出格式")
    estimated_rows: int = Field(description="预计导出行数")
    reason: str = Field(description="导出原因，用于审计")
    priority: Literal["low", "normal", "high"] = Field(
        default="normal", description="优先级"
    )


# 第二步：安全工具不加 args_schema（自动执行）
@tool
def check_export_permission(username: str) -> dict:
    """查询用户导出权限。安全操作，无需审批。"""
    return {"username": username, "can_export": True, "region": "all"}


# 第三步：危险工具加上 args_schema，配合 HITL 拦截
@tool(args_schema=ExportTaskInput)
def create_export_task(
    report_name: str,
    file_format: str,
    estimated_rows: int,
    reason: str,
    priority: str = "normal",
) -> dict:
    """创建报表导出任务。这是一个高风险操作，需要人工审批。"""
    print(
        f"[导出] 创建任务：{report_name} | {file_format} | "
        f"{estimated_rows} 行 | 优先级 {priority}"
    )
    return {
        "task_id": "EXPORT-20260706-002",
        "report_name": report_name,
        "file_format": file_format,
        "estimated_rows": estimated_rows,
        "status": "queued",
    }


# 第四步：创建带 HITL 的 Agent
checkpointer = InMemorySaver()  # HITL 必需：暂停后需要从这里恢复状态

agent = create_agent(
    model=model,
    tools=[check_export_permission, create_export_task],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                # 安全工具：不中断，自动执行
                "check_export_permission": False,
                # 危险工具：中断，提供三种审批选项
                "create_export_task": {
                    "allowed_decisions": ["approve", "edit", "reject"],
                },
            },
        ),
    ],
    checkpointer=checkpointer,
    system_prompt=(
        "你是报表导出平台的智能助手。"
        "用户查询权限时，直接调用 check_export_permission。"
        "只在用户明确要求创建导出任务时调用 create_export_task。"
    ),
)

# 第五步：第一次 invoke——会被 HITL 拦截
config = {"configurable": {"thread_id": "export-001"}}
result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "我是 zhangsan，导出 east 区本月订单报表，xlsx，约 5000 行，月度对账用。",
            }
        ]
    },
    config=config,
)

# result 里会包含中断信息，UI 层可以据此展示审批界面
print("Agent 已暂停，等待审批...")

三种审批操作

# 审批通过：工具以原始参数执行
agent.invoke(
    Command(resume={"decisions": [{"type": "approve"}]}),
    config=config,
)

# 编辑后通过：修改参数再执行（比如把 estimated_rows 从 50000 改成 5000）
agent.invoke(
    Command(resume={
        "decisions": [
            {
                "type": "edit",
                "edited_action": {
                    "name": "create_export_task",
                    "args": {
                        "report_name": "east_region_orders_202607",
                        "file_format": "xlsx",
                        "estimated_rows": 5000,  # 人工修正
                        "reason": "月度对账",
                        "priority": "high",
                    },
                },
            }
        ]
    }),
    config=config,
)

# 拒绝：工具不执行，拒绝原因会反馈给模型
agent.invoke(
    Command(resume={
        "decisions": [
            {
                "type": "reject",
                "message": "审批被驳回：本月对账已由系统自动完成，无需手动导出。",
            }
        ]
    }),
    config=config,
)

⚠️ HITL 的三个要点：

必须有 checkpointer——Agent 暂停后需要持久化状态，InMemorySaver 适合开发调试，生产环境建议用 SqliteSaver 或 PostgresSaver

thread_id** 是恢复的钥匙**——创建任务和审批操作必须用同一个 thread_id

不是所有工具都要审批——只对真正高危的操作加 HITL，给每个操作都加确认会让用户体验很差

2.3 敏感信息自动脱敏

用户经常不经意地在消息里放敏感信息："帮我查一下导出记录，我的手机号 13800138000，身份证 110101199003070013"。

这些信息不应该原样送进模型。PIIMiddleware 在消息进入模型之前做检测和替换：

<!-- 这是一个文本绘图，源码为：flowchart LR

U[👤 用户消息] --> P[🔍 PIIMiddleware]
P -->|检测通过| M[🤖 模型]
P -->|检测到 PII| R{策略?}
R -->|redact| R1["替换为 [REDACTED_PHONE]"]
R -->|hash| R2["替换为一致的 hash 值"]
R -->|mask| R3["部分遮盖如 138****8000"]
R -->|block| R4["直接拒绝，抛出异常"]
R1 --> M
R2 --> M
R3 --> M -->

内置检测类型 + 自定义正则

LangGraph 内置支持 email、信用卡号、IP 地址、MAC 地址、URL 等。中文场景常用的手机号、身份证号需要用自定义正则：

from langchain.agents.middleware import PIIMiddleware

# 中国大陆手机号：1 开头，第二位 3-9，共 11 位
PHONE_PATTERN = r"(?<!\d)1[3-9]\d{9}(?!\d)"

# 身份证号：15 位或 18 位（末尾可能是 X）
ID_CARD_PATTERN = r"(?<!\d)(?:\d{17}[\dXx]|\d{15})(?!\d)"

# 银行卡号：16-19 位数字
BANK_CARD_PATTERN = r"(?<!\d)\d{16,19}(?!\d)"


agent = create_agent(
    model=model,
    tools=[],
    middleware=[
        # 内置检测：邮箱
        PIIMiddleware("email", strategy="redact", apply_to_input=True),
        # 自定义检测：手机号
        PIIMiddleware(
            "phone_number",
            detector=PHONE_PATTERN,
            strategy="redact",
            apply_to_input=True,
        ),
        # 自定义检测：身份证
        PIIMiddleware(
            "id_card",
            detector=ID_CARD_PATTERN,
            strategy="redact",
            apply_to_input=True,
        ),
    ],
    system_prompt="你是报表导出平台的客服助手。回复要简洁，不要提及用户的个人隐私信息。",
)

result = agent.invoke({
    "messages": [
        {
            "role": "user",
            "content": (
                "我的手机号 13800138000，邮箱 zhangsan@example.com，"
                "身份证 110101199003070013。帮我查一下上个月的导出记录。"
            ),
        }
    ]
})
# 模型实际收到的消息：
# "我的手机号 [REDACTED_PHONE_NUMBER]，邮箱 [REDACTED_EMAIL]，
#  身份证 [REDACTED_ID_CARD]。帮我查一下上个月的导出记录。"
print(result)

四种脱敏策略对比

策略	效果示例	适用场景
`redact`	`13800138000` → `[REDACTED_PHONE_NUMBER]`	大多数场景，完全隐藏
`hash`	`13800138000` → `a1b2c3d4...`（同一输入永远同一 hash）	需要去重但不需要原值
`mask`	`13800138000` → `138****8000`	需要保留部分信息（如尾号）
`block`	直接抛出 `PIIDetectionError`	严格禁止任何 PII 进入模型

block 策略实战：直接拒绝含 PII 的请求

block 策略和其他三种不同——它不会替换敏感信息，而是直接抛出 PIIDetectionError，阻止模型调用。适合对合规要求极高的场景（如金融、医疗）：

from langchain.agents.middleware import PIIDetectionError

try:
    result = agent.invoke({
        "messages": [
            {
                "role": "user",
                "content": "我的身份证号是 110101199003070013，帮我查导出记录。",
            }
        ]
    })
    print(result)
except PIIDetectionError as e:
    # block 策略触发后不会调用模型，直接在这里拦截
    print(f"[合规拦截] 检测到敏感信息：{e.matched_entities}")
    # 可以在这里返回通用回复或转人工处理

block 抛出异常后会中断整个 Agent 执行流。如果你的应用需要优雅降级（比如告诉用户"请勿发送敏感信息"），需要在 agent.invoke() 外层 try/except PIIDetectionError。

不止输入：工具返回和输出也要查

agent = create_agent(
    model=model,
    tools=tools,
    middleware=[
        PIIMiddleware(
            "email", strategy="redact",
            apply_to_input=True,           # 用户输入先过滤
            apply_to_tool_results=True,    # 工具返回值也过滤
            apply_to_output=True,          # 模型最终输出再检查一遍
        ),
    ],
    system_prompt="你是报表导出客服助手。",
)

⚠️ 注意：PIIMiddleware 只保护 Agent 消息管道。你自己工具里打的日志、调用的外部 API，不在它的保护范围内——那些需要另外做脱敏。

三、稳定性与上下文治理：让 Agent 长期可靠运行

Agent 在生产环境跑久了，会遇到两类问题：外部依赖偶尔失败（模型超时、接口抖动）和内部状态持续膨胀（对话越来越长、旧工具结果堆积）。这一章用三组 Middleware 解决。

3.1 失败重试与模型降级

网络抖动、API 限流、模型服务临时不可用——这些在生产环境是常态。ToolRetryMiddleware 和 ModelRetryMiddleware 处理"重试"问题，ModelFallbackMiddleware 处理"降级"问题。

<!-- 这是一个文本绘图，源码为：flowchart TD

A[调用开始] --> B{成功?}
B -->|是| C[继续执行]
B -->|否| D{还有重试次数?}
D -->|是| E[等待延迟后重试]
E --> B
D -->|否| F{有备用模型?}
F -->|是| G[切换备用模型]
F -->|否| H[按 on_failure 策略处理] -->

工具重试：处理间歇性超时

from langgraph.agents.middleware import ToolRetryMiddleware


# 模拟一个前两次超时、第三次成功的导出接口
QUERY_ATTEMPTS = {}


@tool
def fetch_export_history(user_id: str) -> dict:
    """查询用户的历史导出记录。"""
    QUERY_ATTEMPTS[user_id] = QUERY_ATTEMPTS.get(user_id, 0) + 1
    attempt = QUERY_ATTEMPTS[user_id]

    if attempt < 3:
        # 前两次模拟网络超时
        raise TimeoutError(f"查询接口超时（第 {attempt} 次）")
    # 第三次成功
    return {
        "user_id": user_id,
        "records": [
            {"task_id": "EXP-001", "report": "月度订单", "time": "2026-07-01"},
            {"task_id": "EXP-002", "report": "客户分析", "time": "2026-07-05"},
        ],
    }


agent = create_agent(
    model=model,
    tools=[fetch_export_history],
    middleware=[
        ToolRetryMiddleware(
            tools=["fetch_export_history"],  # 只对这个工具启用重试
            max_retries=2,                    # 额外重试 2 次（共 3 次机会）
            retry_on=(TimeoutError,),         # 只在超时时重试，权限错误不重试
            initial_delay=0.2,               # 第一次重试前等 0.2 秒
            max_delay=1.0,                    # 重试间隔上限 1 秒（指数退避）
            on_failure="continue",            # 重试耗尽后让模型继续（不抛异常）
        ),
    ],
    system_prompt="你是报表导出平台的查询助手。根据查询结果回答用户。",
)

关键设计理念：重试逻辑不写进工具函数，由 Middleware 统一处理。 你的工具只管"查数据"，超时了怎么办、重试多少次、间隔多久——这些是 Middleware 该操心的事。

模型重试与降级：本地模型挂了切云端

from langchain.agents.middleware import ModelRetryMiddleware, ModelFallbackMiddleware

# 本地模型：Ollama + Qwen3，成本低但在高负载下可能不可用
local_model = init_chat_model("ollama:qwen3:14b")

# 云端模型：DeepSeek，稳定但要花钱
cloud_model = init_chat_model(
    model="deepseek-v4-flash",
    model_provider="openai",
    base_url="https://api.deepseek.com",
    api_key="your_api_key_here",
)

agent = create_agent(
    model=local_model,  # 默认用本地模型
    tools=tools,
    middleware=[
        # 第一层：同一个模型失败后重试
        ModelRetryMiddleware(
            max_retries=3,                  # 最多重试 3 次
            retry_on=(Exception,),          # 任何异常都重试
            on_failure="continue",
        ),
        # 第二层：重试也失败了，切备用模型
        ModelFallbackMiddleware(
            cloud_model,                    # 备用的云端模型
        ),
    ],
    system_prompt="你是报表导出平台的智能助手。",
)

选型建议：重试适合处理临时性抖动（网络波动、短暂限流）；降级适合处理持续性不可用（服务宕机、额度耗尽）。不要把验证错误（如参数校验失败）放进重试范围——重试一万次也不会通过。

3.2 长对话自动摘要：别让上下文吃掉你的 Token

多轮对话中，用户可能前前后后提了十几次需求。所有历史消息都塞进上下文，模型窗口再大也扛不住——而且大部分旧消息对当前回答帮助不大。

SummarizationMiddleware 的思路很朴素：超过阈值后，把旧消息压缩成一段摘要，只保留最近几条原文。

<!-- 这是一个文本绘图，源码为：flowchart TD

A[消息累积中] --> B{达到触发阈值?}
B -->|否| A
B -->|是| C[取旧消息 → 调摘要模型压缩]
C --> D[摘要 + 最近 N 条原文 → 新上下文] -->

from langchain.agents.middleware import SummarizationMiddleware


agent = create_agent(
    model=cloud_model,
    tools=[],
    middleware=[
        SummarizationMiddleware(
            model=model,            # 
            trigger=("messages", 8),        # 消息数达到 8 条时触发
            keep=("messages", 4),           # 保留最近 4 条原文
            # 自定义摘要提示词：强调保留"不做的事情"
            summary_prompt=(
                "请将以下对话历史压缩为一份结构化摘要。必须包含：\n"
                "1. 用户的最终目标\n"
                "2. 已确认的需求要点\n"
                "3. 用户明确表示不需要的功能（否定约束）\n"
                "4. 当前待解决的问题\n"
                "摘要不超过 300 字。"
            ),
        ),
    ],
    system_prompt="你是报表导出平台的需求分析助手。",
)

关键参数说明：

参数	可选值	说明
`trigger`	`("messages", N)` / `("tokens", N)` / `("fraction", 0.8)`	触发摘要的条件
`keep`	`("messages", N)` / `("tokens", N)` / `("fraction", 0.25)`	保留多少最近内容不压缩
`model`	任何 ChatModel	做摘要的模型，可以和主模型不同
`summary_prompt`	自定义 prompt 字符串	默认摘要有时会漏掉"不做的事"

常见坑：默认摘要往往会忽略否定信息。比如用户说过"不支持 PDF 导出"、"移动端不做"，摘要里很可能就丢了。自定义 summary_prompt 时，显式要求保留这类约束。

3.3 清理旧工具结果：别让过期的数据干扰模型

SummarizationMiddleware 压缩的是聊天历史，但还有一种膨胀——工具返回值。

Agent 一个任务调 4-5 次工具很正常。每次工具返回几百上千字，几轮下来上下文里塞满了旧结果。但模型做下一轮决策时，通常只需要最近一次工具的结果。ContextEditingMiddleware + ClearToolUsesEdit 解决的就是这个问题。

from langchain.agents.middleware import ContextEditingMiddleware, ClearToolUsesEdit

agent = create_agent(
    model=model,
    tools=tools,
    middleware=[
        ContextEditingMiddleware(
            edits=[
                ClearToolUsesEdit(
                    trigger=120,                      # token 数超过 120 时触发
                    keep=1,                           # 保留最近 1 次工具结果
                    placeholder="[旧工具结果已被清理，节省上下文空间]",
                    exclude_tools=("critical_audit_log",),  # 审计日志不清理
                    clear_at_least=50,                # 每次至少清理 50 token
                    clear_tool_inputs=True,           # 同时截断 AIMessage 里的 tool_call args
                ),
            ],
        ),
    ],
    system_prompt="你是报表导出平台的智能助手。",
)

关键补充参数：

参数	含义	说明
`clear_tool_inputs`	是否同时清理 AIMessage 中的工具调用参数	设 `True` 后，旧 AIMessage 里的 `tool_calls[].args` 也会被截断，进一步节省空间
`token_count_method`	Token 计数策略	`"approximate"`（默认）：用字符数估算，零开销；`"model"`：调模型精确计数，更准确但有额外成本
`clear_at_least`	触发后最少清理多少 token	防止触发清理但实际回收空间太小（比如刚好跨过阈值 1 token）

摘要 vs 清理：什么时候用哪个

维度	SummarizationMiddleware	ContextEditingMiddleware
处理对象	对话消息（Human + AI）	工具调用结果（ToolMessage）
处理方式	压缩成摘要文本	替换为占位符
适合场景	历史对话信息密度低	工具返回数据量大且时效性强
组合使用	可以，先摘要再清理，各自负责不同维度

四、自定义进阶：写出自己的 Middleware

内置 Middleware 覆盖了大部分通用场景，但每个项目的业务规则千差万别。比如："导出任务超过 10 万行时自动降级为异步"、"模型返回后检查是否包含敏感字段名"——这些就需要自己动手了。

LangGraph 提供了两条路径：装饰器模式和类模式。

4.1 装饰器模式：轻量级，适合单个 Hook

四个装饰器分别对应四个钩子点：

装饰器	适用场景
`@before_model`	调用前作日志、状态检查、上下文注入
`@after_model`	调用后记录结果、更新计数、触发后续动作
`@wrap_model_call`	包裹模型调用，实现重试、降级、超时
`@wrap_tool_call`	包裹工具调用，实现参数校验、权限拦截、耗时统计

下面是一个完整的自定义 Middleware 组，覆盖报表导出平台的业务规则：

import time
from langchain.agents.middleware import (
    before_model, after_model, wrap_model_call, wrap_tool_call,
    AgentState,
)
from langchain_core.messages import ToolMessage


# ---- Middleware 1：模型调用前记录审计信息 ----
@before_model
def audit_before_model(state: AgentState, runtime):
    """在模型调用前，记录当前上下文规模和来源用户。"""
    ctx = runtime.context or {}
    print(
        f"[审计] 模型调用 #{state.get('call_count', 0) + 1} | "
        f"消息数 {len(state['messages'])} | "
        f"用户 {ctx.get('username', 'unknown')}"
    )
    return None


# ---- Middleware 2：模型返回后更新调用计数 ----
@after_model
def update_call_counter(state: AgentState, runtime):
    """模型返回后，调用计数 +1。"""
    return {"call_count": state.get("call_count", 0) + 1}


# ---- Middleware 3：包裹模型调用，实现本地重试 + 云端降级 ----
local_failures = 0
MAX_LOCAL_FAILURES = 3


@wrap_model_call
def retry_local_then_cloud(request, handler):
    """先用本地 Ollama 模型，失败 N 次后切到云端 DeepSeek。"""
    global local_failures

    try:
        result = handler(request)  # 尝试用本地模型
        local_failures = 0         # 成功后重置计数器
        return result
    except Exception as e:
        local_failures += 1
        if local_failures < MAX_LOCAL_FAILURES:
            print(f"[降级] 本地模型失败（{local_failures}/{MAX_LOCAL_FAILURES}），重试中...")
            raise  # 重新抛出，让 ModelRetryMiddleware 处理
        print(f"[降级] 本地模型不可用，切换到云端 DeepSeek")
        # request.override 临时换模型，不影响 Agent 默认配置
        return handler(request.override(model=cloud_model))


# ---- Middleware 4：包裹工具调用，实现业务规则拦截 ----
@wrap_tool_call
def guard_export_tool(request, handler):
    """在 create_export_task 执行前，做业务规则校验。"""
    tool_name = request.tool_call["name"]

    if tool_name == "create_export_task":
        args = request.tool_call.get("args", {})
        estimated_rows = args.get("estimated_rows", 0)

        # 规则 1：超过 100 万行直接拒绝
        if estimated_rows > 1_000_000:
            return ToolMessage(
                content=(
                    f"导出被拦截：预计行数 {estimated_rows} 超过系统上限 1000000。"
                    "建议：缩小时间范围或分批次导出。"
                ),
                tool_call_id=request.tool_call["id"],
            )

        # 规则 2：超过 10 万行发出警告但放行（打印日志供运维关注）
        if estimated_rows > 100_000:
            print(
                f"[警告] 大导出任务：{args.get('report_name')} | "
                f"{estimated_rows} 行 | 用户可能需要等待较长时间"
            )

        # 规则 3：记录工具调用耗时
        start = time.time()
        result = handler(request)
        elapsed = time.time() - start
        print(f"[性能] {tool_name} 执行耗时 {elapsed:.2f}s")

        return result

    # 非目标工具直接放行
    return handler(request)

4.2 类模式：可配置、可复用的 Middleware

装饰器适合快速写一个 hook，但如果你的 Middleware 需要配置参数（比如"不同角色的用户有不同的导出上限"），类模式更合适——继承 AgentMiddleware。

from langchain.agents.middleware import AgentMiddleware


class ExportGovernanceMiddleware(AgentMiddleware):
    """报表导出治理中间件：按角色限制导出规模。

    用法：
        ExportGovernanceMiddleware(
            role_limits={"finance": 500_000, "operation": 100_000, "default": 50_000}
        )
    """

    def __init__(self, role_limits: dict = None):
        super().__init__()
        self.role_limits = role_limits or {"default": 100_000}

    def before_model(self, state, runtime):
        """每次模型调用前，注入当前用户的权限信息到日志。"""
        ctx = runtime.context or {}
        role = ctx.get("user_role", "default")
        limit = self.role_limits.get(role, self.role_limits["default"])
        print(f"[治理] 用户角色 {role}，导出上限 {limit} 行")
        return None

    def wrap_tool_call(self, request, handler):
        """在创建导出任务前，检查是否超过该角色的导出上限。"""
        if request.tool_call["name"] != "create_export_task":
            return handler(request)

        ctx = request.runtime.context or {}
        role = ctx.get("user_role", "default")
        limit = self.role_limits.get(role, self.role_limits["default"])
        args = request.tool_call.get("args", {})
        estimated_rows = args.get("estimated_rows", 0)

        if estimated_rows > limit:
            return ToolMessage(
                content=(
                    f"导出被拦截：预计 {estimated_rows} 行，"
                    f"超过 {role} 角色的上限 {limit} 行。"
                    "请联系上级审批或缩小导出范围。"
                ),
                tool_call_id=request.tool_call["id"],
            )

        return handler(request)


# 使用
agent = create_agent(
    model=model,
    tools=tools,
    middleware=[
        ExportGovernanceMiddleware(
            role_limits={
                "finance": 500_000,
                "operation": 100_000,
                "default": 50_000,
            }
        ),
    ],
    context_schema=RunContext,
    system_prompt="你是报表导出平台的智能助手。",
)

装饰器 vs 类模式：装饰器适合一次性逻辑（写一次、一个场景用）；类模式适合需要参数化、跨 Agent 复用的治理规则。

4.3 动态提示词：让同一个 Agent 适配不同场景

有些 Agent 需要同时处理多种风格的任务——比如同一个报表助手，既要写周报（专业干练），又要写事故复盘（严谨客观），还要回答日常使用问题（简洁友好）。

硬写一个又大又全的 System Prompt 效果很差。@dynamic_prompt 根据每次请求的实际内容动态生成 System Prompt：

from langchain.agents.middleware import ModelRequest, dynamic_prompt


def latest_user_text(request: ModelRequest) -> str:
    """从消息列表中提取最近一条用户消息的文本。"""
    # 倒序遍历，找到最新的 human 类型消息
    for message in reversed(request.messages):
        if message.type == "human":
            return str(message.content)
    return ""


@dynamic_prompt
def adaptive_prompt(request: ModelRequest) -> str:
    """根据用户请求的内容，动态选择 System Prompt 风格。"""
    user_text = latest_user_text(request)

    # 从 runtime.context 读取受众信息（由外部系统传入）
    ctx = request.runtime.context or {}
    audience = ctx.get("audience", "内部团队")

    # 分支 1：事故复盘 → 严谨结构化
    if "事故" in user_text or "复盘" in user_text:
        return (
            "你是一名资深 SRE 工程师，正在撰写线上事故复盘报告。\n"
            f"目标受众：{audience}。\n"
            "请按以下结构组织内容：\n"
            "1. 事故概述（时间、影响范围、持续时间）\n"
            "2. 根因分析（直接原因、间接原因）\n"
            "3. 处理过程（时间线）\n"
            "4. 改进措施（短期 + 长期）\n"
            "语气：客观、严谨；不甩锅，不带情绪。"
        )

    # 分支 2：周报 → 简洁干练
    if "周报" in user_text:
        return (
            "你是一名技术团队的 PM，正在编写项目周报。\n"
            f"目标受众：{audience}。\n"
            "请按以下结构组织内容：\n"
            "1. 本周完成事项（按优先级排列）\n"
            "2. 关键进展（数据支撑）\n"
            "3. 遇到的问题（附解决思路）\n"
            "4. 下周计划\n"
            "要求：简洁、有数据、不写流水账。"
        )

    # 分支 3：默认日常回复
    return (
        "你是报表导出平台的智能助手。\n"
        f"目标受众：{audience}。\n"
        "请用简洁清晰的语言回答用户问题，必要时给出操作指引。"
    )


agent = create_agent(
    model=model,
    middleware=[adaptive_prompt],
    context_schema=RunContext,
)

# 两个不同场景的请求，会得到完全不同的 System Prompt
questions = [
    "帮我写一份本周的报表系统开发周报，涵盖导出接口优化和权限模块改造",
    "帮我对上周六报表导出服务宕机的事故做一份复盘报告",
]

for question in questions:
    result = agent.invoke(
        {"messages": [{"role": "user", "content": question}]},
        context={"audience": "研发管理层"},
    )
    print(result["messages"][-1].content)
    print("---")

@dynamic_prompt** 适用场景**：

同一个 Agent 处理多种不同类型的任务
不同用户角色需要不同的回复风格
Prompt 需要根据上下文实时调整

不适用场景：

审计日志 → 用 @before_model
模型重试 → 用 @wrap_model_call
工具拦截 → 用 @wrap_tool_call

附录：Middleware 选型速查表

按关注维度分类

维度	Middleware	一句话说明	关键参数
🔢 次数限制	`ModelCallLimitMiddleware`	限制模型调用次数	`run_limit`, `thread_limit`, `exit_behavior`
🔢 次数限制	`ToolCallLimitMiddleware`	限制特定工具调用次数	`tool_name`, `run_limit`, `exit_behavior`
🛡️ 安全审批	`HumanInTheLoopMiddleware`	关键操作暂停等人类确认	`interrupt_on`, `allowed_decisions`
🔒 隐私脱敏	`PIIMiddleware`	敏感信息进入模型前脱敏	`detector`, `strategy`, `apply_to_*`
🔄 故障重试	`ToolRetryMiddleware`	工具失败后自动重试	`max_retries`, `retry_on`, `on_failure`
🔄 故障重试	`ModelRetryMiddleware`	模型调用失败后重试	`max_retries`, `retry_on`
⬇️ 模型降级	`ModelFallbackMiddleware`	主模型不可用时切换备用	`fallback_model`
📝 上下文压缩	`SummarizationMiddleware`	长对话自动摘要压缩	`trigger`, `keep`, `model`, `summary_prompt`
🧹 上下文清理	`ContextEditingMiddleware`	清理旧工具结果	`edits=[ClearToolUsesEdit(...)]`
🎯 动态提示	`@dynamic_prompt`	每次模型调用前动态生成 System Prompt	函数返回字符串

按执行阶段分类

<!-- 这是一个文本绘图，源码为：flowchart TB

subgraph 启动阶段
    before_agent
end
subgraph 模型调用阶段
    direction LR
    before_model --> wrap_model_call --> LLM[大模型] --> after_model
end
subgraph 工具调用阶段
    wrap_tool_call --> Tool[工具执行]
end
subgraph 结束阶段
    after_agent
end -->

阶段	可用的 Middleware 类型
Agent 启动前	自定义 `before_agent`
模型调用前	`@before_model`、`PIIMiddleware`（`apply_to_input`）
模型调用中	`@wrap_model_call`、`ModelRetryMiddleware`、`ModelFallbackMiddleware`
模型调用后	`@after_model`、`ModelCallLimitMiddleware`（计数判定）
工具调用中	`@wrap_tool_call`、`ToolRetryMiddleware`、`ToolCallLimitMiddleware`、`HumanInTheLoopMiddleware`
Agent 结束后	自定义 `after_agent`
上下文治理	`SummarizationMiddleware`、`ContextEditingMiddleware`
提示词	`@dynamic_prompt`

组合建议：一份"生产级" Middleware 栈

把前面各章的内容组合起来，一份典型的生产环境 Agent 配置大概长这样：

agent = create_agent(
    model=local_model,
    tools=tools,
    middleware=[
        # —— 安全层（最先执行） ——
        PIIMiddleware("email", strategy="redact", apply_to_input=True),
        HumanInTheLoopMiddleware(
            interrupt_on={"create_export_task": {"allowed_decisions": ["approve", "reject"]}}
        ),

        # —— 可靠性层 ——
        ModelRetryMiddleware(max_retries=3, retry_on=(Exception,)),
        ModelFallbackMiddleware(cloud_model),
        ToolRetryMiddleware(
            tools=["fetch_export_history"], max_retries=2, retry_on=(TimeoutError,)
        ),

        # —— 限流层 ——
        ModelCallLimitMiddleware(run_limit=20, exit_behavior="end"),
        ToolCallLimitMiddleware(tool_name="check_export_progress", run_limit=10),

        # —— 上下文治理层 ——
        SummarizationMiddleware(model=summary_model, trigger=("messages", 12), keep=("messages", 4)),
        ContextEditingMiddleware(edits=[ClearToolUsesEdit(trigger=3000, keep=2)]),

        # —— 自定义业务规则 ——
        ExportGovernanceMiddleware(role_limits={...}),

        # —— 动态提示词（最后，覆盖 System Prompt） ——
        adaptive_prompt,
    ],
    checkpointer=SqliteSaver.from_conn_string("checkpoints.db"),
    state_schema=ExportAgentState,
    context_schema=RunContext,
    system_prompt="你是报表导出平台的智能助手。",  # 兜底 prompt，会被 @dynamic_prompt 覆盖
)

Middleware 顺序很重要。上面的栈里，脱敏在最前面（确保脏数据不进入后续环节），动态提示词在最后面（覆盖最终的 System Prompt）。调换顺序可能导致行为不一致，上线前务必在测试环境验证整条 Middleware 链的执行顺序。

3.+LangGraph实战：Middleware.md 55 KB Geçmiş Ham

前言

一、架构认知：Middleware 是什么，怎么运转

1.1 为什么 Agent 需要被"管"

1.2 第一段 Middleware：从零到能观察到 Agent 在做什么

1.3 深入 request 与 handler

1.3.1 工具级：ToolCallRequest

1.3.2 模型级：ModelRequest

1.4 深入 state 与 runtime

1.4.1 state：Agent 的内部状态

1.4.2 runtime：外部注入的上下文

二、安全风控：给 Agent 装上刹车和护栏

2.1 调用次数限制：防住"无穷循环"

ModelCallLimitMiddleware：限制模型调用次数

ToolCallLimitMiddleware：限制特定工具的调用次数

run_limit vs thread_limit

exit_behavior 的三种模式

2.2 关键操作人工确认：Human-in-the-Loop

实战：为危险操作加上确认环节

三种审批操作

2.3 敏感信息自动脱敏

内置检测类型 + 自定义正则

四种脱敏策略对比

block 策略实战：直接拒绝含 PII 的请求

不止输入：工具返回和输出也要查

三、稳定性与上下文治理：让 Agent 长期可靠运行

3.1 失败重试与模型降级

工具重试：处理间歇性超时

模型重试与降级：本地模型挂了切云端

3.2 长对话自动摘要：别让上下文吃掉你的 Token

3.3 清理旧工具结果：别让过期的数据干扰模型

摘要 vs 清理：什么时候用哪个

四、自定义进阶：写出自己的 Middleware

4.1 装饰器模式：轻量级，适合单个 Hook

4.2 类模式：可配置、可复用的 Middleware

4.3 动态提示词：让同一个 Agent 适配不同场景

附录：Middleware 选型速查表

按关注维度分类

按执行阶段分类

组合建议：一份"生产级" Middleware 栈

3.+LangGraph实战：Middleware.md 55 KB

Geçmiş Ham

1.3 深入 `request` 与 `handler`

1.3.1 工具级：`ToolCallRequest`

1.3.2 模型级：`ModelRequest`

1.4 深入 `state` 与 `runtime`

1.4.1 `state`：Agent 的内部状态

1.4.2 `runtime`：外部注入的上下文

`exit_behavior` 的三种模式