LangGraph是构建有状态Agent的底层操作系统,提供持久化执行、人机协同、全方位记忆等五大能力。其核心是状态在节点间的流转与演化,支持传统DAG不具备的循环结构,适合实现Agent的"思考-行动-观察"循环。支持Graph API和Functional API两种开发模式,通过StateGraph、Nodes和Edges三个核心要素构建Agent。对于需要稳定、长时间运行甚至人工介入的智能体工作流,LangGraph是Python生态中的优选方案。
01 前言
如果说 LangChain 是高层次使用的组件库,那么 LangGraph 就是那个底层的操作系统,专门为构建、管理和部署**长时运行(long-running)且有状态(stateful)**的 Agent 而生。
LangGraph 为了解决生产级 Agent 的痛点,提供了五大能力:
1、持久化执行:智能体可以在失败后从断点恢复,支持长达数天甚至数周的任务运行。
2、人机协同:支持在智能体决策的关键节点进行人工拦截、检查或修改状态。
3、全方位记忆:创建具有状态的智能体,既具备用于持续推理的短期工作记忆,又拥有跨会话的长期记忆。
4、借助LangSmith调试:利用可视化工具深入了解复杂的智能体行为和运行时指标。
5、生产级部署:设计可扩展基础设施,自信地部署有状态、可长期运行的智能体系统。
02 架构解析
LangGraph 的设计深受 Google Pregel 图计算模型和Apache Beam数据流抽象的影响。它的核心运转逻辑并不复杂,本质上是状态(State)在节点(Nodes)之间的流转与演化。
区别于传统的 DAG(有向无环图)工作流,LangGraph 原生支持循环(Cycles),这对于实现 Agent 的“思考-行动-观察”循环至关重要。
LangGraph如何通过共享状态将各个组件串联起来,架构全景图如下:
一个关键概念:State 是不可变的概念,通过 Reducer 演进。在代码层面,我们定义一个TypedDict,当节点返回数据时,LangGraph 并不是直接覆盖原有的 State,而是根据你定义的Reducer(通常是operator.add)将新消息追加到历史记录中。
03 开发范式
LangGraph 支持两套成熟的构建模式,以满足不同开发者的“需求”:
1、Graph API(经典模式):
特点:显式定义 StateGraph,手动执行 add_node 和 add_edge。
适用场景:适合需要极度精细控制路由、或者需要将复杂逻辑可视化的场景。它是这一系列文章的基础,因为最能体现“图”的本质。
2、Functional API(Pythonic 模式):
特点:这是 2025 年推出的特性。通过 @entrypoint 和 @task 装饰器,你完全可以用标准的 if/else 和 while 循环来写逻辑。
适用场景:对软件工程师更友好,代码可读性更高,像写普通 Python 函数一样构建 Agent,底层自动转化为图结构。
建议:先掌握 Graph API,因为它能让你直观地理解 Checkpoint 和 State 的流转原理。
04 核心机制
要构建一个最小可用的图,我们需要关注三个要素:
1、StateGraph:图的骨架。
2、Nodes:具体的业务逻辑函数。
3、Edges:控制流转的规则(特别是条件边)。
一个示例,标准的 ReAct 模式(Reasoning + Acting)的最小实现代码:
# 环境准备: # pip install langgraph langchain-deepseek langchain-core import json import os from typing import Annotated, TypedDict, Literal from langchain_core.messages import BaseMessage, ToolMessage from langchain_core.tools import tool from langchain_deepseek import ChatDeepSeek from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages # --- 1. 定义状态 (State) --- classAgentState(TypedDict): # 使用 add_messages 确保消息是追加而非覆盖 messages: Annotated[list[BaseMessage], add_messages] # --- 2. 定义工具 (Tools) --- @tool defget_weather(city: str) -> str: """查询指定城市的实时天气信息。""" # 模拟真实 API 调用 print(f" 🛠️ [工具调用] 正在查询 {city} 的天气...") if"杭州"in city: return json.dumps({"city": "杭州", "temp": "15℃", "condition": "晴朗"}) elif"上海"in city: return json.dumps({"city": "上海", "temp": "22℃", "condition": "多云"}) return json.dumps({"city": city, "temp": "未知", "condition": "未知"}) tools = [get_weather] # --- 3. 初始化模型并绑定工具 --- # DeepSeek 模型原生支持 Function Calling os.environ["DEEPSEEK_API_KEY"] = "sk-..." llm = ChatDeepSeek( model="deepseek-chat", # 或 deepseek-reasoner (R1) temperature=0 ) # 关键步骤:让模型知道有哪些工具可用 llm_with_tools = llm.bind_tools(tools) # --- 4. 定义节点 (Nodes) --- defreason_node(state: AgentState): """思考节点:LLM 接收历史消息,决定是回复用户还是调用工具""" return {"messages": [llm_with_tools.invoke(state["messages"])]} deftool_node(state: AgentState): """工具节点:解析 LLM 的工具调用请求,执行函数并返回结果""" last_message = state["messages"][-1] # 构造工具执行结果的消息列表 tool_outputs = [] for tool_call in last_message.tool_calls: # 寻找对应的工具函数 selected_tool = {t.name: t for t in tools}[tool_call["name"]] # 执行工具 try: output = selected_tool.invoke(tool_call["args"]) except Exception as e: output = f"Error: {str(e)}" # 构造 ToolMessage 返回给 LLM tool_outputs.append( ToolMessage( content=str(output), tool_call_id=tool_call["id"] ) ) return {"messages": tool_outputs} # --- 5. 定义条件边 (Conditional Edge) --- defshould_continue(state: AgentState) -> Literal["tools", "end"]: """路由逻辑:检查最后一条消息是否包含工具调用""" last_message = state["messages"][-1] if last_message.tool_calls: return"tools" return"end" # --- 6. 组装图 (Graph Construction) --- workflow = StateGraph(AgentState) # 添加节点 workflow.add_node("agent", reason_node) workflow.add_node("tools", tool_node) # 设置入口 workflow.set_entry_point("agent") # 添加边 workflow.add_conditional_edges( "agent", should_continue, { "tools": "tools", # 也就是如果 should_continue 返回 "tools",则去 "tools" 节点 "end": END # 返回 "end",则结束 } ) workflow.add_edge("tools", "agent") # 工具执行完,务必回传给 Agent 进行最终总结 # 编译应用 app = workflow.compile() # Show the agent png_data = app.get_graph().draw_mermaid_png() # 保存为文件 withopen("react_demo2.png", "wb") as f: f.write(png_data) # --- 7. 运行测试 (Run) --- if __name__ == "__main__": from langchain_core.messages import HumanMessage print("Agent 启动...") # 测试问题:包含需要调用工具的意图 inputs = {"messages": [HumanMessage(content="杭州今天天气怎么样?适合穿什么?")]} # stream_mode="values" 可以看到状态流转的每一步 for event in app.stream(inputs, stream_mode="values"): last_msg = event["messages"][-1] sender = "用户"ifisinstance(last_msg, HumanMessage) else"Agent" ifhasattr(last_msg, "tool_calls") and last_msg.tool_calls: sender = "Agent (决定调用工具)" elifisinstance(last_msg, ToolMessage): sender = "系统 (工具返回)" print(f"[{sender}]: {last_msg.content}") print("-" * 30)示例首先定义一个AgentState用于记录状态;然后定义一个查询天气的工具,模拟天气的查询;再定义llm对象并通过bind_tools绑定工具,这一步很重要,告诉大模型有哪些可用的工具;再通过StateGraph定义节点和边,组装成图;最后运行测试。
示例输出如下所示:
Agent 启动... [用户]: 杭州今天天气怎么样?适合穿什么? ------------------------------ [Agent (决定调用工具)]: 我来帮您查询杭州今天的天气情况。 ------------------------------ 🛠️ [工具调用] 正在查询 杭州 的天气... [系统 (工具返回)]: {"city": "\u676d\u5dde", "temp": "15\u2103", "condition": "\u6674\u6717"} ------------------------------ [Agent]: 根据查询结果,杭州今天的天气情况如下: - **温度**:15°C - **天气状况**:晴朗 **穿衣建议**: 15°C的晴朗天气比较舒适,建议穿着: 1. **上衣**:长袖T恤或薄款卫衣,可以搭配一件薄外套(如牛仔外套、风衣或针织开衫) 2. **下装**:长裤(牛仔裤、休闲裤等) 3. **鞋子**:运动鞋或休闲鞋 4. **其他**:早晚温差可能较大,建议带一件薄外套备用 这样的天气非常适合户外活动,但要注意防晒,因为晴朗天气紫外线可能较强。总体来说,今天的杭州天气很不错,适合外出!05 前言
LangGraph的出现,标志着 LLM 应用开发从“Prompt Engineering”进阶到了“Flow Engineering”。
- StateGraph给了我们结构化的数据流视图。
- Cyclic Graph解决了复杂推理的循环依赖。
- Persistence让 Agent 具备了生产级的容错与交互能力。
如果你只是做一个简单的聊天机器人,LangChain 的 Runnable 体系足矣;但如果你在构建一个需要稳定、长时间运行、甚至需要人工介入审批的智能体工作流,LangGraph 是目前 Python 生态中一个不错的选择。
AI大模型从0到精通全套学习大礼包
我在一线互联网企业工作十余年里,指导过不少同行后辈。帮助很多人得到了学习和成长。
只要你是真心想学AI大模型,我这份资料就可以无偿共享给你学习。大模型行业确实也需要更多的有志之士加入进来,我也真心希望帮助大家学好这门技术,如果日后有什么学习上的问题,欢迎找我交流,有技术上面的问题,我是很愿意去帮助大家的!
如果你也想通过学大模型技术去帮助就业和转行,可以扫描下方链接👇👇
大模型重磅福利:入门进阶全套104G学习资源包免费分享!
01.从入门到精通的全套视频教程
包含提示词工程、RAG、Agent等技术点
02.AI大模型学习路线图(还有视频解说)
全过程AI大模型学习路线
03.学习电子书籍和技术文档
市面上的大模型书籍确实太多了,这些是我精选出来的
04.大模型面试题目详解
05.这些资料真的有用吗?
这份资料由我和鲁为民博士共同整理,鲁为民博士先后获得了北京清华大学学士和美国加州理工学院博士学位,在包括IEEE Transactions等学术期刊和诸多国际会议上发表了超过50篇学术论文、取得了多项美国和中国发明专利,同时还斩获了吴文俊人工智能科学技术奖。目前我正在和鲁博士共同进行人工智能的研究。
所有的视频由智泊AI老师录制,且资料与智泊AI共享,相互补充。这份学习大礼包应该算是现在最全面的大模型学习资料了。
资料内容涵盖了从入门到进阶的各类视频教程和实战项目,无论你是小白还是有些技术基础的,这份资料都绝对能帮助你提升薪资待遇,转行大模型岗位。
智泊AI始终秉持着“让每个人平等享受到优质教育资源”的育人理念,通过动态追踪大模型开发、数据标注伦理等前沿技术趋势,构建起"前沿课程+智能实训+精准就业"的高效培养体系。
课堂上不光教理论,还带着学员做了十多个真实项目。学员要亲自上手搞数据清洗、模型调优这些硬核操作,把课本知识变成真本事!
如果说你是以下人群中的其中一类,都可以来智泊AI学习人工智能,找到高薪工作,一次小小的“投资”换来的是终身受益!
应届毕业生:无工作经验但想要系统学习AI大模型技术,期待通过实战项目掌握核心技术。
零基础转型:非技术背景但关注AI应用场景,计划通过低代码工具实现“AI+行业”跨界。
业务赋能 突破瓶颈:传统开发者(Java/前端等)学习Transformer架构与LangChain框架,向AI全栈工程师转型。
👉获取方式:
😝有需要的小伙伴,可以保存图片到wx扫描二v码免费领取【保证100%免费】🆓