LangGraph

AI小编 · 发表于 2026-7-6 14:05:53

作者：CSDN博客
目录

一、LangGraph 是什么？

LangGraph 是 LangChain 生态下的状态驱动图结构工作流框架，专为构建复杂、可恢复、长运行的AI智能体(Agent) 设计。简单说：

流水线

思维导图+流程图

循环、分支、条件判断、状态持久化

核心优势：

状态管理

图结构灵活

生产级可靠

二、快速安装

# 基础安装
pip install langgraph
# 完整安装（含LangChain核心、LLM集成、可视化等）
pip install langgraph[standard]

复制代码

三、核心概念

概念	通俗解释	类比
Graph（图）	整个工作流的容器，定义所有节点和边	一张完整的流程图
Node（节点）	工作流中的单个步骤（如调用LLM、工具、判断）	流程图中的方框
Edge（边）	节点间的连接关系，定义执行顺序	流程图中的箭头
State（状态）	工作流的全局数据存储，所有节点共享读写	流水线的物料箱，每个工位都能取放东西
Conditional Edge（条件边）	根据状态动态选择下一个节点	流程图中的判断分支（是/否）
Checkpoint（检查点）	State状态快照，支持暂停/恢复执行	游戏存档，随时读档继续

四、基础语法API

1. 状态定义（最关键）

状态是LangGraph的灵魂，用TypedDict定义结构化数据：

from typing import TypedDict, List
classAgentState(TypedDict):# 必须包含的字段：消息历史
messages: List[dict]# 自定义字段：当前任务
task:str# 自定义字段：工具调用结果
tool_results:dict# 自定义字段：是否完成
is_finished:bool=False# 带默认值

复制代码

2. 节点函数（业务逻辑）

每个节点是一个函数，输入状态，输出状态更新：

# 1. 简单节点：调用LLM思考defllm_think(state: AgentState)-> AgentState:# 调用LLM生成思考结果
llm_output ="我需要搜索天气信息来回答用户问题"# 更新状态：添加思考消息
state["messages"].append({"role":"assistant","content": llm_output})return state
# 2. 工具调用节点defcall_weather_api(state: AgentState)-> AgentState:# 调用天气API
weather_data ={"temperature":25,"condition":"sunny"}# 更新状态：保存工具结果
state["tool_results"]["weather"]= weather_data
state["messages"].append({"role":"tool","name":"weather_api","content":str(weather_data)})return state
# 3. 条件判断节点（用于条件边）defshould_continue(state: AgentState)->str:# 根据状态判断下一步：继续工具调用还是结束if state["is_finished"]:return"end"# 指向结束节点else:return"continue"# 指向下一个处理节点

复制代码

3. 构建图（核心步骤）

用Graph类组装节点和边：

from langgraph.graph import Graph
# 1. 创建图，指定状态类型
graph = Graph(AgentState)# 2. 添加节点（name, function）
graph.add_node("llm_think", llm_think)
graph.add_node("call_weather", call_weather_api)
graph.add_node("check_finish", should_continue)# 条件判断节点# 3. 添加边（定义执行顺序）# 固定边：从A到B
graph.add_edge("llm_think","call_weather")# 条件边：根据判断结果选择下一个节点
graph.add_conditional_edges("check_finish",# 源节点
should_continue,# 判断函数（返回目标节点名）{"continue":"llm_think",# 循环"end":"__end__"# 内置结束节点})# 4. 设置起始节点
graph.set_entry_point("llm_think")# 5. 编译图（生成可运行的应用）
app = graph.compile()

复制代码

4. 运行图

# 初始状态
initial_state ={"messages":[{"role":"user","content":"今天苏州天气怎么样？"}],"task":"查询苏州天气","tool_results":{}}# 运行图（两种方式）# 方式1：一次性运行到结束
result = app.invoke(initial_state)# 方式2：流式运行（适合实时交互）for step in app.stream(initial_state):print("Step:", step)print("State:", step["messages"][-1])

复制代码

五、实战案例一：智能问答Agent

代码：

from typing import TypedDict, List, Optional
from langgraph.graph import Graph
from langchain_openai import ChatOpenAI
from langchain.tools import tool
import os
# 1. 环境准备（替换为你的API密钥）
os.environ["OPENAI_API_KEY"]="your-api-key"
llm = ChatOpenAI(model="gpt-3.5-turbo")# 2. 定义工具（示例：天气查询工具）@tooldefweather_api(city:str)->str:"""查询指定城市的天气"""# 模拟API调用returnf"{city}天气：25℃，晴，微风"
tools =[weather_api]# 3. 定义状态classAgentState(TypedDict):
messages: List[dict]# 消息历史
tool_name: Optional[str]# 要调用的工具名
tool_input: Optional[str]# 工具输入
tool_result: Optional[str]# 工具结果
is_finished:bool=False# 是否完成# 4. 定义节点函数defllm_decision(state: AgentState)-> AgentState:"""LLM决策：判断是否需要调用工具"""
messages = state["messages"]# 调用LLM生成响应（含工具调用判断）
response = llm.invoke(messages +[{"role":"system","content":"你可以调用weather_api工具查询天气，格式：TOOL[工具名,参数]。不需要工具时直接回答。"}])
content = response.content
state["messages"].append({"role":"assistant","content": content})# 检查是否需要调用工具if"TOOL["in content:# 解析工具调用（简化示例）
tool_call = content.split("TOOL[")[1].split("]")[0]
state["tool_name"], state["tool_input"]= tool_call.split(",")
state["is_finished"]=Falseelse:
state["is_finished"]=Truereturn state
deftool_execution(state: AgentState)-> AgentState:"""工具执行节点"""
tool_name = state["tool_name"]
tool_input = state["tool_input"]# 执行工具for tool in tools:if tool.name == tool_name:
result = tool.run(tool_input)
state["tool_result"]= result
state["messages"].append({"role":"tool","name": tool_name,"content": result})breakreturn state
defshould_continue(state: AgentState)->str:"""判断是否继续循环"""return"end"if state["is_finished"]else"continue"# 5. 构建图
graph = Graph(AgentState)# 添加节点
graph.add_node("llm_decision", llm_decision)
graph.add_node("tool_execution", tool_execution)# 添加边
graph.add_edge("llm_decision","tool_execution")
graph.add_conditional_edges("tool_execution",
should_continue,{"continue":"llm_decision",# 循环：继续决策"end":"__end__"# 结束})# 设置起始节点
graph.set_entry_point("llm_decision")# 编译图
app = graph.compile()# 6. 运行测试if __name__ =="__main__":
initial_state ={"messages":[{"role":"user","content":"苏州今天天气如何？"}],"tool_name":None,"tool_input":None,"tool_result":None}# 流式运行并打印结果for step in app.stream(initial_state):for node, data in step.items():print(f"\n=== {node} ===")if"messages"in data:print("最新消息:", data["messages"][-1])

复制代码

运行流程解析：

用户提问

llm_decision

tool_execution

条件判断

llm_decision

条件判断

六、实战案例二：多智能体协作

核心流程：研究员 → 程序员 → 审核员 → 【不通过→退回程序员 / 通过→总结员】
[code]from typing import TypedDict, Literal
from langgraph.graph import StateGraph

# ===================== 1. 定义全局状态（仅新增2个审核字段，其他不变）=====================classMultiAgentState(TypedDict):# 用户原始需求
query:str# 研究员搜集的资料
research_info:str# 程序员编写的代码
code:str# 审核意见（新增）
review_opinion:str# 代码是否通过审核（新增）
is_code_approved:bool# 总结员的最终结果
final_result:str# ===================== 2. 原有3个智能体（完全没改动！）=====================defresearcher_agent(state: MultiAgentState)-> MultiAgentState:"""研究员：搜集资料"""
query = state["query"]
research_info =f"【研究员资料】关于{query}：核心逻辑是输入处理+函数实现，适配Python语法"print("✅ 研究员完成工作")return{"research_info": research_info}defcoder_agent(state: MultiAgentState)-> MultiAgentState:"""程序员：根据资料/审核意见写代码"""# 如果有审核意见，打印提示（模拟修改代码）if state["review_opinion"]:print(f"

25位大厂高管转战生成式AI创业！吸金猛，不

LangGraph

浏览过的版块

LangGraph

关于我们

产品与服务

解决方案

产品与服务