langgraph agents hitl checkpoints multi-agent production

用于生产代理的LangGraph：检查点、HITL以及多代理图结构

为什么链条会在循环中停滞又恢复运行，以及 LangGraph 的节点、边和检查点保存机制如何让生产代理具备可操作性，并支持人工进行循环审核。

发布者：REACTAPP.TOP加入本站 2026年9月26日阅读约 7 分钟1610 词

来源: LangGraph Explained: Build Production-Ready AI Agents in 2026

什么是LangGraph？

LangGraph是一种运行时环境，用于将具有状态的多步骤智能体构建为明确的图结构：节点负责执行任务，边决定下一步该执行什么操作，而共享状态则会在各步骤之间保持不变——即便在出现故障或人类暂停操作后也是如此。

两年前，许多“智能体”实际上只是围绕大型语言模型运行的while循环，且不具备恢复能力。一旦在任务执行过程中崩溃，就只能从头开始重新启动。而图结构则让内存管理、分支处理以及任务恢复成为可能。

为何链式结构不够用

LCEL链式结构在处理线性流程时表现优异，但当需要循环、条件路由、持久化检查点或任务执行过程中的用户审批时，它们就显得力不从心。这些都属于图结构才能解决的难题。

核心概念：节点、边与状态

状态——任务执行过程中所携带的带类型字段
节点——用于更新状态的函数

边 — 固定或条件性的转换

编译 — 验证结构，添加检查点

调用/流式处理 — 使用配置执行

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

# Define your state
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    next_action: str
    final_answer: str

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

# 1. Define a tool
@tool
def search_web(query: str) -> str:
    """Search the web for current information."""
    # Connect to your actual search API here
    return f"Search results for: {query}"

# 2. Define the LLM with tools bound
llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [search_web]
llm_with_tools = llm.bind_tools(tools)

# 3. Define node functions
def call_llm(state):
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state):
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "run_tools"
    return END

# 4. Build the graph
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_llm)
workflow.add_node("run_tools", ToolNode(tools))

workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue)
workflow.add_edge("run_tools", "agent")

# 5. Compile with a checkpointer
from langgraph.checkpoint.sqlite import SqliteSaver
memory = SqliteSaver.from_conn_string("agents.db")
graph = workflow.compile(checkpointer=memory)

# 6. Run with a thread ID for persistent state
config = {"configurable": {"thread_id": "user-123"}}
result = graph.invoke({"messages": [("user", "What is the latest GPT model?")]}, config)
print(result["messages"][-1].content)

# Add a human approval gate before a risky action
workflow.add_node("human_review", interrupt(before="execute_payment"))

pip install langgraph langchain-openai psycopg2-binary

状态管理的实际运作方式

更新通过还原器合并。配置错误的还原器会导致消息重复或字段丢失。应将还原器的选择视为架构审查的一部分。

第一个智能体（Python）

定义状态 → 添加节点 → 连接边 → 编译 → 使用保存器时附带线程ID进行调用。

人工干预机制

在产生副作用之前中断流程；通过审批数据继续执行。这就是演示版本与生产环境智能体之间的操作差异。

多智能体模式

监督节点与专家节点之间的数据传递、并行分支以及子图都可以清晰地映射到具有共享状态或分区状态的命名节点上。

持久化与恢复

基于数据库的检查点机制能够保证进程重启后的数据不丢失，而内存中的保存方式仅适用于测试。

何时选择 LangGraph

当控制流为非线性、需要暂停或恢复时，可使用 LangGraph；对于无需分支的简单单次 RAG 查询，建议使用更简单的结构。

总结

生产环境中的应用需要明确的状态定义和边关系——而非依赖循环来实现功能。LangGraph 是一种通过代码实现这种结构的可靠方式。

无需关注市场上的宣传噱头，其真正的工程价值在于具备检查点功能的控制流以及 HITL 机制。教程应重点讲解这些基础功能。

需为节点的启动/结束事件添加监控，以便追踪究竟是哪个分支被执行。没有遥测功能的图结构在凌晨两点就会变成黑箱。

提示信息旁应附带版本图定义。若在不进行迁移的情况下重命名节点，就会中断那些仍引用旧名称的线程。

无需理会市场推广中的冗余信息；其工程价值在于带检查点的控制流与HITL机制。教程应聚焦于这些基础概念。