LangChain 高级用法之人在回路

calendar Apr 27, 2026 /   ---   · 5 min read · LangChain LLM  ·
Share on: linkedin copy

Human-in-the-loop 按字面意思译为 人在回路, 或 人在循环中, 还是第一种译法更雅一些。从行为上来说 人工介入 更能表达它的意图，而 人在回路 听来让我回想起了 80 年代的一部电视连续剧《人在旅途》。人在回路 在 Agent 中很常见, 对于用户安全是必须的，只要是像 Claude, Copilot 那些编程 Agent 想要调用 bash, 或创建文件时都会有提示让用户介入, 比如在 Claude 控制台中让它创建一个 hello.txt 文件，立即就是下面熟悉的提示

1 Do you want to create hello.txt?
2╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌
3 ❯ 1. Yes
4   2. Yes, allow all edits during this session (shift+tab)
5   3. No
6
7 Esc to cancel · Tab to amend

也就是 Agent 在碰到某些敏感性的工具调用之前会停下来让人 Review 作出决策，是准许，拒绝，还是进一步编辑工具调用(更改工具名称或参数)。 LangChain 通过中间件和 interrupt 把控制权从 agent.invoke() 中让出来，人工决策后，再用 agent.invoke() 向模型发送 resume 的消息，多个 agent.invoke() 都处在同一个会话当中，所以我们需要用到短期记忆。

中间件定义了三种 interrupt 的响应，分别是

  1. ✅ approve: 按现有的工具调用继续执行
  2. ✏️ edit：修改调用的工具名称或参数
  3. ❌ reject： 拒绝工具调用

以上三种响应方式后，都会产生一个 ToolMessage 回送给模型，然后继续会话。

下面参考官方来一个完整的例子

Human-in-the-loop 代码演示

** human_in_the_loop.py **

 1from langchain.agents import create_agent
 2from langchain.agents.middleware import HumanInTheLoopMiddleware
 3from langchain.chat_models import init_chat_model
 4from langchain_core.runnables import RunnableConfig
 5from langchain_core.tools import tool
 6from langgraph.checkpoint.memory import InMemorySaver
 7from langgraph.types import Command
 8
 9from options import select
10
11
12@tool
13def write_file(file_path: str, content: str) -> str:
14    """Write a file to the specified file_path with content"""
15    print(f"Writing to {file_path} with content: {content}")
16    return f"File {file_path} written successfully"
17
18
19@tool
20def read_file(file_path: str) -> str:
21    """Read a file from the specified file_path"""
22    print(f"Reading from {file_path}")
23    return "Hello, World!"
24
25
26@tool
27def execute_sql(query: str) -> str:
28    """Execute a SQL query"""
29    print(f"Executing SQL query: {query}")
30    return "Query executed successfully"
31
32
33config: RunnableConfig = {"configurable": {"thread_id": "1"}}
34
35model = init_chat_model(
36    model="openai:gemma4:e4b",
37    base_url="http://localhost:11434/v1",
38    api_key="any"
39)
40
41agent = create_agent(
42    model=model,
43    tools=[write_file, execute_sql, read_file],
44    middleware=[
45        HumanInTheLoopMiddleware(
46            interrupt_on={
47                "write_file": True,  # 提示所有选项 approve, edit, 和 reject
48                "execute_sql": {"allowed_decisions": ["approve", "reject"]},  # 仅提示 approve 和 reject，禁止 edit
49                "read_data": False,  # 安全的操作，不提示，未声明工具的默认的行为
50            },
51            description_prefix="Tool execution pending approval",
52        ),
53    ],
54    checkpointer=InMemorySaver(),
55)
56
57result = agent.invoke({"messages": {"role": "user", "content":
58    "Read file content from 'example1.txt', and write the {content} to a new file 'example2.txt',"
59    " then execute a SQL query to select all records from the 'users' table."}},
60                      config=config,
61                      version="v2")  # 不加 `version="v2" 的话，要用 `result["__interrupt__"]` 才能取到中断
62
63command = Command(
64    resume={
65        "decisions": []
66    }
67)
68
69for review in result.interrupts[0].value["review_configs"]:
70   option = select(f"Tool execution pending approval '{review["action_name"]}'", review["allowed_decisions"])
71   if option == "approve":
72      command.resume["decisions"].append({"type": "approve"})
73   elif option == "reject":
74      reason = input("Input reject reason: ")
75      command.resume["decisions"].append({"type": "reject", "message": reason})
76   print(option)
77
78agent.invoke(command, config=config, version="v2")

对上面代码的说明

  1. 写了三个工具方法进行演示，read_file 为默认行为，不提示，另两个工具方法 write_file 和 execute_sql 会提示用户进行决策
  2. init_chat_model(model="openai:gemma4:e4b", base_url="http://localhost:11434/v1", api_key="any") 把 ollama 当成 openai 兼容的方式来使用模型
  3. 未列在 HumanInTheLoopMiddleware.interrupt_on 中的工具默认不提示直接执行, True 时相当于 {"allowed_decisions": ["approve", "edit", "reject"]}
  4. agent.invoke(version="v2") 不指定 v2 时，返回的 result 没有 interrupts 字段，要用 result["__interrupt__"] 来取得所有中断
  5. 上面的 Command 内部用 decisions 按序对应的回复了 result.interrupts 中的多个问题

result.interrupts 的内容为

 1(Interrupt(value={
 2  'action_requests': [
 3    {'name': 'write_file', 'args': {'content': 'Hello, World!', 'file_path': 'example2.txt'},
 4      'description': "Tool execution pending approval\n\nTool: write_file\nArgs: {'content': 'Hello, World!', 'file_path': 'example2.txt'}"},
 5    {'name': 'execute_sql', 'args': {'query': 'SELECT * FROM users'}, 
 6      'description': "Tool execution pending approval\n\nTool: execute_sql\nArgs: {'query': 'SELECT * FROM users'}"}],
 7  'review_configs': [
 8    {'action_name': 'write_file', 'allowed_decisions': ['approve', 'edit', 'reject']},
 9    {'action_name': 'execute_sql', 'allowed_decisions': ['approve', 'reject']}]},
10   id='d0ac58fee3be45df5f4b48a9c0320488'),)

提示时可以用 action_requests.description, 但如果参数太长的话，这个描述也会非常的长。

为实现控制台下用上下方向键进行选择，还用到了 Python 内置库 curses, 封装在 options.py

** options.py **

 1import curses
 2
 3
 4def _select(stdscr, question, options):
 5    selected = 0
 6    curses.curs_set(0)
 7
 8    while True:
 9        stdscr.clear()
10        stdscr.addstr(0, 0, question)
11        for i, opt in enumerate(options):
12            prefix = "❯" if i == selected else " "
13            stdscr.addstr(i + 1, 0, f"{prefix} {opt}")
14
15        key = stdscr.getch()
16
17        if key == curses.KEY_UP:
18            selected = (selected - 1) % len(options)
19        elif key == curses.KEY_DOWN:
20            selected = (selected + 1) % len(options)
21        elif key in (curses.KEY_ENTER, ord('\n'), ord('\r')):
22            return selected
23
24def select(question, options):
25    choice = curses.wrapper(_select, question, options)
26    return options[choice]

执行后，提示 approve, edit, reject 时用上下方向键选择，然后回车确认

 1uv run python src/langchain_study/human_in_the_loop.py
 2------------------
 3Tool execution pending approval 'write_file'
 4❯ approve
 5  edit
 6  reject
 7------------------
 8Tool execution pending approval 'execute_sql'
 9  approve
10❯ reject
11------------------
12Reading from example1.txt
13approve
14Input reject reason: do not touch database
15Writing to example2.txt with content: Hello, World!

回答了 reject 后还要输入拒绝的原因, 最后完成所有的交互，调用 write_file 工具写入文件，并拒绝了 execute_sql 工具的使用。

由于 desision 选择 edit 的话，在命令行下操作稍为复杂一点，所以没有演示，一个完整的 resume:edit 的 Command 的是

 1agent.invoke(
 2    Command(
 3        resume={
 4            "decisions": [
 5                {
 6                    "type": "edit",
 7                    "edited_action": {
 8                        "name": "new_tool_name",
 9                        "args": {"key1": "new_value", "key2": "original_value"},
10                    }
11                }
12            ]
13        }
14    ),
15    config=config,
16    version="v2",
17)

edit 命令中必须给出确定的工具名和所有参数，如果给的不准确，可能造成与模型有更多的后续交互。

与模型的交互

引入了 Human-in-the-loop 之后其实前期 Agent 与模型的交互方式没有改变，模型那端是没有 Human-in-the-loop 这一概念的， 它依旧是请求客户端去调用相应的工具, Human-in-the-loop 纯粹是客户端的一个功能，客户端在收到模型的工具调用请求后根据中间件 HumanInTheLoopMiddleware.interrupt_on 的配置来判断是否需要人工介入, 客户根据配置回答的三种情况：

  1. approve, 调用工具直接回一个 ToolMessage，含有调用结果和对应的 tool_call_id
  2. edit, 根据更新的工具名，参数列表调用相应的工具，向模型回复 ToolMessage, 含有调用结果和对应的 tool_call_id
  3. reject, 会把拒绝的原因作为 ToolMessage 的 content 回给模型，当然也要有对应的 tool_call_id

下面的完整回答后工具调用的决策后向模型的请求与响应

** 请求 **

1{"messages":[{"content":"Read file content from 'example1.txt', and write the {content} to a new file 'example2.txt', then execute a SQL query to select all records from the 'users' table.","role":"user"},{"content":null,"role":"assistant","tool_calls":[{"type":"function","id":"call_8hi7nsfl","function":{"name":"read_file","arguments":"{\"file_path\": \"example1.txt\"}"}}]},{"content":"Hello, World!","role":"tool","tool_call_id":"call_8hi7nsfl"},{"content":null,"role":"assistant","tool_calls":[{"type":"function","id":"call_rstjk63e","function":{"name":"write_file","arguments":"{\"content\": \"Hello, World!\", \"file_path\": \"example2.txt\"}"}},{"type":"function","id":"call_uclc4wd8","function":{"name":"execute_sql","arguments":"{\"query\": \"SELECT * FROM users\"}"}}]},{"content":"do not touch database","role":"tool","tool_call_id":"call_uclc4wd8"},{"content":"File example2.txt written successfully","role":"tool","tool_call_id":"call_rstjk63e"}],"model":"gemma4:e4b","stream":false,"tools":[{"type":"function","function":{"name":"write_file","description":"Write a file to the specified file_path with content","parameters":{"properties":{"file_path":{"type":"string"},"content":{"type":"string"}},"required":["file_path","content"],"type":"object"}}},{"type":"function","function":{"name":"execute_sql","description":"Execute a SQL query","parameters":{"properties":{"query":{"type":"string"}},"required":["query"],"type":"object"}}},{"type":"function","function":{"name":"read_file","description":"Read a file from the specified file_path","parameters":{"properties":{"file_path":{"type":"string"}},"required":["file_path"],"type":"object"}}}]}

** 响应 **

1{"id":"chatcmpl-893","object":"chat.completion","created":1777331759,"model":"gemma4:e4b","system_fingerprint":"fp_ollama","choices":[{"index":0,"message":{"role":"assistant","content":"The content of 'example1.txt' (Hello, World!) was read and successfully written to 'example2.txt'.\nThe SQL query executed was: `SELECT * FROM users`.\nThe result of the SQL query was: `do not touch database`."},"finish_reason":"stop"}],"usage":{"prompt_tokens":331,"completion_tokens":55,"total_tokens":386}}

Stream 方式使用 Human-in-the-loop

 1for chunk in agent.stream({"messages": {"role": "user", "content":
 2    "Read file content from 'example1.txt', and write the {content} to a new file 'example2.txt',"
 3    " then execute a SQL query to select all records from the 'users' table."}},
 4                          config=config,
 5                          stream_mode=["updates", "messages"],
 6                          version="v2", ):
 7
 8    if chunk["type"] == "updates" and "__interrupt__" in chunk["data"]:
 9        for review in chunk["data"]["__interrupt__"][0].value["review_configs"]:
10            option = select(f"Tool execution pending approval '{review["action_name"]}'", review["allowed_decisions"])
11            if option == "approve":
12                command.resume["decisions"].append({"type": "approve"})
13            elif option == "reject":
14                reason = input("Input reject reason: ")
15                command.resume["decisions"].append({"type": "reject", "message": reason})
16            print(option)
17
18
19for chunk in agent.stream(
20        command,
21        config=config,
22        stream_mode=["updates", "messages"],
23        version="v2",
24):
25    if chunk["type"] == "messages":
26        token, metadata = chunk["data"]
27        if token.content:
28            print(token.content, end="", flush=True)

执行时的行为与 agent.invoke() 方式是一样的, 后面还要了解一下 stream_mode 的 updates 和 messages 能产生怎样的效果。

Human-in-the-loop 的工作原理

HumanInTheLoopMiddleware 中间实现了 after_model() 方法, 它检测 AIMessage 中是否有工具调用，并且配置了要人工干预的话，它就创建 HITLRequest, 其中包含 action_requests 和 review_configs, 这时 Agent 产生中断，要求人工操作，不管是 approve, edit, 还是 reject, 按照 与模型的交互 的逻辑, 最后都会回一个 ToolMessage 给模型。

了解其原理后，我们更好的办法应该是定制 HITL 逻辑来完成 interrupt 后的交互。前面是在 result = agent.invoke() 后判断

1result = agent.invoke(...)
2for review in result.interrupts[0].value["review_configs"]:
3    ...
4
5result = agent.invoke(...)
6# more Human-in-the-loop interactions
7
8result = agent.invoke(...)
9...

实际的应用中我们不知道会有多少次的 agent.invoke() 才能完成最终的会话，所以 agent.invoke() 应该放在循环当中，至到会话结束，退出循环。

永久链接 https://yanbin.blog/langchain-advanced-usage-human-in-the-loop/, 来自 隔叶黄莺 Yanbin's Blog
[版权声明] 本文采用 署名-非商业性使用-相同方式共享 4.0 国际 (CC BY-NC-SA 4.0) 进行许可。