LangGraph：基于图结构的大模型智能体开发框架

LangGraph 是LangChainAI开发的一个工具库，用于创建代理和多代理智能体工作流。它提供了以下核心优势：周期、可控性和持久性，对于Agent智能体开发者来说无疑减少了许多工作量。以下篇幅仅从本人角度阐述LangGraph在开发过程中的亮点以及使用方法。

基本介绍

LangGraph的StateGraph是一种状态机，包含了节点和边，节点一般是定义好的函数，边用于连接不同的节点，用于表示图的执行顺序。简单来说，使用LangGraph构建工作流的步骤如下：

1. 初始化模型和工具
2. 定义图的状态信息
3. 定义图节点
4. 定义图的入口节点和边关系
5. 编译图
6. 执行图

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI# 初始化模型
llm = ChatOpenAI()# 定义图的状态信息
class State(TypedDict):# Messages have the type "list". The `add_messages` function# in the annotation defines how this state key should be updated# (in this case, it appends messages to the list, rather than overwriting them)messages: Annotated[list, add_messages]# 定义图节点
def chatbot(state: State):return {"messages": [llm.invoke(state["messages"])]}# 定义图的入口和边
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)# 编译图
graph = graph_builder.compile()# 执行图
user_input = '介绍你自己'
for event in graph.stream({"messages": [("user", user_input)]}):for value in event.values():print("Assistant:", value["messages"][-1].content)

安装

通过Pypi源可以安装LangGraph及相关依赖

pip install -U langgraph

特性

1. 支持循环和分支结构

用普通边add_edge和条件边add_conditional_edges可以在图中搭建循环和分支

from typing import Literaldef route_tools(state: State,
):if isinstance(state, list):ai_message = state[-1]elif messages := state.get("messages", []):ai_message = messages[-1]else:raise ValueError(f"No messages found in input state to tool_edge: {state}")if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:return "tools"return END# The `tools_condition` function returns "tools" if the chatbot asks to use a tool, and "END" if
# it is fine directly responding. This conditional routing defines the main agent loop.
graph_builder.add_conditional_edges("chatbot",route_tools,{"tools": "tools", END: END},
)graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")
graph = graph_builder.compile()

2. 节点状态持久化

对于LangGraph来说，最核心的点在于其实现了图状态的持久化，这使得在构建智能体工作流时可以非常方便实现多轮对话而无需用户保留历史对话信息，同时允许用户在任意位置打断工作流的执行，并且支持修改图的状态以及断点执行。

from langgraph.checkpoint.memory import MemorySaver
memory = MemorySaver()
# 只需要在编译图的时候加上checkpointer这一参数即可
graph = graph_builder.compile(checkpointer=memory)

以下使用了前面基本介绍中使用的图结构，用于阐述加入状态持久化前后的区别。
● 无状态持久化

user_input = "你好，我是小王。你能介绍你自己给我认识吗？"
config = {"configurable": {"thread_id": "1"}}events = graph.stream({"messages": [("user", user_input)]}, config, stream_mode="values"
)
for event in events:if "messages" in event:event["messages"][-1].pretty_print()

================ Human Message =============
你好，我是小王。你能介绍你自己给我认识吗？
================= Ai Message ================
你好，小王！我是一个人工智能助手，名叫 ChatGLM。我基于清华大学 KEG 实验室和智谱 AI 公司于 2024 年共同训练的语言模型 GLM开发而成。我的任务是针对用户的问题和要求提供适当的答复和支持。有什么可以帮助您的吗？

user_input = "还记得我的名字吗？"
config = {"configurable": {"thread_id": "1"}}
# The config is the **second positional argument** to stream() or invoke()!
events = graph.stream({"messages": [("user", user_input)]}, config, stream_mode="values"
)
for event in events:if "messages" in event:event["messages"][-1].pretty_print()

================ Human Message ============
还记得我的名字吗？
================ Ai Message ================
很抱歉，作为一个AI，我没有能力记住个人用户的信息，包括他们的名字。每次与我互动时，我都将作为一个全新的对话开始。如果您愿意，可以再次告诉我您的名字。

● 启用了状态持久化后

# 只需要在编译图的时候加上checkpointer这一参数即可
graph = graph_builder.compile(checkpointer=memory
)

================= Human Message ===============
还记得我的名字吗？
=============== = Ai Message ===================
当然记得，你的名字是小王。如果你有任何问题或者需要帮助，随时告诉我。

MemorySaver会把所有状态都放在内存中，这可能存在内存泄漏的风险。针对这种情况，LangGraph还允许通过数据库方式持久化状态。

支持的数据库

mongodb

postgres

redis

sqlite

duckdb

3. 中断与人为介入

interrupt_before和interrupt_after分别允许在节点执行前后设置断点，等待用户检查状态无误后才恢复图的执行。

def classify_condition(state):category = state["category"]if category == "闲聊":return "闲聊"return "工具"def chatbot(state: State):print('闲聊模式')query = state["query"]response = client.chat(query).choices[-1].message.contentprint(response)def tools(state):print('使用工具')graph_builder.add_node("classify", classify)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", tools)graph_builder.add_edge(START, "classify")graph_builder.add_conditional_edges("classify",classify_condition,{"闲聊": "chatbot", "工具": "tools"}
)graph_builder.add_edge("chatbot", END)
graph_builder.add_edge("tools", END)graph = graph_builder.compile(checkpointer=MemorySaver(),interrupt_after=['classify']
)

● 1. 默认情况

user_input = "指环王的豆瓣评分有多少"
config = {"configurable": {"thread_id": "1"}}
# The config is the **second positional argument** to stream() or invoke()!
events = graph.stream({"query": user_input}, config, stream_mode="values"
)
for event in events:print(event)

该问题会被认为需要工具调用才能回答

snapshot = graph.get_state(config)
snapshot.next

('tools',)

继续图的执行，可以看到调用了模拟的函数

events = graph.stream(None, config, stream_mode="values"
)
for event in events:print(event)

使用工具

● 2. 人为介入

user_input = "指环王的豆瓣评分有多少"
config = {"configurable": {"thread_id": "2"}}
# The config is the **second positional argument** to stream() or invoke()!
events = graph.stream({"query": user_input}, config, stream_mode="values"
)
for event in events:print(event)

snapshot = graph.get_state(config)
snapshot.next

('tools',)

从这里开始，修改图的状态，强制改为「闲聊」

graph.update_state(config,{"category": "闲聊"},
)

可以看到下一步为「闲聊」，说明人为介入成功了

snapshot = graph.get_state(config)
snapshot.next

('chatbot',)

「闲聊」模式是调用模型自身能力回答的结果，因此可以看到有以下输出

events = graph.stream(None, config, stream_mode="values"
)
for event in events:print(event)

《指环王》系列电影在豆瓣上的评分如下：

• 《指环王：护戒使者》（The Lord of the Rings: The Fellowship of the Ring）的豆瓣评分为9.1分。
• 《指环王：双塔奇兵》（The Lord of the Rings: The Two Towers）的豆瓣评分为9.2分。
• 《指环王：王者归来》（The Lord of the Rings: The Return of the King）的豆瓣评分为9.3分。

这三部电影在豆瓣上均获得了极高的评分，是豆瓣上评分很高的电影之一。

总结

私以为LangGraph最大亮点就是提供了状态持久化的特性，这使得开发者无需考虑保存智能体工作流的中间状态，从而把精力放在构建图的过程中。并且该特性还使得用户或开发者可以从指定位置介入工作流的执行，对图的状态进行修改或者中断，从而使得工作流更加可控。