LangGraph 入门要点深度解析

一、LangGraph 的核心概念

LangGraph 是一个专为构建复杂、多智能体（Multi-Agent）语言模型应用而设计的开源框架。它由 LangChain Inc. 开发，灵感来源于 Pregel 和 Apache Beam，接口设计借鉴了 NetworkX。LangGraph 允许开发者以图结构的方式定义和编排代理的行为流程，提供了高度的可控性和灵活性。

1. 图结构的Agent编排

在 LangGraph 中，应用被建模为一个有向图（Directed Graph）：

• 节点（Node）：代表一个操作步骤，如调用 LLM、执行工具函数等。

• 边（Edge）：定义节点之间的执行顺序和条件逻辑，支持循环和分支。

这种结构使得复杂的工作流（如多轮对话、任务分解与执行）可以被清晰地表达和管理。

2. 状态管理与持久化

LangGraph 内置了状态管理机制，允许在图的每一步之后自动保存状态。这支持：

• 错误恢复：在出现异常时从最近的检查点恢复执行。

• 人工干预：在关键节点暂停，等待人工审核或输入。

• “时间旅行”：回溯到某个状态，修改后重新执行。

状态的持久化使得应用具有更高的可靠性和可调试性。

3. 人机协作（Human-in-the-Loop）

LangGraph 支持在执行过程中引入人工决策，特别适用于需要专家判断的场景，如医疗诊断、金融分析等。系统可以在某些节点暂停，等待人工输入后再继续执行。

4. 流式输出与实时反馈

通过支持逐个令牌的流式传输，LangGraph 提供了更好的用户体验。用户可以实时看到代理的思考过程和中间结果，增强了交互性和透明度。

 官网：

https://www.langchain.com/langgraph

github地址：

https://github.com/langchain-ai/langgraph

官方文档（英文）：

https://langchain-ai.github.io/langgraph/

LangGraph的架构说明

https://langchain-ai.github.io/langgraph/concepts/agentic_concepts/

5、langgraph与dify,coze对比

LangGraph、Dify 和 Coze 是三种用于构建 AI 应用的工具，它们在定位、目标用户、开发方式和适用场景等方面各有特色。以下是对这三者的深入对比，帮助你根据自身需求做出选择。 

二、LangGraph 的主要功能

• 循环与分支逻辑：支持复杂的控制流，包括条件判断和循环结构。

• 多代理协作：允许多个代理共享状态，协同完成任务。

• 持久化机制：自动保存和恢复状态，支持长时间运行的任务。

• 人工干预支持：在关键节点引入人工审核，提高决策质量。

• 流式输出：实时传输生成内容，提升用户体验。

• 与 LangChain 集成：无缝结合 LangChain 和 LangSmith，构建完整的 LLM 应用生态。

三、Graph的核心概念

LangGraph 的核心是将代理工作流程建模为图表。您可以使用三个关键组件来定义代理的行为：

• 状态：表示应用程序当前快照的共享数据结构。它可以是任何 Python 类型，但通常是  TypedDict或 Pydantic BaseModel。

• 节点：用于编码代理逻辑的 Python 函数。它们接收当前值State作为输入，执行一些计算或副作用，并返回更新后的State。

• 边 ：根据当前条件确定下一步执行哪个操作的 Python 函数State。它们可以是条件分支或固定转换。

通过组合节点（Nodes）和边（Edges），您可以创建复杂的循环工作流，使其State随时间推移而演化。然而，真正的强大之处在于 LangGraph 对 的管理方式State。需要强调的是：Nodes和Edges只不过是 Python 函数而已——它们可以包含 LLM 代码，也可以只是经典的 Python 代码。

简而言之：节点负责工作，边负责告诉下一步做什么。

LangGraph 的底层图算法使用消息传递来定义通用程序。当一个节点完成其操作时，它会沿着一条或多条边向其他节点发送消息。这些接收节点随后执行其函数，将生成的消息传递给下一组节点，并继续执行该过程。受 Google Pregel系统的启发，该程序以离散的“超级步骤”进行。

超级步骤可以被视为图节点上的单次迭代。并行运行的节点属于同一个超级步骤，而顺序运行的节点则属于不同的超级步骤。

1. 图执行开始时，所有节点都处于同一inactive状态。

2. 当节点在其任何传入边（或“通道”）上收到新消息（状态）时，它将变为active 

3. 然后，活动节点运行其函数并进行更新响应。

4. 在每个超级步骤结束时，没有传入消息的节点通过将其标记为 inactive 来投票 halt。

5. 当所有节点 均为inactive且没有消息在传输时，图执行终止。

状态图StateGraph

该类StateGraph是要使用的主要图形类。它由用户定义的State对象参数化。

编译图

要构建图，首先要定义状态，然后添加节点和边，最后进行编译。编译图究竟是什么？为什么需要编译？

编译是一个非常简单的步骤。它会对图的结构进行一些基本检查（例如，没有孤立节点等）。你还可以在其中指定运行时参数，例如检查点和断点。只需调用以下.compile方法即可编译图： 

graph = graph_builder.compile(...)

您必须先编译您的图表，然后才能使用它。

状态（State）

定义图时，你做的第一件事是定义图的 状态 。 状态 包含图的 模式 以及 归约器函数，它们指定如何将更新应用于状态。 状态 的模式将是图中所有 节点 和 边 的输入模式，可以是 TypedDict 或Pydantic BaseModel。所有 节点 将发出对 状态 的更新，这些更新然后使用指定的 归约器 函数进行应用。

模式（Schema）

指定图模式的主要记录方式是使用TypedDict，也支持使用 Pydantic BaseModel作为图状态，以添加默认值和附加数据验证。

默认情况下，图将具有相同的输入和输出模式。如果您想更改此设置，也可以直接指定显式的输入和输出模式。当您拥有大量键，并且其中一些显式用于输入，另一些用于输出时，这非常有用。

节点（Nodes）

在 LangGraph 中，节点通常是 Python 函数（同步或 async ），其中第一个位置参数是状态，（可选），第二个位置参数是“配置”，包含可选的可配置参数（例如 thread_id ）。

类似于 NetworkX ，您可以使用add_node方法将这些节点添加到图形中。 示例代码如下： 

# 从langgraph.graph模块导入START和StateGraphfrom langgraph.graph import START, StateGraph
# 定义一个节点函数my_node，接收状态和配置，返回新的状态def my_node(state, config):    return {"x": state["x"] + 1,"y": state["y"] + 2}
# 创建一个状态图构建器builder，使用字典类型作为状态类型builder = StateGraph(dict)
# 向构建器中添加节点my_node，节点名称将自动设置为'my_node'builder.add_node(my_node) # node name will be 'my_node'
# 添加一条边，从START到'my_node'节点builder.add_edge(START, "my_node")# 编译状态图，生成可执行的图graph = builder.compile()
# 调用编译后的图，传入初始状态{"x": 1}print(graph.invoke({"x": 3,"y":2}))

在幕后，函数被转换为 RunnableLambda ，它为您的函数添加了批处理和异步支持，以及本地跟踪和调试。

如果您在没有指定名称的情况下将节点添加到图形中，它将被赋予一个默认名称，该名称等同于函数名称。

START 节点

START 节点是一个特殊节点，它代表将用户输入发送到图形的节点。引用此节点的主要目的是确定哪些节点应该首先被调用。

from langgraph.graph import STARTgraph.add_edge(START, "my_node")graph.add_edge("my_node", "other_node")

END 节点

END 节点是一个特殊节点，它代表一个终端节点。当您想要指定哪些边在完成操作后没有动作时，会引用此节点。一个流程通常只有一个开始节点和结束节点。

from langgraph.graph import ENDgraph.add_edge("other_node", END)

边（Edges）

边定义了逻辑如何路由以及图形如何决定停止。这是您的代理如何工作以及不同节点如何相互通信的重要部分。有一些关键类型的边。

• 普通边：直接从一个节点到下一个节点。

• 条件边：调用一个函数来确定下一个要转到的节点。

• 入口点：用户输入到达时首先调用的节点。

• 条件入口点：调用一个函数来确定用户输入到达时首先调用的节点。

一个节点可以有多个输出边。如果一个节点有多个输出边，则所有这些目标节点将在下一个超级步骤中并行执行。

普通边

如果您总是想从节点 A 到节点 B，您可以直接使用add_edge方法。

graph.add_edge("node_a", "node_b")

条件边

如果您想选择性地路由到一个或多个边（或选择性地终止），您可以使用add_conditional_edges方法。此方法接受节点的名称和一个“路由函数”，该函数将在该节点执行后被调用。

graph.add_conditional_edges("node_a", routing_function)

类似于节点， routing_function 接受图形的当前 state 并返回一个值。

默认情况下，返回值 routing_function 用作要将状态发送到下一个节点的节点名称（或节点列表）。

所有这些节点将在下一个超级步骤中并行运行。

您可以选择提供一个字典，该字典将 routing_function 的输出映射到下一个节点的名称。

graph.add_conditional_edges("node_a", routing_function, {True: "node_b", False:"node_c"})

入口点

入口点是图形启动时运行的第一个节点。您可以从虚拟的 START 节点使用 add_edge 方法到要执行的第一 个节点，以指定进入图形的位置。

from langgraph.graph import STARTgraph.add_edge(START, "my_node")

条件入口点

条件入口点允许您根据自定义逻辑从不同的节点开始。您可以从虚拟的 START 节点使用add_conditional_edges 来实现这一点。

from langgraph.graph import STARTgraph.add_conditional_edges(START, routing_function)

您可以选择提供一个字典，该字典将 routing_function 的输出映射到下一个节点的名称。

graph.add_conditional_edges(START, routing_my,{True: "my_node", False: "other_node"})