中文资讯掘金 AI Tue, 28 Jul 2026 正文 ok 589 words

DeepResearchSystem 0x03：引入 HITL 人类在环机制

英文标题：DeepResearchSystem 0x03：HITL

本章为基于 LangGraph 构建的 DeepResearchSystem 增加了 HITL（Human-in-the-loop，人类在环）能力，让用户能够在 Agent 执行过程中审核和修正研究计划。系统在 `generate_plan` 之后新增计划确认环节，只有经过人工确认的计划才会进入 `web_search` 等后续步骤。文章对比了将人工确认伪装为普通节点的轮询方案，以及使用 LangGraph 原生 `interrupt()` 和 `Command(resume=...)` 实现运行时中断与恢复的方案。HITL 可提升 Agent 输出与人类意图的一致性，并特别适用于不可逆操作、证据不足或超出 Agent 授权范围的高风险决策。

打开原文相关 3 · 机会 1

3.2 KB

589 B

11.7 KB

65.2 KB

中文

回顾

此前，我们已经基于 LangGraph 实现了 DeepResearchSystem 的核心流程，也基本达到了最初预期的效果。但似乎忽略了一个问题：

我们是否过度信任 LLM？除了“出题”之外，整个流程中人类没有进行任何干预。
那么，如何保证 LLM 给出的结果符合我们的原始意图？

解决

此前学习 LangGraph 时，曾介绍过一种AI 自动处理 + 人类中断审核的 LangGraph 工作流。也可以参考 LangGraph 官网示例。这里同样基于 LangGraph 加入 HITL（Human-in-the-loop）。

顾名思义，HITL 即人类参与的循环。借助 HITL，可以修正 LLM 输出，直至其完全满足人类意图。这样一来，整个 Agent Loop 中有了 Human 的参与，输出符合人类意图的概率也会提高。

HITL

跳出这个具体 case，先思考一下：为什么 HITL 是 Agent 不可或缺的环节？

HITL 的具体作用
- Monitoring（监控）：实时监控 Agent 的行为。
- Validating（校验）：在结构化检查点中，依据业务规则验证输出。
- Intervening（干预）：当置信度未达到阈值时，进行人工介入。
- Training（训练）：通过系统化反馈训练 Agent，使其性能随时间不断提升。
HITL 的重要性
- 有些决策本来就不应由 Agent 独自完成。
  - 操作不可逆：删除数据、发送邮件、执行变更——一旦完成就很难撤回。
  - 证据不足：Agent 虽然给出了结论，但证据链不完整，或无法区分多个可能的根因。
  - 超出授权范围：Agent 被授权进行诊断，但并未被授权执行变更。

Coding

废话少说，直接上代码。这里对之前的流程做了一项优化：

优化前：用户输入 → 查询规划 → 并行搜索 → 反思评估 → 迭代深化 → 报告生成

优化后：用户输入 → 研究计划 → 查询规划 → 并行搜索 → 反思评估 → 迭代深化 → 报告生成

当前新增的节点位于 generate_plan（研究主题拆解）之后，用于评估拆解计划是否需要人类确认。只有经人类确认的计划，才能进入下一步（web_search）。

先看一下加入 HITL 后的流程图：

HITL 版流程图

OverallState

首先，需要在 OverallState 中添加与人类反馈相关的状态。

plan: str

plan_status: str

plan_messages: Annotated[list, add_messages]

user_feedback: str

generate_plan

ef generate_plan(state: OverallState, config: RunnableConfig) -> OverallState:
    """
    生成研究计划的节点

    基于用户的问题和需求，使用LLM生成一个详细的研究计划。

    Args:
        state: 包含用户问题的当前图状态
        config: 可运行配置，包括LLM提供商设置

    Returns:
        包含计划状态更新的字典，包括plan和plan_status键
    """

    logger.info("正在生成计划...")
    if state.get("plan_status", "unconfirmed") != "unconfirmed":
        return {}

    configurable = Configuration.from_runnable_config(config)
    agent = Agent(model_id=configurable.query_generator_model)
    agent.set_step_prompt(plan_instructions)
    response = agent.step(
        current_date=get_current_date(),
        research_topic=get_research_topic(state["messages"], [msg.content for msg in state["plan_messages"]]),
        research_proposal=state.get("plan", "")
    )
    response = JsonUtils.extract_pattern(response, pattern="markdown")
    logger.info(state)
    logger.info(response)

    return {"messages": [AIMessage(content=response)],
            "plan": response,
            "plan_status": "unconfirmed",
            "plan_messages": [AIMessage(content=response)]}

evaluate_plan：人类参与

def evaluate_plan(state: OverallState, config: RunnableConfig) -> ReflectionState:
    """
    评估研究计划的节点

    分析当前计划并根据用户反馈决定下一步行动：
    - 如果用户确认计划，继续生成查询
    - 如果用户需要修改计划，重新生成
    - 如果用户未确认，等待确认

    Args:
        state: 包含计划信息的当前图状态
        config: 可运行配置

    Returns:
        字符串字面量，指示下一个要访问的节点
    """
    configurable = Configuration.from_runnable_config(config)
    plan = state.get("plan", None)
    if state.get("plan_status", "unconfirmed") == "unconfirmed":
        logger.info("等待用户确认...")
        return AWAITING_PLAN_CONFIRMATION

    if not plan:
        logger.info("没有计划需要评估。")
        return SEARCH_REPLAN
    else:
        context = get_last_user_response(state["messages"])

        if "开始研究" in context or "需求确认" in context:
            return GENERATE_SEARCH_NODE

        agent = JsonAgent(model_id=configurable.query_generator_model, keys=PlanReflection)
        agent.set_step_prompt(plan_reflection_instructions)
        result = agent.step(
            research_proposal=state.get("plan", ""),
            context=context,
        )
        if result.satisfy:
            return GENERATE_SEARCH_NODE
        return SEARCH_REPLAN

HITL

方法 1：人工干预

这是一种**“把 HITL 伪装成普通节点”**的做法：

节点不退出：AWAITING_PLAN_CONFIRMATION 节点被调用后，函数立刻返回当前 state，但图执行并没有真正“结束”。
业务层忙等：前端或 API 层需要不断轮询，或通过某种机制“卡住”在这个节点，等待人类输入。
状态原地打转：evaluate_plan 不断返回 AWAITING_PLAN_CONFIRMATION，图在逻辑上持续“转圈”。

本质：这模拟了一个**“阻塞调用”**，但 LangGraph 的运行时并不知道系统正在等待人类响应。

builder.add_node(AWAITING_PLAN_CONFIRMATION, lambda state, config: state)

builder.add_node(SEARCH_REPLAN, lambda state, config: {"plan_status": "unconfirmed"})

方法 2：`interrupt` 机制

这是 LangGraph 原生支持的**“真正的 HITL”**：

运行时级中断：调用 interrupt() 时，LangGraph Runtime 会：
- 立即挂起整个图执行。
- 将当前状态持久化到 Checkpointer。
- 抛出异常或信号，通知调用方“需要人工介入”。
进程可释放：执行线程或协程可以被释放，无须占用资源空等。

显式恢复：只有通过 graph.invoke/stream 传入 Command(resume=...) 时，图才会从 interrupt 处继续执行。

def awaiting_plan_confirmation(state: OverallState):

 feedback = interrupt({
     "type": "plan_confirm",
     "current_plan":state.get("plan"),
     "prompt": "请确认当前执行计划，同意请输入confirm，拒绝请输入reject"
 })

 logger.info(f"收到人工反馈: {human_response}")

 action = feedback.get("action", "reject")
 feedback = feedback.get("feedback", "")

 if action == "confirm":
     logger.info("计划已确认，标记为 confirmed")
     return {
         "plan_status": "confirmed",
         "user_feedback": feedback,
         "messages": [AIMessage(content=f"用户确认了计划: {feedback}")]
     }
 elif action == "reject":
     logger.info("计划被拒绝，标记为 rejected")
     return {
         "plan_status": "rejected",
         "user_feedback": feedback,
         "messages": [AIMessage(content=f"用户拒绝了计划: {feedback}")]
     }
 else:

     logger.info("收到修改意见，将用于重新规划")
     return {
         "plan_status": "rejected",
         "user_feedback": feedback,
         "messages": [AIMessage(content=f"用户建议修改: {feedback}")]
     }

Command 调用示例

"""
# 1. 首次调用，触发中断
config = {"configurable": {"thread_id": "test_thread_001"}}
initial_state = {"messages": [("user", "研究一下LangGraph的HITL机制")]}
result = graph.invoke(initial_state, config)
# 此时 result 会包含 interrupt 信息，但不会完成执行

# 2. 恢复执行：用户确认计划
resume_command = Command(resume={
    "action": "confirm",
    "feedback": "计划看起来不错，开始吧"
})
result_after_resume = graph.invoke(resume_command, config)
# 此时图会从 interrupt 处继续执行

# 3. 恢复执行：用户拒绝计划
resume_command_reject = Command(resume={
    "action": "reject",
    "feedback": "请增加对生产环境稳定性的分析"
})
result_after_reject = graph.invoke(resume_command_reject, config)
# 此时图会进入 SEARCH_REPLAN 节点，重新生成计划

"""

🤔🤔🤔

目前已经在研究计划节点加入了 HITL。若有需要，同样可以在其他节点加入 HITL。具体是否需要增加，取决于实际场景和 case；增加方式相同，这里不再过多赘述。

English / 原文

回顾

之前已经基于 LangGraph 实现了 DeepRearchSystem 的核心流程，也基本能实现我们初衷想要的效果了。但是，好像我们忽略了一个问题：

我们是否过度信任 LLM 了？整个流程出了 ”出题“，人类并没有做任何干预。
那么，怎么保证 LLM 给出的就是符合我们初衷意愿的呢？

解决

之前学习 LangGraph 时有一个AI 自动处理 + 人类中断审核 的 LangGraph 工作流。也可以参看 LangGraph 官网示例。这里我们也基于 LangGraph 加入 HITL(Human-in-the-loop) 。

顾名思义，即是人类参与的循环。借助 HITL 来对 LLM 输出做修正，直到输出完全满足人类意图。这样在整个 Agent Loop中有了 Human 的参与，可以增大输出更满足人类意图的概率。

HITL

跳出这个具体 case，我们先来思考下为什么 HITL 是 Agent 必不可少的环节？

HITL 具体作用
- Monitoring（监控） ：实时监控 Agent 的行为
- Validating（校验） ：在结构化的检查点对输出进行业务规则验证
- Intervening（干预） ：当置信度阈值未达到时进行人工介入
- Training（训练） ：通过系统化的反馈训练 Agent，使其随时间推移性能不断提升
HITL 重要性
- 有些决策本来就不该由 Agent 独自做。
  - 操作不可逆 ：删数据、发邮件、执行变更——做了就很难撤回
  - 证据不足 ：Agent 给出了结论，但证据链不完整，或无法区分多个可能根因
  - 超出授权范围 ：Agent 被授权做诊断，但没有被授权做变更

Coding

废话少说上代码。我们这里对之前的流程做了一个优化：

优化前：用户输入 → 查询规划 → 并行搜索 → 反思评估 → 迭代深化 → 报告生成优化后：用户输入 → 研究计划 → 查询规划 → 并行搜索 → 反思评估 → 迭代深化 → 报告生成

目前加的节点是在 generate_plan（研究主题拆解）后，评估拆解计划是否需要人类确认，只有经人类确认的计划才能进入下一步(web_search)。

先看下加入 HITL 的流程图：

HITL版.png

OverallState

首先，我们要在 OverallState 添加有关人类反馈的状态。

plan: str

plan_status: str

plan_messages: Annotated[list, add_messages]

user_feedback: str

generate_plan

ef generate_plan(state: OverallState, config: RunnableConfig) -> OverallState:
    """
    生成研究计划的节点

    基于用户的问题和需求，使用LLM生成一个详细的研究计划。

    Args:
        state: 包含用户问题的当前图状态
        config: 可运行配置，包括LLM提供商设置

    Returns:
        包含计划状态更新的字典，包括plan和plan_status键
    """

    logger.info("正在生成计划...")
    if state.get("plan_status", "unconfirmed") != "unconfirmed":
        return {}

    configurable = Configuration.from_runnable_config(config)
    agent = Agent(model_id=configurable.query_generator_model)
    agent.set_step_prompt(plan_instructions)
    response = agent.step(
        current_date=get_current_date(),
        research_topic=get_research_topic(state["messages"], [msg.content for msg in state["plan_messages"]]),
        research_proposal=state.get("plan", "")
    )
    response = JsonUtils.extract_pattern(response, pattern="markdown")
    logger.info(state)
    logger.info(response)

    return {"messages": [AIMessage(content=response)],
            "plan": response,
            "plan_status": "unconfirmed",
            "plan_messages": [AIMessage(content=response)]}

evaluate_plan 人类参与

def evaluate_plan(state: OverallState, config: RunnableConfig) -> ReflectionState:
    """
    评估研究计划的节点

    分析当前计划并根据用户反馈决定下一步行动：
    - 如果用户确认计划，继续生成查询
    - 如果用户需要修改计划，重新生成
    - 如果用户未确认，等待确认

    Args:
        state: 包含计划信息的当前图状态
        config: 可运行配置

    Returns:
        字符串字面量，指示下一个要访问的节点
    """
    configurable = Configuration.from_runnable_config(config)
    plan = state.get("plan", None)
    if state.get("plan_status", "unconfirmed") == "unconfirmed":
        logger.info("等待用户确认...")
        return AWAITING_PLAN_CONFIRMATION

    if not plan:
        logger.info("没有计划需要评估。")
        return SEARCH_REPLAN
    else:
        context = get_last_user_response(state["messages"])
        
        if "开始研究" in context or "需求确认" in context:
            return GENERATE_SEARCH_NODE

        
        agent = JsonAgent(model_id=configurable.query_generator_model, keys=PlanReflection)
        agent.set_step_prompt(plan_reflection_instructions)
        result = agent.step(
            research_proposal=state.get("plan", ""),
            context=context,
        )
        if result.satisfy:
            return GENERATE_SEARCH_NODE
        return SEARCH_REPLAN

HITL

方法1：人工干预

这是一种 "把 HITL 伪装成一个普通节点" 的做法：

节点不退出 ：AWAITING_PLAN_CONFIRMATION 节点被调用后，函数立刻返回当前 state，但图执行并没有真正"结束"
业务层忙等 ：前端/API 层需要不断轮询或通过某种机制"卡住"在这个节点，等待人类输入
状态原地打转 ：evaluate_plan 不断返回 AWAITING_PLAN_CONFIRMATION，图在逻辑上"转圈"

本质：模拟了一个**"阻塞调用"**，但 LangGraph 的运行时并不知道你在等人。

builder.add_node(AWAITING_PLAN_CONFIRMATION, lambda state, config: state)

builder.add_node(SEARCH_REPLAN, lambda state, config: {"plan_status": "unconfirmed"})

方法2：`interrupt` 机制

这是 LangGraph 原生支持的 "真·HITL" ：

运行时级中断 ：interrupt() 被调用时，LangGraph Runtime 会：
- 立即挂起整个图执行
- 将当前状态 持久化到 Checkpointer
- 抛出异常/信号，通知调用方"需要人工介入"
进程可释放 ：执行线程/协程可以被释放，不需要占着资源干等
显式恢复 ：只有通过 graph.invoke/stream 传入 Command(resume=...) 时，才会从 interrupt 处继续执行

def awaiting_plan_confirmation(state: OverallState):
    
    feedback = interrupt({
        "type": "plan_confirm",
        "current_plan":state.get("plan"),
        "prompt": "请确认当前执行计划，同意请输入confirm，拒绝请输入reject"
    })

    
    logger.info(f"收到人工反馈: {human_response}")

    action = feedback.get("action", "reject")
    feedback = feedback.get("feedback", "")

    
    if action == "confirm":
        logger.info("计划已确认，标记为 confirmed")
        return {
            "plan_status": "confirmed",
            "user_feedback": feedback,
            "messages": [AIMessage(content=f"用户确认了计划: {feedback}")]
        }
    elif action == "reject":
        logger.info("计划被拒绝，标记为 rejected")
        return {
            "plan_status": "rejected",
            "user_feedback": feedback,
            "messages": [AIMessage(content=f"用户拒绝了计划: {feedback}")]
        }
    else:
        
        logger.info("收到修改意见，将用于重新规划")
        return {
            "plan_status": "rejected",
            "user_feedback": feedback,
            "messages": [AIMessage(content=f"用户建议修改: {feedback}")]
        }

Command 调用示例

"""
# 1. 首次调用，触发中断
config = {"configurable": {"thread_id": "test_thread_001"}}
initial_state = {"messages": [("user", "研究一下LangGraph的HITL机制")]}
result = graph.invoke(initial_state, config)
# 此时 result 会包含 interrupt 信息，但不会完成执行

# 2. 恢复执行：用户确认计划
resume_command = Command(resume={
    "action": "confirm",
    "feedback": "计划看起来不错，开始吧"
})
result_after_resume = graph.invoke(resume_command, config)
# 此时图会从 interrupt 处继续执行

# 3. 恢复执行：用户拒绝计划
resume_command_reject = Command(resume={
    "action": "reject",
    "feedback": "请增加对生产环境稳定性的分析"
})
result_after_reject = graph.invoke(resume_command_reject, config)
# 此时图会进入 SEARCH_REPLAN 节点，重新生成计划

"""

🤔🤔🤔

我们目前已经在 研究计划 节点加入了 HITL，如果需要，同样可以再在其他节点加入 HITL 。这个取决于具体的场景和 case，增加方式相同，这里不再做过多赘述。

DeepResearchSystem 0x03：引入 HITL 人类在环机制

中文

回顾

解决

HITL

Coding

OverallState

generate_plan

evaluate_plan：人类参与

HITL

方法 1：人工干预

方法 2：interrupt 机制

Command 调用示例

🤔🤔🤔

English / 原文

回顾

解决

HITL

Coding

OverallState

generate_plan

evaluate_plan 人类参与

HITL

方法1：人工干预

方法2：interrupt 机制

Command 调用示例

🤔🤔🤔

方法 2：`interrupt` 机制

方法2：`interrupt` 机制