LangGraph开篇
持续更新中
基于Python 3.13.x
1.概述
LangChain就像一条流水线,对于一步接着一步的任务,用LangChain链式调用可以完美实现,但是很多任务很复杂,并不是一条直线完成的,需要根据各种条件走各种分支,这种复杂任务如果使用LangChain来编程实现,步骤就会很复杂,需要很多的if else判断。
当然解决这个问题还可以通过将任务做成多Agent,让Agent多次推理和决策并执行某个分支,但是这样消耗时间和token都很多,而且这个过程中智能体就是一个黑盒,不可控,不知道得推理多少次,也不知道内部是怎么执行的,过程难以监控和干预,这样一来这种复杂任务LangChain就不适用了,就可以使用LangGraph来实现。
LangGraph是基于LangChain构建的,面向智能体的可实现多轮交互、状态持久化和分支并行执行的图结构工作流框架,是LangChain的高级编排工具,LangGraph中,无论图结构有多复杂,单个任务的执行仍然是线性的,单个任务背后依然是依靠LangChain的链式执行实现的。

LangGraph(图) = 状态(State) + 节点(Node) + 边(Edge)
LangGraph将工作流程抽象为一种的有向图结构,LangGraph的图结构比LangChain的链结构,更加适合这种多分支复杂任务,具体表现为:
LangGraph提供了强大的状态管理控制,允许Agent在不同节点之间传递和维护信息,实现长期记忆和多轮对话能力,通过定义节点和边,可以精确控制Agent执行逻辑,包括条件分支,循环等。
LangGraph能够无缝集成各种外部工具(搜索引擎,数据库,API等),让Agent能够实时获取信息,执行特定操作,拓展了LLM的能力边界。
LangGraph使得Agent的运行路径清晰可见,便于理解Agent的决策过程,出现问题易于定位和调试
LangGraph使得模块具有可复用性,各个节点都是独立的,可复用的组件,维护性高,易于拓展,还能通过子图机制将复杂的工作流拆解为多个可独立开发和测试的模块,提高开发测试效率。
2.快速开始
2.1 构建一个langgraph
安装langgraph
pip install -U langgraph安装grandalf,用于生成可视化图供我们调试测试
pip install grandalf基于langgraph定义一个图,其中:
class DemoState(TypedDict)就是状态(State)
node1(),node2()就是两个节点(Node),节点说白了就是干活的函数
然后节点再add_edge()连接成边(Edge),最后graph.compile()就成为图了
from typing import TypedDict
from langgraph.constants import START
from langgraph.constants import END
from langgraph.graph import StateGraph
# 状态类
class DemoState(TypedDict):
v: str
# 节点
def node1(state: DemoState) -> dict:
return state
# 节点
def node2(state: DemoState) -> dict:
return state
if __name__ == "__main__":
graph = StateGraph(DemoState)
graph.add_node('node1', node1)
graph.add_node('node2', node2)
graph.add_edge(START, 'node1')
graph.add_edge('node1', 'node2')
graph.add_edge('node2', END)
app = graph.compile()
app.get_graph().print_ascii()然后就可以打印看到我们刚刚定义的图了:
+-----------+
| __start__ |
+-----------+
*
*
*
+-------+
| node1 |
+-------+
*
*
*
+-------+
| node2 |
+-------+
*
*
*
+---------+
| __end__ |
+---------+ 普通直线边只能一条线走到某个节点,如果实现真正的分支结构,还需要条件边,而且节点中的代码可以使用大模型还实现某些功能,这样langgraph的能力才真正发挥出来。
2.2 和langchain结合
和langchain整合,将问题传入State,经过节点时由langchain处理,通过add_messages的规约方式保存每次对话信息,最终返回大模型回答
from typing import Annotated, TypedDict, List
from langgraph.constants import START, END
from langgraph.graph import add_messages, StateGraph
import os
from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
load_dotenv(encoding='utf-8')
class MsgState(TypedDict):
messages: Annotated[List, add_messages]
llm = init_chat_model(
model = 'deepseek-chat',
model_provider = 'openai',
api_key = os.getenv('DEEPSEEK_API_KEY'),
base_url = 'https://api.deepseek.com'
)
def chat(state: MsgState):
reply = llm.invoke(state['messages'])
return {'messages': [reply]}
if __name__ == '__main__':
graph = StateGraph(MsgState)
graph.add_node('chat', chat)
graph.add_edge(START, 'chat')
graph.add_edge('chat', END)
app = graph.compile()
res = app.invoke({'messages': '1加1等于几'})
print(res['messages'][-1].content)1加1等于2。这是一个基础的加法运算,在数学中,1和1相加的结果是2。不过,这个问题在某些特定情境下(比如二进制或逻辑运算中)可能有不同解释,但日常数学中就是2。如果你有其他背景或疑问,可以进一步说明! 😊"如果文章对您有帮助,可以请作者喝杯咖啡吗?"
微信支付
支付宝