LangGraph 入门要点深度解析

向量数据库大模型数据库

picture.image

一、LangGraph 的核心概念

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

  1. 图结构的Agent编排

在 LangGraph 中,应用被建模为一个有向图(Directed Graph):

  • 节点(Node):代表一个操作步骤,如调用 LLM、执行工具函数等。
  • 边(Edge):定义节点之间的执行顺序和条件逻辑,支持循环和分支。

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

  1. 状态管理与持久化

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

  • 错误恢复:在出现异常时从最近的检查点恢复执行。
  • 人工干预:在关键节点暂停,等待人工审核或输入。
  • “时间旅行”:回溯到某个状态,修改后重新执行。

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

  1. 人机协作(Human-in-the-Loop)

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

  1. 流式输出与实时反馈

通过支持逐个令牌的流式传输,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 | Dify | Coze | | --- | --- | --- | --- | | 开发方式 | 原生代码(Python) | 低代码(可视化 + 配置) | 无代码(可视化 + 模板) | | 灵活性 | 高 | 中 | 低 | | 学习曲线 | 陡峭 | 中等 | 平缓 | | 适用用户 | 开发者、AI 工程师、研究人员 | 企业用户、开发者、产品经理 | 初级用户、C 端用户、非技术背景人员 | | 适用场景 | 复杂 AI 应用、企业级应用、研究项目 | 企业级 AI 应用开发、生产环境部署 | 快速构建聊天机器人、简单 AI 应用 | | 部署方式 | 本地部署、云部署 | 本地部署、云部署 | 云部署 | | 可定制性 | 高 | 中 | 低 | | 社区支持 | 活跃 | 活跃 | 活跃 |

二、LangGraph 的主要功能

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

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

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

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

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

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

picture.image

三、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和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"}) 

picture.image

0
0
0
0
关于作者
关于作者

文章

0

获赞

0

收藏

0

相关资源
CV 技术在视频创作中的应用
本次演讲将介绍在拍摄、编辑等场景,我们如何利用 AI 技术赋能创作者;以及基于这些场景,字节跳动积累的领先技术能力。
相关产品
评论
未登录
看完啦,登录分享一下感受吧~
暂无评论