AI大模型之旅——LangGraph基础组件
AI大模型之LangGraph（1.0.8）：基础组件
一、基础概念
1.1 什么是LangGraph？
LangGraph是一个基于图结构的 Agent 编排框架，专注于描述和控制 Agent 的执行流程。它可以独立使用，也可以与LangChain 结合，用于更方便地调用模型、工具和记忆组件。
通过LangGraph可以轻松构建出可靠的Agent应用。LangGraph并不依赖于LangChain，但是配合LangChain使用可以发挥更好的效果
1.2 LangGraph的基础组件
LangGraph总体就只有四个大的组件：
state：整个LangGraph的内存数据共享容器，这是一个实体类，实体类中存放我们的所有信息，这个信息能被不同的 Agent 和node 获取和增删改node：工作流的节点，每一个节点都代表了一种处理数据的规则，例如：报告生成节点，数据处理节点，爬虫节点等等edge：边，通过边来连接节点，使得不同的节点能够相互流转，数据通过上方的 state 来共享获取graph：图，整个LangGraph的工作流，就是通过上方的 state+node+edge = graph
1.3 LangGraph 核心能力
LangGraph具有以下核心能力：
Durable execution：持久执行，能够在执行失败后恢复继续执行，而无需从头开始，类似文件传输的断点续传机制，非常适合长时间运行、复杂的多步骤任务。
Human-in-the-loop：人在环路，在任务执行过程中允许人工介入，用于确认是否执行关键操作。
Comprehensive Memory：提供统一的状态与记忆管理能力，通过 State 和 Checkpoint机制支持短期上下文保留，并可与 LangChain 的 Memory或外部存储系统结合，实现长期记忆能力。短期记忆可以让AI记住最近几轮对话的信息，长期记忆可以让AI了解历史对话聊天的总结信息。
Debugging with LangSmith：可视化调试，可以无缝集成LangSmith进行可视化调试和监控
Production-ready Deployment：在配合持久化、错误恢复和监控体系的前提下，LangGraph 非常适合用于生产级
Agent 应用的构建。
1.4 安装依赖
pip install -U langgraph
1.5 LangGraph构建Agent基本步骤
二、实用案例
把大象装进冰箱需要：1.打开冰箱 2.把大象装进去 3.关上冰箱门
from IPython.display import Image, display
from langgraph.constants import START, END
from langgraph.graph import StateGraph
from langgraph.graph.state import CompiledStateGraph
from typing_extensions import TypedDict
# step 1.定义状态
class ElephantInFridgeState(TypedDict):
"""
定义状态
"""
# 冰箱门是否打开
fridge_open: bool
# 大象是否在冰箱里
elephant_inside: bool
# step 2.定义每个步骤处理节点
def open_fridge(state: ElephantInFridgeState) -> ElephantInFridgeState:
"""
打开冰箱门
"""
print("正在打开冰箱门……")
print("冰箱门已打开！")
# 更新状态
state["fridge_open"] = True
# 返回更新后的状态
return state
def put_elephant(state: ElephantInFridgeState) -> ElephantInFridgeState:
"""
把大象装进冰箱
"""
if not state["fridge_open"]:
raise Exception("冰箱门未打开！无法放进大象！")
print("正在把大象装进冰箱……")
print("大象已装进冰箱！")
# 获取状态
state["elephant_inside"] = True
# 返回更新后的状态
return state
def close_fridge(state: ElephantInFridgeState) -> ElephantInFridgeState:
"""
关闭冰箱门
"""
if not state["elephant_inside"]:
raise Exception("警告，冰箱内没有大象！请勿关闭冰箱门！")
print("正在关闭冰箱门……")
print("冰箱门已关闭！")
# 获取状态
state["fridge_open"] = False
# 返回更新后的状态
return state
# step 3.定义流程图
def build_graph() -> CompiledStateGraph:
"""构建把大象装进冰箱的流程图"""
# 创建图对象StateGraph
graph = StateGraph(ElephantInFridgeState)
# 添加节点
graph.add_node("open_fridge", open_fridge)
graph.add_node("put_elephant", put_elephant)
graph.add_node("close_fridge", close_fridge)
# 添加起始边
graph.add_edge(START, "open_fridge")
# 添加结束边，将节点连接起来形成工作流
graph.add_edge("open_fridge", "put_elephant")
graph.add_edge("put_elephant", "close_fridge")
# 添加结束边
graph.add_edge("close_fridge", END)
# 返回编译好的图
return graph.compile()
# step 4.运行流程图
def run_workflow():
"""运行流程图"""
# 构建流程图
graph = build_graph()
# 定义初始状态
initial_state = ElephantInFridgeState(fridge_open=False, elephant_inside=False)
# 运行流程图
result = graph.invoke(initial_state)
print("\n工作流执行完毕,最终状态:")
print(f"冰箱状态: {'打开' if result['fridge_open'] else '关闭'}")
print(f"大象状态: {'在' if result['elephant_inside'] else '不在'}冰箱里")
# 可视化工作流图
display(Image(graph.get_graph().draw_mermaid_png()))
return result
# 执行入口
run_workflow()
打印结果如下：
-
正在打开冰箱门……
-
冰箱门已打开！
-
正在把大象装进冰箱……
-
大象已装进冰箱！
-
正在关闭冰箱门……
-
冰箱门已关闭！
-
工作流执行完毕,最终状态:
-
冰箱状态: 关闭
-
大象状态: 在冰箱里
三、分析流程
3.1 定义状态（ElephantInFridgeState）
# step 1.定义状态
class ElephantInFridgeState(TypedDict):
"""
定义状态
"""
# 冰箱门是否打开
fridge_open: bool
# 大象是否在冰箱里
elephant_inside: bool
3.1.1 核心作用
作为工作流的共享内存空间，跨节点传递和持久化关键状态
3.1.2 状态属性
fridge_open： 实时记录冰箱门的物理状态
elephant_inside: 实时追踪大象的空间位置
3.1.3 设计特点
- 继承自LangGraph的TypedDict，强制类型安全
- 所有节点均可读写，实现跨节点数据共享
- 初始状态: 冰箱门关闭、大象是在冰箱外
3.2 节点功能
每个节点均是独立的状态处理器，遵循“获取状态—>执行业务—>更新状态—>返回状态”的工作流范式
3.2.1 开门节点
def open_fridge(state: ElephantInFridgeState) -> ElephantInFridgeState:
"""
打开冰箱门
"""
print("正在打开冰箱门……")
print("冰箱门已打开！")
# 更新状态
state["fridge_open"] = True
# 返回更新后的状态
return state
- 前置条件： 无（可直接从初始状态启动）
- 核心操作： 变更fridge_open的状态为True
- 执行效果： 输出操作日志确保冰箱门开启
3.2.2 放入大象节点
def put_elephant(state: ElephantInFridgeState) -> ElephantInFridgeState:
"""
把大象装进冰箱
"""
if not state["fridge_open"]:
raise Exception("冰箱门未打开！无法放进大象！")
print("正在把大象装进冰箱……")
print("大象已装进冰箱！")
# 获取状态
state["elephant_inside"] = True
# 返回更新后的状态
return state
- 前置条件： 强制检查fridge_open状态为True否则抛出异常
- 核心操作： 变更elephant_inside的状态为True
- 关键机制： 通过状态依赖确保工作流顺序逻辑
3.2.3 关门节点
def close_fridge(state: ElephantInFridgeState) -> ElephantInFridgeState:
"""
关闭冰箱门
"""
if not state["elephant_inside"]:
raise Exception("警告，冰箱内没有大象！请勿关闭冰箱门！")
print("正在关闭冰箱门……")
print("冰箱门已关闭！")
# 获取状态
state["fridge_open"] = False
# 返回更新后的状态
return state
- 安全机制：当elephant_inside=False时，输出警告
- 核心操作： 变更fridge_open的状态为False
- 容错设计： 允许在无大象时关门，但要给出提示
3.3工作流图构建
# 创建图对象StateGraph
graph = StateGraph(ElephantInFridgeState)
# 添加节点
graph.add_node("open_fridge", open_fridge)
graph.add_node("put_elephant", put_elephant)
graph.add_node("close_fridge", close_fridge)
# 添加起始边
graph.add_edge(START, "open_fridge")
# 添加结束边，将节点连接起来形成工作流
graph.add_edge("open_fridge", "put_elephant")
graph.add_edge("put_elephant", "close_fridge")
# 添加结束边
graph.add_edge("close_fridge", END)
3.3.1 图结构
创建有向无环图控制流
3.3.2 节点关系
- START:LangGraph内置起始节点
- 线性执行链: 开门 —>放入大象—>关门
- END: LangGraph 内置终止节点
3.3.3 编译输出
返回可执行的CompliedStateGraph
# 返回编译好的图
return graph.compile()
3.3.4 工作流执行
# step 4.运行流程图
def run_workflow():
"""运行流程图"""
# 构建流程图
graph = build_graph()
# 定义初始状态
initial_state = ElephantInFridgeState(fridge_open=False, elephant_inside=False)
# 运行流程图
result = graph.invoke(initial_state)
print("\n工作流执行完毕,最终状态:")
print(f"冰箱状态: {'打开' if result['fridge_open'] else '关闭'}")
print(f"大象状态: {'在' if result['elephant_inside'] else '不在'}冰箱里")
# 可视化工作流图
display(Image(graph.get_graph().draw_mermaid_png()))
return result
3.3.5 执行流程
- 编译工作流图
- 初始化状态：fridge_open=False, elephant_inside=False
- 注入初始状态并触发执行
3.3.6 输出内容
- 实时打印各节点操作日志
- 终态数据报告
- Mermaid格式可视化流程图
四、LangGraph组件详解
4.1 LangGraph的StateGraph
StateGraph是Langchain中的一个类，用于表示图的数据结构，并反映系统的当前状态。他类似于状态机，通过节点对状态的更新来驱动流程
# 4.定义State
class State(TypedDict):
# 图运行的消息列表
input: str
all_actions: Annotated[list[str], operator.add]
分析：
input: str:Graph 中保存 用户原始输入all_actions: Annotated[list[str], operator.add]:让 Agent 自动记录自己的全部行为历史。Annotated:给字段附加 合并规则operator.add:自动累加
4.2 LangGraph的节点
节点是图中核心执行单元，每个节点具有唯一名字，且可以是一个函数、可运行的对象或Langchain表达式。
节点通常是python函数，第一个位置参数是state，第二位置参数(可选)是config
def chatbot(state: MessagesState):
# 大模型绑定工具
llm_with_tools = llm.bind_tools(tools)
response = llm_with_tools.invoke(state["messages"])
return {"messages": [response]}
START Node：是一个开始节点，表示将用户输入发送到图的节点，主要作用是确定哪些节点应首先被调用END Node：表示结束节点，表示某些边在完成后没有进一步操作时，可以引用此节点
4.3 LangGraph的边
边定义了逻辑，包括如何路由以及图形如何停止。
三要素：
上游节点：边的起点
路由函数： 根据状态值返回条件结果
状态映射：将路由函数的返回值与具体下游节点绑定
类型：
1.Normal Edges（普通边）： 直接从一个节点转到下一个节点
graph.add_edge("node_a","node_b")
2.Conditional Edges(条件边)：调用一个函数来确定接下来要转到的节点
graph_builder.add_conditional_edges(
"chatbot",
should_continue,
{"tools": "tools", END: END}
)
# 5.判断是否继续（是否有工具调用）
def should_continue(state: MessagesState) -> Literal["tools", END]:
last_msg = state["messages"][-1]
# 只对AI生成的消息检测工具调用
if isinstance(last_msg, AIMessage) and hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
print("有工具调用",last_msg.tool_calls)
return "tools"
return END
3.Entry Point(入口点):当用户输入到达时首先调用的节点
# 设置起始点
graph_builder.set_entry_point("chatbot")
# 或者
graph_builder.add_edge(START,"node_a")
4.Conditional Entry Point(条件入口): 调用一个函数来确定当用户输入到达时首先调用的节点
Conditional Entry Point 允许根据自定义逻辑在不同节点之间开始，可以使用add_conditional_edges从虚拟的START节点来实现graph_builder.add_conditional_edges(START,routing_function）
4.4 看一个LangGraph人机对话案例
from langchain_community.utilities import GoogleSerperAPIWrapper
from langchain_community.tools import GoogleSerperRun
from langchain_core.messages import HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
import dotenv
from langgraph.graph import StateGraph, END, MessagesState
from langgraph.prebuilt import ToolNode
from typing import Literal
from langgraph.checkpoint.memory import MemorySaver
# 1.加载环境变量
dotenv.load_dotenv()
# 2.创建大模型
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.8)
# 3.创建谷歌搜索工具
google_serper_tool = GoogleSerperRun(
name="google_serper_tool",
description="谷歌搜索工具",
api_wrapper=GoogleSerperAPIWrapper(),
)
# 4. 工具节点
tools = [google_serper_tool]
tool_node = ToolNode(tools=tools)
# 5.判断是否继续（是否有工具调用）
def should_continue(state: MessagesState) -> Literal["tools", END]:
last_msg = state["messages"][-1]
# 只对AI生成的消息检测工具调用
if isinstance(last_msg, AIMessage) and hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
print("有工具调用",last_msg.tool_calls)
return "tools"
return END
# 6.Chatbot 节点
def chatbot(state: MessagesState):
# 大模型绑定工具
llm_with_tools = llm.bind_tools(tools)
response = llm_with_tools.invoke(state["messages"])
return {"messages": [response]}
# 7.构建图
graph_builder = StateGraph(MessagesState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", tool_node)
# 设置起始点
graph_builder.set_entry_point("chatbot")
"""
9.以下代码含义:
1. 执行 chatbot(state)
2. 得到新的 state
3. 调用 should_continue(state)
4. 拿到返回值（比如 "tools"）
5. 查映射表：
- "tools" → tools 节点
- END → 结束
6. 跳转执行
"""
# 8.正确设置条件边
graph_builder.add_conditional_edges(
"chatbot",
should_continue,
{"tools": "tools", END: END}
)
# 9.添加执行流程
graph_builder.add_edge("tools", "chatbot")
# 10.内存（关键）初始化内存以在图运行之间持久化状态
checkpointer = MemorySaver()
# 11.编译图，编译成Langchain可运行对象，可以像运行普通Langchain程序一样使用它，在编译时传递内存
graph = graph_builder.compile(checkpointer=checkpointer)
# =========================
# 12.人机对话 REPL
# =========================
THREAD_ID = "chat-001"
print("🤖 对话开始，输入 exit / quit 退出\n")
while True:
user_input = input("User: ").strip()
if user_input.lower() in {"exit", "quit", "q"}:
print("👋 对话结束")
break
final_state = graph.invoke(
{"messages": [HumanMessage(content=user_input)]},
config={"configurable": {"thread_id": THREAD_ID}}
)
last_msg = final_state["messages"][-1]
print("Assistant:", last_msg.content)
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
19
0- 0
已为社区贡献4条内容
所有评论(0)