播面 如何设计一个带有“断点（Interrupt）”的工作流，使大模型在执行敏感操作（如发邮件、转账）前自动暂停，等待人类用户的审批？
设计一个带有“断点（Interrupt）”的大模型工作流，本质上是实现一个 Human-in-the-Loop (HITL，人类在环) 系统。在这个系统中，大模型本身并不具备真正的“暂停”能力，而是通过外围的工程架构与状态机（State Machine）来拦截敏感操作的执行。
以下是设计这样一个工作流的完整指南，包括架构设计、实现路径（以主流框架为例）以及工程最佳实践。
一、 核心架构设计
整个系统可以分为四个阶段：意图识别与规划 -> 拦截与挂起 -> 人类审批 -> 恢复与执行。
- AI 规划层： LLM 接收用户指令，决定需要调用工具（Function Calling）。
- 网关拦截层（核心）： 系统检查 LLM 请求调用的工具名。如果属于“敏感工具名单”（如
send_email,transfer_money），则拦截执行。 - 状态挂起层： 将当前会话的完整上下文（消息历史、准备调用的工具及参数）序列化并持久化保存到数据库（如 Redis、PostgreSQL），释放计算资源，不让程序死等。
- 异步审批层： 通过 Webhook 发送通知给人类审批者（通过钉钉、企业微信、邮件或内部工单后台）。
- 唤醒与执行层： 人类点击“同意”或“拒绝”后，系统从数据库恢复状态，执行操作（或反馈拒绝信息），并将结果作为系统消息交回给 LLM 往下继续。
二、 具体实现方案
方案一：基于 LangGraph 开发（业界标杆，最推荐）
LangGraph 是专门为复杂、有状态的多 Agent 工作流设计的框架，原生支持 checkpointer（状态持久化）和 interrupt（断点）。
设计步骤：
定义图结构与节点：
将工作流拆分为Agent_Node（思考和生成工具调用）和Tool_Node（执行工具）。设置断点：
在编译图（Compile Graph）时，设置在进入Tool_Node之前自动暂停。
伪代码示例：
python
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver
# 1. 定义工具节点和代理节点
def agent_node(state):
# LLM 决定调用 send_money 工具
return {"messages": [llm.invoke(state["messages"])]}
def tool_node(state):
# 实际执行转账
return execute_tool(state)
# 2. 构建状态机
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")
# 如果 LLM 决定调用工具，走到工具节点
workflow.add_conditional_edges("agent", should_continue)
workflow.add_edge("tools", "agent")
# 3. 关键：设置持久化和断点
memory = SqliteSaver.from_conn_string("checkpoints.db")
app = workflow.compile(
checkpointer=memory,
interrupt_before=["tools"] # <-- 关键：在执行工具前强制挂起
)
# 4. 运行工作流（第一次运行，会在 tools 前暂停）
thread = {"configurable": {"thread_id": "user_123"}}
app.invoke({"messages": ["帮我转账 1000 元给张三"]}, config=thread)
# --- 此时程序已经结束，状态保存在数据库中等待审批 ---
# 5. 人类审批后恢复工作流
# 假设人类在后台点击了“同意”
app.invoke(None, config=thread) # 传入 None，它会从上次的断点(tools)继续执行方案二：基于原生 Function Calling 拦截（不依赖复杂框架）
如果你使用原生的 OpenAI API 或轻量级框架，可以手动在业务逻辑层拦截。
执行逻辑：
python
# 1. LLM 返回工具调用请求
response = openai.ChatCompletion.create(
model="gpt-4",
messages=history,
tools=[{"type": "function", "function": {"name": "transfer_money", ...}}]
)
message = response.choices[0].message
# 2. 判断是否是敏感操作
if message.tool_calls:
for tool_call in message.tool_calls:
if tool_call.function.name in ["transfer_money", "send_email"]:
# 3. 敏感操作：持久化上下文并通知人类
save_state_to_db(session_id, history, tool_call)
send_approval_request_to_human(session_id, tool_call.function.arguments)
# 直接 return，结束当前 HTTP 请求/线程
return {"status": "waiting_for_approval"}
else:
# 非敏感操作（如查询天气），直接执行
execute_tool(tool_call)
# ------ 异步审批接口 (API Endpoint) ------
@app.route("/approve", methods=["POST"])
def approve_action():
session_id = request.json['session_id']
is_approved = request.json['approved']
# 从数据库恢复
history, tool_call = load_state_from_db(session_id)
if is_approved:
# 人类同意：执行转账
result = execute_transfer(tool_call.arguments)
else:
# 人类拒绝：伪造一个失败/拒绝的结果
result = "Error: 审批被人类管理员拒绝。"
# 将工具执行结果附加到历史中，再次请求 LLM
history.append(message) # 把之前的 tool_call 消息加上
history.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
})
# 唤醒 LLM，让它根据审批结果继续总结
final_response = openai.ChatCompletion.create(messages=history)
return final_response三、 审批流设计的 4 个关键细节（最佳实践）
为了让这个系统在生产环境中安全可靠，你需要处理以下细节：
1. 审批界面的信息呈现（Contextual UI）
不要只给人类一个枯燥的 JSON。审批通知必须清晰翻译出 LLM 的意图。
- 错误示例： 请求调用
transfer(amount=5000, account=12345)，是否同意？ - 正确示例：
- 触发原因： 用户要求退款给客户 A。
- 操作动作： 转账 5,000 元至账户 12345 (客户 A)。
- 风险提示： 金额超过 1,000 元，需高级权限。
- 上下文回溯： [点击查看 AI 与用户的完整对话记录]
2. 状态机的超时机制（Timeout Handling）
由于是异步等待，人类可能忽略了审批请求。
- 设计： 在数据库中记录
wait_since时间戳。跑一个定时任务（Cron Job），如果超过 24 小时未审批，自动将状态标记为rejected_by_timeout，并将超时信息传回给 LLM，让 LLM 通知用户“由于超时，转账已取消”。
3. 允许人类“修改”参数（Modify in Loop）
高级的审批流不仅能“同意/拒绝”，还能让管理员修改数据后放行。
- 场景： LLM 写好了一封发给大客户的邮件，语气稍微有点生硬。
- 设计： 审批界面提供一个文本框，默认填入 LLM 生成的邮件内容。管理员修改文本后点击“发送”。此时传递给工具的参数不再是 LLM 的原始输出，而是管理员修改后的参数。
4. 权限与身份验证（RBAC）
- 细粒度拦截： 查询数据库的操作可能不需要审批，但更新/删除（UPDATE/DELETE）操作需要审批。
- 人员鉴权： 生成的审批链接必须带有 Token（如 JWT），或者要求审批者登录内部系统（SSO），防止审批链接泄露被恶意点击。
四、 总结
设计带“断点”的工作流，本质上是将 LLM 的“逻辑流”与系统的“控制流”解耦。
LLM 只负责输出意图（我要执行敏感操作），而外围的工程框架（LangGraph 或自研的状态机）负责拦截、持久化、通知、接收反馈和恢复。采用这种设计，可以最大程度地保证 AI 代理在执行现实世界任务时的安全性和可控性。