Agent-10-08.人在回路HITL

发表于 | 分类于
人在回路的设计动机、审批闸门、LangGraph interrupt 与本仓库 orchestrator 实现。

系列：00 索引 · 上一篇：07 Graph RAG · 下一篇：09 LangGraph

1. 行业常见问题

高风险动作 后果
自动提交计算任务（GPU/费用） 成本失控
自动写入公共知识库 污染、合规
自动对外发送邮件/下单 不可撤销
全自动「定稿报告」 幻觉进入决策链

监管与科研场景常要求：关键节点必须有人确认或可追溯驳回。

段末注释：HITL（human-in-the-loop，人在回路）= 流程在指定节点暂停，等待人类输入后再继续。

2. 该技术如何解决

  1. 识别闸门点：submit、ingest、publish、pay
  2. 结构化待办：展示 plan 摘要、artifact 路径、approve/reject
  3. 持久化 state：LangGraph Checkpointer + RunStore 记录 HITL 元数据
  4. 代码层 enforcement：图上的 approved 标志 且 Analyst 工具层的 blocked_hitl

3. 核心原理

3.1 本仓库 full 流水线中的 HITL 位置

1
2
3
4
sequence → literature → method_kb → planner → hitl_plan → analyst → reporter
                                              ↑
                                    interrupt_before=["analyst"]
                                    gate=S4_plan
  • Planner 产出 research_plan.v1（artifact 路径写入 state）
  • hitl_plan 节点：未批准则 RunStore.set_hitl_pending，stage=hitl_plan_pending
  • LangGraph interrupt：图在 analyst 前 挂起，等待 Web/CLI resume
  • Analyst：即使图误放行，execute_plan_steps(require_hitl=True) 仍会 blocked_hitl

3.2 三层职责（不要混在一处）

层 模块 职责
契约 agent/uni/contracts/hitl.py HITLGate、HITLDecision 枚举
编排 agent/orchestrator/graph.py hitl_plan 节点、interrupt_before
持久化 RunStore + SqliteSaver run 元数据 + 图 state 可恢复
执行阻断 agent/analyst/artifacts.py 未 approve 禁止 `submit_template_tool

3.3 interrupt + resume（LangGraph 语义）

1
2
3
app.stream(initial) → 跑到 analyst 前停止 → snap.next 非空 → hitl_pending 事件
app.update_state({"approved": True})
app.stream(Command(resume=True)) → analyst → reporter → END

thread_id = run_id，checkpoint 落盘 workspace/checkpoints.db。

4. 典型实现与代码示例

以下均来自本仓库 agent/orchestrator，示例目的是理解 框架怎么接 HITL，不是重写一套状态机。

4.1 闸门契约（跨 Web / CLI / 图共用）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# agent/uni/contracts/hitl.py
class HITLGate(str, Enum):
    S4_PLAN = "S4_plan"        # full 流水线：research_plan 审批；RunStore.meta.hitl.gate
    S5_EXECUTE = "S5_execute"  # 预留：执行前二次确认

class HITLRequest(BaseModel):
    gate: HITLGate             # 待办类型，Web /pipeline 按 gate 渲染表单
    run_id: str                # 关联 RunStore 与 checkpoint thread_id
    summary: str               # 展示 planner_summary 等摘要文本
    payload_path: str | None = None  # plan_artifact_path，供用户打开 JSON 审阅
    created_at: datetime = ... # 待办创建时间，排序/超时用

class HITLDecision(str, Enum):
    APPROVE = "approve"        # streaming resume → state.approved=True → analyst 可 submit
    REJECT = "reject"          # resume 后图终止或 stage=hitl_rejected
    REVISE = "revise"          # 预留：打回 planner 重规划

RunStore.set_hitl_pending / record_hitl_decision 用同一套 gate id，Web 工作台读 meta.json 里的 hitl 字段展示待办。

4.2 LangGraph：hitl_plan 节点 + interrupt 编译

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# agent/orchestrator/graph.py（节选）

def node_hitl_plan(state: PipelineState) -> PipelineState:
    run_id = state["run_id"]
    store = RunStore(cfg.workspace_root)
    if state.get("approved"):
        store.record_hitl_decision(run_id, gate=HITLGate.S4_PLAN.value, decision="approve")
        return {**state, "stage": "hitl_plan_approved"}
    # 未批准：登记待办，供 Web/CLI 展示 planner_summary + plan_artifact_path
    store.set_hitl_pending(
        run_id,
        gate=HITLGate.S4_PLAN.value,
        summary=state.get("planner_summary", ""),
        payload_path=state.get("plan_artifact_path"),
    )
    return {**state, "stage": "hitl_plan_pending"}

# full + with_hitl 时构图
graph.add_edge("planner", "hitl_plan")
graph.add_edge("hitl_plan", "analyst")
interrupt_before = ["analyst"] if use_interrupt and with_hitl else None
return graph.compile(checkpointer=get_checkpointer(), interrupt_before=interrupt_before)

要点：

  • hitl_plan 是图节点，不是 prompt 里写「请用户确认」
  • interrupt_before=["analyst"] 让 LangGraph 在 analyst 入口挂起，进程可退出
  • get_checkpointer() 用 SQLite，隔夜审批靠 checkpoint 恢复

4.3 执行层二次阻断（Analyst 提交 Argo）

1
2
3
4
5
6
7
8
# agent/analyst/artifacts.py（节选）
def execute_plan_steps(..., require_hitl: bool = True, hitl_approved: bool = False):
    if require_hitl and not hitl_approved:
        return {
            "status": "blocked_hitl",
            "error": "research_plan 未经 HITL approve，禁止 submit",
        }
    # 否则才调用 submit_template_tool ...

编排层传入 hitl_approved=bool(state.get("approved"))（见 node_analyst）。Prompt 约束不够，必须有工具层硬阻断。

4.4 Web / CLI：start → interrupt → resume

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# agent/orchestrator/streaming.py（节选）

# 启动：approved=False，with_hitl=True，use_interrupt=True
initial: PipelineState = {
    "run_id": run_id,              # checkpoint thread_id
    "gene_symbol": symbol,         # 各 Specialist 输入
    "stage": "started",            # 节点推进时更新，SSE 推送
    "approved": False,             # False → hitl_plan 登记待办 + interrupt 挂起
    # 恢复后 update_state({"approved": True})，analyst 读取此字段
}
for chunk in app.stream(initial, config, stream_mode="updates"):
    ...
interrupted = bool(app.get_state(config).next)  # 有 next 即停在 interrupt 点
if interrupted:
    on_event({"type": "hitl_pending", "gate": "S4_plan", "summary": planner_summary})

# 恢复：用户 POST /pipeline/{run_id}/resume {"decision":"approve"}
store.record_hitl_decision(run_id, gate=HITLGate.S4_PLAN.value, decision=decision)
app.update_state(config, {"approved": approved})
for chunk in app.stream(Command(resume=True), config, stream_mode="updates"):
    ...

FastAPI 入口：POST /pipeline/start（SSE）→ POST /pipeline/{run_id}/resume（SSE），见 agent/uni/api/main.py。

4.5 工程验收（推荐）

在项目根目录执行，直接走 LangGraph + Checkpointer：

1
2
3
4
5
6
7
8
9
10
11
# 步骤 1：启动 full 流水线，在 analyst 前 interrupt
uv run python -m agent.orchestrator.cli start-hitl --symbol BRCA1
# 输出 run_id=xxxxxxxx interrupted=True stage=hitl_plan_pending

# 步骤 2：查看 planner 摘要（也可打开 workspace/runs/<run_id>/ 下 artifact）
# 步骤 3：批准后 resume，继续 analyst → reporter
uv run python -m agent.orchestrator.cli resume <run_id> --approve
# 输出 stage=done，analyst_summary、report_artifact_path

# 辅助：打印带 HITL 的 mermaid 图
uv run python -m agent.orchestrator.cli graph --full --hitl

Web 等价路径：启动 uv run python -m agent.uni.api.main → 打开 /pipeline → 审批后点恢复。

4.6 与 Agent-10-09 的分工

主题 本文（08） Agent-10-09
关注点 何时停、谁批准、如何 resume 图怎么建、节点调哪个 Specialist
关键 API interrupt_before、Command(resume=True) StateGraph、build_pipeline
状态字段 approved、plan_artifact_path PipelineState 全字段

5. 替代方案与优缺点

方案 优点 缺点
LangGraph interrupt + SqliteSaver（本仓库） 与编排一体、可恢复 需理解 checkpoint/thread_id
仅 RunStore 布尔位、无 interrupt 实现简单 进程必须常驻或丢进度
外部 BPMN/Camunda 企业审批成熟 与 LLM 步耦合重
仅 Analyst prompt「请勿提交」 零代码 不可审计、不可阻断

6. 自检题

  1. 本仓库为何在 图节点 hitl_plan 之外，Analyst 还要 blocked_hitl？
  2. thread_id 为何等于 run_id？换 run 不换 thread 会怎样？
  3. approve 与 reject 在 streaming._run_resume_pipeline_sync 里如何影响 approved 与下游节点？

7. 延伸阅读

-------------本文结束感谢您的阅读-------------