AI创想

标题: LangGraph——AI应用开发框架 [打印本页]
作者: 米落枫    时间: 4 小时前
标题: 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 只能写：
LangGraph 就是为了解决这些：
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 比作一个公司，那么：
简单来说，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 ，以及添加到其中的各种增强功能（如工具调用、结构化输出和短期记忆）而实现：
(, 下载次数: 0)


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

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


(, 下载次数: 0)


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（节点） - 快递站点

节点就像快递配送⽹络中的各个站点，每个站点负责特定的处理步骤。例如：
揽收站：接收包裹，初始化信息
分拣中心：根据目的地分类包裹
派送站：最终配送至收件人
节点特征：
代码示例（节点只不过是函数）：
  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 中，边就定义了节点之间的连接关系，决定了⼯作流的执⾏顺序。边的类型有：
因此，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，规划运输路线

各节点（站点）准备好后，则需要为快递运输规划。
实际上，这就是为 图 定义 边 。边有几种关键类型：
例如，设置最简单运输路线：快递由揽收站接收，下⼀站固定为分拣中⼼，最后到派送中进⾏派送。
  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 中：
6.StateGraph 图编译，从公司创建到运⾏

在步骤2中，我们仅是构建出 StateGraph ，还无法直接⽤于执⾏。LangGraph 要求：必须先编译图，然后才能使⽤它。编译提供了对图结构的一些基本检查，这会验证：
使⽤ 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"])
复制代码
(, 下载次数: 0)


到此，我们已经构建出了⼀个图式的智能快递配送系统：
3.3 支持搜索的智能代理系统

核心功能如下：
流程设置如下
(, 下载次数: 0)


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

已经定义了两个节点：
因此，可以定义以下逻辑：
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

构建一个智能文档问答系统，核心思路如下:
因此，我们需要：
最终graph示意图：
(, 下载次数: 0)


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.组装"工作流水线"

根据下图，开始组装"工作流水线"。
(, 下载次数: 0)


添加节点与入口点
  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



欢迎光临 AI创想 (http://llms-ai.com/) Powered by Discuz! X3.4