12.3 人类在环
12.2 的循环能自己回炉重试，看起来已经很「自动」了。但自动不等于可以放手——当下一步是发一封对外邮件、扣一笔款、删一批数据时，让模型自己决定并按下去，就是事故的开始。这些操作的共同点是高风险且不可逆：做错了没法简单撤回。对它们，唯一稳妥的做法是暂停，等人点头。这一节把「人」接进图里，作为一个显式的、可审计的环节。
本节把 minagent 的图引擎加上人类在环（human-in-the-loop）：用
interrupt_before标记「执行前必须暂停」的节点，run跑到这里就停下、返回暂停点与当前 state；人给出approve/reject/modify决策后，resume(state, decision)带着决策继续往下走。用假模型真跑一遍「敏感工具调用前暂停 → 人批准 → 继续」。
12.3.1 为什么把「人」做进图里
不给图加任何机制，能不能实现人工确认？能——在节点函数里写一个 input("批准吗？") 就行。但这样做有四个麻烦：
| 麻烦 | 表现 |
|---|---|
| 阻塞 | input() 卡住整个进程，服务端没法服务别的请求 |
| 不可序列化 | 暂停在函数中间，进程重启就丢了 |
| 不可审计 | 有没有等人、等的是谁，日志里查不到 |
| 测试困难 | 每个要确认的节点都得改代码才能自动跑 |
把「暂停」做成引擎的一等能力，这些麻烦就都能解。核心思路是：不要在执行到一半时暂停，而是在进入某个节点之前就停下。引擎在跑每个节点前先看一眼「这个节点在不在 interrupt_before 名单里」，在的话就不执行它，直接返回——此时进程是干净的，state 是完整的、可序列化的，人可以慢慢想，想好了再调 resume 从断点继续。
这就是所谓「先暂停、后执行」的语义。它和「执行到一半再问」有本质区别：前者只在节点边界上暂停，状态永远是一致的；后者要暂停在函数内部，就得处理半成品状态。选前者，实现简单得多，也可靠得多。
12.3.2 三个新类型：HumanDecision、RunOutcome
给引擎加两个 dataclass：
# minagent/workflow/graph.py —— 新增
@dataclass
class HumanDecision:
node: str # 停在哪个节点
action: str # "approve" | "reject" | "modify"
payload: dict[str, Any] = field(default_factory=dict)
@dataclass
class RunOutcome:
state: State
paused_at: str | None = None # 停在哪个节点「之前」
pending: dict[str, Any] | None = None # 暴露给人的待决策项
done: bool = False
HumanDecision 是人给引擎的输入：停在哪个节点、做什么动作、要改什么（payload）。RunOutcome 是引擎给人的输出：跑完的状态，以及「是不是停在了某个节点前、停在哪儿、待决策项是什么」。Graph 只多一个字段 interrupt_before: set[str]，用 interrupt(*names) 往里加节点名。
run 的返回类型从 12.1 的 State 变成了 RunOutcome——因为现在「跑完」有两种：一种是真的走到底（done=True），一种是停在了审批点（paused_at 非空）。调用方必须先判断是哪种，才能决定要不要 resume。
12.3.3 暂停与恢复：run 与 resume
引擎的核心改动是把 12.1 的 run 拆成两半：run 从入口开始，_run_from 是真正干活的循环，resume 从暂停点重新进入。
def run(self, state: State, on_node=None) -> RunOutcome:
return self._run_from(self.entry, state, on_node, frozenset())
def _run_from(self, cur, state, on_node, approved: frozenset) -> RunOutcome:
while cur is not None:
if cur in self.interrupt_before and cur not in approved:
return RunOutcome(state=state, paused_at=cur,
pending={"node": cur, "state": dict(state)})
approved = approved - {cur} # 放行令牌一次性消费
before = dict(state)
update = self.nodes[cur].fn(state) or {}
state.update(update)
if on_node is not None:
on_node(cur, before, dict(state))
cur = self._next(cur, state)
return RunOutcome(state=state, done=True)
def resume(self, outcome: RunOutcome, decision: HumanDecision, on_node=None) -> RunOutcome:
state = outcome.state
if decision.action == "reject":
state["approved"] = False
nxt = self._next(decision.node, state) # 跳过该节点，走它的后继
return self._run_from(nxt, state, on_node, frozenset())
if decision.action == "modify":
state.update(decision.payload)
state["approved"] = True
return self._run_from(decision.node, state, on_node, frozenset({decision.node}))
三处关键设计：
- 暂停发生在节点执行「之前」：
_run_from一进循环就先检查cur in self.interrupt_before，命中就立刻返回，绝不执行该节点。所以返回时 state 里没有这个节点的任何副作用，是干净一致的。 approved是「一次性放行令牌」：resume把要执行的节点名塞进approved，_run_from见到它就不再暂停、放行一次，然后approved - {cur}把令牌消耗掉。这样如果流程后来又绕回同一个节点（比如带环的图），它还是会再次暂停——令牌只放行一次，不会永久开门。resume按动作分流：approve和modify都会执行该节点（modify先把payload合进 state 再执行）；reject则跳过该节点，用_next直接走到它的后继。所以「拒绝」不是报错，而是走另一条路——后继节点会看到state["approved"] == False，据此做出不同反应。
这套设计的精妙处在于：审批逻辑完全在引擎里，节点函数不需要知道有没有人批过。execute 节点只管执行，至于它前面有没有暂停、人批没批，是引擎的事。节点保持纯粹，审批作为横切关注点被统一处理——这正是把「暂停」做进引擎而不是散在节点里的收益。
12.3.4 真跑：敏感工具调用前暂停
场景：模型决定调用 send_email 这个对外、不可逆的工具。我们要求执行它之前必须人工确认。先看图怎么搭：
from minagent.fake import FakeModel
from minagent.types import Message, ToolCall
from minagent.workflow.graph import Graph, HumanDecision
TOOL_SCHEMAS = [{
"name": "send_email",
"description": "发送邮件（对外、不可逆）",
"parameters": {"to": "string", "body": "string"},
}]
def build_model():
return FakeModel([
[ToolCall(id="t1", name="send_email",
arguments={"to": "alice@example.com", "body": "退款已处理"})],
"已按你的批准发送退款邮件。",
])
MODEL = build_model()
def decide(state):
resp = MODEL.complete([Message(role="user", content=state["goal"])], TOOL_SCHEMAS)
return {"tool_call": resp.message.tool_calls[0]}
def execute(state):
tc = state["tool_call"]
args = dict(tc.arguments)
args.update(state.get("override", {})) # 人的修改覆盖模型给的参数
return {"sent_args": args, "result": f"{tc.name}(to={args['to']}) 已执行"}
def summarize(state):
if not state.get("result"):
return {"final": "已按你的决定取消发送。"}
resp = MODEL.complete([Message(role="user", content=state["result"])], [])
return {"final": resp.message.content.strip()}
g = (
Graph()
.add_node("decide", decide)
.add_node("execute", execute)
.add_node("summarize", summarize)
.add_edge("decide", "execute")
.add_edge("execute", "summarize")
.set_entry("decide")
.interrupt("execute") # 执行敏感工具前暂停，等人批
)
decide 节点让模型决定要调 send_email，但它只把工具调用写进 state["tool_call"]，不执行。真正执行的是 execute 节点，而它被 interrupt("execute") 标记了——所以图会停在 execute 之前，把整个待办（含工具参数）暴露给人。这里用一个脚本化的 ScriptedHuman 扮演「人」，避免真的读 stdin：
class ScriptedHuman:
"""脚本化的人：按预设顺序吐出决策，替代真实 stdin。"""
def __init__(self, decisions):
self.decisions = list(decisions)
def decide(self, pending):
print(f" [人] 待决策节点={pending['node']} state keys={sorted(pending['state'])}")
return self.decisions.pop(0)
def show(name, before, after):
print(f"[{name}] 新增 {sorted(set(after) - set(before))}")
def drive(graph, state, human):
out = graph.run(state, on_node=show)
while not out.done:
print(f"⏸ 暂停于 {out.paused_at}，等待人工决策")
out = graph.resume(out, human.decide(out.pending), on_node=show)
return out
def run(decision, label):
global MODEL
MODEL = build_model() # 每个场景重建模型，避免脚本串味
print(f"=== {label} ===")
out = drive(g, {"goal": "给 alice 发退款通知"}, ScriptedHuman([decision]))
print("final:", out.state["final"],
"| sent_args:", out.state.get("sent_args"), "\n")
run(HumanDecision(node="execute", action="approve"), "批准")
run(HumanDecision(node="execute", action="reject"), "拒绝")
run(HumanDecision(node="execute", action="modify",
payload={"override": {"to": "bob@example.com"}}), "修改")
drive 就是宿主程序的骨架：先 run，只要没 done 就说明停在审批点，把待决策项交给「人」，拿到决策后 resume，循环直到 done。三种决策真跑结果（Python 3.14.6）：
=== 批准 ===
[decide] 新增 ['tool_call']
⏸ 暂停于 execute，等待人工决策
[人] 待决策节点=execute state keys=['goal', 'tool_call']
[execute] 新增 ['result', 'sent_args']
[summarize] 新增 ['final']
final: 已按你的批准发送退款邮件。 | sent_args: {'to': 'alice@example.com', 'body': '退款已处理'}
=== 拒绝 ===
[decide] 新增 ['tool_call']
⏸ 暂停于 execute，等待人工决策
[人] 待决策节点=execute state keys=['goal', 'tool_call']
[summarize] 新增 ['final']
final: 已按你的决定取消发送。 | sent_args: None
=== 修改 ===
[decide] 新增 ['tool_call']
⏸ 暂停于 execute，等待人工决策
[人] 待决策节点=execute state keys=['goal', 'tool_call']
[execute] 新增 ['result', 'sent_args']
[summarize] 新增 ['final']
final: 已按你的批准发送退款邮件。 | sent_args: {'to': 'bob@example.com', 'body': '退款已处理'}
三条路径的差异都在「暂停之后」。批准：execute 执行，sent_args 是模型原本给的 alice。拒绝：execute 被跳过（输出里根本没有 [execute] 这一行），直接进 summarize，sent_args 为 None，最终回复是「取消发送」。修改：人把收件人改成 bob，execute 照常执行，但 sent_args 里 to 已变成 bob——人的修改真的覆盖了模型的参数。三种决策走三条路，图都正常收敛，这就是把「人」做成图上一个可控节点的效果。
12.3.5 哪些节点该设人工审批
不是所有节点都值得暂停——处处审批等于没审批，人会麻木地点「通过」。设审批点要靠风险判断，而不是「感觉重要」。三个判据，满足任一就该设：
| 判据 | 含义 | 例子 |
|---|---|---|
| 不可逆 | 做了没法或很难撤回 | 发邮件、扣款、删除数据、发布 |
| 对外 | 会影响到系统之外的人或系统 | 调用第三方 API、给用户发通知 |
| 高代价 | 错了要付出大代价 | 生成合同、开具发票、医疗建议 |
反过来，读操作、可回滚的写操作、纯内部计算都不该设审批——它们错了重来一次就行，暂停只会拖慢流程。一个实用的分层做法：
- 自动：读、检索、计算、草稿生成——模型随便跑。
- 自动 + 事后审计：可回滚的写——先做，但留全日志，出问题能查能撤。
- 人工审批：上表三类——执行前必须停。
注意审批点要设在执行节点之前，而不是「生成决定」的节点之前。让模型自由地「想」要做什么（decide 不暂停），只在「真的要做」时（execute 暂停）拦一道——这样既给了模型发挥空间，又把闸门卡在了最该卡的地方。
12.3.6 真实模型下会怎样
本节的「人」是脚本化的 ScriptedHuman，决策写死在列表里；真实系统里，pending 会推给一个真实的人——可能是 Slack 上的一条待办、一个审批后台的工单、或命令行里的一次确认。这带来几个本节没展开的工程问题。其一，暂停可能持续很久（人可能几小时后才批），所以 state 必须能持久化——这正是第 7 章检查点（checkpoint-resume）的用武之地，RunOutcome 的 state 序列化存盘，进程可以重启。其二，审批要防重复：同一封邮件不能被批两次，resume 需要幂等（第 14 章讲幂等工具时会呼应）。其三，真实模型的 decide 输出更野——它可能一次给出多个工具调用、或参数缺字段，所以 pending 里除了工具参数，还应该带上模型的理由（它为什么要发这封邮件），让人能基于上下文判断，而不是只看到一个收件人地址。
还有一点关于「人」的定位要摆正：人在环不是「人替模型干活」，而是「人在关键闸口做最终裁量」。模型仍然负责把大部分工作做完（分类、检索、起草、决定调用哪个工具），人只负责对少数高风险动作点头或摇头。设计得好的话，人的介入应该是低频、高价值的——如果每一步都要人批，那不是「人在环」，是「人肉执行器」，图编排的价值也就没了。
至此第 12 章闭环：12.1 立起图骨架，12.2 织出分支/合并/循环，12.3 把「人」接进来。minagent 现在既能自动跑固定流程，又能在关键处停下来听人的。下一章 13.1 角色与分工 会把这套单 Agent 的编排，扩展到多个 Agent 之间的协作。
阅读导航：上一节：12.2 分支、合并与循环 · 下一节：13.1 角色与分工 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。