本文详细介绍了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: intadd_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"})如何学习AI大模型？
如果你对AI大模型入门感兴趣，那么你需要的话可以点击这里大模型重磅福利：入门进阶全套104G学习资源包免费分享！
这份完整版的大模型 AI 学习和面试资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】
这是一份大模型从零基础到进阶的学习路线大纲全览，小伙伴们记得点个收藏！
第一阶段：从大模型系统设计入手，讲解大模型的主要方法；
第二阶段：在通过大模型提示词工程从Prompts角度入手更好发挥模型的作用；
第三阶段：大模型平台应用开发借助阿里云PAI平台构建电商领域虚拟试衣系统；
第四阶段：大模型知识库应用开发以LangChain框架为例，构建物流行业咨询智能问答系统；
第五阶段：大模型微调开发借助以大健康、新零售、新媒体领域构建适合当前领域大模型；
第六阶段：以SD多模态大模型为主，搭建了文生图小程序案例；
第七阶段：以大模型平台应用与开发为主，通过星火大模型，文心大模型等成熟大模型构建大模型行业应用。
100套AI大模型商业化落地方案
大模型全套视频教程
200本大模型PDF书籍
👉学会后的收获：👈
• 基于大模型全栈工程实现（前端、后端、产品经理、设计、数据分析等），通过这门课可获得不同能力；
• 能够利用大模型解决相关实际项目需求： 大数据时代，越来越多的企业和机构需要处理海量数据，利用大模型技术可以更好地处理这些数据，提高数据分析和决策的准确性。因此，掌握大模型应用开发技能，可以让程序员更好地应对实际项目需求；
• 基于大模型和企业数据AI应用开发，实现大模型理论、掌握GPU算力、硬件、LangChain开发框架和项目实战技能， 学会Fine-tuning垂直训练大模型（数据准备、数据蒸馏、大模型部署）一站式掌握；
• 能够完成时下热门大模型垂直领域模型训练能力，提高程序员的编码能力： 大模型应用开发需要掌握机器学习算法、深度学习框架等技术，这些技术的掌握可以提高程序员的编码能力和分析能力，让程序员更加熟练地编写高质量的代码。
LLM面试题合集
大模型产品经理资源合集
大模型项目实战合集
👉获取方式：
😝有需要的小伙伴，可以保存图片到wx扫描二v码免费领取【保证100%免费】🆓
