使用 LangGraph 构建人机协同代理工作流

作者
  • avatar
    姓名
    Nino
    职业
    Senior Tech Editor

从简单的线性 LLM 提示词 (Prompt) 转向复杂的多步代理工作流 (Agentic Workflows)，标志着 AI 应用开发进入了一个全新的阶段。虽然完全自主的智能体是我们的最终目标，但由于目前大语言模型在推理可靠性方面的局限性，一种更为严谨的架构——人机协同 (Human-In-The-Loop, HITL) 变得至关重要。通过在代理循环中引入人工干预，开发者可以确保 AI 的输出更加准确、安全且符合用户意图。本教程将详细介绍如何使用 LangGraph 构建这类高级系统，并利用 n1n.ai 提供的极速 API 为底层智能提供动力。

为什么人机协同 (HITL) 是必不可少的

完全自主的智能体经常会陷入“死循环”或“幻觉级联”，即一个错误的决策会导致后续步骤全面崩溃。HITL 模式通过以下方式降低了这些风险：

  1. 审批高风险操作：例如执行代码、发起金融转账或发送外部邮件。
  2. 编辑代理状态：在代理继续执行之前，修正其误解的变量或优化其搜索查询。
  3. 提供反馈引导：当代理进入不确定状态时，由人类提供关键信息。

为了实现这些模式，你需要一个低延迟、高可靠性的后端支持。使用 n1n.ai 可以确保你的代理在交互过程中保持极速响应，为实时的人机协作提供流畅的体验。

LangGraph 实现 HITL 的核心概念

LangGraph 引入了几个关键原语，使 HITL 的实现变得非常简单：

  • 状态管理 (State Management)：一个中心化的对象，记录工作流的所有进度。
  • 检查点 (Checkpoints)：在特定时间点对状态进行的持久化快照。
  • 中断 (Interrupts)：在特定节点运行前或运行后暂停执行，等待外部输入的能力。

实战指南：构建具有 HITL 功能的研究助理

我们将构建一个研究代理，它会进行网页搜索，但在撰写最终报告之前需要人工审核。

1. 定义状态与图结构

首先，我们定义状态模式 (Schema) 和图中的节点。

from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END

class AgentState(TypedDict):
    task: str
    research_notes: List[str]
    approved: bool
    final_report: str

def research_node(state: AgentState):
    # 模拟研究逻辑，实际应用中可调用搜索 API
    return {"research_notes": ["发现了关于 LLM 延迟的数据", "发现了关于 HITL 模式的数据"]}

def human_review_node(state: AgentState):
    # 此节点仅作为人工干预的占位符
    pass

def final_report_node(state: AgentState):
    return {"final_report": f"基于以下笔记生成的报告: {state['research_notes']}"}

2. 配置中断与持久化

HITL 的核心在于 interrupt_before 参数。它告诉 LangGraph 在运行指定节点之前停止执行，并将状态保存到检查点。使用 n1n.ai 提供的稳定 API，可以确保状态更新过程中的并发处理更加稳健。

from langgraph.checkpoint.memory import MemorySaver

builder = StateGraph(AgentState)
builder.add_node("researcher", research_node)
builder.add_node("reviewer", human_review_node)
builder.add_node("writer", final_report_node)

builder.set_entry_point("researcher")
builder.add_edge("researcher", "reviewer")
builder.add_edge("reviewer", "writer")
builder.add_edge("writer", END)

# 初始化内存以保存状态
memory = MemorySaver()
graph = builder.compile(checkpointer=memory, interrupt_before=["reviewer"])

3. 执行工作流并处理人工输入

当我们运行这个图时，它会执行到 researcher 节点后停止。我们必须在提供输入后手动恢复它。

config = {"configurable": {"thread_id": "session_123"}}

# 初始运行
for event in graph.stream({"task": "分析 HITL 趋势", "approved": False}, config):
    print(event)

# 此时图在 'reviewer' 节点前暂停。我们可以检查状态并更新它。
current_state = graph.get_state(config)
print(f"当前笔记内容: {current_state.values['research_notes']}")

# 人工批准并继续
graph.update_state(config, {"approved": True}, as_node="reviewer")
for event in graph.stream(None, config):
    print(event)

高级代理工作流的进阶技巧

  1. 状态编辑与“时光旅行”：LangGraph 允许你“分叉”线程。如果代理犯了错，你可以回溯到之前的检查点，修改状态（例如 state['research_notes'].pop()），然后从该点重新运行。这对于调试复杂的 RAG 管道极其有用。
  2. 动态路由与置信度阈值：可以使用条件边 (Conditional Edges) 来实现自动化升级。例如，当 LLM 的置信度评分 confidence < 0.8 时，自动将任务路由到 human_escalation 节点。
  3. API 性能优化：在构建需要多步 LLM 调用的代理时，API 的响应速度决定了用户体验。通过 n1n.ai，你可以一键接入 DeepSeek-V3、Claude 3.5 Sonnet 等顶级模型，其优化的路由算法能显著降低多轮对话的累计延迟。

总结

构建强大的 AI 代理不仅仅是写一段提示词，更需要建立一个结构化的环境，让渡一部分控制权给人类进行干预、修正和引导。借助 LangGraph 的检查点和中断功能，你可以构建出既强大又安全的生产级代理工作流。对于希望扩展这些工作流的企业来说，选择一个可靠的 API 合作伙伴是迈向成功的下一步。

Get a free API key at n1n.ai

参考来源：https://towardsdatascience.com/building-human-in-the-loop-agentic-workflows/