注：以下所提的“文档”，是指LangGraph官方指南（https://langchain-ai.github.io/langgraph/guides/），参考版本是0.6.8。

以下代码的开发环境：

[project]
name = "my-langgraph"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "dotenv>=0.9.9",
    "langchain>=0.3.27",
    "langchain-openai>=0.3.33",
    "langfuse>=3.5.0",
    "langgraph>=0.6.7",
    "langgraph-checkpoint-sqlite>=2.0.11",
    "numpy>=2.3.3",
]

一、Human-in-the-Loop 在 LangGraph 中的定位与价值

“Human-in-the-Loop” (HITL) 指在自动化流程中插入人工检查 /审批 /干预环节，使得系统可以在人类监督 /决策的协助下运行。LangGraph 在其执行 /持久化 /流程控制体系中设计了对这种场景的支持。概念篇和 How-To 都在不同层面阐述这一能力。

为什么要 Human-in-the-Loop？

• 自动化系统（LLM + 工具 + 图）虽然强大，但在一些高风险 / 高成本 / 需人工保证正确性的场景中，完全自动化不够安全。人工介入能增加可信度和可控性。
• 举例场景包括：内容审核 / 人工校验 / 最终人工审批 / 人工纠错 / 人机协作任务等。
• 在有持久化 / checkpoint / replay 能力的系统中插入人工介入节点，是一种“暂停 → 人工审查 → 继续”的自然模式。LangGraph 将此视作其核心能力之一。

二、概念篇：设计原则与核心机制

指南文档从理念、机制、流程控制等角度讲述如何在 LangGraph 架构中实现 HITL。以下是其主要内容与要点。

核心机制与流程

 1. 暂停点 / 检查点插入（Interruptible Checkpoints）
在图 /节点设计中，可以在某些关键节点之后插入“人工检查点”——即流程在那个节点执行后，不立即继续下一个节点，而是暂停并等待人工输入 /确认。文档称这种节点为 “human node / approval node”。这种机制借助于持久化 / checkpoint 体系：在做人工介入之前，系统已经保存一个稳定 checkpoint，人工介入后可从这个点继续。

 2. 状态展示与人工输入
人工介入节点通常需要展示当前状态 /中间结果 /上下文给人类，让人做判断 /输入。然后把人工输入（如通过 UI 填写的审批意见、注释、纠正值等）写入状态，使得后续节点可基于人工输入继续。

 3. 人工输入注入 & 恢复执行
在人工输入之后，系统会把人工输入数据写入 state（或特定字段），并从那一点继续执行剩余节点。由于 checkpoint 已保存，恢复是可靠的。也可能有“回滚 / 重试 /多人审批”逻辑 —— 比如人工输入后有错误可重新修改。文档指出这种灵活性正是持续流程 + 人工介入机制的价值之一。

4. 与 Persistence / Durable Execution / Update_state 的配合

•  Human-in-the-loop 的暂停 /恢复机制依赖于 checkpoint /持久化能力。若没有 persistence，就无法可靠暂停 /恢复。
• 文档建议设计流程时把“可人工干预点”设在节点边界 /任务边界，而不是在节点内部任意地方作中断。这样恢复点清晰。
• update_state 接口可以用于人工输入之后注入新的状态（例如纠正值、人工标注等），并指定 as_node 以控制继续执行的节点。 

  5. 交互 / UI / 人机界面

虽然文档主要关注后端机制，但它指出：要实现可用的人工介入，需要有界面/人机展示层（dashboard /Web UI /表单界面）展示状态、接收人工输入，并把输入传回系统以继续执行。文档提供的是后端机制支持，不包含完整 UI。

三、How-To 篇：如何插入 Human-In-The-Loop

在指南文档的这部分更偏操作指南，告诉你在 LangGraph 中如何在流程里添加人工节点 / 接入人工输入 /恢复执行。下面是其主要内容和使用步骤。

操作流程与 API 用法

 1. 在节点中检测是否需要人工介入

• 在某些节点（或子图）逻辑里，你可以根据状态 /上下文判断是否要请求人工审核。如若某结果低信度、自动判断不确定，就触发人工交互分支。文档建议在节点里抛一个特定信号或返回一种“待人工判断”状态。
• 节点可以返回一个特殊字段（如 needs_human = True / pause_for_human）在 state 中标记“需要人工”状态。后续处理器（outer loop）会识别这个状态暂停流程。

 2. 暂停 / 等待人工输入 

•  执行到人工节点后，系统应暂停，不自动继续。
• 外部 (UI / 控制台 / web) 获取当前 checkpoint 状态（get_state / get_state_history 等接口），将状态 /中间结果展示给人类。
• 人工输入被封装为新的 state 更新，通过 update_state(...) 接口写回状态。文档示例里演示如何用 update_state 注入人工标注结果。

 3. 人工输入之后恢复执行 

•  在 update_state 时可指定 as_node，以决定下一个要执行的节点。通常你会把 as_node 设置为人工节点（或其前置节点），使得流程从该节点之后继续。
• 恢复执行之后，Graph 或 Agent 会从最近 checkpoint（在人工节点之前）恢复，跳过前面已完成节点，继续执行剩余节点。

 4. 多次人工交互 /循环 

• 流程中可多个节点设计人工介入点。也可以让同一个节点在人工审核后再次触发另一个人工判断。
• 文档建议控制人工节点位置（应在任务边界、关键决策节点后）以避免复杂状态回退。

 5. 示例 / 典型代码片段 

•  文档中给出简化示例，演示如何捕捉 needs_human、暂停、update_state 注入、恢复调用。比如： 

# node 中
if low_confidence(prediction):
    return {"prediction": None, "needs_human": True}

# 外部控制流程（伪代码）
state = graph.invoke(inputs)
if state.get("needs_human"):
    snapshot = graph.get_state(config)
    # 展示 snapshot.values 给人工
    human_label = ask_human(snapshot.values)
    graph.update_state(config, values={"prediction": human_label}, as_node="node_name")
    final = graph.invoke({}, config=config)

文档也说明如何在 .stream 中融合人工暂停与恢复。

四、对比 / 使用建议 + 注意事项

下面是我基于阅读 +实战对这两篇文档内容的综合理解与补充建议。

核心价值 & 使用场景

• HITL 能在高敏感 / 高风险 /需人工监督的系统中，平衡自动化与可控性，是一种非常实用机制。
• 在 Agent / Graph 系统中引入人工节点，使得系统不仅仅是“黑盒自动执行”，还具有“人工检查 /纠正 /策略插入”的能力。

使用建议 &注意点

1. 把人工节点设计在“边界 / 决策节点”
   尽量不要在节点内部复杂逻辑中插入人工判断，而应把人工介入界面放在节点之间。这样 recovery 语义清晰，checkpoint 恢复更安全。
2. 状态展示 /可视化
   人工节点需要展示的状态 /中间结果要足够清晰（例如 node 输出、上下文、日志、工具结果等），以便人类理解做出判断。在设计时要考虑 UI 层的串通。
3. 人工输入设计 & 类型校验
   人工输入最好结构化（如选项、字段、纠正值、注释），而不是完全自由文字。这样更容易写入 state、校验、后续使用。
4. 幂等 / 输入覆盖策略
   人工节点可能会多次修改 / 重试。你要设计 update_state 时的覆盖 /合并逻辑（特别是 state 中已有字段）要清晰、幂等。
5. 恢复逻辑清晰
   在 update_state 时适当指定 as_node，明确接下来要执行哪个节点。缺省可能让系统走回错误路径。
6. 超时 / 人工响应延迟处理
   如果人工审核节点长时间无人响应，系统应有超时 /降级策略（例如自动跳过、默认值、重新提醒等）。文档主要关注机制，不深入 UI 超时处理。
7. 安全 /权限控制
   人工输入可能带来安全 /注入风险，比如用户输入不合法内容。需要在 UI /后端做校验 /权限控制。
8. 版本演进 /文档更替风险
   如你之前注意的那样，LangGraph 在 v1.0 版本中可能重构相关机制，Human-in-the-loop API 可能改动，很有必要预留升级适配空间。

五、例子代码

import argparse
import sqlite3
import time
from typing import TypedDict, Annotated, Dict, Any, List
from operator import add

from langgraph.graph import StateGraph, START, END
from langgraph.config import get_stream_writer
from langgraph.checkpoint.sqlite import SqliteSaver

"""
使用 StateGraph（workflow，不是 agent）
SQLite checkpointer 支持中断后恢复
在审阅节点暂停等待人工输入（“通过/拒绝 + 反馈”）
使用 update_state(..., as_node="review") 注入人工意见并从审阅点继续
控制台流式观察执行（updates/custom/values）
"""

# ---------- 1) 状态（含 HITL 字段） ----------
class FlowState(TypedDict, total=False):
    text: str                 # 初始需求/主题
    draft: str                # 机器起草的内容
    approved: bool            # 人工是否通过
    feedback: str             # 人工给的修改意见
    needs_human: bool         # 是否需要人工介入
    result: str               # 最终产物
    log: Annotated[List[str], add]  # 日志累加（reducer）


# ---------- 2) 节点 ----------
def node_draft(state: FlowState) -> Dict[str, Any]:
    """机器起草（纯计算，无副作用）"""
    w = get_stream_writer()
    topic = state.get("text", "No topic")
    w({"node": "draft", "event": "start", "topic": topic})
    time.sleep(0.1)
    draft = f"- Goal: {topic}\n- Plan: step1, step2, step3\n- Notes: keep it simple"
    w({"node": "draft", "event": "done"})
    return {"draft": draft, "log": [f"draft ready for '{topic}'"]}


def node_review(state: FlowState) -> Dict[str, Any]:
    """
    审阅关卡（需要人工介入）：
    - 如果尚未有 approved（None），标记 needs_human=True，然后“暂停”（由条件边转到 END）
    - 如果已有人工决定：进入下一步
    """
    w = get_stream_writer()
    approved = state.get("approved", None)
    if approved is None:
        # 首次到审阅点 → 请求人工
        w({
            "node": "review",
            "event": "pause_for_human",
            "hint": "call `approve` CLI with --ok yes/no and optional --feedback",
        })
        return {
            "needs_human": True,
            "log": ["awaiting human review (approved? yes/no, with feedback)"],
        }
    else:
        w({
            "node": "review",
            "event": "resume_with_human_input",
            "approved": approved,
            "feedback": state.get("feedback", ""),
        })
        # 明确不需要再等人
        return {"needs_human": False, "log": [f"human says: {approved}"]}


def node_apply_feedback(state: FlowState) -> Dict[str, Any]:
    """根据人工意见修改 draft（示例化处理）"""
    w = get_stream_writer()
    draft = state.get("draft", "")
    approved = state.get("approved", False)
    fb = (state.get("feedback") or "").strip()

    if not approved:
        # 被拒绝：根据反馈重写一个更简版，示意“整改后继续”
        w({"node": "apply_feedback", "event": "rework"})
        new_draft = f"[REWORKED]\n{draft}\n- Reviewer says: {fb or 'No details'}\n- Action: simplify & fix issues"
    else:
        # 通过：根据反馈做轻微润色
        w({"node": "apply_feedback", "event": "light_edit"})
        extra = f"\n- Incorporated note: {fb}" if fb else ""
        new_draft = draft + extra

    return {"draft": new_draft, "log": ["applied feedback"]}


def node_finalize(state: FlowState) -> Dict[str, Any]:
    """输出最终结果（纯计算，无副作用）"""
    w = get_stream_writer()
    w({"node": "finalize", "event": "start"})
    time.sleep(0.05)
    result = f"✅ FINAL DOC\n{state.get('draft','')}\n---\nApproved: {state.get('approved')}"
    w({"node": "finalize", "event": "done"})
    return {"result": result, "log": ["finalized"]}


# ---------- 3) 条件路由 ----------
def route_after_review(state: FlowState) -> str:
    """
    审阅节点后的去向：
    - needs_human=True → 直接结束（暂停等待人工输入）
    - 否则 → 进入 apply_feedback
    """
    return "END" if state.get("needs_human") else "apply_feedback"


# ---------- 4) 组装图 ----------
def build_app(db_path: str = "hitl.sqlite"):
    conn = sqlite3.connect(db_path, check_same_thread=False)
    saver = SqliteSaver(conn)

    g = StateGraph(FlowState)
    g.add_node("draft", node_draft)
    g.add_node("review", node_review)
    g.add_node("apply_feedback", node_apply_feedback)
    g.add_node("finalize", node_finalize)

    g.add_edge(START, "draft")
    g.add_edge("draft", "review")
    # 审阅后的条件边
    g.add_conditional_edges("review", route_after_review, path_map={"END": END, "apply_feedback": "apply_feedback"})
    g.add_edge("apply_feedback", "finalize")
    g.add_edge("finalize", END)

    # 编译：接入 checkpointer
    return g.compile(checkpointer=saver)


# ---------- 5) CLI ----------
def cmd_run(args):
    """
    第一次运行：进入 review 节点后会暂停（needs_human=True → 结束本次运行），
    你会看到控制台提示使用 `approve` 子命令注入人工意见并恢复。
    """
    app = build_app(args.db)
    config = {"configurable": {"thread_id": args.thread}}

    print(f"\n=== RUN (thread={args.thread}) ===")
    for mode, chunk in app.stream(
        {"text": args.text},
        config=config,
        stream_mode=["updates", "custom", "values"],
        durability="sync",
    ):
        if mode == "custom":
            print("🟣 [custom ]", chunk)
        elif mode == "updates":
            print("🔸 [updates]", chunk)
        elif mode == "values":
            print("🟢 [values ]", chunk)

    # 查看当前状态（通常已到 review 并暂停）
    st = app.get_state(config)
    print("\n--- SNAPSHOT ---")
    print("values:", st.values)
    print("next  :", st.next)
    print("\nIf you see needs_human=True, run `approve` to resume.\n")


def cmd_approve(args):
    """
    第二次运行：注入人工意见（approved / feedback），并从审阅节点继续执行。
    """
    app = build_app(args.db)
    config = {"configurable": {"thread_id": args.thread}}

    approved = (args.ok.lower() in ["yes", "y", "true", "1"])
    fb = args.feedback or ""

    print(f"\n=== APPROVE (thread={args.thread}, ok={approved}, feedback={fb!r}) ===")
    # 注入人工输入，并声明是从 review 节点之后继续
    app.update_state(
        config=config,
        values={"approved": approved, "feedback": fb, "needs_human": False},
        as_node="review",
    )

    # 继续执行（从最新 checkpoint 的“下一步”出发）
    for mode, chunk in app.stream(
        None,
        config=config,
        stream_mode=["updates", "custom", "values"],
        durability="sync",
    ):
        if mode == "custom":
            print("🟣 [custom ]", chunk)
        elif mode == "updates":
            print("🔸 [updates]", chunk)
        elif mode == "values":
            print("🟢 [values ]", chunk)

    st = app.get_state(config)
    hist = list(app.get_state_history(config))
    print("\n--- FINAL ---")
    print("result:\n", st.values.get("result", ""))
    print("\nhistory length:", len(hist))

"""
演示 “人工介入 → 恢复执行”
1. 第一次运行（进入审阅并暂停）

python hitl_demo.py run --thread t-42 --text "Prepare incident response checklist"

控制台能看到：
- draft 产出草稿
- 到 review 节点后打印 pause_for_human，状态里有 needs_human=True
- 运行到此结束（等待人工输入）

2. 第二次运行（注入人工意见并恢复）
通过 同一个 thread_id 注入人工判定与反馈：

# 人工“通过”并给一点备注
python hitl_demo.py approve --thread t-42 --ok yes --feedback "Looks good. Add on-call rota."

# 或者“拒绝”，附带修改意见
python hitl_demo.py approve --thread t-42 --ok no  --feedback "Too generic; add escalation path."

你会看到：
- 系统用 update_state(..., as_node="review") 写入人工意见
- 自动从审阅点继续执行 → apply_feedback → finalize
- 打印最终结果与历史长度
"""

class _Args:
    # 一个简单的“命名空间”对象，用来模拟 argparse.Namespace
    def __init__(self, **kwargs):
        for k, v in kwargs.items():
            setattr(self, k, v)

def run_via_code(thread: str="t-42", text: str="Prepare incident response checklist", db: str="hitl_example.sqlite"):
    """
    等价于命令：
    python hitl_demo.py run --thread t-42 --text "Prepare incident response checklist"
    """
    args = _Args(thread=thread, text=text, db=db)
    cmd_run(args)

def approve_via_code(thread: str, ok: str, feedback: str, db:str= "hitl_example.sqlite"):
    """
    等价于命令：
    python hitl_demo.py approve --thread t-42 --ok yes --feedback "Looks good. Add on-call rota."
    python hitl_demo.py approve --thread t-42 --ok no  --feedback "Too generic; add escalation path."
    """
    if ok not in ("yes", "no"):
        raise ValueError("ok 必须是 'yes' 或 'no'")
    args = _Args(thread=thread, ok=ok, feedback=feedback, db=db)
    cmd_approve(args)

def main():
    # run_via_code()
    approve_via_code(thread="t-42", ok="yes", feedback="Looks good. Add on-call rota.")
    # approve_via_code(thread="t-42", ok="no", feedback="Too generic; add escalation path.")

def run_console():
    parser = argparse.ArgumentParser(description="LangGraph Human-in-the-Loop Demo (SQLite + resume)")
    sub = parser.add_subparsers(dest="cmd", required=True)

    p_run = sub.add_parser("run", help="start or continue until human review pause")
    p_run.add_argument("--thread", default="t-1001", help="thread id")
    p_run.add_argument("--text", default="Draft a rollout plan for feature flags", help="task/topic")
    p_run.add_argument("--db", default="hitl_example.sqlite", help="sqlite file path")
    p_run.set_defaults(func=cmd_run)

    p_ok = sub.add_parser("approve", help="inject human decision and resume")
    p_ok.add_argument("--thread", default="t-1001", help="thread id (same as 'run')")
    p_ok.add_argument("--ok", required=True, choices=["yes", "no"], help="human approval")
    p_ok.add_argument("--feedback", default="", help="optional feedback text")
    p_ok.add_argument("--db", default="hitl_example.sqlite", help="sqlite file path")
    p_ok.set_defaults(func=cmd_approve)

    args = parser.parse_args()
    args.func(args)

def hitl_example():
    main()