【2025最新】LangGraph从入门到精通:手把手构建AI智能体的终极指南|保姆级代码+企业级项目实战
LangGraph是LangChain生态中用于构建复杂AI智能体的框架,支持循环、条件分支、状态持久化等特性,解决了传统Agent开发中“黑盒化”和流程控制不足的问题。其核心是通过**状态图(State Graph)**定义智能体的工作流,允许开发者精细控制每一步的执行逻辑,适合开发多智能体协作、自修复系统等场景。
·
一、LangGraph简介:为什么选择它?
LangGraph是LangChain生态中用于构建复杂AI智能体的框架,支持循环、条件分支、状态持久化等特性,解决了传统Agent开发中“黑盒化”和流程控制不足的问题。其核心是通过**状态图(State Graph)**定义智能体的工作流,允许开发者精细控制每一步的执行逻辑,适合开发多智能体协作、自修复系统等场景。
二、环境准备:安装与基础配置
-
安装依赖:
pip install langgraph langchain-openai
确保已配置OpenAI API Key或本地大模型(如DeepSeek、ChatGLM)。
-
初始化项目:
创建Python文件,导入核心模块:from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI from typing import TypedDict, Annotated
三、新手入门:构建第一个聊天机器人
目标:实现一个根据用户输入调用大模型回复的简单智能体。
# 1. 定义状态(State)
class ChatState(TypedDict):
messages: Annotated[list, add_messages] # 消息列表自动追加
# 2. 初始化模型
llm = ChatOpenAI(model="gpt-4o-mini")
# 3. 定义节点函数
def chatbot_node(state: ChatState):
response = llm.invoke(state["messages"])
return {"messages": [response]}
# 4. 构建图
graph_builder = StateGraph(ChatState)
graph_builder.add_node("chatbot", chatbot_node)
graph_builder.add_edge(START, "chatbot") # 从入口节点开始
graph_builder.add_edge("chatbot", END) # 执行后结束
graph = graph_builder.compile()
# 5. 运行测试
result = graph.invoke({"messages": [("user", "你好!")]})
print(result["messages"][-1].content)
效果:用户输入“你好”,智能体返回大模型生成的问候语。
四、进阶实战:工具调用与循环逻辑
场景:让智能体根据问题动态调用天气查询工具,直至满足条件。
-
定义工具与状态:
from langchain_core.tools import tool @tool def get_weather(city: str): """模拟天气查询工具""" return f"{city}今天晴,气温25℃。" class AgentState(TypedDict): messages: Annotated[list, add_messages] tool_calls: list = [] # 记录工具调用历史 # 工具节点 tools = [get_weather] tool_node = ToolNode(tools)
-
定义条件分支与循环:
def should_continue(state: AgentState): last_msg = state["messages"][-1] return "continue" if last_msg.tool_calls else "end" # 构建图 workflow = StateGraph(AgentState) workflow.add_node("llm", llm_node) # 大模型推理节点 workflow.add_node("tools", tool_node) # 工具调用节点 workflow.set_entry_point("llm") workflow.add_conditional_edges( "llm", should_continue, {"continue": "tools", "end": END} ) workflow.add_edge("tools", "llm") # 工具调用后返回LLM
逻辑:LLM判断是否需要调用工具,若需要则进入工具节点,否则结束。
五、企业级项目:太阳能节能计算智能体
需求:根据用户电费计算太阳能板节省金额,支持多轮对话与错误处理。
# 1. 定义工具与状态
@tool
def compute_savings(monthly_cost: float):
"""计算太阳能节省"""
return {"savings": monthly_cost * 0.7} # 模拟计算
class SolarState(TypedDict):
messages: Annotated[list, add_messages]
# 2. 绑定工具到模型
llm = ChatOpenAI().bind_tools([compute_savings])
# 3. 定义带错误处理的工具节点
def tool_node_with_fallback(state):
try:
return ToolNode([compute_savings]).invoke(state)
except Exception as e:
return {"messages": [f"工具调用失败:{str(e)}"]}
# 4. 构建图(含循环)
workflow = StateGraph(SolarState)
workflow.add_node("assistant", llm_node)
workflow.add_node("tools", tool_node_with_fallback)
workflow.add_conditional_edges("assistant", tools_condition) # 根据LLM输出选择分支
workflow.add_edge("tools", "assistant")
六、精通技巧:状态持久化与调试
- 状态持久化:
使用MemorySaver
保存对话历史,支持断点续执行:from langgraph.checkpoint.memory import MemorySaver memory = MemorySaver() graph = workflow.compile(checkpointer=memory)
- 可视化调试:
通过LangGraph Studio实时监控执行流程,设置断点并修改中间状态。
开启你的智能体之旅
LangGraph为AI开发者提供了从简单到复杂的全链路支持。无论是聊天机器人还是企业级多智能体系统,都能通过状态图和工具调用实现精准控制。动手尝试本文代码,你也能成为AI智能体开发高手!
欢迎在评论区分享你的实践心得,或关注我的账号获取更多大模型实战教程! 🚀
相关资源:
更多推荐
所有评论(0)