作者：小郭开发
LangGraph 框架详细介绍与Python代码示例
一、什么是 LangGraph
LangGraph 是一个基于 LangChain 的高级框架，专门用于构建有状态的、多轮对话的、多智能体的应用程序。它允许开发者使用图（Graph）的结构来定义复杂的 AI 工作流，特别适合构建需要记忆、条件分支和循环的智能应用。
核心特性
图结构：使用有向图（DAG）定义应用逻辑，支持循环和分支状态管理：内置状态管理机制，自动处理对话历史多智能体支持：轻松构建多个智能体协作的应用持久化：支持将对话状态持久化到数据库流式处理：支持实时流式输出调试友好：可视化图结构，便于调试和理解
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
节点是图中的处理单元，每个节点接收状态并返回更新后的状态。- defchatbot(state: State):# 处理逻辑return{"messages":[...]}defrouter(state: State):# 路由逻辑return"node_name"
边定义节点之间的连接关系，决定状态如何在节点间流动。- 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)
- 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}")
- 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"})
- 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}")
- 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()
- 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 应用的重要工具。通过合理设计图结构，开发者可以构建出既灵活又可维护的智能应用。
|
广告