🎯 课程主题

HumanInTheLoopMiddleware（人工审核中间件/人在环）——在Agent调用工具前中断执行，由人工决定 approve/reject/edit 后再放行。

📝 核心知识点

1. 人与回环（Human-in-the-loop）机制

  • 概念说明：Agent 调用工具时可能形成「模型→工具→模型」的循环过程。人在环中间件在这个循环中插入人工审核节点，由用户决定工具是否执行、如何执行。
  • 关键细节：在工具调用前中断 Agent 运行，等待用户的决策（approve/拒绝/reject/编辑/edit），然后再继续执行。

2. 三种决策行为

  • 概念说明：用户对中断的工具调用可以有三种操作。
  • 关键细节：
    • approve：同意工具按原参数执行。
    • reject：拒绝该工具调用。
    • edit：编辑工具的参数后再执行（例如将查询"北京"的天气改成"上海"）。

3. 核心参数

  • 概念说明：
    • interrupt_on：指定对哪些工具中断以及中断策略。
    • description_prefix：统一的中断描述信息前缀（默认值存在，可自定义）。
  • 关键细节：
    • interrupt_on 的三种赋值方式：
      • True：对该工具中断，允许 approve/reject/edit 全部三种操作。
      • False：对该工具不中断，直接放行。
      • InterruptOnConfig 对象（字典）：精细控制，内含两个字段：
        • allowed_decisions：只允许指定的决策类型，如 ["approve", "reject"] 禁止编辑。
        • description：针对该工具的单独中断描述文本，优先级高于 description_prefix。

4. 恢复执行的机制（Command / resume）

  • 概念说明：中断后需要将用户的决策传回 Agent 继续执行，使用 LangGraph 的 Command(resume=...) 机制。
  • 关键细节：
    • 通过 response 中的 interrupts 字段获取 action_requests 列表。
    • 遍历 action_requests，对每个被中断的工具构造对应的 decision 字典加入列表。
    • decision 结构：{"action_request": action_request_obj, "decision": "approve" | "reject" | edit_dict}
    • edit 的 decision 结构：{"type": "edit", "edited_action": {"name": "get_weather", "args": {"city": "上海市", "is_forecast": True}}}

5. 短期记忆（Checkpointer）的引入

  • 概念说明：中断后恢复执行需要 Agent 保持同一个会话上下文，因此必须使用 checkpointer（短期记忆）。
  • 关键细节：
    • 使用 InMemorySaver() 开启短期记忆。
    • 通过 config = {"configurable": {"thread_id": "1"}} 指定会话标识，中断前后线程 ID 需保持一致。

🏗️ 架构与工作流

用户消息 → HumanInTheLoopMiddleware 拦截 → 等待用户决策 → 用户放行 → 工具执行 → 工具结果 → 模型总结

Middleware 位于 create_agent 的 middleware 列表中，在 Agent 运行时，模型生成工具调用的 AI Message 后、实际执行工具前生效。仅工具列表中 interrupt_on 不为 False 的工具会被拦截。

💻 代码实战

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langgraph.checkpoint.memory import InMemorySaver

# 模型初始化（略）

# 工具定义（示例）
# get_weather(city: str) / get_news() / read_email_to() / send_email_to()

# === 过程一：创建 Agent 并中断 ===
agent = create_agent(
    model,
    tools=[get_weather, get_news, read_email_to, send_email_to],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "get_weather": True,
                "get_news": True,
                "read_email_to": False,
                "send_email_to": {
                    "allowed_decisions": ["approve", "reject"],
                    "description": "发送邮件中断了",
                },
            },
            description_prefix="中断了",
        )
    ],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "1"}}

response = agent.invoke(
    {
        "messages": [
            HumanMessage(
                content="帮我查天气、查新闻、读邮件，再发一封邮件。"
            )
        ]
    },
    config=config,
)

# === 过程二：解析中断信息，构造决策 ===
interrupts = response.get("__interrupt__", [])
if interrupts:
    interrupt_value = interrupts[0]["value"]
    action_requests = interrupt_value["action_requests"]
    decisions = []

    for action_request in action_requests:
        if action_request["name"] == "get_weather":
            decisions.append({
                "action_request": action_request,
                "decision": {
                    "type": "edit",
                    "edited_action": {
                        "name": "get_weather",
                        "args": {"city": "上海市", "is_forecast": True},
                    },
                },
            })
        elif action_request["name"] == "get_news":
            decisions.append({
                "action_request": action_request,
                "decision": "approve",
            })
        elif action_request["name"] == "send_email_to":
            decisions.append({
                "action_request": action_request,
                "decision": "approve",
            })

    # === 过程三：恢复执行 ===
    from langgraph.types import Command

    resume_response = agent.invoke(
        Command(resume={"decisions": decisions}),
        config=config,
    )

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

⚠️ 常见问题与避坑指南

  • 中断恢复必须使用 checkpointer=InMemorySaver() 并保持 config 中的 thread_id 一致，否则 Agent 会开启全新会话，导致上下文丢失。
  • action_requests 从 interrupts[0]["value"]["action_requests"] 中获取而非直接从 response 拿。
  • edit 的键名是 edited_action（注意是过去式），容易写成 edit_action 导致报错。
  • allowed_decisions 拼写是 allowed_decisions 而非 alone_decisions。

💡 个人总结与延伸

HumanInTheLoopMiddleware 是 LangChain 内置中间件中交互性最强的一个，本质上是将 LangGraph 的 interrupt 机制封装成了 Agent Middleware，让「人工审核」能以声明式配置的方式融入 Agent 调用链。实际生产环境可结合前端审批 UI 实现完整的审核流程。

最近更新:
Contributors: yangwenguang