【万字长文】LangChain MiddleWare深度实战：手把手打造可控智能体！

简介

本文详解LangChain MiddleWare机制，包括六种插入点和两种风格(Node/Wrap)。重点介绍SummarizationMiddleware(对话历史摘要压缩)和HumanInTheLoopMiddleware(工具调用前人工审核)两种内置中间件，通过代码示例展示approve、reject、edit三种决策路径，帮助开发者构建更安全、可控、灵活的智能体系统。

简介

前情回顾

其实我们就已经讲述了关于 ReAct 的基本框架。那因为这节课要讲的 MiddleWare （中间键）和 ReAct 息息相关，所以这里我们先再复习一下具体的概念：

ReAct(Reasoning + Acting)，它是一种 LangChain 默认采用的智能体（agent）执行框架，其执行思路是如下图一样，模型进行推理（Reasoning），然后决定是否调用工具（Acting），这个过程会重复多轮，直到模型觉得可以直接给出最终回答。

整个循环遵循以下步骤：

1. 接收用户输入（messages）
2. 调用模型 → 生成回应（包括是否使用工具）
3. 如果回应中包含 Tool Call → 执行 Tool
4. Tool 返回结果后，再次调用模型进行下一轮
5. 如果模型最终给出 AI 回答，不再调用 Tool → 循环结束

什么是 MiddleWare ？

Middleware 是 LangChain 引入的一种新机制，我们可以想象为是一个🪝“钩子系统（hook system）”，可以在智能体每一步执行前后插入自定义逻辑。

那刚刚我们提到了 ReAct 的执行循环，那这个钩子系统其实就可以插入到这些关键节点的前后，比如：

其实这类似于 LangChain 之前的回调机制，去了解运行过程中的具体信息并打印出来。但是 Middleware 不只是查看信息，还可以做：

• 修改：prompt 改写、替换工具结果、过滤信息等；
• 控制：中断流程、添加条件分支、fallback 等；

所以这使得你可以实现更“可控的智能体”，如：

• ❌ 如果检测到有敏感词 → 阻止模型回答；
• 💡 如果某工具失败 → 自动尝试备选工具；
• 🛑 模型调用太多 → 中止 agent 执行。

除此之外， LangChain 中的 MiddleWare 还支持以下几个特性：

• 多个中间件串联组合执行（按顺序从上往下执行）；
• 配置参数细粒度控制行为；
• 可通过类/函数两种方式定义；
• 内置中间件（LangChain 官方预设好的模版）。

那下面我们就来看看具体的使用吧！

前期准备

代码准备

首先我们需要把我们上节课的代码搬过来使用：

from langchain_community.chat_models import ChatTongyiimport osllm = ChatTongyi(api_key=os.environ.get("DASHSCOPE_API_KEY"), model="qwen-max")from langchain_community.agent_toolkits.load_tools import load_toolstools = load_tools(["arxiv"])from langgraph.checkpoint.memory import InMemorySaver memory = InMemorySaver()from langchain.agents import create_agentagent = create_agent(model=llm, tools=tools, system_prompt="You are a helpful assistant", checkpointer=memory)result1 = agent.invoke({"messages": [{"role": "user", "content": "请使用 arxiv 工具查询论文编号 1605.08386"}]}, config={"configurable": {"thread_id": "user_1"}})print(result1["messages"][-1].content)

后续的内容我们都会在这个代码的基础上去添加相关的一些中间键。

环境准备

除此之外，我们也要和上节课一样配置好对应的环境信息：

pip install -U langchain langgraph langchain-community dashscope arxiv

然后也是到阿里云获取一下 API_Key，并且最好将其设置到环境变量之中（DASHSCOPE_API_KEY）。

那获取到了密钥后，只要能够成功运行以下代码就代表以准备完成：

from langchain_community.chat_models import ChatTongyifrom langchain_core.messages import HumanMessageimport os# 初始化模型llm = ChatTongyi(api_key=os.environ.get("DASHSCOPE_API_KEY"))# 发送消息response = llm.invoke([HumanMessage(content="你好，请用一句话介绍一下你自己。")])print(response.content)

准备好后，我们就可以来开始正式的实战环节了。

MiddleWare 的插入点

插入点详解

那在正式开始讲解 MiddleWare 使用方式之前，我们先重点来看看在 LangChain 里具体有哪些的插入点以及对应的作用是什么。

首先，根据官网对 MiddleWare 介绍的图我们就可以看到总体来说 MiddleWare 的插入有六个位置：

我们可以把这几个 Hook 类型的信息整理一下。那这里有个比较特殊的就是dynamic_prompt这个其实类似于wrap_model_call，我们到时候就一起来进行讲解：

针对所有自学遇到困难的同学们，我帮大家系统梳理大模型学习脉络，将这份LLM大模型资料分享出来：包括LLM大模型书籍、640套大模型行业报告、LLM大模型学习视频、LLM大模型学习路线、开源大模型学习教程等, 😝有需要的小伙伴，可以扫描下方二维码领取🆓↓↓↓

MiddleWare 风格详解

那在表格中，我们可以看到所有的 Hook 整体被总结为两种风格，一种是 Node-style（“前后插针”），这种适用于轻量级的 “观察 / 修改 / 插入”，像是提醒、验证、统计。比如：

def before_model(state, runtime) -> Optional[dict]: ...

我们可以返回一个 字典（dict） 更新 state，或包含jump_to控制流程。

另外一种是 Wrap-style（“整个包起来”），这种类似中间件函数，把整个过程包住，有机会拦截、替代、重试。所以这些 Hook 的形式是：

def wrap_model_call(request, handler) -> ModelResponse: try: return handler(request) # 调用原始模型 except: return ModelResponse(...) # 替代或终止

Node-style Hooks 的输入格式

在前面的示例代码里，我们能够看到 Node-style Hooks 函数里的输入其实是和 Wrap-style Hooks 里的输入是不一样的，那下面我们就拆开详细来看看具体这里的参数含义吧！

首先我们看看 Node-style Hooks：如before_model,after_model：

def before_model( state: AgentState, runtime: Runtime) -> dict[str, Any] | None: ...

这里的输入参数是：

Node-style Hooks 的输出格式

这里的返回值有两种可能：

• None→ 不改变状态，继续执行
• dict→ 更新状态，或跳转执行（包含jump_to字段）

比如：

return { "messages": [AIMessage("我不能回答这个问题。")], "jump_to": "end"}

Wrap-style Hooks 的输入格式

然后我们来看看 Wrap-style Hooks：如wrap_model_call,wrap_tool_call的输入和输出信息：

def wrap_model_call( request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse],) -> ModelResponse: ...

输入的参数有两个，一个是 request，另一个是 handler：

Wrap-style Hooks 的输出格式

那对于返回值而言，我们需要返回的是ModelResponse（wrap_model_call）或ToolMessage | Command（wrap_tool_call）。

当然我们也可以：

• 完全跳过 handler（阻止执行）
• 包裹 handler（重试、打印、缓存等）
• 替代返回结果（模拟/替代输出）

Middleware 的添加方式与执行顺序

那我们如何在原有的代码上添加中间键呢？那其实在create_agent中专门有一个参数去承接这些中间键，并且也就叫的 middleware ：

from langchain.agents import create_agentfrom langchain.agents.middleware import before_model, wrap_model_call, after_modelagent = create_agent( model="openai:gpt-4o", tools=[...], middleware=[ before_model_hook_1, wrap_model_call_hook_2, after_model_hook_1, ],)

所有的 middleware 都是按列表顺序执行的，无论是装饰器函数、内置类，还是自定义类，都可以混合放进列表。比如上面这个代码就会以下面的形式来展开：

→ Before M1⚙️ Wrapping M2 (start) [模型调用]⚙️ Wrapping M2 (end)← After M3

对于 before 类的中间键而言，其作用的时机是在传入之间完成。那假如同时有多个 before_model 的话，就看其在列表的顺序来决定，比如有[before_model_hook_1, before_model_hook_2]的话，那就是先执行完 1 再去执行 2 这样。对于其他的也是一样的原理。

而对于 after 类的中间键，那就是在模型调用后去执行。最特殊的其实属于 Wrapping 类的中间键，其执行是在模型或者工具调用的过程中，相当于我们可以把执行的结果进行进一步的处理，所以假如要对模型的输出进行精细化调整的话，通过都是要用 Wrapping 类的。

内置中间件详解

讲解完基本的原理以后，我们来看看 LangChain 已经给我们准备了哪些中间键！

SummarizationMiddleware：对话历史摘要中间件

简介

从名字其实我们就可以简单的看出，这个中间键主要作用在输入模型前的提示词上的。我们都知道 Agent 在与工作以及大模型的交互过程中会产生大量的对话记录，那这些对话记录的话我们不可能一五一十的都保存起来，因为可能会：

• ❌ 超出模型 token 上限（比如 GPT-4o 的 128k）
• ❌ 上下文越来越混乱，模型“找不到重点”
• ❌ 成本飙升

所以这个时候，这个中间件自动帮你处理：

• 检测是否超过 token 阈值（比如 4000）
• 把历史消息进行摘要压缩
• 保留最新若干条消息（比如 20 条）
• 让模型“继续记得发生了什么”，但只用更少的 token 表达

在官方文档中，这张图非常直观地展示了SummarizationMiddleware的工作原理：

当对话上下文的长度超过设定的阈值时，中间件会自动将较早的对话历史交给大模型进行摘要生成。

随后，它会将这条摘要消息、最近一轮的完整对话（即上一轮的 Human + AI message）、以及当前用户的新问题共同组成新的上下文，重新交由模型进行回复。

这种机制的好处在于：

• ✅ 一方面，保留了上一轮的上下文信息，确保连续对话不会“断片”；
• ✅ 另一方面，又能有效控制上下文长度，避免 token 溢出或成本增加。

那我们假如要使用这个中间键的话，首先我们需要从langchain.agents.middleware中先导入SummarizationMiddleware。然后将其放到create_agent的middleware参数中。

from langchain.agents import create_agentfrom langchain.agents.middleware import SummarizationMiddlewareagent = create_agent(model=llm, tools=tools, system_prompt="You are a helpful assistant", middlewares=[SummarizationMiddleware()], checkpointer=memory)

参数信息

然后呢，这个组件有很多可选的参数，包括：

使用示例

我们可以根据需要去添加对应的内容，比如对于我们现在这种基于 arXiv 的知识问答型 Agent，其有几个显著的特征：

所以这种情况下，我们可以让最多 1000 token 的内容保存在上下文中，并且每次保留三轮的对话。同时因为中文的 token 计算不太准确，所以我们这里以字符的方式计算总的 token 数量。

middleware = SummarizationMiddleware( model=llm, max_tokens_before_summary=1000, messages_to_keep=3, summary_prefix="📌摘要：", token_counter=lambda messages: sum(len(m.content) for m in messages if hasattr(m, "content")))

然后我们也可以把这个中间层放入到create_agent中：

agent = create_agent( model=llm, tools=tools, system_prompt="You are a helpful assistant", middleware=[middleware], checkpointer=InMemorySaver())

这样设置完的话，在每轮agent.invoke()后，LangChain 会执行以下流程：

↓检查当前 state["messages"] 总 token 数↓超出阈值？ 是 → 用模型生成摘要（压缩旧消息） 插入 AIMessage("📌摘要：……") 清除老消息，只保留 recent messages↓模型继续按新 messages 回复

HumanInTheLoopMiddleware：工具调用前的人工审核中间件

代码中断

其实在很多场景下，假如直接让大模型自己完成所有工作是很难的。这里最主要的原因就是大模型本质上就是一个黑盒模型，会存在幻觉的问题，很可能会做出一些人类无法理解的操作。

当在一些安全敏感的操作，比如修改数据库、发送邮件、调用金融接口等，直接让其修改的话可能会导致信息混乱的情况发生。因此在大模型调用某些特殊工具以前，先让人类进行审核还是非常有必要的。这也是HumanInTheLoopMiddleware这个中间键存在的重要原因。

该中间键其实就是让 agent 在执行工具调用之前先暂停，等待人工“批准 / 编辑 / 拒绝”。但有个点需要注意的是，这个必须要设置记忆 checkpointer，否则中断状态无法持久化。

那这个中间键有几个关键的参数：

比如对于我们这个 arxiv 工具来说，假如我们希望能够在调用前加上一个人工审核的话，我们可以这样来写：

from langchain.agents.middleware import HumanInTheLoopMiddlewaremiddleware = HumanInTheLoopMiddleware( interrupt_on={ "arxiv": { # 工具名 "allowed_decisions": ["approve", "edit", "reject"] } })

这样的话，对于 arxiv 这个工具就会在调用前拦截了。那假如我们希望拦截的时候告诉操作人员该如何判断的话，也可以加上描述的内容：

middleware = HumanInTheLoopMiddleware( interrupt_on={ "arxiv": { "allowed_decisions": ["approve", "edit", "reject"], "description": "请确认是否调用 arXiv 工具查询论文。", } }, description_prefix="🚦 工具调用正在等待人工批准")

但假如我们希望 arxiv 这个工具不要被拦截（默认情况），那我们可以这样来写：

from langchain.agents.middleware import HumanInTheLoopMiddlewaremiddleware = HumanInTheLoopMiddleware( interrupt_on={ "arxiv": False, })

当我们把这个中间键加入代码中并运行后：

import osfrom langchain_community.chat_models import ChatTongyifrom langchain_community.agent_toolkits.load_tools import load_toolsfrom langgraph.checkpoint.memory import InMemorySaverfrom langchain.agents import create_agentfrom langchain.agents.middleware import HumanInTheLoopMiddlewarellm = ChatTongyi(api_key=os.environ.get("DASHSCOPE_API_KEY"), model="qwen-max")tools = load_tools(["arxiv"])memory = InMemorySaver()middleware = HumanInTheLoopMiddleware( interrupt_on={ "arxiv": { "allowed_decisions": ["approve", "edit", "reject"], "description": "请确认是否调用 arXiv 工具查询论文。", } }, description_prefix="🚦 工具调用正在等待人工批准")# ✅ 构建 Agentagent = create_agent( model=llm, tools=tools, system_prompt="You are a helpful assistant", middleware=[middleware], checkpointer=InMemorySaver())result1 = agent.invoke({"messages": [{"role": "user", "content": "请使用 arxiv 工具查询论文编号 1605.08386"}]}, config={"configurable": {"thread_id": "user_1"}})print(result1)

发现其并不是我们想象的那样 LangChain 会在终端让我们输入对应的信息，而是输出以下结果：

{'messages': [HumanMessage(content='请使用 arxiv 工具查询论文编号 1605.08386', additional_kwargs={}, response_metadata={}, id='819816b7-d389-43c6-b2e3-d31b92ee3b62'), AIMessage(content='', additional_kwargs={'tool_calls': [{'function': {'arguments': '{"query": "1605.08386"}', 'name': 'arxiv'}, 'id': 'call_3e74ecf9c2034eb3a575fd', 'index': 0, 'type': 'function'}]}, response_metadata={'model_name': 'qwen-max', 'finish_reason': 'tool_calls', 'request_id': '06082536-d9a3-4288-9ce2-9666b0a8051c', 'token_usage': {'input_tokens': 241, 'output_tokens': 29, 'prompt_tokens_details': {'cached_tokens': 0}, 'total_tokens': 270}}, id='lc_run--7b835e7f-36da-4bc1-bf8d-b8d6228d9685-0', tool_calls=[{'name': 'arxiv', 'args': {'query': '1605.08386'}, 'id': 'call_3e74ecf9c2034eb3a575fd', 'type': 'tool_call'}])], '__interrupt__': [Interrupt(value={'action_requests': [{'name': 'arxiv', 'args': {'query': '1605.08386'}, 'description': '请确认是否调用 arXiv 工具查询论文。'}], 'review_configs': [{'action_name': 'arxiv', 'allowed_decisions': ['approve', 'edit', 'reject']}]}, id='78394dc41dc5dbae39f979a993247c0e')]}

我们可以看到打印出来的result1是个字典的结构：

{ 'messages': [...], # 人类和模型的前两轮消息 '__interrupt__': [Interrupt(...)] # ✅ HITL 中断信息}

在messages中有两条的信息，一条是我提出的问题（HumanMessage），另一条是 AI 给出的工具调用回复（AIMessage）。然后在__interrupt__中是打断的信息，包括调用的请求，终端的描述以及一些其他的元数据信息。这也意味着中断发生了！

但是我们如何输入我们是否同意或者是拒绝或者是修改呢，在 LangChain 的文档中其实也给出了明确的指示，我们需要手动调用agent.invoke(Command(...))来“恢复执行”。下面我们就分别来看一下这不同的执行方式是如何实现的。

Approve 同意

那首先其实我们还是要先把 Command 这个命令进行载入：

from langgraph.types import Command

然后假如我们要模拟输入这个场景，在 python 里我们可以使用 input() 来在终端输入信息进行模拟，这样我们就可以输入对应的 approve/edit/reject 信息了：

# 检查是否触发中断if"__interrupt__"in result1: print("🟠 中断已触发，人工审核中...") # ✅ 打印 description 信息 for interrupt in result1["__interrupt__"]: for req in interrupt.value["action_requests"]: print(f"\n📝 中断说明：{req.get('description', '无描述信息')}") decisions = input("请输入您的决定（approve/edit/reject）：").strip().lower() # 使用 APPROVE 批准调用 result_approve = agent.invoke( Command( resume={"decisions": [{"type": decisions}]} ), config={"configurable": {"thread_id": "user_1"}} ) # 打印回复内容 print("🟢 批准后模型回复：") print(result_approve["messages"][-1].content)

在代码里我们先看中断是否存在，假如存在的话先提供一些预设好的信息帮助用户去判断调用这个工具是否合适。然后就可以让用户进行决定的输入，并给出一系列选项。当前我们预演的是 approve 选项。假如在这个情况下就直接通过 Command 传入我们的决定，并且也把对应的记忆标识符也传入，这样就能够基于上一轮的记忆继续进行对话了。

中断已触发，人工审核中...📝 中断说明：请确认是否调用 arXiv 工具查询论文。请输入您的决定（approve/edit/reject）：approve🟢 批准后模型回复：该论文标题为 "Heat-bath random walks with Markov bases"，作者是 Caprice Stanley 和 Tobias Windisch，发布日期为 2016 年 5 月 26 日。论文摘要如下：本文研究了格点上的图，其边来自任意长度的有限允许移动集。我们证明了这些 在固定整数矩阵纤维上的图的直径可以从上方被一个常数所限制。然后，我们研究了这些图上的 热浴随机行走的混合行为。我们还提出了移动集合的显式条件，使得热浴随机行走（Glauber动力学的一种推广）在固定维度上是一个扩展器。

可以看到结果也确实如此。

reject（拒绝）

当我们输入拒绝的时候，我们不仅仅要输入reject，还有一个要输入的就是为什么我们reject，也就是下面的reject_message，这样大模型才能够知道其犯错的原因。同时假如我们需要准确的给出错误原因的回复，我们也需要把详细的内部信息打印出来才行，包括调用的具体工具以及询问的具体问题这样。

并且假如用户传入的决策超出了 approve, edit, reject 的范围，那我们可以默认将其设置为reject，这样就能够确保不出现错误了。

from langgraph.types import Command# 检查是否触发中断if"__interrupt__"in result1: print("🟠 中断已触发，人工审核中...") # 🗣 显示触发中断的用户提问 user_msg = next((m.content for m in result1["messages"] if m.type == "human"), "无") print(f"\n🧑 用户问题：{user_msg}") # 🔍 打印中断详情 for interrupt in result1["__interrupt__"]: for i, req in enumerate(interrupt.value["action_requests"]): print(f"\n🔧 工具：{req['name']}") print(f"📦 参数：{req['args']}") print(f"📝 描述：{req.get('description', '无')}") print(f"✅ 可选操作：{interrupt.value['review_configs'][i]['allowed_decisions']}") # 👤 人工决定 decisions = input("请输入您的决定（approve/edit/reject）：").strip().lower() if decisions notin ["approve", "edit", "reject"]: print("⚠️ 无效的决定，默认选择 reject。") decisions = "reject" if decisions == "reject": reject_message = input("请提供拒绝的理由：") # 🚦 发送人工决策结果 result = agent.invoke( Command( resume={ "decisions": [ { "type": decisions, "message": reject_message if decisions == "reject"elseNone } ] } ), config={"configurable": {"thread_id": "user_1"}} ) # 📤 显示模型后续回复 print("\n🟢 模型后续回复：") print(result["messages"][-1].content)

那这样输出的结果是：

🟠 中断已触发，人工审核中...🧑 用户问题：请使用 arxiv 工具查询论文编号 1605.08386🔧 工具：arxiv📦 参数：{'query': '1605.08386'}📝 描述：请确认是否调用 arXiv 工具查询论文。✅ 可选操作：['approve', 'edit', 'reject']请输入您的决定（approve/edit/reject）：reject请提供拒绝的理由：查询编号错误🟢 拒绝后模型回复：对不起，使用编号 "1605.08386" 搜索时没有找到相关的论文。请确认您输入的编号是否正确。 如果有其他问题或需要进一步的帮助，请告诉我！

我们输入 reject 并输入拒绝理由后，模型就会直接跳到最终结束的位置，并不会继续查询工具。

edit（编辑）

假如是查询编号错误的情况，可能我们不会直接去拒绝输出。正常合理的情况是我们去编辑出错的位置，然后把正确的内容传入进去，这时我们就可以用edit方式进行人工修正。

比如在 Command 中传入下面的内容：

Command( resume={ "decisions": [ { "type": "edit", "edited_action": { "name": "arxiv", # 工具名，通常保持不变 "args": {"query": "1605.08389"} # 修改后的参数 } } ] })

那在前面我们也要设置一个 input 来传入修改后的调用信息，所以该部分的代码如下：

from langgraph.types import Command# 检查是否触发中断if"__interrupt__"in result1: print("🟠 中断已触发，人工审核中...") # 🗣 显示触发中断的用户提问 user_msg = next((m.content for m in result1["messages"] if m.type == "human"), "无") print(f"\n🧑 用户问题：{user_msg}") # 🔍 打印中断详情 for interrupt in result1["__interrupt__"]: for i, req in enumerate(interrupt.value["action_requests"]): print(f"\n🔧 工具：{req['name']}") print(f"📦 参数：{req['args']}") print(f"📝 描述：{req.get('description', '无')}") print(f"✅ 可选操作：{interrupt.value['review_configs'][i]['allowed_decisions']}") # 👤 人工决定 decisions = input("请输入您的决定（approve/edit/reject）：").strip().lower() if decisions notin ["approve", "edit", "reject"]: print("⚠️ 无效的决定，默认选择 reject。") decisions = "reject" if decisions == "reject": reject_message = input("请提供拒绝的理由：") elif decisions == "edit": query = input("请输入正确的查询命令（query）：") # 🚦 发送人工决策结果 result = agent.invoke( Command( resume={ "decisions": [ { "type": decisions, "edited_action" : { "name": "arxiv", # 工具名，通常保持不变 "args": {"query": query} # 修改后的参数 } } ] } ), config={"configurable": {"thread_id": "user_1"}} ) # 📤 显示模型后续回复 print("\n🟢 模型后续回复：") print(result["messages"][-1].content)

运行后我们就会发现，当我们修改了调用的论文编号，查询的结果真的会发生改变！

🟠 中断已触发，人工审核中...🧑 用户问题：请使用 arxiv 工具查询论文编号 1605.08386🔧 工具：arxiv📦 参数：{'query': '1605.08386'}📝 描述：请确认是否调用 arXiv 工具查询论文。✅ 可选操作：['approve', 'edit', 'reject']请输入您的决定（approve/edit/reject）：edit请输入正确的查询命令（query）：1605.08387🟢 模型后续回复：这篇论文的标题是"Notes on frequencies and timescales in nonequilibrium Green's functions"，作者是Takaaki Ishii。论文发表于2016年6月5日。论文摘要如下：我们讨论了在具有全息对偶的强耦合理论中，非平衡格林函数的衰减行为，重点 研究了准正规模式的平衡过程。我们详细研究了Vaidya-AdS时空中的探针标量的时间分辨光谱函 数，作为对先前工作arXiv:1603.06935的补充，并使用非常非绝热温度变化的进一步数值结果。 结果表明，通过Wigner变换获得的非平衡光谱函数的松弛由最低准正规模式频率控制。背景温度 变化的时间尺度也在频率分析中被观察到。然后，我们考虑了一个受准正规模式行为启发的玩具 模型，并讨论了这些主要特征在数值结果中的简单实现。

组合三个方法

那上面三个内容其实都是分开验证三个方法的可行性的，下面我们就一起来看看怎么把三个内容组合到一起：

from langgraph.types import Command# ✅ 检查是否触发中断if"__interrupt__"in result1: print("🟠 中断已触发，人工审核中...") # 🗣 提取并显示用户提问内容 user_msg = next((m.content for m in result1["messages"] if m.type == "human"), "无") print(f"\n🧑 用户提问：{user_msg}") # 🔍 遍历中断详情，展示工具调用信息 for interrupt in result1["__interrupt__"]: for i, req in enumerate(interrupt.value["action_requests"]): print(f"\n🔧 工具名：{req['name']}") print(f"📦 参数：{req['args']}") print(f"📝 描述：{req.get('description', '无')}") print(f"✅ 可选操作：{interrupt.value['review_configs'][i]['allowed_decisions']}") # 👤 人工输入决策类型 decision_type = input("\n请输入您的决定（approve / edit / reject）：").strip().lower() if decision_type notin ["approve", "edit", "reject"]: print("⚠️ 无效输入，默认设置为 reject。") decision_type = "reject" # 🧩 构造决策对象 decision_payload = {"type": decision_type} # 若是 edit，询问新的参数 if decision_type == "edit": new_query = input("请输入新的论文编号（query）：") decision_payload["edited_action"] = { "name": "arxiv", "args": {"query": new_query} } # 若是 reject，提供反馈信息 elif decision_type == "reject": decision_payload["message"] = input("请输入拒绝理由：") # 🚀 发送人工决策结果 result2 = agent.invoke( Command(resume={"decisions": [decision_payload]}), config={"configurable": {"thread_id": "user_1"}} ) # 📤 展示最终回复 print("\n🟢 模型后续回复：") print(result2["messages"][-1].content)

那在这个方里，我们创建了一个决策信息保存的字典 decision_playload，从而帮助我们能够应对三种不同的场景。当是 edit 的场景下，那我们需要往这个字典里添加一个 edited_action 的字段，并把更新后的调用信息放入。当时 reject 的场景，得要新加一条 message 信息把拒绝的理由写入。假如是 approve 那就直接传入 decision_playoad 就好了，也就是 {“type”: “approve”} 这样一个字典。

最后这个字典就传入到 Command 中和 config 一起进行重新启动了。所以在 approve 场景我们可以看到：

🟠 中断已触发，人工审核中...🧑 用户提问：请使用 arxiv 工具查询论文编号 1605.08386🔧 工具名：arxiv📦 参数：{'query': '1605.08386'}📝 描述：请确认是否调用 arXiv 工具查询论文。✅ 可选操作：['approve', 'edit', 'reject']请输入您的决定（approve / edit / reject）：approve🟢 模型后续回复：以下是您查询的论文信息：- 发布日期：2016年5月26日- 标题：基于马尔科夫基的热浴随机游走- 作者：Caprice Stanley, Tobias Windisch- 摘要：研究了格点上的图，其边来自有限集合中的允许移动，这些移动可以是任意长度。我们 展示了在固定整数矩阵的纤维上，这些图的直径可以从上方由一个常数来限定。然后，我们研究 了这些图上的热浴随机游走的混合行为。此外，我们还提出了移动集合的具体条件，以使热浴随 机游走（Glauber动力学的一种推广）在固定维度中成为扩展器。如果您需要更多详细信息或有其他问题，请告诉我！

在 reject 场景我们可以看到：

🟠 中断已触发，人工审核中...🧑 用户提问：请使用 arxiv 工具查询论文编号 1605.08386🔧 工具名：arxiv📦 参数：{'query': '1605.08386'}📝 描述：请确认是否调用 arXiv 工具查询论文。✅ 可选操作：['approve', 'edit', 'reject']请输入您的决定（approve / edit / reject）：reject请输入拒绝理由：编号错误🟢 模型后续回复：对不起，您提供的论文编号1605.08386在arXiv.org上找不到。请确认编号是否正确，或者提供其他的搜索关键词让我帮您查找。

在 edit 场景可以看到：

🟠 中断已触发，人工审核中...🧑 用户提问：请使用 arxiv 工具查询论文编号 1605.08386🔧 工具名：arxiv📦 参数：{'query': '1605.08386'}📝 描述：请确认是否调用 arXiv 工具查询论文。✅ 可选操作：['approve', 'edit', 'reject']请输入您的决定（approve / edit / reject）：edit请输入新的论文编号（query）：1605.08386🟢 模型后续回复：这篇论文的标题是 "Heat-bath random walks with Markov bases"，作者为 Caprice Stanley 和 Tobias Windisch，发布于 2016 年 5 月 26 日。论文摘要如下：该研究探讨了格点上的图，其边来自具有任意长度的有限允许移动集。研究表明，对于给定整数 矩阵的纤维上这些图的直径可以从上方限定为一个常数。然后，研究了在这些图上的热浴随机游 走的混合行为。还明确提出了移动集合的具体条件，以便热浴随机游走（Glauber动力学的一种推广）在固定维度下成为扩展器。

这样我们就实现了在特定工具下的人工介入了！

总结

本节课我们系统地介绍了Middleware（中间件）在 LangChain 中的概念与核心机制。通过回顾 ReAct 框架，我们了解到智能体在执行过程中会经历“推理—调用工具—再推理”的循环，而 Middleware 的出现正是为了让开发者能够在这个循环的任意阶段插入自定义逻辑，实现对 Agent 执行流程的精准控制。

我们重点学习了两种内置中间件的使用方式：

• SummarizationMiddleware：

  通过自动摘要压缩长对话历史，帮助我们控制上下文长度，避免 token 溢出，同时保持对话的连贯性和上下文记忆。

• HumanInTheLoopMiddleware：

  让人类在模型执行关键操作（如调用外部工具）前进行人工审核与决策，实现“人机协同”的智能体控制。我们还演示了三种典型的决策路径——approve（批准）、reject（拒绝）与edit（修改），并实现了完整的交互式审批流程。

通过这些示例，我们可以看到 Middleware 的强大之处：它不仅能让模型更安全可控，还能让智能体的行为更加灵活、稳健、可解释。

如何学习AI大模型？

大模型时代，火爆出圈的LLM大模型让程序员们开始重新评估自己的本领。 “AI会取代那些行业？”“谁的饭碗又将不保了？”等问题热议不断。

不如成为「掌握AI工具的技术人」，毕竟AI时代，谁先尝试，谁就能占得先机！

想正式转到一些新兴的 AI 行业，不仅需要系统的学习AI大模型。同时也要跟已有的技能结合，辅助编程提效，或上手实操应用，增加自己的职场竞争力。

但是LLM相关的内容很多，现在网上的老课程老教材关于LLM又太少。所以现在小白入门就只能靠自学，学习成本和门槛很高

那么针对所有自学遇到困难的同学们，我帮大家系统梳理大模型学习脉络，将这份LLM大模型资料分享出来：包括LLM大模型书籍、640套大模型行业报告、LLM大模型学习视频、LLM大模型学习路线、开源大模型学习教程等, 😝有需要的小伙伴，可以扫描下方二维码领取🆓↓↓↓

学习路线

第一阶段： 从大模型系统设计入手，讲解大模型的主要方法；

第二阶段： 在通过大模型提示词工程从Prompts角度入手更好发挥模型的作用；

第三阶段： 大模型平台应用开发借助阿里云PAI平台构建电商领域虚拟试衣系统；

第四阶段： 大模型知识库应用开发以LangChain框架为例，构建物流行业咨询智能问答系统；

第五阶段： 大模型微调开发借助以大健康、新零售、新媒体领域构建适合当前领域大模型；

第六阶段： 以SD多模态大模型为主，搭建了文生图小程序案例；

第七阶段： 以大模型平台应用与开发为主，通过星火大模型，文心大模型等成熟大模型构建大模型行业应用。

👉学会后的收获：👈

• 基于大模型全栈工程实现（前端、后端、产品经理、设计、数据分析等），通过这门课可获得不同能力；

• 能够利用大模型解决相关实际项目需求： 大数据时代，越来越多的企业和机构需要处理海量数据，利用大模型技术可以更好地处理这些数据，提高数据分析和决策的准确性。因此，掌握大模型应用开发技能，可以让程序员更好地应对实际项目需求；

• 基于大模型和企业数据AI应用开发，实现大模型理论、掌握GPU算力、硬件、LangChain开发框架和项目实战技能， 学会Fine-tuning垂直训练大模型（数据准备、数据蒸馏、大模型部署）一站式掌握；

• 能够完成时下热门大模型垂直领域模型训练能力，提高程序员的编码能力： 大模型应用开发需要掌握机器学习算法、深度学习框架等技术，这些技术的掌握可以提高程序员的编码能力和分析能力，让程序员更加熟练地编写高质量的代码。

1.AI大模型学习路线图
2.100套AI大模型商业化落地方案
3.100集大模型视频教程
4.200本大模型PDF书籍
5.LLM面试题合集
6.AI产品经理资源合集

👉获取方式：
😝有需要的小伙伴，可以保存图片到wx扫描二v码免费领取【保证100%免费】🆓