LangGraph快速入门

课程说明：

体验课内容节选自《2025大模型Agent智能体开发实战》(夏季班) 完整版付费课程

体验课时间有限，若想深度学习大模型技术，欢迎大家报名由我主讲的《2025大模型Agent智能体开发实战》(夏季班)

《2025大模型Agent智能体开发实战》(夏季班) 为【100+小时】体系大课，总共20大模块精讲精析，零基础直达大模型企业级应用！

部分项目成果演示

from IPython.display import Video

MateGen项目演示

Video("https://ml2022.oss-cn-hangzhou.aliyuncs.com/MG%E6%BC%94%E7%A4%BA%E8%A7%86%E9%A2%91.mp4", width=800, height=400)

智能客服项目演示

Video("https://ml2022.oss-cn-hangzhou.aliyuncs.com/%E6%99%BA%E8%83%BD%E5%AE%A2%E6%9C%8D%E6%A1%88%E4%BE%8B%E8%A7%86%E9%A2%91.mp4", width=800, height=400)

Dify项目演示

Video("https://ml2022.oss-cn-hangzhou.aliyuncs.com/2f1b47f42c65fd59e8d3a83e6cb9f13b_raw.mp4", width=800, height=400)

LangChain&LangGraph搭建Multi-Agnet

Video("https://ml2022.oss-cn-hangzhou.aliyuncs.com/%E5%8F%AF%E8%A7%86%E5%8C%96%E6%95%B0%E6%8D%AE%E5%88%86%E6%9E%90Multi-Agent%E6%95%88%E6%9E%9C%E6%BC%94%E7%A4%BA%E6%95%88%E6%9E%9C.mp4", width=800, height=400)

此外，若是对大模型底层原理感兴趣，也欢迎报名由我和菜菜老师共同主讲的《2025大模型原理与实战课程》(夏季班)

两门大模型课程夏季班目前上新特惠+618年中钜惠双惠叠加，合购还有更多优惠哦~详细信息扫码添加助教，回复“大模型”，即可领取课程大纲&查看课程详情👇

LangGraph快速入门与Agent开发实战

Part 1.LangGraph快速入门

一、LangGraph 生态介绍

随着大模型应用日趋复杂，开发者需要一种既能利用 Agent 灵活性又能保证可靠性的基础设施。LangChain通过 “Chain” 工作流（固定顺序步骤）虽然简单但缺乏适应性，而完全自主的 Agent 虽灵活却容易不可控。LangGraph 正是在这种背景下出现，核心定位是低门槛构建“可控且持久”的智能代理流程，以平衡确定性和自主性的取舍

LangGraph 是由 LangChain AI 团队于 2023 年 8 月 11 日正式开源的 AI Agent 框架，旨在支撑和编排复杂的生成式 AI Agent 工作流。LangGraph 主要针对构建长时间运行、可回溯的智能体流程所面临的挑战提供一些特定的解决方案，包括：

可靠性与持续性： 提供持久化执行能力，使 Agent 即使在失败后也能从中断处恢复，支持长时任务不中断运行，例如在复杂业务流程或多轮对话中，Agent 可以保存中间状态，避免上下文丢失。
状态与记忆： 内置短期工作记忆和长期存储记忆，允许 Agent 跨对话会话保留上下文，这解决了以往仅依赖模型上下文窗口导致的记忆局限，使 Agent 能“记住”历史互动，提供更连贯的服务。
人机协作与监督： 引入Human-in-the-loop（人类在环）机制，支持在关键步骤由人审核或干预，以防止Agent偏离正确轨道，这使企业能在敏感任务中加入人工把关，提升最终结果可信度。
调试与可观测性： 通过与 LangSmith 、LangGraph Studio等工具集成，实现对复杂 Agent 行为的可视化追踪和调试，开发者可以观察每步决策、工具调用、状态变化，方便定位问题，优化 Agent 策略。
生产级扩展： 提供面向生产的部署支持，包括水平扩展、任务队列、缓存重试等容错机制，帮助 Agent 应对大规模并发和长流程的稳定运行。这使企业能在敏感任务中加入人工把关，提升最终结果可信度。

其官方开源地址：https://github.com/langchain-ai/langgraph

LangGraph 本质上是对原有 LangChain 功能的演进和补充。LangChain 一直以简洁的接口帮助开发者将大模型应用串联（链式）起来，实现提示模板、外部工具调用、Retrieval等功能。如下图所示：

然而，随着用例复杂度提升，LangChain 原有的链式代理在状态管理、多分支流程等方面显得局限。LangGraph 因此作为独立库发布（Python 和 JS 版本均有），但与 LangChain 无缝集成：开发者仍可利用 LangChain 丰富的模型接口、向量库、工具集，只是用 LangGraph 来编排更复杂的控制流。可以理解为 “LangChain 提供组件，LangGraph 负责编排”，两者协同构建完整应用。

顾名思义，LangGraph 框架的核心理念是图，它的出现是要解决线性序列的局限性问题，而解决的方法就是循环图 在LangGraph框架中，用图取代了LangChain的AgentExecutor（代理执行器），用来管理代理的生命周期并在其状态内将暂存器作为消息进行跟踪，增加了以循环方式跨各种计算步骤协调多个链或参与者的功能。这就与 LangChain 将代理视为可以附加工具和插入某些提示的对象不同，对于图来说，意味着我们可以从任何可运行的功能或代理或链作为一个程序的起点。

因此，其实LangChain 与 LangGraph 是互补关系：但相较于LangChain，LangGraph是一个适用范围更广的AI Agent开发框架。在大模型的支持方面，LangGraph支持DeepSeek、Qwen 3等，还兼容其他多种在线或开源模型，例如 OpenAI、llama3等，可以说热门的大模型均可以接入到该框架中进行AI Agent应用程序的开发。而关于大模型的接入方式，我们既可以通过传统的openai api等原生方式将大模型集成到LangGraph构建的AI Agent流程中，也可以利用ollma、vllm等大模型推理加速库，实现更加便捷和高效的集成。除此之外，在AI Agent的构建范式上，LangGraph不仅提供了预配置的ReAct代理机制，还支持更多自定义的如Planning策略的接入，以满足不同应用场景的需求。

当前业界除了 LangGraph，还有多种框架支持 Agent 应用开发，如微软的 AutoGen、开源平台 Dify、谷歌的 Agent Development Kit (ADK) 等等，与其他框架相比，LangGraph 最大的优势在于流程编排的灵活性、原生支持持久化记忆和强大的人机协同机制。它适合企业级、任务复杂、需要过程审计的业务场景。

LangGraph 与其它框架对比

框架	架构特点	主要优势	适用场景	局限/劣势
LangGraph	图流程编排+状态持久	流程灵活、可控、原生记忆、人机协同	企业复杂流程、定制化Agent	考验开发人员的研发能力
AutoGen	多Agent对话协作	自由协作、创新AI社会实验	多智能体协作/AI研究	流程难控，生产可用性弱
Dify	平台化+零代码	快速开发、可视化、插件丰富	快速原型/简单业务集成	自由度低，复杂流程定制受限
Google ADK	云原生+企业集成	易部署、性能强	大型企业/谷歌云用户	生态绑定谷歌，灵活度相对较低

LangChain公开课参考：https://www.bilibili.com/video/BV1pYKgzAE5C/

二、LangGraph图结构对象创建方法与核心语法

本次公开课使用的版本是：langgraph==0.4.8

! pip install langgraph -i https://pypi.tuna.tsinghua.edu.cn/simple

! pip show langgraph

1. LangChain图结构概念说明

在以图构建的框架中，任何可执行的功能都可以作为对话、代理或程序的启动点。这个启动点可以是大模型的 API 接口、基于大模型构建的 AI Agent，通过 LangChain 或其他技术建立的线性序列等等，即下图中的 "Start" 圆圈所示。无论哪种形式，它都首先处理用户的输入，并决定接下来要做什么。下图展示了在 LangGraph 概念下，最基本的一种代理模型：👇

如上图所示，在启动点定义的可运行功能会根据收到的输入决定是否进行检索以及如何响应。 比如在执行过程中，如果需要检索信息，则可以利用搜索工具来实现，如Web Search（网络搜索）、Query Database（查询数据库）、RAG等获取必要的信息（图中的 "Action" 圆圈）。接下来，再使用一个大语言模型（“LLM”）处理工具提供的信息，结合用户最初传入的初始查询，生成最终的响应（图中的 "LLMs" 圆圈）。最终，这个响应被传递至终点节点（图中的 "End" 圆圈）。

这个流程就是在LangGraph框架中一个非常简单的代理构成形式。非常关键且我们必须清楚的概念是：每个圆圈代表一个“节点”（Nodes），每个箭头表示一条“边”（Edges）。在 LangGraph 中，无论代理的构建是简单还是复杂，它最终都是由节点和边通过特定的组合形成的图。这样的构建形式形成的工作流原理就是：当每个节点完成工作后，通过边告诉下一步该做什么，所以也就得出了：LangGraph的底层图算法就是在使用消息传递来定义通用程序。当节点完成其操作时，它会沿着一条或多条边向其他节点发送消息。然后，这些接收节点执行其功能，将结果消息传递给下一组节点，然后该过程继续。如此循环往复。

这就是LangGraph底层架构设计中图算法的根本思想。

LangGraph框架是通过组合Nodes和Edges去创建复杂的循环工作流程，通过消息传递的方式串联所有的节点形成一个通路。那么维持消息能够及时的更新并向该去的地方传递，则依赖langGraph构建的State概念。 在LangGraph构建的流程中，每次执行都会启动一个状态，图中的节点在处理时会传递和修改该状态。这个状态不仅仅是一组静态数据，而是由每个节点的输出动态更新，然后影响循环内的后续操作。如下所示：👇

为了帮助大家更好的理解，我们先尝试在不接入大模型的情况下，构建一个如上图所示的简单工作流。

2.手动构建图流程

定义图时要做的第一件事是定义图的State。状态表示会随着图计算的进行而维护和更新的上下文或记忆。它用来确保图中的每个步骤都可以访问先前步骤的相关信息，从而可以根据整个过程中积累的数据进行动态决策。这个过程通过状态图StateGraph类实现，它是由LangGraph框架提供的核心类之一，专门用来创建state状态。

构建state的方法非常简答。我们可以将图的状态设计为一个字典，用于在不同节点间共享和修改数据，然后使用StateGraph类进行图的实例化。代码如下：

from langgraph.graph import StateGraph

# 使用 stategraph 接收一个字典
builder = StateGraph(dict)

builder

builder.schema

这里需要注意的是，builder也是后面要用到的图构建器（Graph Builder）对象，用于逐步添加节点、边、控制流逻辑，最终编译成可执行的 LangGraph 图。而这个图构建器需要通过带入一个状态对象来创建。

接下来，定义两个节点。addition节点是一个加法逻辑，接收当前状态StateGraph(dict)，将字典中x的值增加1，并返回新的状态。而subtraction节点是一个减法逻辑，接收从addition节点传来的状态StateGraph(dict)，从字典中的x值减去2，创建并返回一个新的键y。代码如下：

def addition(state):
    # 注意：这里接收到的是初始状态
    print(f"init_state: {state}")
    return {"x": state["x"] + 1}

def subtraction(state):
    # 注意：这里接收到的是上一个节点的状态
    print(f"addition_state: {state}")
    return {"x": state["x"] - 2}

然后，进行图结构的设计。具体来看，我们添加名为addition和subtraction的节点，并关联到上面定义的函数。设定图的起始节点为addition，并从addition到subtraction设置一条边，最后从subtraction到结束节点设置另一条边。代码如下：

# START 和 END 是两个特殊的节点，分别表示图的开始和结束。
from langgraph.graph import START, END

# 向图中添加两个节点
builder.add_node("addition", addition)
builder.add_node("subtraction", subtraction)

# 构建节点之间的边
builder.add_edge(START, "addition")
builder.add_edge("addition", "subtraction")
builder.add_edge("subtraction", END)

注，这里我们通过节点的同名字符串作为节点的名称。

builder.edges

builder.nodes

最后，执行图的编译。需要通过调用compile()方法将这些设置编译成一个可执行的图。代码如下所示：

graph = builder.compile()

除了上述通过打印的方式查看构建图的结构，LangGraph还提供了多种内置的图形可视化方法，能够将任何Graph以图形的形式展示出来，帮助我们更好地理解节点之间的关系和流程的动态变化。可视化最大的好处是：直接从代码中生成图形化的表示，可以检查图的执行逻辑是否符合构建的预期。

我们首先需要执行如下代码安装图可视化依赖的第三方库：

! pip install pyppeteer ipython -i https://pypi.tuna.tsinghua.edu.cn/simple

生成图结构的可视化非常直接，只需一行代码即可完成。具体代码如下：

from IPython.display import Image, display

display(Image(graph.get_graph(xray=True).draw_mermaid_png()))

当通过 builder.compile() 方法编译图后，编译后的 graph 对象提供了 invoke 方法，该方法用于启动图的执行。我们可以通过 invoke 方法传递一个初始状态（如 initial_state = {"x": 10}），这个状态将作为图执行的起始输入。代码如下：

# 定义一个初始化的状态
initial_state = {"x":10}

graph.invoke(initial_state)

LangGraph 的执行模型并不强制要求图中必须有 END 节点。只要执行路径在某个节点“无后继边”（即到达“终点”），那么该节点就被视为隐式终点。

也就是说，将一个字典作为状态对象带入到图中，即可进行图的实际运行。

在图的执行过程中，每个节点的函数会被调用，并且接收到前一个节点返回的状态作为输入。每个函数处理完状态后，会输出一个新的状态，传递给下一个节点。

上述代码执行过程中图的运行状态如下图所示：👇

这里需要注意的一个关键信息是：节点函数不需要返回整个状态，而是仅返回它们更新的部分。 也就是说：在每个节点的函数内部逻辑中，需要使用和更新哪些State中的参数中，只需要在return的时候指定即可，不必担心未在当前节点处理的State中的其他值会丢失，因为LangGraph的内部机制已经自动处理了状态的合并和维护。

# 定义一个初始化的状态
initial_state = {"x":10, "y": 9}

graph.invoke(initial_state)

总体来看，该图设置了一个简单的工作流程。其中值首先在第一个节点通过加法函数增加，然后在第二个节点通过减法函数减少。这一流程展示了节点如何通过图中的共享状态进行交互。需要注意的是，状态在任何给定时间只包含来自一个节点的更新信息。这意味着当节点处理状态时，它只能访问与其特定操作直接相关的数据，从而确保每个节点的逻辑是隔离和集中的。使用字典作为状态模式非常简单，由于缺乏预定义的模式，节点可以在没有严格类型约束的情况下自由地读取和写入状态，这样的灵活性有利于动态数据处理。 然而，这也要求开发者在整个图的执行过程中保持对键和值的一致性管理。因为如果在任何节点中尝试访问State中不存在的键，会直接中断整个图的运行状态。

3. 借助Pydantic对象创建图

Pydantic 是一个用于创建“数据模型”的 Python 库，它可以自动校验数据类型，并将字典数据转换为结构化对象。它就像是给字典加了一个“类型安全 + 自动验证”的外壳，是现代 Python 项目中最主流的“数据结构定义工具”。

state = {"x": 1, "y": "hello"}  # 没有类型限制

state["x"] = "abc"  # 不小心改错也不会报错

state

from pydantic import BaseModel

class MyState(BaseModel):
    x: int
    y: str = "default"   # 设置默认值

# 自动校验
state = MyState(x=1)
print(state.x)       # 输出 1
print(state.y)       # 输出 default

# 错误类型会报错
# state = MyState(x="abc")  # ❌ 会抛出 ValidationError

然后我们可以在 StateGraph(MyState) 传入一个 Pydantic 模型：

from pydantic import BaseModel
from langgraph.graph import StateGraph, START, END

# ✅ 1. 定义结构化状态模型
class CalcState(BaseModel):
    x: int

# ✅ 2. 定义节点函数，接收并返回 CalcState
def addition(state: CalcState) -> CalcState:
    print(f"[addition] 初始状态: {state}")
    return CalcState(x=state.x + 1)

def subtraction(state: CalcState) -> CalcState:
    print(f"[subtraction] 接收到状态: {state}")
    return CalcState(x=state.x - 2)

# ✅ 3. 构建图
builder = StateGraph(CalcState)

builder.add_node("addition", addition)
builder.add_node("subtraction", subtraction)

builder.add_edge(START, "addition")
builder.add_edge("addition", "subtraction")
builder.add_edge("subtraction", END)

graph = builder.compile()

# ✅ 4. 执行图：传入结构化状态对象
initial_state = CalcState(x=10)
final_state = graph.invoke(initial_state)

# ✅ 5. 打印最终结果
print("\n[最终结果] ->", final_state)

但是需要注意的是，无论输入端输入什么结构的对象，最终图计算返回结果是一个字典类型对象。

4.创建条件分支图

from typing import Optional
from pydantic import BaseModel
from langgraph.graph import StateGraph, START, END

# ✅ 定义结构化状态
class MyState(BaseModel):
    x: int
    result: Optional[str] = None

# ✅ 定义各节点处理逻辑（接受 MyState，返回 MyState）
def check_x(state: MyState) -> MyState:
    print(f"[check_x] Received state: {state}")
    return state

def is_even(state: MyState) -> bool:
    return state.x % 2 == 0

def handle_even(state: MyState) -> MyState:
    print("[handle_even] x 是偶数")
    return MyState(x=state.x, result="even")

def handle_odd(state: MyState) -> MyState:
    print("[handle_odd] x 是奇数")
    return MyState(x=state.x, result="odd")

# ✅ 构建图
builder = StateGraph(MyState)

builder.add_node("check_x", check_x)
builder.add_node("handle_even", handle_even)
builder.add_node("handle_odd", handle_odd)

# ✅ 添加条件分支
builder.add_conditional_edges("check_x", is_even, {
    True: "handle_even",
    False: "handle_odd"
})

# ✅ 衔接起始和结束
builder.add_edge(START, "check_x")
builder.add_edge("handle_even", END)
builder.add_edge("handle_odd", END)

# ✅ 编译图
graph = builder.compile()

# ✅ 执行测试
print("\n✅ 测试 x=4（偶数）")
graph.invoke(MyState(x=4))

print("\n✅ 测试 x=3（奇数）")
graph.invoke(MyState(x=3))

在本示例中，我们基于 LangGraph 框架构建了一个简单的有状态条件分支图，用于演示如何使用结构化状态（通过 Pydantic 模型定义）在多步骤的决策流程中进行状态传递与条件控制。

我们首先定义了一个名为 MyState 的 Pydantic 模型，用于描述图中每个节点共享的上下文状态信息。该状态包含两个字段：x 表示输入数值，result 表示最终处理结果。通过使用 Pydantic，能够显著增强状态管理的类型安全性、可读性和扩展性。

图的结构由三个核心节点组成：

check_x：作为图的第一步处理节点，接收初始状态并进行输出转发，不修改任何字段；
handle_even：处理偶数情况，标记 result = "even"；
handle_odd：处理奇数情况，标记 result = "odd"。

在 check_x 节点之后，我们引入了基于 is_even 判断函数的条件分支控制逻辑。LangGraph 提供的 add_conditional_edges 方法允许我们根据状态中的值动态跳转至不同的执行路径，从而实现类似于传统编程语言中的 if-else 分支控制。

整体流程图如下：

FENCE0

该图的编排逻辑清晰地展现了 LangGraph 的优势所在：即在状态流转的基础上，灵活地控制流程分支，并通过结构化模型传递和管理上下文信息，为构建具备复杂控制流的智能体（Agent）打下了基础。

其中 FENCE0 代码解释如下

字段	类型定义	是否必填	默认值	含义
`x`	`int`	✅ 必填	无默认值	必须提供的整数字段
`result`	`Optional[str]` = `Union[str, None]`	❌ 可选	默认是 `None`	表示一个可选的字符串，常用于延迟赋值或非必要字段

而完整的图结构执行流程如下：

参数位置	传入值	含义
第1个参数	`"check_x"`	当前执行完的节点名称（分支判断起点）
第2个参数	`is_even`	一个接收状态 `state` 并返回布尔值的函数，用于判断分支条件
第3个参数	`{True: "handle_even", False: "handle_odd"}`	条件结果 → 目标节点的映射表

from IPython.display import display, Image
display(Image(graph.get_graph(xray=True).draw_mermaid_png()))

4.创建条件循环图

from pydantic import BaseModel
from langgraph.graph import StateGraph, START, END

# ✅ 1. 定义结构化状态模型
class LoopState(BaseModel):
    x: int

# ✅ 2. 定义节点逻辑
def increment(state: LoopState) -> LoopState:
    print(f"[increment] 当前 x = {state.x}")
    return LoopState(x=state.x + 1)

def is_done(state: LoopState) -> bool:
    return state.x > 10

# ✅ 3. 构建图
builder = StateGraph(LoopState)
builder.add_node("increment", increment)

# ✅ 4. 设置循环控制：is_done 为 True 则结束，否则继续
builder.add_conditional_edges("increment", is_done, {
    True: END,
    False: "increment"
})

builder.add_edge(START, "increment")
graph = builder.compile()

# ✅ 5. 测试执行
print("\n✅ 执行循环直到 x > 10")
final_state = graph.invoke(LoopState(x=6))
print(f"[最终结果] -> x = {final_state['x']}")

注意，这里需要注意，add_conditional_edges中构建了循环

final_state

display(Image(graph.get_graph(xray=True).draw_mermaid_png()))

from pydantic import BaseModel
from typing import Optional
from langgraph.graph import StateGraph, START, END

# ✅ 1. 定义状态模型
class BranchLoopState(BaseModel):
    x: int
    done: Optional[bool] = False

# ✅ 2. 定义各节点逻辑
def check_x(state: BranchLoopState) -> BranchLoopState:
    print(f"[check_x] 当前 x = {state.x}")
    return state

def is_even(state: BranchLoopState) -> bool:
    return state.x % 2 == 0

def increment(state: BranchLoopState) -> BranchLoopState:
    print(f"[increment] x 是偶数，执行 +1 → {state.x + 1}")
    return BranchLoopState(x=state.x + 1)

def done(state: BranchLoopState) -> BranchLoopState:
    print(f"[done] x 是奇数，流程结束")
    return BranchLoopState(x=state.x, done=True)

# ✅ 3. 构建图
builder = StateGraph(BranchLoopState)

builder.add_node("check_x", check_x)
builder.add_node("increment", increment)
builder.add_node("done_node", done)

builder.add_conditional_edges("check_x", is_even, {
    True: "increment",
    False: "done_node"
})

# ✅ 4. 循环逻辑：偶数 → increment → check_x
builder.add_edge("increment", "check_x")

# ✅ 5. 起始与终点
builder.add_edge(START, "check_x")
builder.add_edge("done_node", END)

graph = builder.compile()

# ✅ 6. 测试执行
print("\n✅ 初始 x=6（偶数，进入循环）")
final_state1 = graph.invoke(BranchLoopState(x=6))
print("[最终结果1] ->", final_state1)

print("\n✅ 初始 x=3（奇数，直接 done）")
final_state2 = graph.invoke(BranchLoopState(x=3))
print("[最终结果2] ->", final_state2)

display(Image(graph.get_graph(xray=True).draw_mermaid_png()))

LangGraph 会把所有节点名、状态字段、通道名放在一个命名空间中处理，为了避免歧义，它会严格检查有没有冲突，最保险的做法是：节点名不要与字段名重复，既如果使用 state.result = "done"，也不要有 "result" 这个节点。

理解了LangGraph构建图的基本流程后，接下来我们就尝试接入大模型构建一个对话机器人。

三、搭建基于LangGraph的多轮对话问答机器人

1. LangGraph中多轮对话实现方法

在接下来的这个案例中，我们进一步将大模型接入到 LangGraph 工作流程中，并允许动态消息处理以及与模型的交互。

大模型应用都是接受消息列表作为输入，就像LangChain中的Chat Model，需要接收Message对象列表作为输入。这些消息有多种形式，例如HumanMessage （用户输入）或AIMessage （大模型响应）。这种消息格式其实就与我们之前介绍的StateGraph(dict)结构有一些区别。

因此对于消息序列格式的state，一种更简单的方法就是使用LangGraph预构建的add_messages函数，这个更高级的状态所实现的是：对于全新的消息，它会附加到现有列表，同时它也会正确处理现有消息的更新。 代码如下所示：

from typing import Annotated

from typing_extensions import TypedDict

from langgraph.graph import StateGraph
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]

graph_builder = StateGraph(State)

add_messages的核心逻辑是合并两个消息列表，按 ID 更新现有消息。默认情况下，状态为“仅附加”，当新消息与现有消息具有相同的 ID时，进行更新。具体参数是：

left （ Messages ） – 消息的基本列表。
right （ Messages ） – 要合并到基本列表中的消息列表（或单个消息）。

其返回值是一个消息列表，其中的合并逻辑则是：如果right的消息与left的消息具有相同的 ID，则right的消息将替换left的消息，否则作为一条新的消息进行追加。通过这种形式维护一个messages列表，可以很方便的实现消息的合并和更新。其基本形式如下：

FENCE0

我们可以使用add_messages函数来进行快速验证。如果消息的ID不一样，则会进行追加。代码如下：

from langgraph.graph.message import add_messages
from langchain_core.messages import AIMessage, HumanMessage

msgs1 = [HumanMessage(content="你好。", id="1")]
msgs2 = [AIMessage(content="你好，很高兴认识你。", id="2")]

add_messages(msgs1, msgs2)

如果ID相同，则会对消息内容进行更新。代码如下：

msgs1 = [HumanMessage(content="你好。", id="1")]
msgs2 = [HumanMessage(content="你好呀。", id="1")]

add_messages(msgs1, msgs2)

需要注意的是，不能直接在普通 Python 代码中测试 add_messages 的合并功能。

原因是： ✅ add_messages 并不会在你创建 State 字典时自动生效， ⛔ 它只会在 LangGraph 的内部状态更新系统中被识别和调用。

因此，当通过messages: Annotated[list, add_messages]去定义状态时，我们就可以很方便的实现聊天机器人场景下消息序列的处理和维护。

from typing import Annotated

from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]

graph_builder = StateGraph(State)

元素	含义	是什么？
`list`	字段的数据类型	✅ Python 内置的类型（列表类型）
`add_messages`	字段的“附加语义”	✅ LangGraph 提供的特殊函数（合并器/reducer），不是 list 的方法

接下来我们需要创建一个大模型节点，接收用户的输入，并返回大模型的响应。因此首先需要准备一个可以进行调用的大模型，这里我们选择使用DeepSeek的大模型，并使用DeepSeek官方的API_KEK进行调用。如果初次使用，需要现在DeepSeek官网上进行注册并创建一个新的API_Key，其官方地址为：https://platform.deepseek.com/usage

注册好DeepSeek的API_KEY后，首先在项目同级目录下创建一个env文件，用于存储DeepSeek的API_KEY，如下所示：

接下来通过python-dotenv库读取env文件中的API_KEY，使其加载到当前的运行环境中，代码如下：

! pip install python-dotenv -i https://pypi.tuna.tsinghua.edu.cn/simple

import os
from dotenv import load_dotenv 
load_dotenv(override=True)

DeepSeek_API_KEY = os.getenv("DEEPSEEK_API_KEY")

# print(DeepSeek_API_KEY)  # 可以通过打印查看

我们在当前的运行环境直接使用DeepSeek的API进行网络连通性测试，测试代码如下：

! pip install openai -i https://pypi.tuna.tsinghua.edu.cn/simple

from openai import OpenAI

# 初始化DeepSeek的API客户端
client = OpenAI(api_key=DeepSeek_API_KEY, base_url="https://api.deepseek.com")

# 调用DeepSeek的API，生成回答
response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "system", "content": "你是乐于助人的助手，请根据用户的问题给出回答"},
        {"role": "user", "content": "你好，请你介绍一下你自己。"},
    ],
)

# 打印模型最终的响应结果
print(response.choices[0].message.content)

如果可以正常收到DeepSeek模型的响应，则说明DeepSeek的API已经可以正常使用且网络连通性正常。

对于LangGraph框架，接入大模型最简单的方法就是借助langChain中的ChatModel组件。因此，我们首先需要安装LangChain的DeepSeek组件，安装命令如下：

! pip install langchain-deepseek

安装好LangChain集成DeepSeek模型的依赖包后，需要通过一个init_chat_model函数来初始化大模型，代码如下：

from langchain.chat_models import init_chat_model

model = init_chat_model(model="deepseek-chat", model_provider="deepseek")

其中model用来指定要使用的模型名称，而model_provider用来指定模型提供者，当写入deepseek时，会自动加载langchain-deepseek的依赖包，并使用在model中指定的模型名称用来进行交互。

question = "你好，请你介绍一下你自己。"

result = model.invoke(question)
print(result.content)

这里可以看到，仅仅通过两行代码，我们便可以在LangChain中顺利调用DeepSeek模型，并得到模型的响应结果。相较于使用DeepSeek的API，使用LangChain调用模型无疑是更加简单的。同时，不仅仅是DeepSeek模型，LangChain还支持其他很多大模型，如OpenAI、Qwen、Gemini等，我们只需要在init_chat_model函数中指定不同的模型名称，就可以调用不同的模型。其工作的原理是这样的：

理解了这个基本原理，如果大家想在用LangChain进行开发时使用其他大模型如Qwen3系列，则只需要先获取到Qwen3模型的API_KEY，然后安装Tongyi Qwen的第三方依赖包，即可同样通过init_chat_model函数来初始化模型，并调用invoke方法来得到模型的响应结果。关于LangChain都支持哪些大模型以及每个模型对应的是哪个第三方依赖包，大家可以在LangChain的官方文档中找到，访问链接为：https://python.langchain.com/docs/integrations/chat/

当然，除了在线大模型的接入，langChain也只是使用Ollama、vLLM等框架启动的本地大模型，关于如何使用不同的框架启动如DeepSeek R1、Qwen3等模型，大家可以参考我的往期公开课视频：

LangChain公开课参考：https://www.bilibili.com/video/BV1pYKgzAE5C/

掌握了如何使用langChain接入大模型后，接下来我们可以直接把langChain的Chat Model接入LangGraph中作为一个图节点（Node）, 并使用LangGraph的StateGraph来管理消息序列。代码如下：

def chatbot(state: State):
    return {"messages": [model.invoke(state["messages"])]}

接下来，添加一个chatbot节点，将当前State作为输入并返回一个字典，该字典中更新了messages中的状态信息。

# 添加节点
graph_builder.add_node("chatbot", chatbot)

# 添加边
graph_builder.add_edge(START, "chatbot")

graph = graph_builder.compile()

使用可视化工具，可以很方便的查看图的结构和节点之间的连接关系。代码如下：

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

这个图的基本逻辑是：第一个节点调用大模型并生成一个输出，该输出是一个AIMessage对象类型，然后，第二个节点直接将前一个节点的 AIMessage 提取为具体的JSON格式，完成JSON的解析。

final_state = graph.invoke({"messages": ["你好，我叫陈明，好久不见。"]})
print(final_state)

一次对话内容如下所示：

final_state['messages']

final_state['messages'][0]

final_state['messages'][0].id

final_state['messages'][0].content

final_state['messages'][1]

final_state['messages'][1].id

final_state['messages'][1].content

from langchain_core.messages import AIMessage, HumanMessage, SystemMessage

messages_list = [
    HumanMessage(content="你好，我叫陈明，好久不见。"),
    AIMessage(content="你好呀！我是小智，一名乐于助人的AI助手。很高兴认识你！"),
    HumanMessage(content="请问，你还记得我叫什么名字么？"),
]

final_state = graph.invoke({"messages": messages_list})
print(final_state)

final_state['messages'][-1].content

据此，我们可以直接调用这个编译好的图graph，即可实现一个可交互式的聊天机器人。代码如下：

messages_list = []

while True:
    try:
        user_input = input("用户提问: ")
        if user_input.lower() in ["exit"]: 
            print("下次再见！")
            break
        messages_list.append(HumanMessage(content=user_input))
        final_state = graph.invoke({"messages": messages_list})
        print("🤖 小智：", final_state['messages'][-1].content)
        messages_list.append(final_state['messages'][-1])
        messages_list = messages_list[-50:]
    except:
        break

借助MemorySaver高效搭建多轮对话机器人

🧠 MemorySaver 的核心功能

短期记忆（线程级记忆） MemorySaver 为每个 thread_id 保存和恢复对话状态（State），实现在同一会话中的历史上下文记忆。
状态持久化 在每个节点运行后，State 会自动存储；再次调用时，如果使用相同的 thread_id，MemorySaver 会恢复此前保存的状态，无需手动传递历史信息。
多会话隔离 通过不同的 thread_id 可实现会话隔离，允许多个用户并发交互且各自的对话互不干扰。
图状态快照与恢复 不仅包括对话历史，还保存整个工作流状态，可用于错误恢复、时间旅行、断点续跑、Human‑in‑the‑loop 等高级场景。

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import MemorySaver

# 1. 定义状态类（会自动合并 messages）
class State(TypedDict):
    messages: Annotated[list, add_messages]

# 2. 初始化模型
model = init_chat_model(model="deepseek-chat", model_provider="deepseek")

# 3. 定义聊天节点
def chatbot(state: State) -> State:
    reply = model.invoke(state["messages"])
    return {"messages": [reply]}

# 4. 构建带 MemorySaver 的图
builder = StateGraph(State)
builder.add_node("chatbot", chatbot)
builder.add_edge(START, "chatbot")
memory = MemorySaver()
graph = builder.compile(checkpointer=memory)

display(Image(graph.get_graph().draw_mermaid_png()))

此时测试效果如下：

# 5. 运行多轮对话，使用相同 thread_id 实现记忆
thread_config = {"configurable": {"thread_id": "session_10"}}

# 第一轮对话
state1 = graph.invoke({"messages": [{"role":"user","content":"你好，好久不见，我叫陈明。"}]}, config=thread_config)

state1

state1['messages']

state1['messages'][-1].content

state2 = graph.invoke({"messages":[{"role":"user","content":"你好呀，你还记得我的名字吗？"}]}, config=thread_config)

state2['messages']

state2['messages'][-1].content

# 使用不同 thread_id，会开启全新对话
state3 = graph.invoke({"messages":[{"role":"user","content":"记得我的名字吗？"}]}, 
                      config={"configurable":{"thread_id":"session_2"}})
state3['messages'][-1].content

查看记忆

latest = graph.get_state(thread_config)

print(latest.values["messages"])  # 包含全部轮次对话

2. LangGraph流式打印实现方法

在实际应用中，流式输出尤其适用于需要快速反馈的业务场景，典型的应用场景就是聊天机器人，因为大语言模型可能需要几秒钟才能生成对查询的完整响应，这远远慢于应用程序对最终用户的响应速度约为 200-300 毫秒的阈值，如果是涉及多个大模型调用的复杂应用程序，这种延时会变得更加明显。让应用程序感觉响应更快的关键策略是显示中间进度；即，通过 token 流式传输大模型Token的输出，以此来显著提升用户体验。而在开发阶段，利用流式输出功能可以准确追踪到事件的具体执行阶段，并捕获相关数据，从而接入不同逻辑的数据处理和决策流程。是我们在应用开发中必须理解和掌握的技术点。

流式输出，这种方式允许客户端逐渐接收到大模型生成的每一部分内容，而不是等待整个响应完成后一次性接收。

这里我们可以先看下使用deepseek-chat模型的流式调用代码及输出情况。在流式输出的实现方式中，我们需要在调用 client.chat.completions.create() 时添加 stream=True 参数，用以指定启用流式输出，从而允许 API 逐步发送生成的文本片段，然后使用 for 循环来迭代 completion 对象，每次循环接收到的 chunk.choices[0].delta 中会包含最新生成的文本片段。完整代码如下所示：

import os
from dotenv import load_dotenv 
load_dotenv(override=True)

DeepSeek_API_KEY = os.getenv("DEEPSEEK_API_KEY")

# print(DeepSeek_API_KEY)  # 可以通过打印查看

from openai import OpenAI
client = OpenAI( base_url="https://api.deepseek.com",api_key=DeepSeek_API_KEY)

completion = client.chat.completions.create(
  model="deepseek-chat",
  messages=[
    {"role": "system", "content": "你是一位乐于助人的人工智能小助理。"},
    {"role": "user", "content": "你好，请你介绍一下你自己。"}
  ],
  stream=True
)

for chunk in completion:
  print(chunk.choices[0].delta)

从输出上看，每个 ChoiceDelta 对象代表了大模型生成的一小段文本。所以很明显，这种流式调用的方式可以让我们看到 DeepSeek 模型是如何一步一步构建出最终的回答的。例如，从 "你好" 开始，接着是 "！"，然后是 "我是"，一直到整个回答完整地构建出来，每个 ChoiceDelta 可能包含几个字符到一个或几个词。

接下来，我们还是重点关注LangGraph中的流式输出应用方法。

因为LangGraph框架的底层和 LangChain一样都是基于LCEL语法而构建的，所有直接把LangChain中的回调系统拿过来便可以使用。在LangChain中的流式输出是：以块的形式传输最终输出，即一旦监测到有可用的块，就直接生成它。 如下流程所示：

model = init_chat_model(model="deepseek-chat", model_provider="deepseek")

chunks = []
async for chunk in model.astream("你好，请你详细的介绍一下你自己。"):
    chunks.append(chunk)
    print(chunk.content, end="|", flush=True)

chunks[0]

每一个块，都是一个AIMessageChunk对象，用来代表AIMessage对象的一部分。消息块在设计上是可加的，比如：

chunks[0] + chunks[1] + chunks[2] + chunks[3] + chunks[4]

LangGraph框架中的工作流中由各个步骤的节点和边组成。这里的流式传输涉及在各个节点请求更新时跟踪图状态的变化。这样可以更精细地监控工作流中当前处于活动状态的节点，并在工作流经过不同阶段时提供有关工作流状态的实时更新。其实现方式也是和LangChain一样通过.stream和.astream方法执行流式输出，只不过适配到了图结构中。调用.stream和.astream方法时可以指定几种不同的模式，即：

values ：在图中的每个步骤之后流式传输状态的完整值。
updates ：在图中的每个步骤之后将更新流式传输到状态。如果在同一步骤中进行多个更新（例如运行多个节点），则这些更新将单独流式传输。
debug ：在整个图的执行过程中流式传输尽可能多的信息，主要用于调试程序。
messages：记录每个messages中的增量token。
custom：自定义流，通过LangGraph 的 StreamWriter方法实现。

如果我们想流式传输每个过程中的 Tokens，对当前图：

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

这里调用graph.stream和graph.astream方法时, 指定模式为messages，即流式传输每个过程中的 Tokens。代码如下：

from langchain_core.messages import HumanMessage

async for msg, metadata in graph.astream({"messages": ["你好，请你详细的介绍一下你自己"]}, stream_mode="messages"):
    if msg.content and not isinstance(msg, HumanMessage):
        print(msg.content, end="", flush=True)

掌握State的定义模式和消息传递是LangGraph中最关键，也是构建应用最核心的部分，所有的高阶功能，如工具调用、上下文记忆，人机交互等依赖State的管理和使用，所以大家务必理解并掌握上述相关内容。这一部分我们在下一节课中展开详细的介绍，接下来我们先给大家介绍一个实用的开发工具：LangSmith。

课程说明：​

LangGraph快速入门与Agent开发实战

Part 1.LangGraph快速入门

一、LangGraph 生态介绍​

二、LangGraph图结构对象创建方法与核心语法​

1. LangChain图结构概念说明​

2.手动构建图流程​

3. 借助Pydantic对象创建图​

4.创建条件分支图​

4.创建条件循环图​

三、搭建基于LangGraph的多轮对话问答机器人​

1. LangGraph中多轮对话实现方法​

2. LangGraph流式打印实现方法​