LangGraph 快速入门

大模型向量数据库云通信
  1. LangGraph 简介 ===============

随着大模型技术的不断演进, AI Agent 成为了大模型领域的焦点。开发者们纷纷投身其中,期望构建出能够自主决策、灵活交互的智能体。在这片竞争激烈的技术蓝海中,LangGraph 作为从 LangChain 孵化而来的独立平台,迅速崛起为 Multi-Agent 开发框架中的佼佼者。

LangGraph 基于有向无环图(DAG)这一核心数据结构,通过节点(Node)和边(Edge)的组合,使得开发者能够快速地搭建起一个Multi-Agent 系统。

对于追求高效、可靠开发的工程师们来说,LangGraph 的核心优势在于:

  • 敏捷高效:凭借简洁统一的开发架构,将复杂 AI Agent 开发流程标准化,降低开发门槛;开发者可快速复用节点与工作流模板,减少重复性开发工作,大幅缩短项目周期,实现从构思到落地的高效转化 。
  • 可靠性和可控性:通过审核检查与人工介入机制,确保智能体行动精准合规,同时能在长期工作流中保留上下文,让智能体行动始终保持正确方向。
  • 高可扩展性:开发者可摆脱固有框架束缚,使用完全描述性的底层原语自由定制智能体,轻松设计满足各类需求的可扩展多智能体系统。
  • 原生流式支持:支持逐个 token 流式传输和中间步骤流式输出,实时呈现智能体推理过程,增强交互透明度。

从企业级复杂业务流程自动化,到创新型智能交互应用开发,LangGraph 都能成为开发者手中的得力工具。接下来,就让我们开启 LangGraph 的快速入门之旅,探索其强大功能与魅力。

  1. LangGraph 的核心概念 ==================

LangGrapg 有三个最核心的概念,分别是:State、 NodeGraph ,对这几个概念的具体理解如下:

  • State:状态,其中记录了智能体运行过程中的所有信息,涵盖输入数据、中间计算结果和最终输出,为智能体的持续运行提供上下文。 状态内的参数可以在多个节点间保存和流转,是节点间通信的核心机制。

  • Node:节点,它是构成 LangGraph 的基本单元。每个 Node 可实现特定功能,例如数据处理、逻辑判断或调用外部 API。

  • Graph:Graph 就是最终的图结构,由众多 Node 通过特定逻辑连接而成,定义了智能体的整体工作流程,各 Node 在其中协同运作,推动 AI Agent 完成复杂任务。

  1. 把大象装进冰箱 ==========

接下来,我们通过一个例子,展示下如何使用LangGraph快速构建一个简单的多步骤工作流。

相信大家一定看过赵本山和宋丹丹的经典小品《钟点工》,里面有这样一段台词:

要把大象装冰箱,总共分三步:第一步,把冰箱门打开;第二步,把大象装进去;第三步,把冰箱门带上。

picture.image

首先,让我们看看完整的实现代码:

  
from IPython.display import Image, display  
from langgraph.constants import START, END  
from langgraph.graph import StateGraph  
from langgraph.graph.state import CompiledStateGraph  
from typing_extensions import TypedDict  
  
# 1. 定义状态  
class ElephantInFridgeState(TypedDict):  
    """  
    定义状态——大象在冰箱中  
    状态内的参数可以在多个节点间保存和流转  
    """  
    fridge_open: bool  # 冰箱门是否打开  
    elephant_inside: bool  # 大象是否在冰箱内  
  
# 2. 定义每个步骤的处理节点  
def open_fridge(state: ElephantInFridgeState) -> ElephantInFridgeState:  
    """第一步: 把冰箱门打开"""  
    print("正在打开冰箱门...")  
    # 执行当前节点业务逻辑  
    print("冰箱门已打开!")  
    # 更新状态  
    state["fridge_open"] = True  
    # 当前节点处理完成,更新状态  
    return state  
  
def put_elephant(state: ElephantInFridgeState) -> ElephantInFridgeState:  
    """第二步: 把大象放进去"""  
    # 获取当前状态  
    if not state["fridge_open"]:  
        raise ValueError("冰箱门未打开,无法放入大象!")  
    # 执行当前节点业务逻辑  
    print("正在把大象放入冰箱...")  
    print("大象已放入冰箱!")  
    # 更新状态  
    state["elephant_inside"] = True  
    # 当前节点处理完成,更新状态  
    return state  
  
def close_fridge(state: ElephantInFridgeState) -> ElephantInFridgeState:  
    """第三步: 把冰箱门带上"""  
    # 获取当前状态  
    if not state["elephant_inside"]:  
        print("警告:冰箱内没有大象,是否确认关闭?")  
    # 执行当前节点业务逻辑  
    print("正在关闭冰箱门...")  
    print("冰箱门已关闭!")  
    # 更新状态  
    state["fridge_open"] = False  
    # 当前节点处理完成,更新状态  
    return state  
  
# 3. 构建Graph工作流图  
def build_graph() -> CompiledStateGraph:  
    """构建"把大象装冰箱"的工作流程图"""  
    # 创建图对象StateGraph  
    graph = StateGraph(ElephantInFridgeState)  
    # 添加节点Node  
    graph.add_node("open_fridge", open_fridge)  
    graph.add_node("put_elephant", put_elephant)  
    graph.add_node("close_fridge", close_fridge)  
    # 添加起始边  
    graph.add_edge(START, "open_fridge")  
    # 添加边Edge,将节点连接起来,形成工作流  
    graph.add_edge("open_fridge", "put_elephant")  
    graph.add_edge("put_elephant", "close_fridge")  
    # 添加结束边  
    graph.add_edge("close_fridge", END)  
    # 返回编译好的图Graph  
    return graph.compile()  
  
# 4. 运行Graph工作流  
def run_workflow():  
    """运行"把大象装冰箱"的工作流"""  
    # 构建图  
    graph = build_graph()  
    # 定义初始状态  
    initial_state = ElephantInFridgeState(fridge_open=False, elephant_inside=False)  
    # 运行图  
    result = graph.invoke(initial_state)  
    # 输出最终状态  
    print("\n工作流执行完毕,最终状态:")  
    print(f"冰箱门状态: {'打开' if result["fridge_open"] else '关闭'}")  
    print(f"大象是否在冰箱内: {'是' if result["elephant_inside"] else '否'}")  
    # 可视化工作流图  
    display(Image(graph.get_graph().draw_mermaid_png()))  
    return result  
  
# 执行入口  
if __name__ == "__main__":  
    final_state = run_workflow()

接下来,我们详细分析下这段代码的功能。

定义状态类

我们首先创建了 ElephantInFridgeState类,它继承自 LangGraph的 TypedDict基类,是一种 key-value 的结构,用于管理整个工作流的状态。这个类包含两个 bool 类型属性:

  • fridge\_open:表示冰箱门是否打开

  • elephant\_inside:表示大象是否在冰箱内

    这些状态变量将在各个节点之间传递,通过节点间的通信与协作,确保工作流的连贯性。

定义工作流节点

我们为三个步骤分别创建了函数作为节点:

  • open\_fridge:负责将冰箱门状态设置为打开。

  • put\_elephant:检查冰箱门是否打开,并将大象放入冰箱。

  • close\_fridge:关闭冰箱门,并在冰箱内没有大象时发出警告。

    我们可以看到,每个节点都遵循这样的处理范式:

  1. 从状态中获取所需数据。

  2. 执行当前节点的业务逻辑。

  3. 更新状态。

  4. 返回处理后的状态。

这种设计确保了状态在整个工作流中的一致性和可追踪性。

构建工作流图

build\_graph 函数负责将各个节点连接成完整的工作流:

  • 创建 StateGraph 实例

  • 添加三个工作步骤作为节点

  • 通过 add_edge 方法添加边,定义节点之间的执行顺序

  • 添加起始边和结束边。这里 START 和 END 是 LangGraph 内置的常量,分表代表图的起始和结束节点。

    这种图结构的设计让我们可以轻松定义复杂的工作流,包括条件分支、循环、回退、异步等高级特性。

运行工作流

run\_workflow 函数演示了如何编译和执行工作流:

  • 构建图

  • 初始化状态

  • 编译图并执行

  • 输出最终状态

通过这种方式,我们可以清晰地看到每个步骤的执行结果,以及整个工作流的最终状态。

这里我还通过可视化的方式,打印了整个图的结构

picture.image

运行结果

最后,我们执行入口函数,就可以看到如下的效果:

picture.image

这个简单的例子展示了LangGraph的核心功能: 通过状态管理、节点定义和图构建,我们可以创建复杂的工作流,每个步骤都可以独立开发和测试,同时保持整体的一致性和可控性。

  1. 小结 =====

通过对 LangGraph 的优势解析、核心概念拆解及 “把大象装冰箱” 的实战演练,我们清晰看到: LangGraph 以简洁高效的架构,将复杂 AI Agent 开发流程化繁为简。它通过 State、Node 和 Graph 的有机结合,赋予开发者精准掌控智能体行为的能力,无论是基础任务的有序推进,还是复杂场景的多智能体协作,都能从容应对。在 AI Agent 蓬勃发展的当下,LangGraph 不仅是开发者提升开发效率的利器,更为创新应用的落地提供了坚实支撑,有望推动人工智能迈向更智能、更灵活的新阶段。

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

文章

0

获赞

0

收藏

0

相关资源
在火山引擎云搜索服务上构建混合搜索的设计与实现
本次演讲将重点介绍字节跳动在混合搜索领域的探索,并探讨如何在多模态数据场景下进行海量数据搜索。
相关产品
评论
未登录
看完啦,登录分享一下感受吧~
暂无评论