AGENT

构建人在回路的智能体工作流

了解如何在LangGraph中设置人在回路中的智能体工作流

admin

• 12 min read
构建人在回路的智能体工作流
AI模型价格对比 | AI工具导航 | ONNX模型库 | Vibe Coding教程 | PLC在线仿真器 | Tripo 3D | Meshy AI | ElevenLabs | KlingAI | ArtSpace | Phot.AI | InVideo

像OpenAI的GPT-5.4和Anthropic的Opus 4.6这样的最新LLM在执行长时间运行的智能体任务方面展示了出色的能力。

因此，我们看到在个人和企业环境中越来越多地使用LLM智能体来完成复杂的任务，例如运行财务分析、构建应用程序和进行广泛的研究。

这些智能体，无论是高度自主设置的一部分还是预定义的工作流，都可以使用工具执行多步任务以实现目标，只需最少的人工监督。

然而，"最少"并不意味着零人工监督。

相反，人工审查仍然很重要，因为LLM具有固有的概率性质和潜在的出错可能。

这些错误可能会沿着工作流传播和放大，尤其是当我们将众多智能体组件串联在一起时。

你可能已经注意到智能体在编码领域取得的令人印象深刻的进展。原因是代码相对容易验证（即它要么运行成功要么失败，反馈立即可见）。

但在内容创作、研究或决策等领域，正确性往往是主观的，更难自动评估。

这就是为什么**人在回路中（HITL）**设计仍然至关重要。

在本文中，我们将介绍如何使用LangGraph为内容生成和在Bluesky上的发布设置人在回路中的智能体工作流。你可以在这里找到配套的GitHub仓库。

1、LangGraph入门

LangGraph（LangChain生态系统的一部分）是一个用于构建智能体工作流的低级智能体编排框架和运行时。

它是我的首选框架，因为它具有高度的控制和可定制性，这对于生产级解决方案至关重要。

虽然LangChain提供了一个中间件对象（HumanInTheLoopMiddleware）来轻松开始在智能体调用中引入人工监督，但它是在高抽象层次上完成的，掩盖了底层机制。

相比之下，LangGraph不会抽象提示或架构，从而让我们获得所需的更精细的控制程度。它明确允许我们定义：

  • 数据如何在步骤之间流动
  • 在何处进行决策和代码执行
  • 在何处需要人工干预

因此，我们将使用LangGraph在智能体工作流中演示HITL概念。

区分智能体工作流和自主AI智能体也很有帮助。

智能体工作流有预定的路径，设计为按定义的顺序执行，LLM和/或智能体集成到一个或多个组件中。另一方面，AI智能体自主地规划、执行和迭代以实现目标。

在本文中，我们专注于智能体工作流，在其中我们有目的地将人工检查点插入到预定义的流程中。

无
比较智能体工作流和LLM智能体

2、示例工作流

对于我们的示例，我们将构建一个社交媒体内容生成工作流，如下所示：

无
内容生成工作流

1) 用户输入感兴趣的主题（例如，"Anthropic的最新消息"）。

  • 网页搜索节点使用Tavily工具在线搜索匹配主题的文章。
  • 选择搜索结果中的顶部条目，并输入到内容创作节点中的LLM，以生成社交媒体帖子。
  • 在审核节点中，有两个人工审核检查点：(i) 向人类展示生成的内容以批准、拒绝或编辑；(ii) 经批准后，工作流触发Bluesky API发布工具，并请求最终确认后再将其发布到网上。

这是从终端运行时的样子：

无
终端中的工作流运行

这是我Bluesky个人资料上的实时帖子：

无
从工作流生成的Bluesky社交媒体帖子 
Bluesky是一个类似于Twitter (X)的社交平台，之所以在本演示中选择它，是因为它的API更容易访问和使用。

3、关键概念

LangGraph中HITL设置的核心机制是中断的概念。

中断（在LangGraph中使用interrupt()和Command）使我们能够在特定点暂停图执行，向人类显示某些信息，并在恢复工作流之前等待他们的输入。

Command是一个多功能对象，允许我们更新图状态（update）、指定要执行的下一个节点（goto），或捕获用于恢复图执行的值（resume）。

流程如下所示：

(1) 一旦到达interrupt()函数，执行暂停，传入它的payload将显示给用户。在interrupt中传入的payload通常是JSON或字符串格式，例如，

decision = interrupt("我们午饭应该吃KFC吗？")  # 向用户显示的字符串

(2) 用户响应后，我们将响应值传递给图以恢复执行。它涉及在重新调用图时使用Command及其resume参数：

if human_response == "yes":
    return graph.invoke(Command(resume="KFC"))
else:
    return graph.invoke(Command(resume="McDonalds"))

(3) resume中的响应值在decision变量中返回，节点将使用它进行节点执行的其余部分和后续图流：

if decision == "KFC":
    return Command(goto="kfc_order_node", update={"lunch_choice": "KFC")
else:
    return Command(goto="mcd_order_node", update={"lunch_choice": "McDonalds")

中断是动态的，可以放置在代码中的任何位置，而静态断点是在特定节点之前或之后固定的。

也就是说，我们通常将中断放置在节点内或在图执行期间调用的工具内。

最后，让我们谈谈检查点器。当工作流在中断处暂停时，我们需要一种方法来保存其当前状态，以便以后可以恢复。

因此，我们需要一个检查点来持久化状态，以便在中断暂停期间状态不会丢失。将检查点视为图状态在给定时间点的快照。

对于开发，将状态保存在内存中与InMemorySaver检查点器是可接受的。

对于生产，最好使用像Postgres或Redis这样的存储。考虑到这一点，在本示例中我们将使用**SQLite检查点**而不是内存存储。

为了确保图准确地在中断发生的位置恢复，我们需要传递和使用相同的线程ID。

将线程视为单个执行会话（就像一个独立的对话），每个会话都有一个唯一的ID，并维护自己的状态和历史。

线程ID在每次图调用时传递给config，以便LangGraph知道在中断后从哪个状态恢复。

现在我们已经涵盖了中断、Command、检查点和线程的概念，让我们进入代码详解。

4、代码详解

由于重点是人在回路中的机制，我们不会涵盖完整的代码设置。请访问GitHub仓库获取完整实现。

4.1 初始设置

我们首先安装所需的依赖项，并为Bluesky、OpenAI、LangChain、LangGraph和Tavily生成API密钥。

# requirements.txt
langchain-openai>=1.1.9
langgraph>=1.0.8
langgraph-checkpoint-sqlite>=3.0.3
openai>=2.20.0
tavily-python>=0.7.21

# env.example
export OPENAI_API_KEY=your_openai_api_key
export TAVILY_API_KEY=your_tavily_api_key
export BLUESKY_HANDLE=yourname.bsky.social
export BLUESKY_APP_PASSWORD=your_bluesky_app_password

4.2 定义状态

我们设置State，这是一个共享的、结构化的数据对象，作为图的中心内存。它包括捕获关键信息的字段，如帖子内容和批准状态。

post_data键是存储生成帖子内容的位置。

4.3 节点级别中断

我们之前提到，中断可以发生在节点级别或工具调用内。让我们通过设置人工审核节点来了解前者是如何工作的。

审核节点的目的是暂停执行并向用户展示草稿内容以供审核。

在这里，我们看到了interrupt()的实际应用（第8到13行），图执行在节点函数的第一部分暂停。

传入interrupt()的details键包含生成的内容，而action键触发一个处理函数（handle_content_interrupt()）来支持审核：

生成的内容在终端中打印供用户查看，他们可以原样批准、直接拒绝或在批准前直接在终端中编辑。

基于决策，处理函数返回三个值之一：

  • True（批准），
  • False（拒绝），或
  • 对应于用户编辑内容的字符串值（edited）。

这个返回值使用graph.invoke(Command=resume…)传回审核节点，它从调用interrupt()的地方恢复执行（第15行），并决定接下来去哪个节点：批准、拒绝，或编辑内容然后继续批准。

4.4 工具级别中断

中断也可以在工具调用级别定义。这在内容在Bluesky上在线发布之前的批准节点中的下一个人工审核检查点中演示。

我们不是将interrupt()放在节点内，而是将其放在通过Bluesky API创建帖子的publish_post工具内：

就像我们在节点级别看到的那样，我们调用一个处理函数（handle_publish_interrupt）来捕获人类决策：

此审核步骤的返回值是：

  • {"action": "confirm"}，或
  • {"action": "cancel"}，

publish_post工具中的代码的后半部分（即第19行起）使用这个返回值来决定是否在Bluesky上继续发布帖子。

4.5 使用检查点器设置图

接下来，我们将节点连接在一个图中进行编译，并引入一个SQLite检查点器，在每个中断处捕获状态的快照。

SQLite默认只允许创建数据库连接的线程使用它。由于LangGraph使用线程池进行检查点写入，我们需要设置check_same_thread=False以允许这些线程也访问该连接。

4.6 使用配置设置完整工作流

图准备好后，我们现在将其放入一个启动内容生成管道的工作流中。

此工作流包括配置一个线程ID，该ID传递给每个graph.invoke()。这个ID是将调用联系在一起的链接，以便图在中断处暂停并从停止的地方恢复。

你可能已经注意到上面代码中的__interrupt__键。它只是一个LangGraph在命中interrupt()时添加到结果中的特殊键。

换句话说，它是指示图执行已暂停并正在等待人工输入再继续的主要信号。

通过在while循环中放置__interrupt__，意味着循环持续检查中断是否仍在进行。一旦中断解决，键就会消失，while循环退出。

工作流完成后，我们可以这样运行它：

run_hitl_workflow(query="latest news about Anthropic")

5、中断的最佳实践

虽然中断在启用HITL工作流方面非常强大，但如果使用不当，它们可能会造成中断。

因此，我推荐阅读这篇LangGraph文档。以下是一些需要记住的实用规则：

  • 不要在中断调用外部包装try/except块，否则它们将无法正确暂停执行
  • 每次保持中断调用的顺序相同，不要跳过或重新排列它们
  • 只向中断传递JSON安全值，避免复杂对象
  • 确保中断前的任何代码可以安全地运行多次（即幂等性），或将其移到中断后

例如，我在网页搜索节点中遇到了一个问题，我在Tavily搜索后立即放置了一个中断。意图是暂停并允许用户审核用于内容生成的搜索结果。

但因为中断通过重新运行调用它们的节点来工作，节点只是重新运行了网页搜索，并传递了一组与我之前批准的不同搜索结果。

因此，中断在作为操作之前的门时效果最佳，但如果我们在非确定性步骤（如搜索）之后使用它们，我们需要持久化结果，否则在恢复时有得到不同结果的风险。

6、结束语

人工审查在智能体工作流中似乎是一个瓶颈，但它仍然至关重要，尤其是在结果难以验证或主观的领域。

LangGraph通过中断和检查点使构建HITL工作流变得简单明了。

因此，挑战在于决定在哪里放置这些人工决策点，以在监督和效率之间取得良好的平衡。

原文链接: Building Human-In-The-Loop Agentic Workflows

汇智网翻译整理，转载请标明出处