LangGraph——AI应用开发框架

作者：CSDN博客
目录
一，LangGraph介绍
1.1 认识智能服务(Agent Server)
1.2 LangGraph -- Agent Server 的“操作系统”
1.3 LangGraph生态系统
二，LangGraph核心概念
2.1 Agent
2.2 工作流
2.3 实现工作流的核心概念
State（状态） - 快递的"包裹信息"
Nodes（节点） - 快递站点
Edges（边） - 快递运输路线
三，LangGraph简单示例
3.1 LangGraph编码思路
3.2 智能快递配送系统
1.定义 State，设置快递的"包裹信息"
2.定义 StateGraph 图，成⽴快递公司
3.定义 Nodes，创建配送站点
4.添加 Nodes，建设配送站点
5.添加 Edges，规划运输路线
6.StateGraph 图编译，从公司创建到运⾏
7.完整代码+测试用例
3.3 支持搜索的智能代理系统
1.设置 Nodes
2.设置 State
3.设置 Edges
4.完整代码
3.4基于LangGraph实现的RAG
1.装备知识库并创建检索工具
2.创建节点
3.组装"工作流水线"
4.完整代码

---

一，LangGraph介绍

LangGraph 是 LangChain 生态中用于构建有状态、图驱动、支持循环 / 分支 / 多智能体协作的 LLM工作流编排框架，核心是用「状态 + 节点 + 边」建模复杂 AI 流程，解决传统线性链难以处理的循环、多轮决策、多角色协作问题。
在 LangChain 中，我们接入 LLM 构建的 AI 系统有以下特点：

单次对话，“记不住”太多上下文
主要是问答模式
相对简单的任务

以前用 LangChain 只能写：

线性流程：A→B→C→结束
没法轻松写：循环、多轮决策、多智能体、分支判断、长上下文记忆

LangGraph 就是为了解决这些：

自主思考的 Agent
多轮工具调用（搜索→计算→查资料→再搜索）
多角色协作（研究员 + 分析师 + 写作）
带循环、分支、重试的复杂工作流

1.1 认识智能服务(Agent Server)

什么是 Agent Server？它是更聪明的 AI 助⼿
想象一个虚拟的“小助手”，它不仅能回答问题，还能：

记住和你聊过的所有事情
执行一连串任务（比如：查天气 → 推荐穿搭 → 提醒带伞）
中途可以等你给出反馈
运行很长时间不会“忘记“

构建 Agent Server 时遇到的四⼤难题：

难题	描述
状态丢失	AI处理长任务时容易"忘记"前面步骤
难以调试	不知道AI为什么做出某个决定
无法干预	不能中途给AI指导
部署困难	复杂Agent Server难以上线运行

1.2 LangGraph -- Agent Server 的“操作系统”

LangGraph 就像是一个强大且灵活的“Agent Server 操作系统内核”。
它不关心我们具体用什么模型或提示词，而是为我们解决构建复杂、可靠、可交互的 Agent Server 时所面临的状态管理、流程编排、持久化和人工监督等底层⼯程难题。如果我们需要构建超越简单问答的、具备复杂逻辑和长期记忆的AI 应用，LangGraph 就是为此设计的⼯具。
如果把 Agent Server 比作一个公司，那么：

AI 模型 = 员工（干活的人）
工具= 办公设备（电脑、电话）
LangGraph = 项⽬经理 + 流程系统

简单来说，LangGraph 是一个专门用于构建和管理 Agent Server 的底层框架。
1.3 LangGraph生态系统

LangGraph 是 LangChain 产品家族的一部分。可以与 LangSmith（用于追踪、评估、监控）、LangGraph 部署平台（⽤于轻松部署和扩展Agent Server）以及 LangChain（提供大量集成和组件）结合使⽤，形成完整的开发、调试、部署⼯作流。
LangGraph 与 LangChain 顺利集成，但也可以在没有 LangChain 的情况下使用。
二，LangGraph核心概念

2.1 Agent

Agent (智能体) 是⼀个能够感知环境输入、⾃主决策、规划⾏动路径, 并可调⽤⼯具或执⾏操作以达成目标的自主性软件实体。
其核心在于：由大语言模型 (LLM) 动态控制流程走向

想象你有一个超级聪明的程序员助手，你只需要说一句："把这个项目的单元测试覆盖率提升到 80%."
然后他就自己去看代码、写测试、运⾏构建、检查结果……直到达标为为止⸺中间不需要你插手.
这个"能独立完成任务"的程序, 就是我们说的 Agent (智能体) .
想象一个私人助理机器人：
你说："帮我安排下周去上海的差旅. "
它不会傻傻地只回复"好的", 而是会：
a. 查天气→ 决定带什么衣服
b. 查航班/高铁时间 → 比较价格
c. 预订酒店 → 发送日程到你的手机
d. 提醒你带身份证
整个过程没有固定的流程图, 它自己"想"出来的步骤 ⸺ 这就是典型的 Agent 行为

2.2 工作流

Workflow (⼯作流) 是⼀个将复杂过程分解为定义明确、顺序执⾏的任务流程, 并在其中⾃动化传递数据与任务的状态, 用户完成特定⽬标. 如果说 Agent 是聪明的执⾏者，那 ⼯作流就是它的⾏动蓝图 ⸺它定义了“先做什么、后做什么、在什么条件下跳转或终止”。
其核心在于：预设的、可重复的流程路径.

比如公司的报销流程
提交发票 → 领导审批 → 财务审核 → 财务结算 → 归档完成
或者一个自动化客服工单系统
用户提交问题 → 系统⾃动分类 → 分配给对应客服 → 客服处理 → 用户评价 → 工单关闭
这个固定路径, 整个过程沿着预设的流程运⾏, 就是典型的⼯作流⾏为.

LangGraph 是实现的就是⼯作流！
在 LangGraph中，工作流基于 LLM ，以及添加到其中的各种增强功能（如工具调用、结构化输出和短期记忆）而实现：

LangGraph 是一个强大且灵活的“Agent Server 操作系统内核”。它不关心我们具体用什么模型或提示词，⽽是为我们解决构建复杂、可靠、可交互的 Agent Server 时所面临的状态管理、流程编排、持久化和人工监督等底层⼯程难题。
2.3 实现工作流的核心概念

如何实现工作流逻辑？图计算是一种⽤节点和边来表示复杂系统的方法。在 AI 领域，它特别适合构建多步骤、有状态的智能工作流。
想象一个快递配送系统，下图展示了包裹从输入（揽收站）到输出（配送站）的过程：

配送站、揽收站、分拣中⼼ = 节点（Node）
运输路线 = 边（Edge）
包裹信息 = 状态（State）
完整配送网 = 图（Graph）

State（状态） - 快递的"包裹信息"

State 就像快递包裹上的标签，记录着包裹的当前位置、目的地、配送状态等信息。在整个配送过程中，所有站点都能查看和更新这个信息。
状态特性：

共享性：所有节点（快递站点）都能读取和修改
持久性：在整个工作流（快递运输）执行期间持续存在
结构化：有明确的字段定义

代码示例：

1. from typing import TypedDict
3. class PackageState(TypedDict):
4.     # 包裹的状态
5.     package_id:str   #包裹ID
6.     origin:str       #包裹的始发地
7.     destination:str  #包裹的目的地
9.     #配送状态("揽收中","配送中",...)
10.     status:str
12.     # 包裹的流转历史
13.     # history:list[str]     #覆盖更新
14.     history:Annotated[list[str],operator.add]   #追加更新
16.     # 总里程
17.     total_instance: Annotated[int,operator.add]
19.     # 配送详情("加急","普通")
20.     priority:str

复制代码

所有配送站点共享的包裹信息卡就是 LangGraph 中的 State！
Nodes（节点） - 快递站点

节点就像快递配送⽹络中的各个站点，每个站点负责特定的处理步骤。例如：
揽收站：接收包裹，初始化信息
分拣中心：根据目的地分类包裹
派送站：最终配送至收件人
节点特征：

单一职责：每个节点只做一件事
输入输出：接收状态，返回状态更新
独⽴性：节点间不直接通信，通过 State 交互

代码示例（节点只不过是函数）：

1. import operator
2. from typing import TypedDict, Annotated
4. # 揽收站节点
5. def receive_package(state:PackageState):
6.     """揽收站"""
7.     #获取出发地
8.     origin = state["origin"]
9.     return {
10.         "status": "已揽收",
11.         "history": [f"在{origin}揽收"]
12.     }
14. # 分拣中心节点:根据目的地进行分拣
15. def sort_package(state:PackageState):
16.     """分拣中心站"""
17.     #获取目的地
18.     destination = state["destination"]
19.     if "北京" in destination:
20.         next = "北京分拣中心"
21.     elif "西安" in destination:
22.         next = "西安分拣中心"
23.     else:
24.         next = "其他分拣中心"
25.     return {
26.         "status" : "已分拣",
27.         "history": [f"在{next}分拣"]
28.     }
30. # 派送站节点
31. def final_delivery(state:PackageState):
32.     """派送站"""
33.     destination = state["destination"]
34.     return {
35.         "status":"已签收",
36.         "history":[f"在{destination}签收"]
37.     }
39. # 普通配送站节点
40. def standard_delivery(state:PackageState):
41.     """普通配送"""
42.     return {
43.         "status":"运输中",
44.         "history":["标准陆运"],
45.         "total_instance":500
46.     }
48. # 加急配送站节点
49. def express_delivery(state:PackageState):
50.     """加急配送"""
51.     return {
52.         "status":"加急运输",
53.         "history":["加急空运"],
54.         "total_instance":800
55.     }

复制代码

Edges（边） - 快递运输路线

边定义了包裹在站点之间的流动路径，就像快递公司的运输路线图。对于路线，一般类型有：

开始/结束路线：流程的开始和结束点（包裹的开始站，与结束站）。
固定路线：包裹可以从揽收站→分拣中⼼（所有包裹都走这条路），而不能从配送站→揽收站，而是配送站→家。
条件路线：根据⽬的地选择不同的分拣中心

实际上在 LangGraph 中，边就定义了节点之间的连接关系，决定了⼯作流的执⾏顺序。边的类型有：

固定边：总是从 A 到 B
条件边：根据状态决定下一步
以及图的开始和结束点，标志了工作流的入口和出口。

因此，LangGraph 通过节点（每个处理步骤）、边（步骤之间的连接）和状态（保存执行过程），就可以构建出一个任务工作流（图）。
三，LangGraph简单示例

3.1 LangGraph编码思路

构建 Graph 图，首先需要【定义状态】，然后【定义并添加节点和边】，最后【编译】它。编译提供了对图形结构的一些基本检查（没有孤立节点等）。
LangGraph 所谓的“编译” 与 传统意义上的语言编译完全不同，LangGraph 编译本质是在运行时动态构建和验证一个复杂的图，而非翻译代码。
3.2 智能快递配送系统

1.定义 State，设置快递的"包裹信息"

定义图时要做的第一件事是定义图的 状态 。 状态 将是图中所有 节点 和 边 的输入，可以是TypedDict 或 Pydantic 模型（ Pydantic 的性能不如 TypedDict ）。
如下所示：

1. class PackageState(TypedDict):
2.     # 包裹的状态
3.     package_id:str   #包裹ID
4.     origin:str       #包裹的始发地
5.     destination:str  #包裹的目的地
7.     #配送状态("揽收中","配送中",...)
8.     status:str
10.     # 包裹的流转历史
11.     # history:list[str]     #覆盖更新
12.     history:Annotated[list[str],operator.add]   #追加更新
14.     # 总里程
15.     total_instance: Annotated[int,operator.add]
17.     # 配送详情("加急","普通")
18.     priority:str

复制代码

状态支持更新的关键是 Reducers。在示例中，我们使用 Annotated 类型为 history 和total_distance 指定了一个 reducer 函数 （ operator.add ）。这定义了如何更新包裹信息，特别是当多个站点都要记录信息时。当状态更新时，新的值会追加到现有值中，而不是替换。
如下所示：

1. # 覆盖更新：每次新状态替换旧状态
2. status: str
4. # 追加更新：新的流转记录添加到历史列表
5. history: Annotated[list[str], add]
7. # 数值累加：⾥程数累加
8. total_distance: Annotated[int, add]

复制代码

2.定义 StateGraph 图，成⽴快递公司

StateGraph 是一个有状态图计算框架，它基于有向图（Directed Graph） 模型构建，专门设计用于处理多步骤、有状态的工作流程。
StateGraph 用来将复杂的工作流程可视化、模块化，让开发者能够像设计快递配送网路一样设计软件系统。通过这种思维方式，即使是复杂的多步骤 AI 应⽤也变得清晰可控。
我们需要使用 langgraph.graph.state.StateGraph 来定义。 StateGraph 仅是⼀个构建器类，可以使⽤State来构建。
如下所示：

1. from langgraph.graph import StateGraph
2. # 2. 成⽴快递公司
3. delivery = StateGraph(PackageState)

复制代码

3.定义 Nodes，创建配送站点

接着我们可以定义各个配送站点（节点）。在 LangGraph 中， 节点就是一个Python 函数（同步或异步）。注意

节点 接收 状态 作为参数。
节点 不需要返回整个 状态 模式，只需一个更新。

如下所示：

1. # 揽收站节点
2. def receive_package(state:PackageState):
3.     """揽收站"""
4.     #获取出发地
5.     origin = state["origin"]
6.     return {
7.         "status": "已揽收",
8.         "history": [f"在{origin}揽收"]
9.     }
11. # 分拣中心节点:根据目的地进行分拣
12. def sort_package(state:PackageState):
13.     """分拣中心站"""
14.     #获取目的地
15.     destination = state["destination"]
16.     if "北京" in destination:
17.         next = "北京分拣中心"
18.     elif "西安" in destination:
19.         next = "西安分拣中心"
20.     else:
21.         next = "其他分拣中心"
22.     return {
23.         "status" : "已分拣",
24.         "history": [f"在{next}分拣"]
25.     }
27. # 派送站节点
28. def final_delivery(state:PackageState):
29.     """派送站"""
30.     destination = state["destination"]
31.     return {
32.         "status":"已签收",
33.         "history":[f"在{destination}签收"]
34.     }
36. # 普通配送站节点
37. def standard_delivery(state:PackageState):
38.     """普通配送"""
39.     return {
40.         "status":"运输中",
41.         "history":["标准陆运"],
42.         "total_instance":500
43.     }
45. # 加急配送站节点
46. def express_delivery(state:PackageState):
47.     """加急配送"""
48.     return {
49.         "status":"加急运输",
50.         "history":["加急空运"],
51.         "total_instance":800
52.     }

复制代码

4.添加 Nodes，建设配送站点

接下来，我们需要将各节点，组织进图中，即添加节点到 StateGraph 中。可以使用add_node() 将新节点添加到 StateGraph 。

1. # 3.定义图
2. delivery = StateGraph(PackageState)
4. # 4.添加节点
5. delivery.add_node("揽收站",receive_package)
6. delivery.add_node("分拣中心",sort_package)
7. delivery.add_node("派送站",final_delivery)
8. delivery.add_node("普通派送站",standard_delivery)
9. delivery.add_node("加急派送站",express_delivery)

复制代码

5.添加 Edges，规划运输路线

各节点（站点）准备好后，则需要为快递运输规划。
实际上，这就是为 图 定义 边 。边有几种关键类型：

普通边/固定边（Normal Edges）：直接从一个节点转到下一个节点。
条件边（Conditional Edges）：调⽤函数来确定下一步要转到哪个节点。

例如，设置最简单运输路线：快递由揽收站接收，下⼀站固定为分拣中⼼，最后到派送中进⾏派送。

1. # 固定边
2. delivery.add_edge(START,"揽收站")
3. delivery.add_edge("揽收站","分拣中心")
5. def select_delivery(state:PackageState):
6.     priority=state["priority"]
7.     if priority=="加急":
8.         return "备注加急"
9.     elif priority=="普通":
10.         return "无备注"
12. # 添加条件边
13. delivery.add_conditional_edges(
14.     "分拣中心",      #条件起始节点
15.     select_delivery,      #确定下个节点的可调用对象
16.     {
17.         "备注加急":"加急派送站",
18.         "无备注":"普通派送站"
19.     }
20. )
22. # 添加固定边
23. delivery.add_edge("加急派送站","派送站")
24. delivery.add_edge("普通派送站","派送站")
25. delivery.add_edge("派送站",END)

复制代码

在 LangGraph 中：

START 节点：是一个特殊节点，表⽰将⽤⼾输入发送到图形的节点。引⽤此节点的主要目的是确定应该⾸先调用哪些节点。
END 节点：是一个表示终端节点的特殊节点。当想要指示哪些边在完成后没有后续动作时，将引用此节点。
条件入口点（Conditional Entry Point）：调⽤一个函数来确定在⽤⼾输入到达时⾸先调⽤哪个节点

6.StateGraph 图编译，从公司创建到运⾏

在步骤2中，我们仅是构建出 StateGraph ，还无法直接⽤于执⾏。LangGraph 要求：必须先编译图，然后才能使⽤它。编译提供了对图结构的一些基本检查，这会验证：

从 START 到所有节点的可达性
从所有节点到 END 的可达性
没有孤⽴节点或死循环

使⽤ compile() 方法即可编译图。该⽅法将 StateGraph 编译为 CompiledStateGraph 对象。编译后的图实现了 Runnable 接⼝，可以异步调⽤、流式传输、批处理和运⾏。
代码如下：

1. # 7. 编译系统
2. delivery_system = delivery.compile()

复制代码

7.完整代码+测试用例

1. # 智能快递派送系统
2. import operator
3. from typing import TypedDict, Annotated
4. from langgraph.constants import START, END
5. from langgraph.graph import StateGraph
7. # 1.状态定义(定义包裹的状态)
8. class PackageState(TypedDict):
9.     # 包裹的状态
10.     package_id:str   #包裹ID
11.     origin:str       #包裹的始发地
12.     destination:str  #包裹的目的地
14.     #配送状态("揽收中","配送中",...)
15.     status:str
17.     # 包裹的流转历史
18.     # history:list[str]     #覆盖更新
19.     history:Annotated[list[str],operator.add]   #追加更新
21.     # 总里程
22.     total_instance: Annotated[int,operator.add]
24.     # 配送详情("加急","普通")
25.     priority:str
27. # 2.节点定义(定义函数)
28. # 节点之间使用状态(state)进行通信
30. # 揽收站节点
31. def receive_package(state:PackageState):
32.     """揽收站"""
33.     #获取出发地
34.     origin = state["origin"]
35.     return {
36.         "status": "已揽收",
37.         "history": [f"在{origin}揽收"]
38.     }
40. # 分拣中心节点:根据目的地进行分拣
41. def sort_package(state:PackageState):
42.     """分拣中心站"""
43.     #获取目的地
44.     destination = state["destination"]
45.     if "北京" in destination:
46.         next = "北京分拣中心"
47.     elif "西安" in destination:
48.         next = "西安分拣中心"
49.     else:
50.         next = "其他分拣中心"
51.     return {
52.         "status" : "已分拣",
53.         "history": [f"在{next}分拣"]
54.     }
56. # 派送站节点
57. def final_delivery(state:PackageState):
58.     """派送站"""
59.     destination = state["destination"]
60.     return {
61.         "status":"已签收",
62.         "history":[f"在{destination}签收"]
63.     }
65. # 普通配送站节点
66. def standard_delivery(state:PackageState):
67.     """普通配送"""
68.     return {
69.         "status":"运输中",
70.         "history":["标准陆运"],
71.         "total_instance":500
72.     }
74. # 加急配送站节点
75. def express_delivery(state:PackageState):
76.     """加急配送"""
77.     return {
78.         "status":"加急运输",
79.         "history":["加急空运"],
80.         "total_instance":800
81.     }
83. # 3.定义图
84. delivery = StateGraph(PackageState)
86. # 4.添加节点
87. delivery.add_node("揽收站",receive_package)
88. delivery.add_node("分拣中心",sort_package)
89. delivery.add_node("派送站",final_delivery)
90. delivery.add_node("普通派送站",standard_delivery)
91. delivery.add_node("加急派送站",express_delivery)
93. # 5.添加边
94. # 固定边
95. delivery.add_edge(START,"揽收站")
96. delivery.add_edge("揽收站","分拣中心")
98. def select_delivery(state:PackageState):
99.     priority=state["priority"]
100.     if priority=="加急":
101.         return "备注加急"
102.     elif priority=="普通":
103.         return "无备注"
105. # 添加条件边
106. delivery.add_conditional_edges(
107.     "分拣中心",      #条件起始节点
108.     select_delivery,      #确定下个节点的可调用对象
109.     {
110.         "备注加急":"加急派送站",
111.         "无备注":"普通派送站"
112.     }
113. )
115. # 添加固定边
116. delivery.add_edge("加急派送站","派送站")
117. delivery.add_edge("普通派送站","派送站")
118. delivery.add_edge("派送站",END)
120. # 6.编译图
121. delivery_system = delivery.compile()
123. # 7.执行图(输入初始状态,输出最终状态)
124. test_packages=[
125.     {
126.         "package_id":"p001",
127.         "origin":"北京",
128.         "destination":"上海",
129.         "priority":"加急",
130.         "history":[],
131.         "total_instance":0
132.     },
133.     {
134.         "package_id": "p002",
135.         "origin": "西安",
136.         "destination": "杭州",
137.         "priority": "普通",
138.         "history": [],
139.         "total_instance": 0
140.     }
141. ]
143. for package in test_packages:
144.     print(f"\n配送包裹:{package['package_id']}")
145.     #执行图
146.     result = delivery_system.invoke(package)
147.     print("最终状态:",result["status"])
148.     print("配送历史",result["history"])
149.     print("总里程",result["total_instance"])

复制代码

到此，我们已经构建出了⼀个图式的智能快递配送系统：

State = 包裹信息卡（记录所有状态）
Nodes = 配送站点（执⾏具体操作）
Edges = 运输路线（控制流转顺序）
Reducers = 信息更新规则（如何记录变更）
编译 = 从路线图到运营系统的转换

3.3 支持搜索的智能代理系统

核心功能如下：

智能对话与工具调用：基于聊天模型，能够理解用户问题并决定是否需要调用搜索⼯具
自动搜索整合：通过 Tavily 搜索工具获取实时信息，并将搜索结果整合到回答中
循环决策机制：能够多次调⽤工具和模型，直到获得满意答案

流程设置如下

1.设置 Nodes

根据节点的单一职责特性，即每个节点只做一件事，我们可以设置：
节点1（ tool_node ）：专门负责搜索，获取搜索结果。
节点2（ llm_call ）：专门负责调用LLM，获取最终结果。
根据节点的独立性特性，即每个节点间不直接通信，⽽是通过 State 交互。它们接收 State，返回State 的更新。
2.设置 State

根据状态的共享性特性，即所有节点都能读取和修改状态。LangGraph 在许多情况下，将以前的对话历史记录存储为 State 中的消息列表会很有帮助，这样就能通过 State 中的消息列表来跟踪整个对话的完整历史，这是构建对话系统的关键。
为此，我们可以在图状态中添加一个 messages 键，如下所示：

1. class MessageState(TypedDict):
2.     #消息列表(AIMessage,ToolMessage,HumanMessage)
3.     messages:Annotated[list[AnyMessage],operator.add]
4.     #LLM调用次数
5.     llm_calls:Annotated[int,operator.add]

复制代码

3.设置 Edges

已经定义了两个节点：

节点1（ tool_node ）：专门负责搜索，获取搜索结果。
节点2（ llm_call ）：专门负责调用 LLM，获取最终结果。

因此，可以定义以下逻辑：

对于开始逻辑：是当用户进行输入后，直接让 LLM 进行处理。是否需要搜索⼯具调用，也是通过LLM 进行判断。因此图的入口点则是 llm_call 节点。
对于结束逻辑：可以根据 LLM 返回的结构判断是否调用工具，来决定我们是应该继续循环还是停止循环：
如果 LLM 要调用工具，则进入tool_node节点进行处理；
如果 LLM 不要调用工具，则结束。

4.完整代码

1. # 支持搜索的智能代理系统
2. import operator
3. from typing import TypedDict, Annotated
5. from langchain.chat_models import init_chat_model
6. from langchain_core.messages import AnyMessage, SystemMessage, ToolMessage, HumanMessage
7. from langchain_tavily import TavilySearch
8. from langgraph.constants import START, END
9. from langgraph.graph import StateGraph
11. # 定义工具,模型绑定工具
12. search=TavilySearch(max_results=4)
13. tools =[search]
14. model = init_chat_model(model="gpt-4o-mini")
15. #绑定工具
16. model_with_tools=model.bind_tools(tools)
18. # 1.定义状态
19. class MessageState(TypedDict):
20.     #消息列表(AIMessage,ToolMessage,HumanMessage)
21.     messages:Annotated[list[AnyMessage],operator.add]
22.     #LLM调用次数
23.     llm_calls:Annotated[int,operator.add]
25. # 2.定义节点
27. # 定义调用llm节点
28. # 可能是由START节点过来的
29. # 也可能是由"工具调用"节点过来的
30. def llm_call(state:MessageState):
31.     """LLM决定是否调用工具"""
32.     # 如果是由START节点过来的,state中的messages就只包含用户输入的消息(HumanMessage)
33.     # 如果是由"工具调用"节点过来的,state中的messages就包含(HumanMessage,AIMessage,ToolMessage)
34.     messages = state["messages"]
36.     # result可能1: 带有tool_calls的AIMessage
37.     # result可能2: 不带tool_calls的AIMesssage
38.     result = model_with_tools.invoke(
39.         [
40.             SystemMessage("你是一个乐于助人的助手,支持调用工具进行搜索")
41.         ]
42.         + messages
43.     )
44.     return {
45.         "messages": [result],
46.         "llm_calls":1
47.     }
49. # 获取所有工具的名称
50. # 工具名称和工具的映射
51. tools_by_name = {tool.name:tool for tool in tools}
53. # 定义调用工具节点
54. def tool_node(state:MessageState):
55.     # 此时的AIMessage中一定包含工具调用的信息(tool_calls)
56.     # 我们需要找到对应的工具并调用
57.     # 当前state中的messages中的AIMessage就包含工具调用的信息(tool_call)
58.     # 例如: 工具的ID,工具的参数,工具的名称
59.     result = []    # 填充ToolMessage(工具的执行结果)
60.     for tool_call in state["messages"][-1].tool_calls:
61.         tool = tools_by_name[tool_call["name"]] # 由工具名称找到对应的工具
62.         obs = tool.invoke(tool_call["args"])    # 传入参数,调用工具
63.         result.append(ToolMessage(content=obs,tool_call_id=tool_call["id"]))
65.     return {
66.         "messages":result
67.     }
69. # 3.定义图,添加节点和边
70. agent_builder = StateGraph(MessageState)
71. agent_builder.add_node(llm_call)
72. agent_builder.add_node(tool_node)
74. # 设置固定边
75. agent_builder.add_edge(START,"llm_call")
77. def should_continue(state:MessageState):
78.     # 如果state中的messages的AIMessage中包含tool_call,则要走tool_node节点
79.     last_message = state["messages"][-1]
80.     if last_message.tool_calls:
81.         return "tool_node"
82.     return END
84. # 设置条件边
85. agent_builder.add_conditional_edges(
86.     "llm_call",
87.     should_continue,
88.     ["tool_node",END]
89. )
91. agent_builder.add_edge("tool_node","llm_call")
93. # 4.编译图
94. agent_search = agent_builder.compile()
96. # 5.执行图
97. result = agent_search.invoke({
98.     "messages":[HumanMessage(content="今天西安的天气怎么样?")]
99. })
101. print(f"一共调用llm{result['llm_calls']}次")
102. for msg in result["messages"]:
103.     msg.pretty_print()

复制代码

3.4基于LangGraph实现的RAG

构建一个智能文档问答系统，核心思路如下:

模块化设计：每个节点只做一件事，职责清晰
质量闭环：检索 → 检查 → 优化 → 再检索，确保答案质量
智能路由：AI 自主决定下一步⾏动，无需人工干预

因此，我们需要：

第一步：准备"知识库"（数据加载与处理）
第二步：创建"检索工具"
第三步：设计"工作流程节点"

节点1：决策节点 generate_query_or_respond
节点2：检索器工具节点 retrieve
节点3：问题优化节点 rewrite_question
节点4：答案生成节点 generate_answer

第四步：组装"工作流水线"

条件边1：LLM 决策是否需要进行知识库检索
条件边2：检测【检索到的文档】是否与【问题】相关

最终graph示意图：

1.装备知识库并创建检索工具

1. # 1.准备知识库,并创建检索工具
3. # 创建聊天模型与嵌入式模型
4. model = init_chat_model("gpt-4o-mini")
5. embedding_model = OpenAIEmbeddings(model="text-embedding-3-large")
7. # 文档列表
8. paths =[
9.     "../Docs/MarkDown/城市展厅项目.md",
10.     "../Docs/MarkDown/智能PPT导览项目.md",
11.     "../Docs/MarkDown/记忆岛项目.md"
12. ]
14. # 文档加载器
15. # 得到的一个结果的结构类似：
16. # [Document(page_content="文档1内容", metadata={"source": "doc1.md"})]
17. docs = [UnstructuredMarkdownLoader(path).load() for path in paths]
18. # docs = [
19. #     [Document(page_content="test1的内容", metadata={"source": "test1.md"})],
20. #     [Document(page_content="test2的内容", metadata={"source": "test2.md"})]
21. # ]
23. # 把嵌套列表转换成扁平列表
24. docs_list = [item for sublist  in docs for item in sublist]
25. # docs_list = [
26. #     Document(page_content="test1的内容", metadata={"source": "test1.md"}),
27. #     Document(page_content="test2的内容", metadata={"source": "test2.md"})
28. # ]
30. # 使用分词器对文档进行分割
31. text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
32.     encoding_name="cl100k_base",
33.     chunk_size=1000,
34.     chunk_overlap=50
35. )
36. doc_splits = text_splitter.split_documents(docs_list)
38. # 使用嵌入模型将文档转化为向量,使用内存数据库存储
39. vector_store = InMemoryVectorStore.from_documents(
40.     documents=doc_splits,
41.     embedding=embedding_model
42. )
44. # 使⽤ LangChain 的预构建 create_retriever_tool 创建检索器⼯具
45. retriever = vector_store.as_retriever(search_kwargs={"k": 2})
46. retriever_tool = create_retriever_tool(
47.     retriever,
48.     "retrieve",
49.     "搜索并返回有关城市展厅/智能PPT导览/记忆岛项目开发的信息。",
50. )

复制代码

2.创建节点

generate_query_or_respond节点
该节点核心设计：
决定是直接回答还是检索文档
可以使用 model.bind_tools([retriever_tool]) 让模型能够调用检索工具

1. # 2.定义状态
2. # 此处可以自定义消息状态
3. # 也可以直接使用LangGraph包中提供的MessagesState
4. # class MessageState(TypedDict):
5. #     #消息列表(AIMessage,ToolMessage,HumanMessage)
6. #     messages:Annotated[list[AnyMessage],operator.add]
9. # 3.定义节点
10. # 定义决策节点(AI决策是否需要检索)
11. # 调用LLM绑定检索工具的节点,判断是否需要调用检索工具
12. def generate_query_or_respond(state:MessagesState):
13.     """调用模型基于当前的状态生成响应,给定问题，它将决定是否使用工具进行检索,或者是直接给用户返回答案"""
14.     result = model.bind_tools([retriever_tool]).invoke(state["messages"])
15.     # result是AIMessage,可能包含tool_calls,也可能不包含
16.     return {
17.         "messages":[result]
18.     }

复制代码

检索器工具节点retrieve
前⾯已经创建好了检索器工具。对于节点，可以使用 ToolNode 类来定义：

1. # 定义工具节点,调用检索工具
2. # 可以实现一个函数,函数中调用工具
3. # 也可以直接使用LangGraph提供的ToolNode来生成工具节点
4. retriever_node = ToolNode([retriever_tool])

复制代码

问题优化节点 rewrite_question
该节点主要负责当文档不相关时，重写问题以改进检索效果.

1. # 让LLM重写问题的提示词
2. REWRITE_PROMPT = (
3.     "查看输⼊并尝试推断潜在的语义意图/含义。\n"
4.     "这是最初的问题："
5.     "\n ------- \n"
6.     "{question}"
7.     "\n ------- \n"
8.     "提出⼀个改进后的问题："
9. )
10. # 定义重写节点(问题优化节点,优化用户提出的问题)
11. def rewrite_node(state:MessagesState):
12.     """重新用户原始问题,对用户的问题进行改进"""
13.     messages = state["messages"]
14.     # 此时的messages中包含
15.     # HumanMessage(用户输入的问题)
16.     # AIMessage(第一次调用LLM的结果,其中包含tool_calls)
17.     # ToolMessage(工具调用的结果)
18.     # 现在需要获取到HumanMessage,并调用LLM进行优化
19.     question = messages[0].conent
20.     prompt = REWRITE_PROMPT.format(question=question)
21.     response = model.invoke([HumanMessage(content=prompt)])
22.     #更新节点状态,把修改后的用户问题,设置为新的用户消息
23.     return {
24.         "messages":[HumanMessage(content=response.content)]
25.     }

复制代码

答案⽣成节点 generate_answer
该节点将基于相关文档生成最终答案。

1. # 定义答案节点
2. # 定义答案生成的提示词
3. GENERATE_PROMPT = (
4.     "你是负责回答问题的助⼿。 "
5.     "使⽤以下检索到的上下⽂⽚段来回答问题。 "
6.     "如果你不知道答案，就说你不知道。 "
7.     "最多只⽤三句话，回答要简明扼要。\n"
8.     "Question: {question} \n"
9.     "Context: {context}"
10. )
11. def generate_answer(state:MessagesState):
12.     # 此时的消息中包含[HumanMessage,AIMessage,ToolMessage]
13.     # 问题+检索结果
14.     question = state["messages"][0].content   #HumanMessage
15.     context = state["messages"][-1].content   #ToolMessage
16.     prompt = GENERATE_PROMPT.format(question=question,context=context)
17.     #调用LLM构建AIMessage,返回最终的结果
18.     return {
19.         "messages":[model.invoke(HumanMessage(context=prompt))]
20.     }

复制代码

3.组装"工作流水线"

根据下图，开始组装"工作流水线"。

添加节点与入口点

1. # 4.创建图,添加节点、边
2. workflow = StateGraph(MessagesState)
3. workflow.add_node(generate_query_or_respond)    #决策节点
4. workflow.add_node("retriever",retriever_node)   #检索节点
5. workflow.add_node(rewrite_node)                 #重写问题节点
6. workflow.add_node(generate_answer)              #生成答案节点
8. workflow.add_edge(START,"generate_query_or_respond")
9. workflow.add_conditional_edges(
10.     "generate_query_or_respond",
11.     tools_condition,   #LangGraph提供的函数,检查是否需要调用工具
12.     {
13.         "tools":"retriever",
14.         "__end__":END
15.     }
16. )
19. #定义LLM调用后的结构化输出结果
20. class GradeDocumnet(BaseModel):
21.     score:str=Field(description="相关性评分,如果相关则为“yes”,如果不相关则为“no”")
23. GRADE_PROMPT = (
24.     "你是⼀个评分员，评估检索到的⽂档与⽤⼾问题的相关性。 \n "
25.     "以下是检索到的⽂档： \n\n {context} \n\n"
26.     "以下是⽤⼾的问题： {question} \n"
27.     "如果⽂档包含与⽤⼾问题相关的关键字或语义，则将其评为相关。 \n"
28.     "给出⼀个⼆元分数“yes”或“no”，以表明该⽂档是否与问题相关。"
29. )
30. #判断检索到的结果是否正确(正确则生成答案,有问题怎重写问题)
31. def grade_documents(state:MessagesState)-> Literal["rewrite_node", "generate_answer"]:
32.     """确定检索到的文档是否与问题相关"""
33.     #将问题和检索到的文档交给LLM,判断是否相关(yes->rewrite_node,no->generate_answer)
34.     #需要过滤出用户最新的消息和检索工具调用的结果
35.     user_messages = filter_messages(state["messages"])
36.     question = user_messages[-1].content
37.     tool_message = state["messages"][-1]
38.     context = tool_message.content
39.     prompt = GRADE_PROMPT.format(context=context,question=question)
40.     result = model.with_structured_output(GradeDocumnet).invoke(
41.         [HumanMessage(content=prompt)]
42.     ).content
43.     if result.str=="yes":
44.         return "generate_answer"
45.     else:
46.         return "rewrite_node"
49. workflow.add_conditional_edges(
50.     "retriever",
51.     grade_documents,
52.     ["generate_answer","rewrite_node"]
53. )
55. workflow.add_edge("generate_answer",END)
56. workflow.add_edge("rewrite_node","generate_query_or_respond")

复制代码

编译并运行图

1. # 5.编译图
2. graph = workflow.compile()
4. # 执行图
5. # 流式输出
6. for chunk in graph.stream(
7.     {
8.     "messages":[HumanMessage(content="记忆岛这个项目的开发用到了哪些技术?")]
9.     }
10. ):
11.     print(chunk)

复制代码

4.完整代码

1. # 基于LangGraph实现的RAG(检索增强生成)
2. from typing import Literal
4. from langchain.chat_models import init_chat_model
5. from langchain_community.document_loaders import UnstructuredMarkdownLoader
6. from langchain_core.messages import HumanMessage, filter_messages
7. from langchain_core.tools import create_retriever_tool
8. from langchain_core.vectorstores import InMemoryVectorStore
9. from langchain_openai import OpenAIEmbeddings
10. from langchain_text_splitters import RecursiveCharacterTextSplitter
11. from langgraph.constants import START, END
12. from langgraph.graph import MessagesState, StateGraph
13. from langgraph.prebuilt import ToolNode, tools_condition
14. from pydantic import BaseModel, Field
16. # 1.准备知识库,并创建检索工具
18. # 创建聊天模型与嵌入式模型
19. model = init_chat_model("gpt-4o-mini")
20. embedding_model = OpenAIEmbeddings(model="text-embedding-3-large")
22. # 文档列表
23. paths =[
24.     "../Docs/MarkDown/城市展厅项目.md",
25.     "../Docs/MarkDown/智能PPT导览项目.md",
26.     "../Docs/MarkDown/记忆岛项目.md"
27. ]
29. # 文档加载器
30. # 得到的一个结果的结构类似：
31. # [Document(page_content="文档1内容", metadata={"source": "doc1.md"})]
32. docs = [UnstructuredMarkdownLoader(path).load() for path in paths]
33. # docs = [
34. #     [Document(page_content="test1的内容", metadata={"source": "test1.md"})],
35. #     [Document(page_content="test2的内容", metadata={"source": "test2.md"})]
36. # ]
38. # 把嵌套列表转换成扁平列表
39. docs_list = [item for sublist  in docs for item in sublist]
40. # docs_list = [
41. #     Document(page_content="test1的内容", metadata={"source": "test1.md"}),
42. #     Document(page_content="test2的内容", metadata={"source": "test2.md"})
43. # ]
45. # 使用分词器对文档进行分割
46. text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
47.     encoding_name="cl100k_base",
48.     chunk_size=1000,
49.     chunk_overlap=50
50. )
51. doc_splits = text_splitter.split_documents(docs_list)
53. # 使用嵌入模型将文档转化为向量,使用内存数据库存储
54. vector_store = InMemoryVectorStore.from_documents(
55.     documents=doc_splits,
56.     embedding=embedding_model
57. )
59. # 使⽤ LangChain 的预构建 create_retriever_tool 创建检索器⼯具
60. retriever = vector_store.as_retriever(search_kwargs={"k": 2})
61. retriever_tool = create_retriever_tool(
62.     retriever,
63.     "retrieve",
64.     "搜索并返回有关城市展厅/智能PPT导览/记忆岛项目开发的信息。",
65. )
67. # 2.定义状态
68. # 此处可以自定义消息状态
69. # 也可以直接使用LangGraph包中提供的MessagesState
70. # class MessageState(TypedDict):
71. #     #消息列表(AIMessage,ToolMessage,HumanMessage)
72. #     messages:Annotated[list[AnyMessage],operator.add]
75. # 3.定义节点
76. # 定义决策节点(AI决策是否需要检索)
77. # 调用LLM绑定检索工具的节点,判断是否需要调用检索工具
78. def generate_query_or_respond(state:MessagesState):
79.     """调用模型基于当前的状态生成响应,给定问题，它将决定是否使用工具进行检索,或者是直接给用户返回答案"""
80.     result = model.bind_tools([retriever_tool]).invoke(state["messages"])
81.     # result是AIMessage,可能包含tool_calls,也可能不包含
82.     return {
83.         "messages":[result]
84.     }
85. # 定义工具节点,调用检索工具
86. # 可以实现一个函数,函数中调用工具
87. # 也可以直接使用LangGraph提供的ToolNode来生成工具节点
88. retriever_node = ToolNode([retriever_tool])
89. # 让LLM重写问题的提示词
90. REWRITE_PROMPT = (
91.     "查看输⼊并尝试推断潜在的语义意图/含义。\n"
92.     "这是最初的问题："
93.     "\n ------- \n"
94.     "{question}"
95.     "\n ------- \n"
96.     "提出⼀个改进后的问题："
97. )
98. # 定义重写节点(问题优化节点,优化用户提出的问题)
99. def rewrite_node(state:MessagesState):
100.     """重新用户原始问题,对用户的问题进行改进"""
101.     messages = state["messages"]
102.     # 此时的messages中包含
103.     # HumanMessage(用户输入的问题)
104.     # AIMessage(第一次调用LLM的结果,其中包含tool_calls)
105.     # ToolMessage(工具调用的结果)
106.     # 现在需要获取到HumanMessage,并调用LLM进行优化
107.     question = messages[0].conent
108.     prompt = REWRITE_PROMPT.format(question=question)
109.     response = model.invoke([HumanMessage(content=prompt)])
110.     #更新节点状态,把修改后的用户问题,设置为新的用户消息
111.     return {
112.         "messages":[HumanMessage(content=response.content)]
113.     }
114. # 定义答案节点
115. # 定义答案生成的提示词
116. GENERATE_PROMPT = (
117.     "你是负责回答问题的助⼿。 "
118.     "使⽤以下检索到的上下⽂⽚段来回答问题。 "
119.     "如果你不知道答案，就说你不知道。 "
120.     "最多只⽤三句话，回答要简明扼要。\n"
121.     "Question: {question} \n"
122.     "Context: {context}"
123. )
124. def generate_answer(state:MessagesState):
125.     # 此时的消息中包含[HumanMessage,AIMessage,ToolMessage]
126.     # 问题+检索结果
127.     question = state["messages"][0].content   #HumanMessage
128.     context = state["messages"][-1].content   #ToolMessage
129.     prompt = GENERATE_PROMPT.format(question=question,context=context)
130.     #调用LLM构建AIMessage,返回最终的结果
131.     return {
132.         "messages":[model.invoke(HumanMessage(context=prompt))]
133.     }
136. # 4.创建图,添加节点、边
137. workflow = StateGraph(MessagesState)
138. workflow.add_node(generate_query_or_respond)    #决策节点
139. workflow.add_node("retriever",retriever_node)   #检索节点
140. workflow.add_node(rewrite_node)                 #重写问题节点
141. workflow.add_node(generate_answer)              #生成答案节点
143. workflow.add_edge(START,"generate_query_or_respond")
144. workflow.add_conditional_edges(
145.     "generate_query_or_respond",
146.     tools_condition,   #LangGraph提供的函数,检查是否需要调用工具
147.     {
148.         "tools":"retriever",
149.         "__end__":END
150.     }
151. )
154. #定义LLM调用后的结构化输出结果
155. class GradeDocumnet(BaseModel):
156.     score:str=Field(description="相关性评分,如果相关则为“yes”,如果不相关则为“no”")
158. GRADE_PROMPT = (
159.     "你是⼀个评分员，评估检索到的⽂档与⽤⼾问题的相关性。 \n "
160.     "以下是检索到的⽂档： \n\n {context} \n\n"
161.     "以下是⽤⼾的问题： {question} \n"
162.     "如果⽂档包含与⽤⼾问题相关的关键字或语义，则将其评为相关。 \n"
163.     "给出⼀个⼆元分数“yes”或“no”，以表明该⽂档是否与问题相关。"
164. )
165. #判断检索到的结果是否正确(正确则生成答案,有问题怎重写问题)
166. def grade_documents(state:MessagesState)-> Literal["rewrite_node", "generate_answer"]:
167.     """确定检索到的文档是否与问题相关"""
168.     #将问题和检索到的文档交给LLM,判断是否相关(yes->rewrite_node,no->generate_answer)
169.     #需要过滤出用户最新的消息和检索工具调用的结果
170.     user_messages = filter_messages(state["messages"])
171.     question = user_messages[-1].content
172.     tool_message = state["messages"][-1]
173.     context = tool_message.content
174.     prompt = GRADE_PROMPT.format(context=context,question=question)
175.     result = model.with_structured_output(GradeDocumnet).invoke(
176.         [HumanMessage(content=prompt)]
177.     ).content
178.     if result.str=="yes":
179.         return "generate_answer"
180.     else:
181.         return "rewrite_node"
184. workflow.add_conditional_edges(
185.     "retriever",
186.     grade_documents,
187.     ["generate_answer","rewrite_node"]
188. )
190. workflow.add_edge("generate_answer",END)
191. workflow.add_edge("rewrite_node","generate_query_or_respond")
193. # 5.编译图
194. graph = workflow.compile()
196. # 执行图
197. # 流式输出
198. for chunk in graph.stream(
199.     {
200.     "messages":[HumanMessage(content="记忆岛这个项目的开发用到了哪些技术?")]
201.     }
202. ):
203.     print(chunk)

复制代码

原文地址：https://blog.csdn.net/2401_82677021/article/details/159131621