LangGraph 框架详细介绍与Python代码示例

米落枫 · 发表于 6 天前

作者：小郭开发
LangGraph 框架详细介绍与Python代码示例

一、什么是 LangGraph

LangGraph 是一个基于 LangChain 的高级框架，专门用于构建有状态的、多轮对话的、多智能体的应用程序。它允许开发者使用图（Graph）的结构来定义复杂的 AI 工作流，特别适合构建需要记忆、条件分支和循环的智能应用。
核心特性

图结构

状态管理

多智能体支持

持久化

流式处理

调试友好

LangGraph vs LangChain

特性	LangChain	LangGraph
结构	线性链式	图结构
状态管理	手动管理	自动管理
多轮对话	需要显式 memory	内置状态
复杂工作流	代码控制	图结构
智能体协作	较复杂	原生支持

二、安装与配置

安装 LangGraph

pip install langgraph
pip install langchain-openai
pip install langchain

复制代码

基础配置

import os
os.environ["OPENAI_API_KEY"]="your-api-key-here"

复制代码

三、核心概念

1. State（状态）

LangGraph 使用状态（State）来管理应用的所有数据。状态可以是字典、数据类等。

from typing import Annotated
from typing_extensions import TypedDict
classState(TypedDict):
messages:list
user_input:str
context:str

复制代码

2. Nodes（节点）

节点是图中的处理单元，每个节点接收状态并返回更新后的状态。

defchatbot(state: State):# 处理逻辑return{"messages":[...]}defrouter(state: State):# 路由逻辑return"node_name"

复制代码

3. Edges（边）

边定义节点之间的连接关系，决定状态如何在节点间流动。

graph.add_edge("node1","node2")
graph.add_conditional_edges("node1", router,{"path1":"node2","path2":"node3"})

复制代码

四、基本示例

示例1：简单的对话图

from langgraph.graph import StateGraph
from langchain_openai import ChatOpenAI
from typing import TypedDict
# 定义状态classState(TypedDict):
messages:list# 创建 LLM
llm = ChatOpenAI(model="gpt-3.5-turbo")# 定义节点defchatbot(state: State):
messages = state["messages"]
response = llm.invoke(messages)return{"messages": messages +[response]}# 构建图
workflow = StateGraph(State)
workflow.add_node("chatbot", chatbot)
workflow.set_entry_point("chatbot")
workflow.add_edge("chatbot","chatbot")# 编译图
app = workflow.compile()# 运行
result = app.invoke({"messages":[{"role":"user","content":"你是谁"}]})print(result["messages"][-1].content)

复制代码

示例2：带条件分支的图

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
classState(TypedDict):
messages:list
current_node:strdefchatbot(state: State):
messages = state["messages"]
response = ChatOpenAI(model="gpt-3.5-turbo").invoke(messages)return{"messages": messages +[response]}defroute_message(state: State):
last_message = state["messages"][-1].content
if"天气"in last_message:return"weather"elif"笑话"in last_message:return"joke"else:return"chatbot"
workflow = StateGraph(State)
workflow.add_node("chatbot", chatbot)
workflow.add_node("weather",lambda s:{"messages": s["messages"]+[{"role":"assistant","content":"今天晴转多云"}]})
workflow.add_node("joke",lambda s:{"messages": s["messages"]+[{"role":"assistant","content":"为什么程序员分不清万圣节和圣诞节？因为 Oct 31 == Dec 25！"}]})
workflow.add_conditional_edges("chatbot",
route_message,{"weather":"weather","joke":"joke","chatbot": END
})
workflow.set_entry_point("chatbot")
app = workflow.compile()# 运行
result = app.invoke({"messages":[{"role":"user","content":"讲个笑话"}]})print(result["messages"][-1].content)

复制代码

五、高级功能

1. 循环图（聊天机器人）

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
classState(TypedDict):
messages:listdefchatbot(state: State):
llm = ChatOpenAI(model="gpt-3.5-turbo")
response = llm.invoke(state["messages"])return{"messages": state["messages"]+[response]}# 创建图
workflow = StateGraph(State)
workflow.add_node("chatbot", chatbot)
workflow.set_entry_point("chatbot")# 添加循环边
workflow.add_edge("chatbot", END)# 编译
app = workflow.compile()# 多轮对话
messages =[{"role":"user","content":"你好"}]
result1 = app.invoke({"messages": messages})print(f"用户: {messages[-1]['content']}")print(f"AI: {result1['messages'][-1].content}")
messages = result1["messages"]+[{"role":"user","content":"你能做什么？"}]
result2 = app.invoke({"messages": messages})print(f"用户: {messages[-1]['content']}")print(f"AI: {result2['messages'][-1].content}")

复制代码

2. 多智能体系统

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated
import operator
classState(TypedDict):
messages:list
current_agent:strdefresearcher(state: State):
llm = ChatOpenAI(model="gpt-3.5-turbo")
response = llm.invoke(state["messages"])return{"messages": state["messages"]+[response],"current_agent":"researcher"}defwriter(state: State):
llm = ChatOpenAI(model="gpt-3.5-turbo")
response = llm.invoke(state["messages"])return{"messages": state["messages"]+[response],"current_agent":"writer"}defrouter(state: State):if"写"in state["messages"][-1].content:return"writer"return"researcher"# 创建多智能体图
workflow = StateGraph(State)
workflow.add_node("researcher", researcher)
workflow.add_node("writer", writer)
workflow.add_conditional_edges("researcher",
router,{"researcher":"researcher","writer":"writer"})
workflow.set_entry_point("researcher")
app = workflow.compile()# 运行
result = app.invoke({"messages":[{"role":"user","content":"写一篇关于人工智能的文章"}],"current_agent":"researcher"})

复制代码

3. 带记忆的聊天机器人

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from typing import TypedDict
classState(TypedDict):
messages:listdefchatbot(state: State):
llm = ChatOpenAI(model="gpt-3.5-turbo")
response = llm.invoke(state["messages"])return{"messages": state["messages"]+[response]}# 创建带内存检查点的图
workflow = StateGraph(State)
workflow.add_node("chatbot", chatbot)
workflow.set_entry_point("chatbot")
workflow.add_edge("chatbot", END)# 使用 MemorySaver 保存对话历史
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)# 配置线程 ID
config ={"configurable":{"thread_id":"user_1"}}# 第一次对话
result1 = app.invoke({"messages":[{"role":"user","content":"我叫张三"}]},
config
)print(f"AI: {result1['messages'][-1].content}")# 第二次对话（保留上下文）
result2 = app.invoke({"messages":[{"role":"user","content":"我叫什么名字？"}]},
config
)print(f"AI: {result2['messages'][-1].content}")

复制代码

4. 复杂的工作流（研究助手）

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated
classState(TypedDict):
topic:str
research:str
outline:str
draft:strdefresearch_node(state: State):
llm = ChatOpenAI(model="gpt-3.5-turbo")
prompt =f"研究主题：{state['topic']}，提供详细的研究内容"
research = llm.invoke(prompt)return{"research": research.content}defoutline_node(state: State):
llm = ChatOpenAI(model="gpt-3.5-turbo")
prompt =f"基于以下研究内容，创建文章大纲：\n{state['research']}"
outline = llm.invoke(prompt)return{"outline": outline.content}defwrite_node(state: State):
llm = ChatOpenAI(model="gpt-3.5-turbo")
prompt =f"根据以下大纲撰写文章：\n大纲：{state['outline']}\n主题：{state['topic']}"
draft = llm.invoke(prompt)return{"draft": draft.content}# 创建工作流
workflow = StateGraph(State)
workflow.add_node("research", research_node)
workflow.add_node("outline", outline_node)
workflow.add_node("write", write_node)# 定义流程
workflow.set_entry_point("research")
workflow.add_edge("research","outline")
workflow.add_edge("outline","write")
workflow.add_edge("write", END)# 编译
app = workflow.compile()# 运行
result = app.invoke({"topic":"人工智能在医疗领域的应用"})print("研究内容：", result["research"])print("文章大纲：", result["outline"])print("最终文章：", result["draft"])

复制代码

六、可视化图结构

from langgraph.graph import StateGraph
from IPython.display import Image, display
# 创建图
workflow = StateGraph(State)
workflow.add_node("chatbot", chatbot)
workflow.set_entry_point("chatbot")
workflow.add_edge("chatbot", END)# 编译
app = workflow.compile()# 可视化try:
display(Image(app.get_graph().draw_mermaid_png()))except:print(app.get_graph().draw_ascii())

复制代码

七、实际应用场景

1. 智能客服系统

from langgraph.graph import StateGraph, END
from typing import TypedDict
classState(TypedDict):
messages:list
intent:strdefdetect_intent(state: State):# 意图检测
last_msg = state["messages"][-1].content
if"退货"in last_msg:return{"intent":"return"}elif"咨询"in last_msg:return{"intent":"consult"}return{"intent":"general"}defgeneral_handler(state: State):
llm = ChatOpenAI(model="gpt-3.5-turbo")
response = llm.invoke(state["messages"])return{"messages": state["messages"]+[response]}defreturn_handler(state: State):return{"messages": state["messages"]+[{"role":"assistant","content":"退货流程：1. 订单确认 2. 原因说明 3. 物流寄回"}]}defconsult_handler(state: State):return{"messages": state["messages"]+[{"role":"assistant","content":"请描述您的具体问题，我会为您解答"}]}
workflow = StateGraph(State)
workflow.add_node("detect", detect_intent)
workflow.add_node("general", general_handler)
workflow.add_node("return", return_handler)
workflow.add_node("consult", consult_handler)
workflow.set_entry_point("detect")
workflow.add_conditional_edges("detect",lambda s: s["intent"],{"general":"general","return":"return","consult":"consult"})
workflow.add_edge("general", END)
workflow.add_edge("return", END)
workflow.add_edge("consult", END)
app = workflow.compile()

复制代码

2. 代码审查助手

from langgraph.graph import StateGraph, END
from typing import TypedDict
classState(TypedDict):
code:str
issues:list
suggestions:strdefanalyze_code(state: State):
llm = ChatOpenAI(model="gpt-3.5-turbo")
prompt =f"""分析以下代码，找出潜在问题：
```python
{state['code']}

复制代码

请列出所有发现的问题"“”
response = llm.invoke(prompt)
return {“issues”: response.content.split(“\n”)}
def suggest_fixes(state: State):
llm = ChatOpenAI(model=“gpt-3.5-turbo”)
prompt = f"""针对以下代码问题，提供修复建议：
问题：
{state[‘issues’]}
代码：
{state[‘code’]}“”"
response = llm.invoke(prompt)
return {“suggestions”: response.content}
workflow = StateGraph(State)
workflow.add_node(“analyze”, analyze_code)
workflow.add_node(“suggest”, suggest_fixes)
workflow.set_entry_point(“analyze”)
workflow.add_edge(“analyze”, “suggest”)
workflow.add_edge(“suggest”, END)
app = workflow.compile()
result = app.invoke({“code”: “def add(a, b): return a + b”})
print(result[“issues”])
print(result[“suggestions”])

## 八、LangGraph 的优势
1. **清晰的结构**：图结构使复杂逻辑更易理解和维护
2. **状态管理**：自动处理状态传递和持久化
3. **可扩展性**：轻松添加新节点和边
4. **调试方便**：可视化图结构，便于排查问题
5. **灵活性**：支持条件分支、循环、并行等复杂逻辑
6. **多智能体**：原生支持多个智能体协作
## 九、总结
LangGraph 为构建复杂的 AI 应用提供了强大的图结构支持。相比 LangChain 的线性链式结构，LangGraph 的图结构更适合以下场景：
- **需要状态管理的多轮对话应用**
- **多个智能体协作的工作流**
- **包含条件分支和循环的复杂逻辑**
- **需要持久化和恢复的应用状态**
随着 AI 应用复杂度的不断提升，LangGraph 正成为构建生产级 AI 应用的重要工具。通过合理设计图结构，开发者可以构建出既灵活又可维护的智能应用。

复制代码

原文地址：https://blog.csdn.net/weixin_43126767/article/details/158609671

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

LangGraph 框架详细介绍与Python代码示例

langchain入门系列之一初识langchain

关于我们

产品与服务

解决方案

产品与服务

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

LangGraph 框架详细介绍与Python代码示例

langchain入门系列之一 初识langchain

关于我们

产品与服务

解决方案

产品与服务

langchain入门系列之一初识langchain