AI创想

标题: LangGraph从入门到精通(一)——LangGraph介绍与基础组件 [打印本页]

作者: 米落枫 时间: 11 小时前
标题: LangGraph从入门到精通(一)——LangGraph介绍与基础组件
作者：小陈phd
01. LangGraph 介绍与功能

LangGraph 是一个构建具有状态、多角色应用程序的库，用于创建智能体和多智能体工作流，与其他 LLM 框架相比，LangGraph 提供了以下核心优势：循环、可控制性和持久性。LangGraph 允许定义设计循环、条件判断的流程，这对于高级 Agent 非常重要，这和传统的有向无换图（DAG）解决方案区分开。
因为 LangGraph 作为最底层的框架，所以涉及的组件都是最基础的，并没有过度封装，允许开发者实现对应用程序流程和状态的精细控制（自由度极高），而且 LangGraph 可以便捷集成持久化方案、任意节点中断交互、修改状态等特性，该框架虽然由 LangChain 团队构建，但是却可以在没有 LangChain 的情况下单独使用。
对比 LCEL 表达式，LangGraph 的主要功能：
• 循环和分支：在 LLM/Agent 应用程序中使用便捷的方式实现循环和条件语句，而无需配置额外的程序。
• 持久化：在图的每个步骤之后自动保存状态，在运行的任意一个阶段都支持暂停和恢复图执行，以支持错误恢复、人机交互工作流、时间旅行等。
• 人机交互：中断图的执行以批准或者编辑状态计划去执行下一计划。
• 流支持：图结构的每个节点都支持流式输出（包括 token 流式输出）。
• 与LangChain集成：LangGraph 可以和 LangChain 和 LangSmith 无缝集成

(, 下载次数: 0)

在一个最基础 LangGraph 应用程序中，涵盖了 3 种基本组件：

边：边在图中定义了路由逻辑，即不同节点之间是如何传递的，传递给谁，以及图节点从哪里开始，从哪里结束，并且一个节点可以设置多条边，如果有多条边，则下一条边连接的所有节点都会并行运行。
由于 LangGraph 是一个独立的框架，使用前必须先安装，命令如下

pip install-U langgraph

复制代码

02. LangGraph 基础组件使用

通过上面的图示例，其实可以很容易知道一个 LangGraph 应用程序的组成部分：节点、边、状态，接下来我们来利用这 3 个组件来构建一个最最基础的图架构应用，仅包含开始节点、中间节点和结束节点的聊天机器人应用，其图结构如下：

(, 下载次数: 0)

在这个图架构应用中，开始节点连接 LLM大语言模型节点，并且该节点接收状态数据，并生成对应内容，随后传递给结束节点。简单来说，节点完成工作。边指示下一步要做什么。
在 LangGraph 中，开始/结束节点作为特殊节点，并预定义了，导入后即可使用
from langgraph.graph import START, END
对于状态的声明，LangGraph 推荐使用的是 TypedDict 或者 Pydantic模型，并且在 LangGraph 中，还支持对声明的状态使用归纳函数，归纳函数可以修改数据的更新方式，例如下方是一个没有为任何键添加归纳函数的状态:

from typing import TypedDict
classState(TypedDict):
foo:int
bar:list[str]

复制代码

假设在上述图结构中，输入是 {“foo”: 1, “bar”: [“hi”]}，并且第一个节点返回 {“foo”: 2}，LangGraph 会根据返回的数据对状态进行更新，并且节点不需要返回所有数据，这样下一个节点接收到的状态数据为 {“foo”: 2, “bar”: [“hi”]}，可以看到默认更新方式是覆盖更新。
如果要使用归纳函数，我们可以使用 Annotated 类型来为特定的键添加，例如为第二个键添加 operator.add 函数，代码更新如下:

from typing import TypedDict, Annotated
from operator import add
classState(TypedDict):
foo:int
bar: Annotated[list[str], add]

复制代码

这个时候，假设图的输入是 {“foo”: 1, “bar”: [“hi”]}，然后，第一个节点返回 {“bar”: [“bye”]}，此时状态数据变为 {“foo”: 2, “bar”: [“hi”, “bye”]}，即数据会使用 Annotated 标注的函数来进行更新（add 函数会将两个列表的数据相加）。
创建好状态后，就可以根据状态来实例化图结构了，在 LangGraph 中有两种类型的图，一种是 StateGraph，另外一种是 MessageGraph，前者的状态是自定义字典，后者的状态是消息列表，使用示例
from langgraph.graph import StateGraph
graph_builder = StateGraph(State)
创建好图结构后，即可为图添加节点与边，使用的函数也非常简单：
• add_node(节点名称, 节点对应函数)：将一个函数添加到图上，并设置对应的节点名称。
• add_edge(开始节点名称, 结束节点名称)：将两个节点进行拼接，第一个参数是拼接的开始节点，第二个参数是拼接的结束节点。
完成上述的步骤后，这个时候的图结构仍然不可使用，我们需要将其转换成 Runnable可运行组件，这个时候就可以调用 .compile() 函数进行编译，编译返回的数据就是 Runnable可运行组件，可以调用统一的接口，例如：invoke、stream、batch 等。
通过上述的分析，可以知道在 LangGraph 中，无论多么复杂的项目，都可以拆分成对应的 6 个步骤，只需按照这 6 个步骤来执行即可：

调用编译后的 Runnable 可运行组件执行图（graph.invoke 函数调用图）。
例如将上述的 3节点聊天机器人实现后，示例代码如下:

from typing import TypedDict, Annotated
import dotenv
from langchain_openai import ChatOpenAI
from langgraph.graph import START, END
from langgraph.graph import StateGraph
from langgraph.graph.message import add_messages
dotenv.load_dotenv()classGraphState(TypedDict):"""图状态数据结构，类型为字典"""
messages: Annotated[list, add_messages]# 设置消息列表，并添加归纳函数
node_name:str
llm = ChatOpenAI(model="gpt-4o-mini")defchatbot(state: GraphState)-> GraphState:"""聊天机器人函数"""# 1.获取状态里存储的消息列表数据并传递给LLM
ai_message = llm.invoke(state["messages"])# 2.返回更新/生成的状态return{"messages":[ai_message],"node_name":"chatbot"}# 1.创建状态图，并使用GraphState作为状态数据
graph_builder = StateGraph(GraphState)# 2.添加节点
graph_builder.add_node("llm", chatbot)# 3.添加边
graph_builder.add_edge(START,"llm")
graph_builder.add_edge("llm", END)# 4.编译图为Runnable可运行组件
graph = graph_builder.compile()# 5.调用图架构应用print(graph.invoke({"messages":[("human","你好，你是？")],"node_name":"graph"}))

复制代码

输出内容:

{'messages': [HumanMessage(content='你好，你是？', id='b1bf444b-ad7f-4cf7-bccb-510c85787d32'), AIMessage(content='你好！我是一个人工智能助手，随时准备回答你的问题或帮助你解决问题。有什么我可以帮你的吗？', response_metadata={'token_usage': {'completion_tokens': 27, 'prompt_tokens': 11, 'total_tokens': 38}, 'model_name': 'gpt-4o-mini', 'system_fingerprint': 'fp_80a1bad4c7', 'finish_reason': 'stop', 'logprobs': None}, id='run-97b1f830-8b49-4531-af41-fa26f86b1ded-0', usage_metadata={'input_tokens': 11, 'output_tokens': 27, 'total_tokens': 38})], 'node_name': 'chatbot'}

复制代码

得益于 LangSmith 日志信息的采集，并且编译后的图架构应用也是一个 Runnable 可运行组件，所以可以完整观察图的整个运行流程，为后续复杂图架构应用开发调试流程可观察提供了基础。

(, 下载次数: 0)

from typing import TypedDict, Annotated, Any
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.runnables import RunnableConfig # 导入 RunnableConfigimport dotenv
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
dotenv.load_dotenv()
llm = ChatOpenAI(model="qwen-plus")# 1. 创建状态图，并使用GraphState作为状态数据classState(TypedDict):"""图结构的状态数据"""
messages: Annotated[list, add_messages]
use_name:strdefchatbot(state: State, config: RunnableConfig)-> Any:# 修改 config 类型
response = llm.invoke(state["messages"])# response 已是 AIMessage 类型return{"messages":[response],"use_name":"chatbot"}# 4. 构建 Graph
graph_builder = StateGraph(State)
graph_builder.add_node("llm", chatbot)
graph_builder.add_edge(START,"llm")
graph_builder.add_edge("llm", END)
graph = graph_builder.compile(checkpointer=MemorySaver())# 5. 输入消息
state_input ={"messages":[HumanMessage(content="你好，你是谁，我喜欢篮球")],"use_name":"graph"}# 创建 RunnableConfig 实例
config = RunnableConfig(configurable_id="chenweifeng", thread_id="1")# 运行
result = graph.invoke(
state_input,
config=config,# 使用 RunnableConfig 实例)print(result)

复制代码

(, 下载次数: 0)

原文地址：https://blog.csdn.net/weixin_42917352/article/details/150543995

欢迎光临 AI创想 (https://www.llms-ai.com/)