超越 input()：使用 LangGraph 构建生产就绪的人机交互 AI 代理
Source: Dev.to
请提供您希望翻译的正文内容，我将按照要求将其翻译为简体中文，并保留原始的格式、Markdown 语法以及技术术语。谢谢！
介绍
如果你已经构建了一个简单的聊天机器人或 CLI 工具，你可能已经习惯使用 Python 可靠的 input() 函数。它非常适合快速脚本：提出问题，等待答案，完成。但当你需要构建更复杂的东西时会怎样？如果你的 AI 代理需要在工作流中途暂停，等待人工批准（可能是几小时甚至几天），然后恰好在它离开的地方继续运行，该怎么办？
这正是 input() 完全失效的地方。
input() 的问题
说实话：input() 是一个同步阻塞器。它会冻结整个程序，等待有人输入内容并按下 Enter。这在生产环境中实际上意味着：
想象一下，你正在构建一个用于金融交易的 AI 审批系统。某位经理在星期五下午 4 点收到一条关于需要审查的 5 万美元交易的通知。他正准备下班去度周末。使用 input() 时，你的流程就会……停在那里，阻塞住。
而使用 LangGraph 时，工作流会 暂停，释放所有资源，并耐心等待。当经理在星期一上午批准后，流程会无缝继续。
Source:
等等，我们之前见过这个吗？BizTalk 连接
如果你使用过 Microsoft BizTalk Server，这可能会让你感到熟悉。LangGraph 的检查点系统在概念上类似于 BizTalk 的脱水/再水化机制——这并非偶然。两者都在解决同一个根本问题：如何在不浪费资源的情况下暂停一个长时间运行的工作流？
BizTalk 的做法：脱水与再水化
在 BizTalk Server 中，当一个编排（工作流）需要等待消息、超时或人工批准时，BizTalk 并不会让它一直驻留在内存中。相反，它会：
- 脱水 编排实例，将其完整状态序列化并存入 MessageBox 数据库。
- 将实例从内存中移除，释放服务器资源。
- 当触发条件到达（收到消息、超时结束）时，BizTalk 通过从数据库加载状态来 再水化 实例。
- 从中断的地方继续执行。
BizTalk 脱水 vs. LangGraph 检查点
我从 BizTalk 学到的东西
在使用 BizTalk Server 的过程中，我可以告诉你 dehydration/rehydration（脱水/再水化）模式对企业工作流至关重要。原因如下：
- Purchase Order Approval – 订单进入后会被验证，然后等待经理批准。在 BizTalk 中，这个编排会脱水（状态保存）到数据库。经理可能在数小时、数天甚至数周后才批准；当他们批准时，编排会再水化并继续处理。
- Long‑Running Transactions – 跨越数天、数周甚至数月的多步骤业务流程（例如保险理赔、合同批准、监管工作流）无法一直驻留在内存中。BizTalk 将状态存储在 SQL Server 中，并跟踪关联集（correlation sets）以将传入的消息匹配到正确的编排实例。我见过编排脱水后等待外部批准或第三方响应长达数周。
- Server Restarts – 如果 BizTalk Server 崩溃或重启，所有已脱水的编排都会因为已持久化在数据库中而存活下来。服务器恢复后，它们会自动继续运行。
LangGraph 将同样经过实战检验的模式引入 AI 工作流。不同之处在于，XML 消息和关联集被 AI 代理状态和线程 ID 取代。消息箱数据库（MessageBox database）被 PostgreSQL 或 SQLite 替代。但核心概念——持久化状态、释放资源、稍后恢复——是完全相同的。
趣闻: BizTalk 的 MessageBox 数据库本质上是一个巨大的状态机。每个编排实例的状态连同其关联属性一起存储，使 BizTalk 能够将传入的消息路由到正确的等待中的编排。LangGraph 的检查点（checkpointer）对 AI 代理工作流执行相同的功能——thread_id 就是你的关联集！
人机交互循环的三大支柱
构建生产级的人机交互循环系统需要三块关键组成部分协同工作。下面我们逐一拆解。
1. 检查点（Checkpointing）：你的代理的记忆
在代理能够暂停并恢复之前，它需要记忆。不是普通的记忆，而是持久、可靠的记忆，能够在崩溃、重启，甚至迁移到不同服务器后依然存在。
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph
# In‑memory checkpointer (great for development)
checkpointer = MemorySaver()
# For production, use PostgresSaver or SQLiteSaver
graph = workflow.compile(checkpointer=checkpointer)关键洞见： 没有 checkpointer，中断根本无法工作。就这么简单。checkpointer 保存完整的执行状态——变量、上下文、进度——因此当你数小时或数天后回来时，任何东西都不会丢失。
把它想象成保存电子游戏进度。当你设定检查点时，你不仅仅保存了一个变量；你保存了精准的位置、背包、生命值、任务进度——所有一切。恢复时，你会直接回到原来的位置。
2. In
(原始内容在此被截断。请根据需要继续后续章节。)
中断：暂停按钮
既然我们已经有了记忆，就需要实际暂停的能力。LangGraph 为你提供了两种实现方式：
静态中断 – “始终在此暂停”
当你知道某些节点始终需要人工审查时使用这些。
# Pause BEFORE a node executes
graph = workflow.compile(
checkpointer=checkpointer,
interrupt_before=["sensitive_action"]
)
# Pause AFTER a node executes (useful for review)
graph = workflow.compile(
checkpointer=checkpointer,
interrupt_after=["generate_response"]
)静态中断非常适用于合规场景。
每一次内容生成可能在发布前需要人工批准，或者每一次数据库删除可能需要第二双眼睛审查。
动态中断 – 条件暂停
当是否暂停取决于运行时的实际情况时使用这些。
from langgraph.types import interrupt
def process_transaction(state):
amount = state["transaction_amount"]
# Only pause for high‑value transactions
if amount > 10_000:
human_decision = interrupt({
"question": f"Approve transaction of ${amount}?",
"transaction_details": state["details"]
})
if human_decision.get("approved") != True:
return {
"status": "rejected",
"reason": human_decision.get("reason")
}
# Continue with transaction
return {"status": "approved", "processed": True}这非常强大。你的 AI 可以智能地决定何时需要帮助：
- 低置信度分数 → 暂停并询问。
- 交易金额 > $10 K → 暂停并询问。
- 其他情况 → 继续执行。
何时使用哪种？
| 类型 | 典型使用场景 |
|---|---|
| 静态 | 合规审查、最终批准、任何“始终在此暂停”的情形 |
| 动态 | 高价值交易、低置信度预测、边缘案例 |
命令：恢复按钮
所以你已经暂停了工作流，人工审查了它，并作出了决定。接下来怎么办？这就是 Command 发挥作用的地方：
from langgraph.types import Command
# Resume the paused workflow with the human's input
result = graph.invoke(
Command(resume={
"approved": True,
"notes": "Verified customer identity"
}),
config={"configurable": {"thread_id": "transaction-123"}}
)工作原理
BEFORE (workflow paused at interrupt):
human_input = interrupt({...}) # ← Waiting here, state saved
RESUME (human provides decision):
graph.invoke(Command(resume={"approved": True}), config)
↓
Data flows back to interrupt()
↓
AFTER:
human_input = {"approved": True} # ← Now has the human's response!工作流正好从 此处 继续。图不会重新启动或重放所有内容；它会在中断的地方继续执行，人工输入已经在节点中就位。
可视化概览
综合示例
以下是实际生产工作流的示例：
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
from langgraph.graph import StateGraph
# 1. Set up checkpointing
checkpointer = MemorySaver()
# 2. Define your workflow with dynamic interrupts
def review_node(state):
if state["risk_score"] > 0.8:
decision = interrupt({
"message": "High‑risk detected. Review required.",
"data": state["analysis"]
})
return {"approved": decision["approved"]}
return {"approved": True}
# 3. Compile with persistence
graph = workflow.compile(checkpointer=checkpointer)
# 4. Invoke the workflow
config = {"configurable": {"thread_id": "workflow-123"}}
result = graph.invoke(initial_state, config)
# 5. Check if paused
state = graph.get_state(config)
if bool(state.next):
print("Workflow paused, waiting for human input")
# Later, when the human responds...
graph.invoke(
Command(resume={"approved": True}),
config
)为什么这很重要
在生产环境的 AI 系统中，你无法承受无限期阻塞。你需要能够实现以下功能的工作流：
- 在不消耗资源的情况下，暂停以进行人工判断 without。
- 能够在重启后继续运行，并处理数百个并发执行。
- 允许在星期五暂停，在星期一恢复 without losing any context。
这并不是 input() 所设计的用途，但正是 LangGraph 的 human‑in‑the‑loop 系统所构建来处理的。区别不仅在于技术层面——而是在于它是玩具还是你真正可以部署的工具。
最后思考
构建 human‑in‑the‑loop AI 并不仅仅是给工作流加一个暂停按钮，而是要创建能够尊重人类决策的异步性和不可预测性的系统。你的管理者不会按照代码的时间表工作；他们会在周末休息，需要时间思考。借助 LangGraph 的静态和动态中断以及 Command 恢复机制，你可以为他们提供这种灵活性，同时保持 AI 流水线的稳健、可扩展和可投产。
结论
在做决定之前，你可能需要征求他人的意见。
LangGraph 的架构——由 checkpointing（检查点）、interrupts（中断） 和 commands（命令） 三大支柱构成——正是对这一现实的回应。它为你提供了构建与人类协同工作的 AI 系统的工具，而不是与之对立的系统。
因此，下次当你准备使用 input() 时，问问自己：我是在编写脚本，还是在构建一个 AI（AI Agents）系统？ 如果是后者，你已经知道该怎么做。
祝构建愉快！
谢谢，
Sreeni Ramadorai