LangGraph基础知识(Graph-GraphState)

背景

相信大家都听过LangChain,也相信大家有个疑问：为什么有LangChain 了 还需要有LangGraph呢？
用专业的术语来说，使用`LangChain` 构建的是 DAG（有向无环图）。而之所以会出现`LangGraph`框架，根本原因是在于随着AI应用（特别是AI Agent）的发展，**对于大语言模型的使用不仅仅是作为执行工具，而更多作为推理引擎的需求在日益增长**。这种转变带来的是更多的重复（循环）和复杂条件的交互需求，这就导致**基于`LCEL`的线性序列构建方式在构建更复杂、更智能的系统时显示出了明显的局限性。

而LangGraph要解决线性序列的局限性问题，而解决的方法就是循环图

Graph 基类

对于任意一个简单或者复杂的图来说，都是基于`Graph`类来构建和管理图结构的。在`Graph`类中允许添加节点、边，并定义节点间的动态流转逻辑。如下是`Graph`类的主要组成部分和功能：

from collections import defaultdict
from typing import Any, Callable, Dict, Optional, Set, Tuple, Union, Awaitable, Hashableclass Graph:def __init__(self) -> None:self.nodes: Dict[str, Any] = {}  # 一个字典，用于存储图中的所有节点。每个节点可以是一个字符串标识或者是一个可调用对象self.edges: Set[Tuple[str, str]] = set()  # 一个集合，用来存储图中所有的边，边由一对节点名称组成，表示从一个节点到另一个节点的直接连接。self.branches: defaultdict = defaultdict(dict)  # 一个默认字典，用于存储条件分支，允许从一个节点根据特定条件转移到多个不同的节点。self.support_multiple_edges = False  # 一个布尔值，指示图是否支持同一对节点间的多条边。self.compiled = False    #  一个布尔值，表示图是否已经被编译。编译是指图的结构已经设置完毕，准备进行执行。@propertydef _all_edges(self) -> Set[Tuple[str, str]]:"""添加一个新节点到图中。节点可以有附加的元数据，这些元数据存储在节点的字典中。"""return self.edgesdef add_node(self, node: Union[str, Callable], action: Optional[Callable] = None, *, metadata: Optional[Dict[str, Any]] = None) -> 'Graph':"""添加一个新节点到图中。节点可以有附加的元数据，这些元数据存储在节点的字典中。"""passdef add_edge(self, start_key: str, end_key: str) -> 'Graph':"""在图中添加一条边，连接两个指定的节点。"""passdef add_conditional_edges(self, source: str, path: Callable, path_map: Optional[Dict[Hashable, str]] = None, then: Optional[str] = None) -> 'Graph':"""添加一个条件边，允许在执行时根据某个条件从一个节点动态地转移到一个或多个节点。"""passdef set_entry_point(self, key: str) -> 'Graph':"""设置图的入口点，即定义图执行的起始节点。"""passdef set_conditional_entry_point(self, path: Callable, path_map: Optional[Dict[Hashable, str]] = None, then: Optional[str] = None) -> 'Graph':"""设置一个条件入口点，允许根据条件动态决定图的起始执行点。"""passdef set_finish_point(self, key: str) -> 'Graph':"""设置结束点，定义图执行到此节点时将停止。"""passdef validate(self, interrupt: Optional[Set[str]] = None) -> 'Graph':"""验证图的结构是否正确，确保所有节点和边的定义都符合逻辑和图的规则。"""passdef compile(self, checkpointer=None, interrupt_before: Optional[Set[str]] = None, interrupt_after: Optional[Set[str]] = None, debug: bool = False) -> 'Graph':"""编译图，确认图的结构合法且可执行后，准备图以供执行。"""pass

GraphState

定义图时要做的第一件事是定义图的`State`。状态表示会随着图计算的进行而维护和更新的上下文或记忆。它用来确保图中的每个步骤都可以访问先前步骤的相关信息，从而可以根据整个过程中积累的数据进行动态决策。这个过程通过状态图`StateGraph`类实现，它继承自 `Graph` 类，这意味着 `StateGraph` 会使用或扩展基类的属性和方法。

rom collections import defaultdict
from typing import Any, Callable, Dict, Optional, Set, Tuple, Type, Unionclass StateGraph(Graph):"""StateGraph 是一个管理状态并通过定义的输入和输出架构支持状态转换的图。"""def __init__(self, state_schema: Optional[Type[Any]] = None, config_schema: Optional[Type[Any]] = None) -> None:super().__init__()self.state_schema = state_schema      # 一个可选的类型参数，定义图状态的结构。这是用于定义和验证图中节点处理的状态数据的模式。self.config_schema = config_schema    # 一个可选的类型参数，用于定义配置的结构。这可以用于定义和验证图的配置参数。self.nodes: Dict[str, Any] = {}       # 一个字典，用于存储图中的节点。每个节点可以关联特定的动作和其他数据。self.edges: Set[Tuple[str, str]] = set()   # 一个集合，存储图中所有的边。每条边由一对字符串组成，表示从一个节点到另一个节点的连接。self.branches: defaultdict = defaultdict(dict)   # 一个默认字典，用于管理节点间的条件分支。这使得从一个节点基于某些条件跳转到不同的节点成为可能。def add_node(self, node: Union[str, Callable], action: Optional[Callable] = None, *, metadata: Optional[Dict[str, Any]] = None) -> 'StateGraph':"""向图中添加一个新节点。节点可以是一个具名字符串或一个可调用对象（如函数）, 如果node是字符串，则action应为与节点关联的可调用动作。"""passdef add_edge(self, start_key: str, end_key: str) -> 'StateGraph':"""在图中添加一条边，连接两个节点。"""passdef compile(self) -> 'CompiledStateGraph':"""编译图，将其转换成可运行的形式。包括验证图的完整性、预处理数据等。"""pass

TypeDict

`TypedDict` 是 `Python` 类型注解系统中的一个工具，它**允许为字典中的键指定期望的具体类型.

示例代码：

from typing import TypedDictclass Contact(TypedDict):name: stremail: strphone: strdef send_email(contact: Contact) -> None:print(f"Sending email to {contact['name']} at {contact['email']}")# 使用定义好的 TypedDict 创建字典
contact_info: Contact = {'name': 'Alice','email': '[e-mail]','phone': '123-456-7890'
}send_email(contact_info)

我们尝试在LangGraph中使用：

from langgraph.graph import StateGraph
from typing_extensions import TypedDict# 定义输入的模式
class InputState(TypedDict):question: str# 定义输出的模式
class OutputState(TypedDict):answer: str# 将 InputState 和 OutputState 这两个 TypedDict 类型合并成一个字典类型。
class OverallState(InputState, OutputState):pass
# 明确指定它的输入和输出数据的结构或模式
builder = StateGraph(OverallState, input=InputState, output=OutputState)

Nodes(节点)

Nodes 是图中的基本执行单元，代表工作流中的一个步骤或操作.

在 LangGraph 中，节点主要有两种类型：

1. 函数节点：封装了特定的功能或操作

   • 可以是同步或异步函数

   • 接收输入并返回输出

   • 示例：调用语言模型、数据转换等

2. 条件节点：控制流程的分支

   • 根据条件决定下一步执行哪个分支

   • 通常与边配合使用

from langgraph.graph import Graphgraph = Graph()# 添加函数节点
graph.add_node("model_call", lambda x: llm.invoke(x))# 添加条件节点
def should_continue(state):if state["needs_more_info"]:return "get_more_info"return "end"

现在有了图结构，并且图结构中也存在两个孤立的节点`agent_node`和`action_node`，接下来我们要做的事就是需要将图中的节点按照我们所期望的方式进行连接，这需要用到的就是`Edges` - 边

Edeges(边)

Edges（边）用来定义逻辑如何路由以及图何时开始与停止。这是代理工作以及不同节点如何相互通信的重要组成部分。有几种关键的边类型：

- 普通边：直接从一个节点到下一个节点。

- 条件边：调用函数来确定下一个要转到的节点。

- 入口点：当用户输入到达时首先调用哪个节点。

- 条件入口点：调用函数来确定当用户输入到达时首先调用哪个节点。

# 添加普通边
graph.add_edge("node1", "node2")# 添加条件边
graph.add_conditional_edges("decision_node",should_continue,  # 条件函数{"continue": "next_node","end": END}
)# 设置起始和结束节点
graph.set_entry_point("start_node")
graph.set_finish_point("end_node")

Graph的调用方法

要调用图中的方法，可以使用 `invoke` 方法

示例代码：

from langgraph.graph import StateGraph
from typing_extensions import TypedDict
from langgraph.graph import START, END# 定义输入的模式
class InputState(TypedDict):question: str
# 定义输出的模式
class OutputState(TypedDict):answer: str
# 将 InputState 和 OutputState 这两个 TypedDict 类型合并成一个更全面的字典类型。
class OverallState(InputState, OutputState):pass
def agent_node(state: InputState):print("我是一个AI Agent。")return {"question": state["question"]}
def action_node(state: InputState):print("我现在是一个执行者。")step = state["question"]return {"answer": f"我接收到的问题是：{step}，读取成功了！"}
# 明确指定它的输入和输出数据的结构或模式
builder = StateGraph(OverallState, input=InputState, output=OutputState)# 添加节点
builder.add_node("agent_node", agent_node)
builder.add_node("action_node", action_node)# 添加边
builder.add_edge(START, "agent_node")
builder.add_edge("agent_node", "action_node")
builder.add_edge("action_node", END)# 编译图
graph = builder.compile()
# 运行图方法
graph.invoke({"question":"今天的天气怎么样？"})

结果输出：我是一个AI Agent。
我现在是一个执行者。

{'answer': '我接收到的问题是：今天的天气怎么样？，读取成功了！'}

最后使用LangGraph构建大模型的问答流程

from langgraph.graph import StateGraph
from typing_extensions import TypedDict
from langgraph.graph import START, END
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from dotenv import load_dotenv
load_dotenv()# 定义输入的模式
class InputState(TypedDict):question: str# 定义输出的模式
class OutputState(TypedDict):answer: str# 将 InputState 和 OutputState 这两个 TypedDict 类型合并成一个更全面的字典类型。
class OverallState(InputState, OutputState):pass
def llm_node(state: InputState):messages = [("system","你是一位乐于助人的智能小助理",),("human", state["question"])]llm = ChatOpenAI(model="deepseek-chat",temperature=0,)response = llm.invoke(messages) return {"answer": response.content}
# 明确指定它的输入和输出数据的结构或模式
builder = StateGraph(OverallState, input=InputState, output=OutputState)# 添加节点
builder.add_node("llm_node", llm_node)# 添加便
builder.add_edge(START, "llm_node")
builder.add_edge("llm_node", END)# 编译图
graph = builder.compile()
graph.invoke({"question":"你好，我用来测试"})

{'answer': '你好！看来你是在测试我是否正常工作呢～ 😊 我状态良好，随时可以回答你的问题、聊天或帮忙处理各种任务。需要我为你做些什么吗？比如： \n- 解答某个具体问题？ \n- 生成一段文字/代码？ \n- 或者单纯想测试我的反应？ \n\n请随意吩咐～ （如果这是误触发，也欢迎告诉我哦） \n\n**小提示**：你可以试着问我 *“今天天气如何？”* 或 *“用Python写个冒泡排序”* 这类具体指令，我会更高效地响应你！ 🌟'}

环境变量.env配置文件内容：

OPENAI_API_KEY=sk-xxxxx
openai_api_base=https://api.deepseek.com