第 8 章：人在回路（HITL）

对有外部副作用的工具使用 HumanInTheLoopMiddleware。它在模型提出匹配工具调用后生成 LangGraph interrupt；调用方展示 action，再以 Command(resume={"decisions": [...]}) 恢复。必须配 checkpointer，否则没有可恢复状态。

agent = create_agent(
    model, tools=[send_message], checkpointer=InMemorySaver(),
    middleware=[HumanInTheLoopMiddleware({"send_message": True})],
)
first = await agent.ainvoke(input, config=config)
resumed = await agent.ainvoke(
    Command(resume={"decisions": [{"type": "approve"}]}), config=config
)

interrupt_on 是 {工具名: 配置}：True 允许 approve/edit/reject/respond；也可指定 allowed_decisions 和动态 when。每个 interrupt 中可能有多个 action，decisions 数量与顺序必须匹配。approve 执行原调用；edit 执行编辑后的参数；reject 生成错误 ToolMessage；respond 由人工直接提供工具结果。

本地示例 12_hitl.py 只调用内存中的 draft_notice，不会发送网络消息。它先断言出现 interrupt，再批准，最后确认工具确实执行。生产实现还要将审批人、时间、原参数、编辑后参数和结果记入审计系统，并在工具本身再次授权。

相关资源

查看示例代码：docs/langchain/examples/12_hitl.py

"""Chapter 8: pause for local approval before executing an in-memory action."""

import asyncio

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_core.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

from open_deep_research.configuration import Configuration


load_dotenv()

EXECUTED = []


@tool
def draft_notice(text: str) -> str:
    """Record a notice draft in local memory. This does not send anything."""
    EXECUTED.append(text)
    return "草稿已记录，未发送。"


async def main() -> None:
    settings = Configuration.from_env()
    agent = create_agent(
        init_chat_model(model=settings.research_model, max_tokens=80),
        tools=[draft_notice],
        checkpointer=InMemorySaver(),
        middleware=[HumanInTheLoopMiddleware({"draft_notice": True})],
        system_prompt="必须调用 draft_notice 记录用户提供的草稿。",
    )
    config = {"configurable": {"thread_id": "lesson-hitl"}}
    first = await agent.ainvoke(
        {"messages": [("user", "调用 draft_notice 记录：明天上午十点开会。 ")]}, config=config
    )
    assert first.get("__interrupt__"), first
    final = await agent.ainvoke(
        Command(resume={"decisions": [{"type": "approve"}]}), config=config
    )
    assert EXECUTED and "明天" in EXECUTED[0]
    print(f"审批后结果: {final['messages'][-1].content}")


if __name__ == "__main__":
    asyncio.run(main())