LangGraph入门指南：构建大模型应用的核心组件与实战技巧

本文详细介绍了LangGraph框架的核心概念与使用方法。LangGraph通过状态(State)、节点(Nodes)和边(Edges)构建有状态应用程序。文章讲解了StateGraph类的使用、状态定义与reducer函数、节点实现方式以及普通边和条件边的应用，为开发者提供了构建大模型应用的完整技术指南。

Graph API 是一个灵活、强大的工具，用于构建有状态的、基于事件的应用程序。在 LangGraph 中，一切都是图！

图由以下核心组件组成：

状态（State）：图的全局状态，包含所有节点共享的数据。它可以是任何 Python 类型，但通常是 TypedDict、dataclass 或 Pydantic BaseModel。

节点（Nodes）：图中的计算单元(Python 函数)，处理输入状态并返回更新的状态。它们接收当前的 State 作为输入参数，执行一些计算或副作用，并返回更新后的 State。

边（Edges）：定义节点之间的转换规则，控制数据流向。根据当前 State 决定接下来执行哪个 Node 的 Python 函数。它们可以是条件分支或固定转换。

通过组合节点和边，可以构建复杂的、循环的工作流。节点和边仅仅是普通的 Python 函数，可以在函数体内调用 LLM 或仅有普通的 Python 代码。

👀

StateGraph

LangGraph 提供了 StateGraph 类，是构建图的主要接口。StateGraph 允许您定义状态结构、添加节点和边，并编译图以供执行。

要构建图，您首先定义状态，然后添加节点和边，最后进行编译。

class State(BaseModel): input: str results: str builder = StateGraph(State) def node1(state: State): # 业务逻辑 return { "results": "..." } # 添加节点 builder.add_node("node1", node1) # 添加边， START(开始节点，内置变量) -> node1 -> END(结束节点，内置变量) builder.add_edge(START, "node1") builder.add_edge("node1", END) # 编译成图 graph = builder.compile(...)

编译后的图是一个Runnable对象，您可以像使用 Langchain 那样，调用 invoke、stream 等函数调用它。

👀

State 状态

定义图时，您要做的第一件事是定义图的 State。State 由图的 schema 以及指定如何将更新应用到状态的 reducer 函数组成。State 的 schema 将作为图中所有 Nodes 和 Edges 的输入 schema，可以是 TypedDict、dataclass 或 Pydantic 模型。所有 Nodes 将发出对 State 的更新，然后使用指定的 reducer 函数应用这些更新。

from langgraph.graph.message import add_messages class State(TypedDict): messages: Annotated[list[AnyMessage], add_messages] extra_field: int

add_messages 函数就是 reducer 函数

reducer 函数

理解节点更新如何应用于 State，Reducer 是关键。State 中的每个键都有其独立的 reducer 函数。如果未明确指定 reducer 函数，则假定对该键的所有更新都应覆盖它。Reducer 有几种不同类型：

默认类型的 reducer，节点的输出会覆盖全局状态。

from typing_extensions import TypedDict class State(TypedDict): foo: int bar: list[str]

在此示例中，未为任何键指定 reducer 函数。假设图的输入是 {“foo”: 1, “bar”: [“hi”]}，

假设第一个 Node 返回 {“foo”: 2}，这被视为对状态的更新。**Node 不需要返回整个 State 的所有键，而只需返回需要更新的键值即可。**应用此更新后，State 将变为 {“foo”: 2, “bar”: [“hi”]}。

第二个节点返回 {“bar”: [“bye”]}，则 State 将变为 {“foo”: 2, “bar”: [“bye”]}。

消息 reducer，用于合并消息列表。

from typing import Annotated from typing_extensions import TypedDict from langgraph.graph import add_messages class State(TypedDict): foo: int messages: Annotated[list[str], add_messages]

在此示例中，我们使用 Annotated 类型为第二个键 (bar) 指定了一个 reducer 函数 (add_messages)。假设图的输入是 {“foo”: 1, “messages”: [“hi”]}。

第一个 Node 返回 {“foo”: 2}，应用此更新后，State 将变为 {“foo”: 2, “messages”: [“hi”]}。

第二个节点返回 {“messages”: [“bye”]}，则 State 将变为 {“foo”: 2, “messages”: [“hi”, “bye”]}，这里的 messages 键是通过将两个列表相加来更新的。

3.自定义 reducer，编写处理消息的逻辑。

def custom_reducer(left: int, right: int) -> int: return left + right class State(TypedDict): # 使用 add_messages reducer messages: Annotated[List[str], add_messages] # 使用自定义 reducer count: Annotated[int, custom_reducer]

MessagesState

由于在状态中包含消息列表非常常见，因此存在一个预构建的状态 MessagesState。MessagesState 定义了一个单一的 messages 键，它是一个 AnyMessage 对象列表，并使用 add_messages reducer，我们可以继承 MessagesState 来定义状态。

from langgraph.graph import MessagesState class State(MessagesState): documents: list[str] # 额外增加的key

👀

Node 节点

在 LangGraph 中，节点通常是 Python 函数（同步或异步）。

第一个参数是状态

第二个参数是“config”RunnableConfig，包含可选的可配置参数（例如 thread_id）

第三个参数是 “runtime”Runtime，包含 Context、store、stream_writer 等

from typing_extensions import TypedDict from langchain_core.runnables import RunnableConfig from langgraph.graph import StateGraph class State(TypedDict): input: str results: str builder = StateGraph(State) def my_node(state: State, config: RunnableConfig): print("In node: ", config["configurable"]["user_id"]) return {"results": f"Hello, {state['input']}!"} def my_other_node(state: State): return state # 添加节点，并设置节点缓存 builder.add_node("my_node", my_node, cache_policy=CachePolicy(ttl=3)) builder.add_node("other_node", my_other_node) # 不要忘记传递 cache graph = builder.compile(cache=InMemoryCache())

START 节点是一个特殊节点，它代表将用户输入发送到图的节点。引用此节点的主要目的是确定应首先调用哪些节点。

END 节点是一个特殊节点，代表一个终止节点。当您想表示哪些边在完成后没有后续操作时，会引用此节点。

节点缓存，默认缓存 key 通过 pickle 对输入进行哈希生成，可以通过key_func自定义缓存 key 的生成策略。

👀

Edge 边

边定义了逻辑如何路由以及图如何决定停止。这是代理工作方式以及不同节点之间如何通信的重要组成部分。LangGraph 支持以下类型的边：

普通边直接从一个节点连接到另一个节点

builder.add_edge("node_a", "node_b")

条件边根据自定义逻辑决定下一个要执行的节点

# 直接返回节点名称 def routing_function(state: State): if state["count"] > 0: return "node_b" else: return "node_c" builder.add_conditional_edges("node_a", routing_function)

# 不直接返回节点名称 def routing_function(state: State): if state["count"] > 0: return False return True # 这里要做映射 graph.add_conditional_edges("node_a", routing_function, {True: "node_b", False: "node_c"})