LangGraph 09：HITL 人机协同

发表于 2026-08-20 分类于 ai

基于 interrupt 与 Command resume，讲解审批、编辑、并行中断与 checkpoint 续跑，并简述 LangGraph Studio 接入。

前言

Agent 跑到「调外部工具」「改用户数据」「生成对外文案」时，往往需要人点头或改一改再继续。
LangGraph 1.x 用节点内的 interrupt() 挂起图，再用 Command(resume=...) 按同一 thread_id 续跑。
本文覆盖动态 HITL：单点中断、审批路由、审核编辑、并行多中断，以及检查点上的分步恢复。
静态断点 interrupt_before / interrupt_after 见下一篇。
示例对接 火山方舟 Coding Plan，聊天模型用 ark-code-latest。
下文需要 Python 3.12+，依赖用 uv 管理。

依赖

建议使用 Python 3.12 及以上。
用 uv 初始化工程并声明依赖。

uv init langgraph-hitl
cd langgraph-hitl
uv venv --python 3.12
uv add "langgraph>=1.0,<2.0" "langchain>=1.0,<2.0" langchain-openai python-dotenv rich

在项目根目录创建 .env，写入 Coding Plan 的 Key 与专用 Base URL。

1 2	OPENAI_API_KEY=你的火山方舟 API Key OPENAI_BASE_URL=https://ark.cn-beijing.volces.com/api/coding/v3

请勿把 Base URL 写成普通方舟 .../api/v3，以免无法抵扣 Coding Plan 额度。
HITL 必须挂 checkpointer，否则中断后无法按线程恢复。

单点中断

节点里调用 interrupt(payload) 会立刻暂停。
invoke 返回值里带 __interrupt__，其中有提示文案与中断 id。
用同一 config 再 invoke(Command(resume=用户输入)) 即可续跑。

from typing import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt
from rich import print as rprint

class OverAllState(TypedDict):
    username: str

def node_a(state: OverAllState) -> OverAllState:
    username = interrupt("请输入您的姓名：")
    return {"username": username}

builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_a", node_a)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", END)

graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "hitl-1"}}

paused = graph.invoke({}, config=config)
rprint(paused)
prompt = paused["__interrupt__"][0].value
# 演示里写死；真实场景用前端表单或 CLI input(prompt)
done = graph.invoke(Command(resume="小明"), config=config)
rprint(done)

resume 的值会原样成为本次 interrupt() 的返回值，再写回状态。

审批路由

审批节点可返回 Command(goto=..., update=...)，按人的决定跳到不同分支。
下面用 Coding Plan 生成短诗：同意走 LLM，拒绝走默认文案。

import os
from typing import Literal, TypedDict

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt
from rich import print as rprint

load_dotenv()

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
)

class OverAllState(TypedDict):
    topic: str
    poem: str
    is_approved: bool

def approve_node(state: OverAllState) -> Command[Literal["llm_node", "default_node"]]:
    is_approved = interrupt("是否同意调用模型？")
    goto = "llm_node" if is_approved else "default_node"
    return Command(goto=goto, update={"is_approved": is_approved})

def llm_node(state: OverAllState) -> OverAllState:
    topic = state["topic"]
    text = model.invoke(
        [HumanMessage(content=f"写一首关于{topic}的两行短诗，只写诗句")]
    ).content
    return {"poem": text}

def default_node(state: OverAllState) -> OverAllState:
    return {"poem": "请求被拒绝"}

builder = StateGraph(state_schema=OverAllState)
builder.add_node("approve_node", approve_node)
builder.add_node("llm_node", llm_node)
builder.add_node("default_node", default_node)
builder.add_edge(START, "approve_node")
builder.add_edge("llm_node", END)
builder.add_edge("default_node", END)

graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "approve-1"}}

paused = graph.invoke({"topic": "春花"}, config=config)
rprint(paused["__interrupt__"])
# True / False 决定 goto
done = graph.invoke(Command(resume=True), config=config)
rprint(done)

approve_node 不必再画死边到 LLM：路由写在 Command.goto 里。

审核编辑

interrupt 的 payload 可以是 dict，把草稿一并交给人改。
resume 回传修改后的字符串（或原样），写入 reviewed_poem。

import os
from typing import TypedDict

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt
from rich import print as rprint

load_dotenv()

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
)

class OverAllState(TypedDict):
    topic: str
    poem: str
    reviewed_poem: str

def llm_node(state: OverAllState) -> OverAllState:
    text = model.invoke(
        [HumanMessage(content=f"写一首关于 {state['topic']} 的两行短诗，只写诗句")]
    ).content
    return {"poem": text}

def review_node(state: OverAllState) -> OverAllState:
    reviewed = interrupt(
        {
            "instruction": "请审核并修改下面短诗",
            "poem": state["poem"],
        }
    )
    return {"reviewed_poem": reviewed}

builder = StateGraph(state_schema=OverAllState)
builder.add_node("llm_node", llm_node)
builder.add_node("review_node", review_node)
builder.add_edge(START, "llm_node")
builder.add_edge("llm_node", "review_node")
builder.add_edge("review_node", END)

graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "review-1"}}

paused = graph.invoke({"topic": "橘猫"}, config=config)
draft = paused["__interrupt__"][0].value["poem"]
rprint(draft)
edited = draft + "\n（人工润色）"
done = graph.invoke(Command(resume=edited), config=config)
rprint(done["reviewed_poem"])

工具调用前审批也可把 interrupt 写在 @tool 内部：工具真正执行前先等人确认。

并行中断

从 START 同时连出多个节点时，可能一次暂停里出现多条 __interrupt__。
此时 Command(resume=...) 应传 dict：键是各中断的 id，值是对应回复。

from time import sleep
from typing import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt
from rich import print as rprint

class OverAllState(TypedDict):
    username: str
    age: int

def node_a(state: OverAllState) -> OverAllState:
    username = interrupt("请输入您的姓名:")
    return {"username": username}

def node_b(state: OverAllState) -> OverAllState:
    sleep(0.1)
    age = interrupt("请输入您的年龄:")
    return {"age": age}

builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_edge(START, "node_a")
builder.add_edge(START, "node_b")
builder.add_edge("node_a", END)
builder.add_edge("node_b", END)

graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "parallel-1"}}

paused = graph.invoke({}, config=config)
rprint(paused["__interrupt__"])

resume_map = {}
for item in paused["__interrupt__"]:
    if "年龄" in str(item.value):
        resume_map[item.id] = 18
    else:
        resume_map[item.id] = "小明"

done = graph.invoke(Command(resume=resume_map), config=config)
rprint(done)

若只有一条中断，传标量即可；多条必须用 id -> value 映射，否则对不上并行任务。

检查点续跑

同一节点内连续两次 interrupt 时，每次恢复只推进一格，中间状态都落在 checkpointer。
可用 get_state_history 观察「姓名已填、年龄未填」这类半完成快照。

from typing import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt
from rich import print as rprint

class OverAllState(TypedDict):
    username: str
    age: int

def node_a(state: OverAllState) -> OverAllState:
    username = interrupt("请输入您的姓名：")
    age = interrupt("请输入您的年龄：")
    return {"username": username, "age": age}

builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_a", node_a)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", END)

graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "ckpt-1"}}

step1 = graph.invoke({}, config=config)
rprint(step1)
step2 = graph.invoke(Command(resume="小明"), config=config)
rprint(step2)
step3 = graph.invoke(Command(resume=20), config=config)
rprint(step3)
rprint(list(graph.get_state_history(config=config)))

并行场景下，未中断的分支可先跑完并写入 checkpoint；只挂起带 interrupt 的那条边，恢复后合并状态。

Studio 接入

可用 langgraph.json 把图挂到 LangGraph Studio，例如：

{
  "dependencies": ["."],
  "graphs": {
    "graph": "./src/agent.py:graph",
    "chat_graph": "./src/chat_agent.py:chat_graph"
  },
  "env": ".env"
}

agent.py 里可在节点中连续 interrupt 收集姓名、年龄、性别；chat_agent.py 则可在 tool_node 里一次性 interrupt 一批工具审批（含 approve / reject / edit）。
本地可用 Studio 可视化暂停点与 resume，不必手写 CLI。
生产环境把同一协议接到 Web 表单或工单系统即可。

总结

动态 HITL：节点内 interrupt(payload)，同 thread_id 用 Command(resume=...) 续跑。
审批可结合 Command(goto=...)；编辑场景把草稿放进 payload。
并行多中断用 resume={id: value}。
必须配置 checkpointer；可用 get_state_history 排查半完成状态。
Studio 通过 langgraph.json 加载图定义，便于联调人机协同。