LangChain 05：中间件教程

发表于 更新于 分类于
基于 LangChain 1.x 的 create_agent 中间件，覆盖摘要、限流、HITL、PII、工具重试与自定义钩子，示例对接火山方舟 Coding Plan。

前言

create_agent 默认已经能跑通「推理 → 调工具 → 再推理」循环。
一旦对话变长、工具会失败、或敏感操作需要人工确认，就不宜把逻辑塞进系统提示词里硬扛。
中间件（Middleware） 挂在 Agent 循环的钩子上，用统一方式做摘要、限流、脱敏、重试与审计，而不必手写整张 LangGraph。
本文以 LangChain 1.x 为准，侧重内置中间件与自定义钩子；Agent 基础见《LangChain 1.x Agent 教程》。
示例继续对接 火山方舟 Coding Plan，模型用 ark-code-latest。
下文每个 Python 示例都是完整可运行脚本：复制到项目根目录（与 .env 同级）后执行 uv run python xxx.py 即可。
下文需要 Python 3.12+，依赖用 uv 管理。

依赖

在项目目录声明依赖（版本请按项目实际调整）。

1
uv add "langchain>=1.0,<2.0" langchain-openai langgraph python-dotenv rich

rich 用于在终端里更清楚地查看结构化结果：彩色高亮、自动缩进嵌套 dict / list / 消息对象，比内置 print 更适合调试中间件改写后的 messages。
示例里常用 from rich import print as rprint，把原来的 print(...) 换成 rprint(...) 即可。

在项目根目录创建 .env，写入 Coding Plan 的 API Key 与专用 Base URL。
不要把 .env 提交进 Git。

1
2
OPENAI_API_KEY=你的火山方舟 API Key
OPENAI_BASE_URL=https://ark.cn-beijing.volces.com/api/coding/v3

示例脚本开头用 load_dotenv() 加载该文件。
请勿把 Base URL 写成普通方舟 .../api/v3，以免无法抵扣 Coding Plan 额度。

实现

钩子概览

中间件在 Agent 循环的固定位置插入逻辑，大致分两类。

节点式钩子在步骤前后执行，可返回状态更新，甚至跳转到结束：

  1. before_agent / after_agent：整次调用开始与结束各一次。
  2. before_model / after_model：每次模型调用前后。

包裹式钩子包住真正的模型或工具调用，可决定调用 0 次、1 次或多次（短路、重试、改写请求）：

  1. wrap_model_call：包裹每次模型调用。
  2. wrap_tool_call：包裹每次工具调用。

内置中间件多数是上述钩子的封装；自定义时可用装饰器或继承 AgentMiddleware。

挂载方式

所有中间件通过 create_agent(..., middleware=[...]) 传入，按列表顺序组合。
下面是一个最小可运行示例：不挂业务中间件，只验证 Coding Plan 与 create_agent 通路。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from rich import print as rprint

load_dotenv()

api_key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=api_key,
    base_url=base_url,
)

@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气（示例数据）。"""
    return f"{city} 今天晴，25°C。"

agent = create_agent(
    model=model,
    tools=[get_weather],
    middleware=[],
    system_prompt="你是中文助手。需要天气时调用 get_weather。",
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "上海天气怎么样？"}]
})
rprint(result["messages"][-1])

对话摘要

长对话会把上下文窗口撑满。
SummarizationMiddleware 在触发条件满足后压缩历史，并保留近期若干条消息。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.memory import InMemorySaver
from rich import print as rprint

load_dotenv()

api_key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=api_key,
    base_url=base_url,
)


@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气（示例数据）。"""
    return f"{city} 今天晴，25°C。"


agent = create_agent(
    model=model,
    tools=[get_weather],
    checkpointer=InMemorySaver(),
    middleware=[
        SummarizationMiddleware(
            model=model,
            trigger=("tokens", 4000),
            keep=("messages", 20),
        ),
    ],
    system_prompt="你是中文助手。需要天气时调用 get_weather。",
)

config: RunnableConfig = {"configurable": {"thread_id": "mw-summary-1"}}

r1 = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫小明。"}]},
    config,
)
rprint(r1["messages"][-1])

r2 = agent.invoke(
    {"messages": [{"role": "user", "content": "上海天气怎么样？"}]},
    config,
)
rprint(r2["messages"][-1])

r3 = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫什么？刚才问的是哪个城市？"}]},
    config,
)
rprint(r3["messages"][-1])

trigger 控制何时摘要，keep 控制摘要后保留多少近期消息。
也可写成多条件 OR（列表）或 AND（字典）；阈值请按模型窗口再调。
多轮记忆仍依赖 checkpointer + 稳定的 thread_id。

调用限额

防止 Agent 死循环或单次任务烧太多 token 时，用 ModelCallLimitMiddleware。
thread_limit 限制整条会话线程内的模型调用次数，run_limit 限制单次 invoke。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import ModelCallLimitMiddleware
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langchain_core.runnables.config import RunnableConfig
from langgraph.checkpoint.memory import InMemorySaver
from rich import print as rprint

load_dotenv()

api_key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=api_key,
    base_url=base_url,
)


@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气（示例数据）。"""
    return f"{city} 今天晴，25°C。"


agent = create_agent(
    model=model,
    tools=[get_weather],
    checkpointer=InMemorySaver(),
    middleware=[
        ModelCallLimitMiddleware(
            thread_limit=20,
            run_limit=6,
            exit_behavior="end",
        ),
    ],
    system_prompt="你是中文助手。需要天气时调用 get_weather。",
)

config: RunnableConfig = {"configurable": {"thread_id": "mw-limit-1"}}
result = agent.invoke(
    {"messages": [{"role": "user", "content": "查询一下杭州的天气。"}]},
    config,
)
rprint(result["messages"][-1])

线程级限额需要 checkpointer 才能跨多次 invoke 累计。
若想观察触顶行为，可把 run_limit 调到 1 或 2 后再跑需要工具调用的问题。
exit_behavior 可按文档选择结束运行或抛错等策略，请以当前 langchain 小版本为准。

人工审批

高风险工具（发邮件、写库、删文件）执行前可暂停，等人工 approve / edit / reject。
HumanInTheLoopMiddleware 必须配合 checkpointer，才能在中断后恢复状态。
下面示例先触发中断，再用 Command(resume=...) 自动批准，便于本地一次性跑通。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
import os
import uuid

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
from rich import print as rprint

load_dotenv()

api_key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=api_key,
    base_url=base_url,
)

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件（示例：真正环境请接邮件服务）。"""
    return f"已发送给 {to}：{subject}"

@tool
def read_email(email_id: str) -> str:
    """读取邮件内容（示例数据）。"""
    return f"邮件 {email_id} 的正文……"

agent = create_agent(
    model=model,
    tools=[read_email, send_email],
    checkpointer=InMemorySaver(),
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "send_email": {
                    "allowed_decisions": ["approve", "edit", "reject"],
                },
                "read_email": False,
            }
        ),
    ],
    system_prompt="你是邮件助手。用户要求发信时必须调用 send_email。",
)

config = {"configurable": {"thread_id": str(uuid.uuid4())}}

interrupted = agent.invoke(
    {
        "messages": [{
            "role": "user",
            "content": "给 demo@example.com 发一封主题为测试、正文为你好的邮件。",
        }]
    },
    config,
)
rprint(interrupted)

# 本地演示：自动批准中断中的工具调用；真实产品应接入人工 UI
resumed = agent.invoke(
    Command(resume={"decisions": [{"type": "approve"}]}),
    config,
)
rprint(resumed["messages"][-1])

interrupt_on 的键是工具名（@tool 默认用函数名）。
若你安装的小版本里 resume 字段名不同，请对照官方 Human-in-the-loop 文档微调 Command 参数。

PII 脱敏

用户输入里常有邮箱、卡号等敏感信息。
PIIMiddleware 可在进入模型前做 redact / mask / block / hash。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from rich import print as rprint

load_dotenv()

api_key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=api_key,
    base_url=base_url,
)

agent = create_agent(
    model=model,
    tools=[],
    middleware=[
        PIIMiddleware("email", strategy="redact", apply_to_input=True),
        PIIMiddleware("credit_card", strategy="mask", apply_to_input=True),
    ],
    system_prompt="你是客服助手，不要主动索要完整卡号。",
)

result = agent.invoke(
    {
        "messages": [
            HumanMessage("我的邮箱是 demo@example.com，帮我写一句礼貌的自动回复。")
        ]
    }
)
rprint(result["messages"])

也可传入自定义 detector（正则或函数）识别业务侧密钥、工号等。
流式输出侧的脱敏能力与版本有关，若需要请对照官方 PII 文档确认 apply_to_output 等参数。

工具重试

网络抖动或偶发超时适合交给 ToolRetryMiddleware，避免模型因一次失败就放弃。
下面用一个「前两次抛 ConnectionError、第三次成功」的工具，验证重试是否生效。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import ToolRetryMiddleware
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from rich import print as rprint

load_dotenv()

api_key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=api_key,
    base_url=base_url,
)

_calls = {"n": 0}

@tool
def flaky_weather(city: str) -> str:
    """查询指定城市的天气（示例：前两次故意失败）。"""
    _calls["n"] += 1
    if _calls["n"] < 3:
        raise ConnectionError(f"temporary network error (attempt {_calls['n']})")
    return f"{city} 今天晴，25°C。"

agent = create_agent(
    model=model,
    tools=[flaky_weather],
    middleware=[
        ToolRetryMiddleware(
            max_retries=3,
            backoff_factor=2.0,
            initial_delay=0.2,
            retry_on=(ConnectionError, TimeoutError),
            on_failure="continue",
        ),
    ],
    system_prompt="你是中文助手。需要天气时调用 flaky_weather。",
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "上海天气怎么样？"}]
})
rprint(result["messages"][-1])
print(f"工具实际调用次数：{_calls['n']}")

tools=[...] 可限定只对部分工具重试；on_failure 控制重试耗尽后是继续把错误交给模型，还是按其它策略处理。
也可再叠 ToolErrorMiddleware，把异常改写成模型可理解的提示。

自定义钩子

单钩子场景用装饰器最快：例如在每次模型调用前打日志，或用 wrap_model_call 做简易重试。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
import os
from typing import Any, Callable

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import (
    AgentState,
    ModelRequest,
    ModelResponse,
    before_model,
    wrap_model_call,
)
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langgraph.runtime import Runtime
from rich import print as rprint

load_dotenv()

api_key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=api_key,
    base_url=base_url,
)

@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气（示例数据）。"""
    return f"{city} 今天晴，25°C。"

@before_model
def log_before_model(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    print(f"即将调用模型，当前消息数：{len(state['messages'])}")
    return None

@wrap_model_call
def retry_model(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
    for attempt in range(3):
        try:
            return handler(request)
        except Exception as exc:
            if attempt == 2:
                raise
            print(f"模型调用失败，重试 {attempt + 1}/3：{exc}")

agent = create_agent(
    model=model,
    tools=[get_weather],
    middleware=[log_before_model, retry_model],
    system_prompt="你是中文助手。需要天气时调用 get_weather。",
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "北京天气怎么样？"}]
})
rprint(result["messages"][-1])

需要多钩子、可配置阈值或同时提供 sync / async 实现时，改继承 AgentMiddleware。
下面把消息上限设得很小，并连续多轮调用，便于本地看到提前结束。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
import os
from typing import Any

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware, AgentState
from langchain.chat_models import init_chat_model
from langchain.messages import AIMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.runtime import Runtime
from rich import print as rprint

load_dotenv()

api_key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=api_key,
    base_url=base_url,
)

class MessageLimitMiddleware(AgentMiddleware):
    def __init__(self, max_messages: int = 6):
        super().__init__()
        self.max_messages = max_messages

    def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        if len(state["messages"]) >= self.max_messages:
            return {
                "messages": [AIMessage(content="对话过长，已停止本次继续调用。")],
                "jump_to": "end",
            }
        return None

agent = create_agent(
    model=model,
    tools=[],
    checkpointer=InMemorySaver(),
    middleware=[MessageLimitMiddleware(max_messages=6)],
    system_prompt="你是中文助手，回答要短。",
)

config = {"configurable": {"thread_id": "mw-msg-limit-1"}}

for i, text in enumerate(["你好", "再聊一句", "继续", "还在吗"], start=1):
    print(f"--- 第 {i} 轮 ---")
    result = agent.invoke(
        {"messages": [{"role": "user", "content": text}]},
        config,
    )
    rprint(result["messages"][-1])

节点式钩子返回 dict 会合并进 Agent 状态；需要提前结束时可配合 jump_to（具体可用跳转目标以当前版本文档为准）。

组合使用

真实项目常把多个中间件叠在一起。
列表顺序有意义：一般把「改写输入 / 脱敏」放外层，把「重试」放更靠近实际调用的一侧；具体以官方组合说明与实测为准。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import (
    ModelCallLimitMiddleware,
    PIIMiddleware,
    SummarizationMiddleware,
    ToolRetryMiddleware,
)
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from rich import print as rprint

load_dotenv()

api_key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=api_key,
    base_url=base_url,
)

@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气（示例数据）。"""
    return f"{city} 今天晴，25°C。"

agent = create_agent(
    model=model,
    tools=[get_weather],
    checkpointer=InMemorySaver(),
    middleware=[
        PIIMiddleware("email", strategy="redact", apply_to_input=True),
        SummarizationMiddleware(
            model=model,
            trigger=("tokens", 4000),
            keep=("messages", 20),
        ),
        ModelCallLimitMiddleware(run_limit=8, exit_behavior="end"),
        ToolRetryMiddleware(max_retries=2, on_failure="continue"),
    ],
    system_prompt="你是中文助手。需要天气时调用 get_weather。",
)

config = {"configurable": {"thread_id": "mw-combo-1"}}
result = agent.invoke(
    {
        "messages": [{
            "role": "user",
            "content": "我的邮箱是 demo@example.com。上海天气怎么样？",
        }]
    },
    config,
)
rprint(result["messages"][-1])

调试时用 rprint(result) 或打印完整 messages，确认中间件是否按预期改写了输入、是否触发了摘要或限额。

验证

按下面顺序确认中间件路径可用。

  1. 执行 uv run python -c "import importlib.metadata as m; print(m.version('langchain'))"，主版本应为 1.x。
  2. 确认 .env 中 OPENAI_BASE_URL 为 Coding Plan 专用地址，并已 load_dotenv()。
  3. 逐段复制「挂载方式」「对话摘要」「调用限额」等完整脚本到 .py 文件，用 uv run python 直接运行。
  4. 跑「工具重试」脚本，终端打印的工具调用次数应大于等于 3。
  5. 跑「自定义钩子」脚本，终端应出现「即将调用模型」类日志。
  6. 跑「人工审批」脚本，应先看到中断态结果，再看到批准后的最终回复。

若 SummarizationMiddleware 的 trigger / keep 报错，对照当前小版本文档调整写法。
若 HITL 的 Command(resume=...) 报错，对照官方 Human-in-the-loop 文档核对 resume 字段结构。

扩展

中间件跑通后，可继续这些方向。

  • 动态提示词：用 @dynamic_prompt 按用户角色或路由结果切换系统提示。
  • 模型回退：内置 fallback 类中间件可在主模型失败时切换备用模型（参数以文档为准）。
  • 可观测性：接入 LangSmith，对照钩子日志与每次模型 / 工具调用。
  • 子 Agent：更复杂的委派可用 Deep Agents 等更高层封装中的 SubAgent 中间件。
  • 生产 checkpointer：用 SQLite / Postgres 替代 InMemorySaver。

总结

  1. 中间件通过 create_agent(..., middleware=[...]) 挂载，插在模型与工具循环的钩子上，而不是改写整张图。
  2. 长对话用 SummarizationMiddleware，防失控用 ModelCallLimitMiddleware，高风险工具用 HumanInTheLoopMiddleware。
  3. 输入侧敏感信息用 PIIMiddleware；偶发工具失败用 ToolRetryMiddleware。
  4. 单钩子用装饰器，多钩子或可配置逻辑用 AgentMiddleware 子类。
  5. 组合时注意顺序与 checkpointer；用 rprint 核对 messages 是否被中间件按预期改写。