LangGraph 的出现
在 LangChain 框架中,智能体(Agent)从数据结构的角度来看等同于一个有向无环图。这意味着传统的 Chain 在推理过程中无法被循环调用。尽管 AgentExecutor(代理执行器)支持循环机制,但它缺乏精确的控制能力,时常导致失控并陷入死循环。
使用代理执行器实现循环调用大语言模型(LLM)的能力,其调用过程主要有两步:
- 通过大模型来决定采取什么行动、使用什么工具,或对用户输出响应。
- 执行步骤 1 中的行动,并将结果继续交给大模型进行决策。
AgentExecutor 存在的问题是决策过程隐藏在背后,过于黑盒,缺乏更精细的控制能力,这在构建复杂的 Agent 时受到限制:
- 难以控制工具的使用顺序。
- 在执行过程中添加人机交互较为困难。
- 灵活更换 Prompt 或背后的 LLM 不够便捷。
在 LangChain 中,简单的链不具备循环能力,而 AgentExecutor 调用 Agent 又过于黑盒,因此需要一个具备更精细控制能力的框架来支持复杂场景的 LLM 应用。LangGraph 的出现标志着 LangChain 进入多智能体框架领域。LangGraph 基于图论运作,提供了一种状态机技术,可驱动循环代理调用,实现有向有环图。因此,LangGraph 有三个关键元素:
StateGraph:状态图
这是 LangChain 的一个类,表示图的数据结构并反映其状态。节点会更新图的状态。
Node:节点
图中关键元素之一。每个 LangGraph 节点都有一个名称值,可以是 LangChain 表达式中的函数或者可运行项(Runnable)。每个节点接收一个字典类型的数据,返回具有相同结构的更新状态。有一个名为 END 的特殊节点,用于识别状态机的结束状态。
Edge:边缘
边维系节点之间的关系,分为三种类型:开始边(没有上游节点)、普通边和条件边。
- 普通边:定义上游节点应始终调用的下游节点。
- 条件边:通过函数(路由器)来确定下游节点。条件边需要三个元素:
- 上游节点:边的起点,表示转换的起点。
- 路由函数:此函数根据返回值有条件地确定应进行转换的下游节点。
- 状态映射:根据路由函数的返回值,指定下游节点。它将路由函数可能的返回值与相应的下游节点相关联。
构建 LangGraph 实例
LangGraph 的状态
类似于状态机(State Machine),由一组状态(State)和状态之间的转换(Transition)组成,用于表示系统在不同状态之间的转换和响应事件的行为。
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph
from langgraph.graph.message import add_messages
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]
graph_builder = StateGraph(State)
这个状态定义了随时间更新的核心状态对象,它接收一些操作以及属性定义,会被节点更新;会在每一个 Node 之间传递不同的状态信息。然后每一个节点会根据自己定义的逻辑去更新这个状态信息。
节点信息
Node 可以是一个 LangChain 的 Runnable 或者是一个可执行的函数,也可以是一个 Graph。构建完成的 Graph 也是一个 LangChain 的 Runnable,这也正是 LangGraph 作为 LangChain 的扩展可以与 LangChain 完美衔接的关键。
添加节点直接使用 Graph 实例中的 add_node 方法添加即可,当然这个节点应当有一个名字:
graph_builder.add_node("node", node)
这里的 node 就是 LangChain 的 Runnable 对象或者可执行函数,具体在开发中定义。
绘制图的边
边(Edge)描述的是节点与节点之间的关系,可以是普通的或者是有条件的。它们都有方向,Edge 描述的上游节点与下游节点的关系(开始边除外)。
Edge 的实现由 Graph 实例中的 add_edge 方法添加,同样这个边也有一个名字,这个名字就是节点的名字,代表的是上游节点:
graph_builder.add_edge("node", next_node)
next_node 就是 node 节点的下游节点。
条件边的实现由 Graph 实例中的 add_conditional_edge 进行添加:
graph.add_conditional_edge(
"node",
should_continue,
{
"end": END,
"continue": "next_node"
}
)
should_continue 就是条件边三个组成元素的路由函数了,用于确定下一个的可调用对象是一个或多个节点。
编译图
到现在就完成一个图所需要的基本条件了,这也是一个最简单的 LangGraph 例子。编译图也是由 Graph 实例对象的方法实现:
graph = graph_builder.compile()
编译之后的 graph 是一个 LangChain 的 Runnable 对象,同样具有 .invoke() 和 .stream() 方法,也具备成为一个节点的能力。
由 LangGraph 构建的简单聊天机器人
以下是一个完整的示例,展示了如何配置环境变量、定义状态、构建节点以及运行循环对话:
import os
from dotenv import find_dotenv, load_dotenv
load_dotenv(find_dotenv())
# 确保已设置 OPENAI_API_BASE 和 OPENAI_API_KEY
OPENAI_API_BASE = os.environ.get('OPENAI_API_BASE')
OPENAI_API_KEY = os.environ.get('OPENAI_API_KEY')
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
class State(TypedDict):
messages: Annotated[list, add_messages]
# 初始化 LLM
llm = ChatOpenAI(model="gpt-3.5-turbo")
def chatbot(state: State):
# 调用 LLM 生成回复
response = llm.invoke(state["messages"])
return {"messages": [response]}
# 构建图
graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
# 设置入口点和结束点
graph_builder.set_entry_point("chatbot")
graph_builder.set_finish_point("chatbot")
# 编译图
graph = graph_builder.compile()
# 运行对话循环
while True:
user_input = input("User: ")
if user_input.lower() in ["quit", "exit", "q"]:
print("Goodbye!")
break
# 流式输出处理
for event in graph.stream({"messages": [("user", user_input)]}):
for value in event.values():
print("Assistant:", value["messages"][-1].content)
LangGraph 的应用与优势
想要让单个 Agent 干很多活儿,就必须有极强的推理能力以及定义好工具之间的关系,和精确到控制能力。而单个 Agent 的执行过程过于黑盒等问题导致单个 Agent 很难集多种本领于一身,同时 LangChain 本身没有一个能直接实现多 Agent 协作的方法或技巧。LangGraph 作为 LangChain 的扩展就可以实现多 Agent(Multi-agent)。
当然,单个 Agent 有单个的好处,分开也有分开的优势:
- 单 Agent 的构建:较为简单,不仅编码简单,逻辑也比较清晰。LangGraph 与之相比就太过于复杂了,不仅要有清晰的逻辑关系,还要有一个图的数据流向。构建一个简单的图人脑还是能应对,一旦节点、边多起来了,就需要借助工具进行设计了,在复杂的逻辑链路中抽丝剥茧。
- 多 Agent 的优势:LangGraph 构建多 Agent 在编码的过程中或许会比较难受,但是实现效果不是单个 Agent 能够相比的。它允许开发者显式地定义 Agent 之间的通信路径、状态共享机制以及决策流程。
AutoGen 中的多 Agent 是基于会话实现的,有三个 Agent 角色,构建过程要比 LangGraph 简单。与 LangGraph 不同,LangGraph 具有更清晰的逻辑条理,更适合需要严格状态管理和流程控制的复杂业务场景。
当前的 LangGraph 还处在一个初期发展的阶段,在 LangChain 的 0.2.x 版本迭代中,Agent 应该会占据一个重要地位。相信在未来的 LangGraph 构建会更容易,文档和社区支持也会更加完善。
最佳实践与注意事项
在使用 LangGraph 构建复杂应用时,建议遵循以下最佳实践:
- 状态设计:State 的设计应尽量精简,只包含必要的上下文信息。避免在 State 中存储过大的数据,以免增加内存开销和序列化延迟。
- 错误处理:在节点函数内部捕获异常,并决定是将错误传递给下一个节点还是终止图。可以通过自定义异常类来区分不同类型的错误。
- 调试模式:利用 LangGraph 提供的可视化功能(如
graph.get_graph().draw_mermaid_png())来检查图的结构,确保节点连接符合预期。 - 并发控制:如果涉及多个并行节点,注意状态更新的原子性,避免竞态条件导致状态不一致。
通过合理运用 LangGraph 的状态机和图结构,开发者可以构建出既灵活又可控的智能体系统,为复杂的 AI 应用场景提供坚实的基础。


