Human-in-the-loop 如何拯救智能体的骚操作？

今天继续讨论 LangGraph 的实践，这次我们来聊聊 human-in-the-loop。毕竟智能体再智能，也难免会出 “骚操作”—— 比如瞎编数据、做超出权限的决策，而 human-in-the-loop 就是给智能体装个 “刹车”，关键时刻让人类接手决策。

这篇文章里，我们会从基础概念聊起，搞清楚它和多轮对话到底有啥区别，然后通过 LangGraph 实现这个功能，看看它在咱们 TinyCodeBase 项目里的实战用法。

什么是 human-in-the-loop？

Human-in-the-Loop（后面我们统一用 人类在环 这个术语）的核心思想，就是让机器在执行任务的过程中，主动与人类协作，而不是一条道走到黑。

它不是等 AI 跑完所有流程我们再给个“好评”或“差评”，而是在高风险、易出错的关键节点主动暂停，由人类来确认决策。只有得到人类的“绿灯”后，AI 才能继续执行后续流程。

编码助手里的“人类在环”，你每天都在用

其实，很多程序员朋友现在离不开的编码助手（比如 Cursor、Claude Code），就内置了非常典型的 人类在环 逻辑。

就拿“生成并执行 Shell 命令”这个功能来说，它从不会“自作主张”直接操作你的电脑，而是会乖乖停下来，等你拍板：

你可以选择执行一次、直接执行且不再询问、拒绝执行。

Cursor 人类在环 示例

Claude Code 人类在环 示例

当然，你也可以选择关掉这个安全锁，让编码助手自动执行所有命令，那你也需要承担一定的风险，理论上编码助手可以做出任何意想不到的操作。

比如 Claude Code 提供了 --dangerously-skip-permissions 参数。但官方也给出了严肃的警告：

风险，不言而喻。

和多轮对话有什么区别？

我刚开始经常把 人类在环 和 多轮对话 搞混，毕竟看起来都是人类对 AI 进行了一系列的干涉嘛。但它们的核心区别在于控制权：

• 多轮对话：AI 为了更好地理解你的意图而追问，控制权在 AI。
  • 例子：你问“明天能去公园吗？”，AI 追问“你在哪个城市？”，目的是为了完善信息。
• 人类在环：AI 在执行一个关键动作前停下等你决策，控制权在人类。
  • 例子：AI 准备帮你购买公园门票，它会停下来问“确认支付 50 元吗？”，目的是为了确认决策。

人类在环所涉及的权限往往更加敏感，出错的代价也更大。多轮对话出错，最多是信息不准，人类在环出错，可能是系统崩溃或钱包被掏空。

️ 在 LangGraph 中实现人类在环

好，理论说完了，上代码！我们来看看如何在 LangGraph 中实现 人类在环。

LangGraph 提供了一个关键函数 Interrup() ，它可以在图（Graph）的任何节点上暂停流程，等待外部（也就是我们人类）的输入。

官方教程的例子是把“请求人类帮助”作为一个工具，但我们来做一个更贴近实战的场景：为 Agent 提供一个搜索工具，但每次调用前都必须由我们来确认，就像 Claude Code 那样。

首先，我们定义一个需要人类确认的搜索函数：

def search_with_confirmation(query: str) -> str:
    """使用 Tavily 搜索，但需要人类确认才能执行搜索。"""
    confirmation = interrupt({"query": query})
    if not confirmation.get("approved", False):
        return "搜索被用户取消。"
    return tavily_search(query)

首先我们定义一个 search_with_confirmation 函数，它接受一个查询参数，然后通过 interrupt() 函数暂停流程，等待人类输入。

人类的输入结果会被解析，并按照 comfirmatino 的结构返回。

如果人类输入了取消，则返回搜索被用户取消。，否则继续执行搜索。

接下来我们将该工具和 Agent 结合起来，形成一个完整的流程。

# ... (省略了 State 定义和 LLM 初始化) ...
tools = [search_with_confirmation]
llm_with_tools = llm.bind_tools(tools)

def chat_node(state: State):
    result = llm_with_tools.invoke(state["messages"])
    # 确保只有一个工具调用
    assert len(result.tool_calls) <= 1
    return {"messages": [result]}

graph.add_node("chat", chat_node)
graph.add_node("tools", ToolNode(tools))

graph.add_edge(START, "chat")
graph.add_conditional_edges(
    "chat",
    tools_condition,  # Routes to "tools" or "__end__"
    {"tools": "tools", "__end__": "__end__"}
)
graph.add_edge("tools", "chat")
graph.add_edge("chat", END)

app = graph.compile(checkpointer=memory)

这一段的逻辑和之前都是差不多的，需要注意的是，LangGraph 的工具默认会并发执行，assert len(result.tool_calls) <= 1 可以确保每次只有一个工具调用。

接下来我们来写推理流程。首先想大模型发送用户请求，在大模型调用工具前会通过 interrupt() 函数暂停流程，等待人类输入。

然后等待人类输入，如果人类输入了同意，则继续执行流程，工具会返回给大模型搜索后的结果，如果拒绝则返回搜索被用户取消。

大模型通过返回的内容再进行推理，最终给出答案。

user_input = "请帮我搜索一下 LangGraph 的最新功能和使用教程"
config = {"configurable": {"thread_id": "search_demo"}}

# 第一次执行：AI 会分析用户请求并准备调用搜索工具
events = list(app.stream(
    {"messages": [{"role": "user", "content": user_input}]},
    config,
    stream_mode="values",
))

# 等待用户真实输入
while True:
    user_input_confirm = input("请输入 (y/n 或 是/否): ").strip().lower()
    if user_input_confirm in ['y', 'yes', '是', '同意']:
        approved = True
        break
    else:
        approved = False
        break

human_command = Command(resume={"approved": approved})
# 第二次执行：使用人类批准的结果
events = app.stream(human_command, config, stream_mode="values")

这段代码需要解释下，在调用 interrupt() 函数后，后续的逻辑就都被挂起不执行了，如果我们要恢复执行，必须传入 Command 对象给智能体，来注入人类输入的结果。

同时我们也能发现，之前大模型推理都是调用 app.invoke() 函数，现在则变成了 app.stream() 。

我们简单列一下两种执行方式的区别：

app.invoke (同步执行)：

• 一次性执行整个工作流程
• 等待所有步骤完成后返回最终结果
• 适合需要获取完整结果的场景

app.stream (流式执行)：

• 逐步返回工作流程中的每个状态变化
• 可以实时观察执行过程
• 适合需要实时反馈或交互的场景

这个差异也是挺重要的，因为只有 app.stream() 支持 interrupt 函数的调用和流程恢复，大家在写代码的时候一定要注意。

最后我们给出代码执行结果的示例：

============================================================
第一阶段：AI 分析用户请求并准备搜索...
============================================================
 用户: 请帮我搜索一下 LangGraph 的最新功能和使用教程
 AI: 用户需要搜索LangGraph的最新功能和使用教程，调用search_with_confirmation函数进行搜索。
 AI 准备调用工具: [{'name': 'search_with_confirmation', 'args': {'query': 'LangGraph的最新功能和使用教程'}, 'id': 'call_9v1by8xpwx9zwnycg7w1okzq', 'type': 'tool_call'}]
============================================================
第二阶段：等待用户确认搜索...
============================================================
 AI 想要搜索: 'LangGraph 的最新功能和使用教程'
是否允许执行这次搜索？
 用户决定: 同意搜索
============================================================
第三阶段：执行搜索并返回结果...
============================================================
 AI: 用户需要搜索LangGraph的最新功能和使用教程，调用search_with_confirmation函数进行搜索。
 AI: 搜索结果：
标题: 【2025最新】LangGraph从入门到精通：手把手构建AI智能体的终极 ...
链接: xxxxx
摘要: LangGraph from langgraph.graph import StateGraph, START, END response = llm.invoke(state["messages"]) graph_builder.add_node("chatbot", chatbot_node) graph_builder.add_edge(START, "chatbot")  # 从入口节点开...
--------------------------------------------------
标题: LangGraph 官方文档翻译1 - 快速入门及示例教程（聊天、工具、记忆
链接: xxxxx
摘要: May 16, 2025·本指南将向您展示如何设置和使用LangGraph 提供的预构建、可复用组件，这些组件旨在帮助您快速可靠地构建智能代理系统。 前提条件. 在开始本教程前，请确保...
--------------------------------------------------
标题: Langgraph 的教程，任何来源都会有帮助。 : r/LangChain - Reddit
链接: xxxxx
摘要: Langgraph 的教程，任何来源都会有帮助。 : r/LangChain Skip to main contentLanggraph 的教程，任何来源都会有帮助。 : r/LangChain Open menu Open navigationGo to Reddit Home r/LangChain A chip A close button Log InLog in to Reddit Ex...
--------------------------------------------------
 AI: 为你找到一些关于LangGraph的最新功能和使用教程的相关信息：
- 【2025最新】LangGraph从入门到精通：手把手构建AI智能体的终极 ...
- LangGraph官方文档翻译1 - 快速入门及示例教程（聊天、工具、记忆
- Langgraph的教程，任何来源都会有帮助。

这些链接可能包含你需要的详细信息，你可以点击链接进一步查看。 

完整的可运行代码示例，已经放在 TinyCodeBase 仓库的 agent_langgraph_human_in_the_loop.py 文件中，欢迎大家去探索！

项目链接： (github.com/codemilesto…)

总结

今天我们不仅理清了 人类在环 的概念，还亲手通过 LangGraph 实现了它。

人类在环是智能体中非常必要的功能，它可以让智能体在关键节点停下来，等待人类决策，从而避免造成意想不到的损失。

希望这篇教程能帮助你更好地理解人类在环的概念，并应用到你的实际工作中。

参考资料

[1] LangGraph文档 (langchain-ai.github.io/langgraph/t…)