微信扫码
添加专属顾问
我要投稿
LangGraph — 构建多智能体语言模型应用的新利器。核心内容:1. LangGraph的核心概念及其图结构Agent编排2. LangGraph的状态管理机制与持久化功能3. LangGraph支持人机协作及流式输出的用户体验优化4. LangGraph与Dify和Coze的对比分析
一、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系统的启发,该程序以离散的“超级步骤”进行。
超级步骤可以被视为图节点上的单次迭代。并行运行的节点属于同一个超级步骤,而顺序运行的节点则属于不同的超级步骤。
图执行开始时,所有节点都处于同一inactive状态。
当节点在其任何传入边(或“通道”)上收到新消息(状态)时,它将变为active
然后,活动节点运行其函数并进行更新响应。
在每个超级步骤结束时,没有传入消息的节点通过将其标记为 inactive 来投票 halt。
当所有节点 均为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和StateGraph
from 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 START
graph.add_edge(START, "my_node")
graph.add_edge("my_node", "other_node")
END 节点
END 节点是一个特殊节点,它代表一个终端节点。当您想要指定哪些边在完成操作后没有动作时,会引用此节点。一个流程通常只有一个开始节点和结束节点。
from langgraph.graph import END
graph.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 START
graph.add_edge(START, "my_node")
条件入口点
条件入口点允许您根据自定义逻辑从不同的节点开始。您可以从虚拟的 START 节点使用add_conditional_edges 来实现这一点。
from langgraph.graph import START
graph.add_conditional_edges(START, routing_function)
您可以选择提供一个字典,该字典将 routing_function 的输出映射到下一个节点的名称。
graph.add_conditional_edges(START, routing_my,{True: "my_node", False: "other_node"})
53AI,企业落地大模型首选服务商
产品:场景落地咨询+大模型应用平台+行业解决方案
承诺:免费场景POC验证,效果验证后签署服务协议。零风险落地应用大模型,已交付160+中大型企业
2024-10-10
2024-07-13
2024-06-03
2024-04-08
2024-09-04
2024-04-08
2024-08-18
2024-03-28
2024-06-24
2024-07-10
2025-05-08
2025-05-06
2025-04-22
2025-04-18
2025-03-22
2025-03-22
2025-03-15
2025-02-05