最近在尝试构建复杂的AI应用时你是否遇到过这样的困境单个大模型调用无法处理多步骤任务不同工具和模型之间的状态流转混乱不堪代码里充满了难以维护的if-else逻辑这正是传统AI应用开发中普遍存在的痛点。LangGraph的出现就是为了解决这类问题而生。它不是一个新的大模型而是一个基于LangChain的框架专门用于构建有状态、多步骤的AI智能体Agent。本文将带你从零开始彻底掌握LangGraph的核心概念、设计模式与实战应用无论你是想入门AI应用开发还是希望将现有单点AI能力升级为自动化工作流都能在这里找到完整的答案。1. LangGraph是什么为什么你需要它在深入代码之前我们必须先理解LangGraph要解决的核心问题。简单来说LangGraph是一个用于构建有状态、多步骤工作流的库。它将你的AI应用建模为一个“图”Graph图中的节点Node代表一个执行单元如调用一次大模型、执行一个工具边Edge则定义了节点之间的流转逻辑。1.1 从LangChain到LangGraph思维的转变你可能已经熟悉LangChain它通过Chain将多个组件如LLM、Prompt、Output Parser链接起来。但Chain本质上是线性的、无状态的。对于“根据用户问题搜索网络然后分析结果最后生成报告并保存”这样的复杂场景线性Chain就显得力不从心。LangGraph引入了“图”和“状态”的概念图Graph 描述了应用的整体流程和决策路径。状态State 在整个工作流执行过程中持久化并传递的数据。这是LangGraph的灵魂它使得工作流可以记住之前步骤的结果并根据这些结果决定下一步走向。1.2 核心应用场景AI智能体AgentLangGraph最典型的应用就是构建智能体。一个智能体通常包含规划Planning 理解目标拆解步骤。执行Action 调用工具如搜索、计算、查数据库或大模型。观察Observation 获取工具执行结果。循环Loop 根据观察结果决定是继续执行还是结束。这个经典的“规划-执行-观察”循环用LangGraph可以非常优雅地实现。此外它还非常适合构建需要分支、循环、并行或人工审核的复杂业务流程。2. 环境准备与核心概念在动手之前我们需要搭建好开发环境并厘清几个最关键的概念。2.1 环境搭建本文示例使用Python环境。请确保已安装Python建议3.8版本。首先安装必要的库。LangGraph通常与LangChain配合使用并需要一个LLM提供商如OpenAI。pip install langgraph langchain langchain-openai如果你使用其他LLM如通义千问、DeepSeek等请安装对应的LangChain集成包。2.2 理解核心抽象State, Node, Edge这是学习LangGraph必须跨越的三个门槛。State状态这是一个贯穿整个图执行过程的字典。你需要预先定义这个字典的“结构”Schema即有哪些字段分别是什么类型。每个节点都可以读取和修改State中的字段。例如一个聊天机器人的State可能包含messages对话历史user_query当前问题search_results中间搜索结果。Node节点图中的一个功能单元。它本质上是一个函数接收当前的State执行一些操作如调用LLM然后返回一个包含更新后State字段的字典。例如一个“调用搜索工具”的节点会读取State中的user_query调用搜索引擎API然后将结果写入State的search_results字段。Edge边决定执行完一个节点后下一步该去哪个节点。分为两种条件边Conditional Edge 根据State中的某个值通常是LLM的判断来决定下一个节点。这是实现分支逻辑的关键。普通边Normal Edge 无条件地指向下一个节点。理解了这三个概念LangGraph的整个设计思路就清晰了定义好数据容器State安排好处理单元Node并用逻辑线Edge把它们连接起来。3. 第一个LangGraph应用智能对话助手让我们通过一个最简单的例子将上述概念具象化。我们将构建一个能根据用户问题决定是否要联网搜索的对话助手。3.1 定义状态State首先我们定义工作流需要维护哪些数据。这里使用TypedDict来获得更好的类型提示。from typing import TypedDict, List, Annotated import operator from langchain_core.messages import HumanMessage, AIMessage class AgentState(TypedDict): # 对话消息历史 messages: Annotated[List[HumanMessage | AIMessage], operator.add] # 是否需要搜索由LLM判断 needs_search: bool # 搜索结果 search_result: strmessages 使用Annotated和operator.add是LangGraph的一个约定表示这个字段是“追加”模式。每次节点返回的messages列表都会自动追加到历史中而不是覆盖。needs_search 一个布尔标志。search_result 存储搜索到的文本。3.2 创建节点Nodes我们需要三个节点router路由判断、search_node搜索、general_chat_node普通聊天。from langchain_openai import ChatOpenAI from langgraph.graph import END, StateGraph # 初始化LLM llm ChatOpenAI(modelgpt-3.5-turbo) def router(state: AgentState) - dict: 判断用户问题是否需要联网搜索 last_message state[“messages”][-1].content # 构建一个Prompt让LLM判断 prompt f”用户的问题是‘{last_message}’。这个问题需要实时或最新的网络信息才能准确回答吗只需要回答‘是’或‘否’。” response llm.invoke(prompt) decision response.content.strip() needs_search decision “是” return {“needs_search”: needs_search} def search_node(state: AgentState) - dict: 模拟搜索节点实际项目中替换为真实搜索API调用 query state[“messages”][-1].content # 这里模拟搜索过程 print(f”[搜索节点] 正在搜索: {query}”) # 模拟返回搜索结果 mock_result f”关于‘{query}’的模拟搜索结果当前最新版本是v2.0发布于2023年底。” return {“search_result”: mock_result, “messages”: [AIMessage(contentf”我已搜索到信息{mock_result}”)]} def general_chat_node(state: AgentState) - dict: 普通聊天节点直接回答 last_message state[“messages”][-1] response llm.invoke([last_message]) return {“messages”: [response]}3.3 构建图Graph并连接边Edges现在我们把节点组装起来并定义它们之间的流转逻辑。from langgraph.graph import StateGraph, END # 1. 创建图构建器并指定状态结构 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(“router”, router) workflow.add_node(“search”, search_node) workflow.add_node(“general_chat”, general_chat_node) # 3. 设置入口点所有流程都从router开始 workflow.set_entry_point(“router”) # 4. 添加条件边根据router的结果决定下一步 def decide_next_step(state: AgentState) - str: if state.get(“needs_search”): return “search” # 需要搜索则前往search节点 else: return “general_chat” # 否则直接聊天 workflow.add_conditional_edges( “router”, # 从哪个节点出发 decide_next_step, # 决定下一个节点的函数 {“search”: “search”, “general_chat”: “general_chat”} # 可能的目的地映射 ) # 5. 添加普通边search节点执行完后应进入general_chat节点来总结回答 workflow.add_edge(“search”, “general_chat”) # 6. 添加普通边general_chat节点执行完后工作流结束 workflow.add_edge(“general_chat”, END) # 7. 编译图得到可执行对象 app workflow.compile()3.4 运行与测试现在我们可以运行这个图了。# 初始化状态 initial_state AgentState( messages[HumanMessage(content“LangGraph的最新版本是什么”)], needs_searchFalse, search_result“” ) # 执行图 final_state app.invoke(initial_state) # 查看最终结果 for msg in final_state[“messages”]: print(f”{msg.type}: {msg.content}”)预期输出[搜索节点] 正在搜索: LangGraph的最新版本是什么 ai: 我已搜索到信息关于‘LangGraph的最新版本是什么’的模拟搜索结果当前最新版本是v2.0发布于2023年底。 ai: 这里会是general_chat节点基于搜索结果生成的最终友好回答这个简单的例子展示了LangGraph的核心流程状态流转、节点执行和条件分支。router节点修改了needs_search状态条件边根据这个状态值决定路径最终所有结果都汇聚到messages中。4. 构建高级智能体ReAct模式实战ReActReasoning Acting是智能体的经典模式它让LLM循环地进行“思考-行动-观察”。用LangGraph实现ReAct智能体非常直观。4.1 设计状态与工具我们将创建一个能使用计算器和搜索工具的智能体。from typing import TypedDict, Annotated, List, Union import operator from langchain_core.messages import BaseMessage from langchain.tools import tool from langchain_community.tools.tavily_search import TavilySearchResults # 定义工具 tool def calculator(expression: str) - str: “”“计算一个数学表达式。例如‘3 * 7 5’“”“ try: # 警告实际项目中请使用更安全的eval替代方案如ast.literal_eval或数学库 result eval(expression) return f”计算结果: {result}” except Exception as e: return f”计算错误: {e}” # 假设我们有一个搜索工具需要API Key # search_tool TavilySearchResults() # 为演示我们创建一个模拟工具 tool def web_search(query: str) - str: “”“在网络上搜索信息。”“” return f”关于‘{query}’的模拟搜索结果这是一个模拟的搜索结果摘要。” # 可用工具列表 tools [calculator, web_search] # 定义状态 class ReActState(TypedDict): messages: Annotated[List[BaseMessage], operator.add] # 用于记录LLM的“思考”过程 scratchpad: Annotated[List[BaseMessage], operator.add] # 当前轮次防止无限循环 step: int4.2 创建核心节点Agent和Tools在ReAct图中通常有两个主要节点agent负责思考并决定使用哪个工具和tools执行工具。from langchain.agents import create_react_agent from langchain_core.prompts import ChatPromptTemplate # 1. 创建Agent节点 def agent_node(state: ReActState): # 构建Prompt包含指令、工具描述、历史scratchpad和当前问题 prompt ChatPromptTemplate.from_messages([ (“system”, “你是一个有帮助的助手可以使用工具。请严格按以下格式回应\n思考你的推理过程\n行动要使用的工具名\n行动输入工具的输入”), (“placeholder”, “{chat_history}”), # 这里会填入scratchpad (“human”, “{input}”) ]) # 将工具绑定到LLM llm_with_tools llm.bind_tools(tools) chain prompt | llm_with_tools # 调用LLM agent_response chain.invoke({ “chat_history”: state[“scratchpad”], “input”: state[“messages”][-1].content }) # 将LLM的响应记录到scratchpad中 return { “scratchpad”: [agent_response], “step”: state[“step”] 1 } # 2. 创建Tools节点 def tools_node(state: ReActState): # 从scratchpad中取出最新的LLM响应 last_message state[“scratchpad”][-1] # 解析LLM响应提取“行动”和“行动输入” # 这里简化处理实际应使用更健壮的解析器 content last_message.content action_line [l for l in content.split(‘\n’) if l.startswith(‘行动’)][0] action_input_line [l for l in content.split(‘\n’) if l.startswith(‘行动输入’)][0] tool_name action_line.replace(‘行动’, ‘’).strip() tool_input action_input_line.replace(‘行动输入’, ‘’).strip() # 找到对应的工具并执行 tool_to_use next((t for t in tools if t.name tool_name), None) if tool_to_use: tool_result tool_to_use.invoke(tool_input) result_msg AIMessage(contentf”观察{tool_result}”) else: result_msg AIMessage(contentf”观察工具‘{tool_name}’未找到。”) # 将工具执行结果观察记录到scratchpad return {“scratchpad”: [result_msg]}4.3 构建并运行ReAct图现在构建一个循环让Agent和Tools交替执行直到LLM决定结束。from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver # 创建图 workflow StateGraph(ReActState) workflow.add_node(“agent”, agent_node) workflow.add_node(“tools”, tools_node) workflow.set_entry_point(“agent”) # 定义边逻辑Agent之后总是去Tools workflow.add_edge(“agent”, “tools”) # 关键从Tools节点出来的条件边决定是继续循环还是结束 def should_continue(state: ReActState) - str: last_message state[“scratchpad”][-1] # 如果LLM在思考中说了“最终答案”或者步骤超过5步则结束 if “最终答案” in last_message.content or state[“step”] 5: return “end” else: return “continue” workflow.add_conditional_edges( “tools”, should_continue, {“continue”: “agent”, “end”: END} # 继续循环回Agent或结束 ) # 编译图 app workflow.compile()4.4 测试ReAct智能体# 使用检查点存储器便于调试和观察中间状态 memory MemorySaver() app_with_memory workflow.compile(checkpointermemory) config {“configurable”: {“thread_id”: “test_react_1”}} initial_state ReActState( messages[HumanMessage(content“请先计算15的平方然后搜索一下LangGraph的作者信息。”)], scratchpad[], step0 ) # 以流式方式执行可以看到每一步的输出 for event in app_with_memory.stream(initial_state, config, stream_mode“values”): event[“scratchpad”][-1].pretty_print()这个ReAct智能体会先调用计算器工具得到225然后调用搜索工具获取信息最后LLM综合两者给出最终答案。通过scratchpad我们可以清晰地看到LLM“思考-行动-观察”的完整链条。5. 高级特性与工程化实践掌握了基础构建后我们来看一些让LangGraph应用更健壮、更强大的高级特性。5.1 持久化与检查点Checkpoints对于长时间运行或需要中断恢复的智能体检查点功能至关重要。它能在每个节点执行后自动保存完整状态。from langgraph.checkpoint.sqlite import SqliteSaver import tempfile import os # 使用SQLite持久化检查点 db_file tempfile.NamedTemporaryFile(deleteFalse).name checkpointer SqliteSaver.from_conn_string(f”sqlite:///{db_file}”) # 在编译图时传入checkpointer persistent_app workflow.compile(checkpointercheckpointer) config {“configurable”: {“thread_id”: “user_session_123”}} try: # 第一次执行 state1 persistent_app.invoke(initial_state, config) print(“第一次执行完成。”) # 模拟应用重启后可以从检查点恢复继续执行 # 通过指定config中的相同thread_id可以加载历史状态 # 这对于实现多轮对话、异步任务恢复非常有用 finally: os.unlink(db_file) # 清理临时文件5.2 人工干预节点Human-in-the-Loop在某些关键业务流程中需要引入人工审核。LangGraph可以轻松地将“等待人工输入”作为一个节点。def human_review_node(state: AgentState): 模拟人工审核节点。在实际应用中这里会连接到一个Web界面或通知系统。 task_description state[“messages”][-1].content print(f”\n 需要人工审核 \n任务{task_description}\n”) # 在实际系统中这里会阻塞等待用户输入。 # 为演示我们模拟用户输入“批准” human_decision “批准” # 模拟输入 return {“human_decision”: human_decision} # 在图中加入人工审核节点并设置条件边只有审核通过才进入下一步。5.3 子图Subgraphs与模块化复杂的应用可以拆分成多个子图提高可维护性和复用性。子图本身也是一个StateGraph可以被主图当作一个节点来调用。# 定义一个处理“数据查询”的子图 subgraph_builder StateGraph(AgentState) # ... 构建子图内部的节点和边 ... subgraph subgraph_builder.compile() # 在主图中可以将子图作为一个节点添加 main_workflow.add_node(“data_query_subgraph”, subgraph)这种方式非常适合微服务架构每个团队可以负责开发一个独立的子图。6. 常见问题与调试技巧在开发LangGraph应用时你可能会遇到以下典型问题。6.1 状态State更新不符合预期这是最常见的问题。请牢记追加Add vs 覆盖Replace 使用Annotated[List[...], operator.add]的字段是追加否则是覆盖。确保你的节点返回的字典键值与State定义匹配且更新逻辑符合预期。调试方法 在每个节点函数开始和结束时打印State。或者使用LangGraph的stream模式观察状态变化。6.2 图陷入无限循环智能体可能无法自行终止。解决方案设置最大步数 在State中设置一个step计数器在条件边判断函数中检查是否超过阈值。优化Prompt 明确指示LLM在何时输出“最终答案”等终止标记。使用超时机制 在图编译或调用时设置超时。6.3 工具调用解析失败LLM可能不按你指定的格式返回工具调用信息。使用LangChain的bind_tools和with_structured_output 它们能强制LLM返回结构化的工具调用对象比解析文本更可靠。在Prompt中提供更清晰的示例 在系统提示词中给出2-3个完美的响应格式示例。6.4 性能优化对于复杂图性能可能成为瓶颈。异步执行 如果节点间没有严格的先后依赖可以考虑使用add_node的异步版本并利用asyncio。缓存 对于纯函数节点或昂贵的LLM调用可以考虑引入缓存机制。精简State 只保留必要的数据在State中避免传递过大的对象。7. 最佳实践与项目架构建议将LangGraph用于实际项目时遵循以下实践能避免很多坑。7.1 状态设计原则最小化 State只存储工作流真正需要流转和共享的数据。中间计算结果尽量放在节点局部变量中。明确类型 坚持使用TypedDict或Pydantic模型来定义State这能极大提高代码可读性和减少运行时错误。区分会话与流程 将会话ID、用户ID等元数据放在config中而不是State里。State应专注于业务流程数据。7.2 节点设计原则单一职责 每个节点只做一件事。一个节点要么调用LLM要么调用一个工具要么处理数据。幂等性 尽可能让节点函数是幂等的即相同输入产生相同输出这有利于调试和重试。错误处理 节点内部应做好异常捕获并选择是将错误信息放入State向下传递还是直接抛出异常终止图。7.3 图的版本管理与测试版本化 当图的结构节点、边发生变化时应视为一次版本升级。考虑如何平滑迁移正在运行中的旧图实例的状态。单元测试 为每个节点函数编写单元测试模拟输入State验证输出State。集成测试 为整个图编写集成测试使用典型输入验证最终输出和状态变化。7.4 生产环境部署可观测性 在每个节点的入口和出口记录日志包括State的快照。这对于排查生产问题至关重要。监控与告警 监控图的执行时长、错误率、循环次数等指标。配置化 将LLM的模型名称、API Key、工具开关等配置外置便于不同环境开发、测试、生产的切换。LangGraph代表了一种构建复杂AI应用的新范式它将流程控制与AI能力解耦通过声明式的“图”来管理复杂度。从简单的条件分支对话助手到拥有复杂推理循环的ReAct智能体其核心思想始终是状态驱动和流程可视化。学习LangGraph的最佳方式就是从一个具体需求开始画出你的工作流程图然后将其转化为节点和边。随着你对状态管理和条件流转越来越熟悉你会发现构建强大、可靠且可维护的AI应用不再是一件令人畏惧的事情。 LangGraph实战：构建有状态多步骤AI智能体的完整指南
最近在尝试构建复杂的AI应用时你是否遇到过这样的困境单个大模型调用无法处理多步骤任务不同工具和模型之间的状态流转混乱不堪代码里充满了难以维护的if-else逻辑这正是传统AI应用开发中普遍存在的痛点。LangGraph的出现就是为了解决这类问题而生。它不是一个新的大模型而是一个基于LangChain的框架专门用于构建有状态、多步骤的AI智能体Agent。本文将带你从零开始彻底掌握LangGraph的核心概念、设计模式与实战应用无论你是想入门AI应用开发还是希望将现有单点AI能力升级为自动化工作流都能在这里找到完整的答案。1. LangGraph是什么为什么你需要它在深入代码之前我们必须先理解LangGraph要解决的核心问题。简单来说LangGraph是一个用于构建有状态、多步骤工作流的库。它将你的AI应用建模为一个“图”Graph图中的节点Node代表一个执行单元如调用一次大模型、执行一个工具边Edge则定义了节点之间的流转逻辑。1.1 从LangChain到LangGraph思维的转变你可能已经熟悉LangChain它通过Chain将多个组件如LLM、Prompt、Output Parser链接起来。但Chain本质上是线性的、无状态的。对于“根据用户问题搜索网络然后分析结果最后生成报告并保存”这样的复杂场景线性Chain就显得力不从心。LangGraph引入了“图”和“状态”的概念图Graph 描述了应用的整体流程和决策路径。状态State 在整个工作流执行过程中持久化并传递的数据。这是LangGraph的灵魂它使得工作流可以记住之前步骤的结果并根据这些结果决定下一步走向。1.2 核心应用场景AI智能体AgentLangGraph最典型的应用就是构建智能体。一个智能体通常包含规划Planning 理解目标拆解步骤。执行Action 调用工具如搜索、计算、查数据库或大模型。观察Observation 获取工具执行结果。循环Loop 根据观察结果决定是继续执行还是结束。这个经典的“规划-执行-观察”循环用LangGraph可以非常优雅地实现。此外它还非常适合构建需要分支、循环、并行或人工审核的复杂业务流程。2. 环境准备与核心概念在动手之前我们需要搭建好开发环境并厘清几个最关键的概念。2.1 环境搭建本文示例使用Python环境。请确保已安装Python建议3.8版本。首先安装必要的库。LangGraph通常与LangChain配合使用并需要一个LLM提供商如OpenAI。pip install langgraph langchain langchain-openai如果你使用其他LLM如通义千问、DeepSeek等请安装对应的LangChain集成包。2.2 理解核心抽象State, Node, Edge这是学习LangGraph必须跨越的三个门槛。State状态这是一个贯穿整个图执行过程的字典。你需要预先定义这个字典的“结构”Schema即有哪些字段分别是什么类型。每个节点都可以读取和修改State中的字段。例如一个聊天机器人的State可能包含messages对话历史user_query当前问题search_results中间搜索结果。Node节点图中的一个功能单元。它本质上是一个函数接收当前的State执行一些操作如调用LLM然后返回一个包含更新后State字段的字典。例如一个“调用搜索工具”的节点会读取State中的user_query调用搜索引擎API然后将结果写入State的search_results字段。Edge边决定执行完一个节点后下一步该去哪个节点。分为两种条件边Conditional Edge 根据State中的某个值通常是LLM的判断来决定下一个节点。这是实现分支逻辑的关键。普通边Normal Edge 无条件地指向下一个节点。理解了这三个概念LangGraph的整个设计思路就清晰了定义好数据容器State安排好处理单元Node并用逻辑线Edge把它们连接起来。3. 第一个LangGraph应用智能对话助手让我们通过一个最简单的例子将上述概念具象化。我们将构建一个能根据用户问题决定是否要联网搜索的对话助手。3.1 定义状态State首先我们定义工作流需要维护哪些数据。这里使用TypedDict来获得更好的类型提示。from typing import TypedDict, List, Annotated import operator from langchain_core.messages import HumanMessage, AIMessage class AgentState(TypedDict): # 对话消息历史 messages: Annotated[List[HumanMessage | AIMessage], operator.add] # 是否需要搜索由LLM判断 needs_search: bool # 搜索结果 search_result: strmessages 使用Annotated和operator.add是LangGraph的一个约定表示这个字段是“追加”模式。每次节点返回的messages列表都会自动追加到历史中而不是覆盖。needs_search 一个布尔标志。search_result 存储搜索到的文本。3.2 创建节点Nodes我们需要三个节点router路由判断、search_node搜索、general_chat_node普通聊天。from langchain_openai import ChatOpenAI from langgraph.graph import END, StateGraph # 初始化LLM llm ChatOpenAI(modelgpt-3.5-turbo) def router(state: AgentState) - dict: 判断用户问题是否需要联网搜索 last_message state[“messages”][-1].content # 构建一个Prompt让LLM判断 prompt f”用户的问题是‘{last_message}’。这个问题需要实时或最新的网络信息才能准确回答吗只需要回答‘是’或‘否’。” response llm.invoke(prompt) decision response.content.strip() needs_search decision “是” return {“needs_search”: needs_search} def search_node(state: AgentState) - dict: 模拟搜索节点实际项目中替换为真实搜索API调用 query state[“messages”][-1].content # 这里模拟搜索过程 print(f”[搜索节点] 正在搜索: {query}”) # 模拟返回搜索结果 mock_result f”关于‘{query}’的模拟搜索结果当前最新版本是v2.0发布于2023年底。” return {“search_result”: mock_result, “messages”: [AIMessage(contentf”我已搜索到信息{mock_result}”)]} def general_chat_node(state: AgentState) - dict: 普通聊天节点直接回答 last_message state[“messages”][-1] response llm.invoke([last_message]) return {“messages”: [response]}3.3 构建图Graph并连接边Edges现在我们把节点组装起来并定义它们之间的流转逻辑。from langgraph.graph import StateGraph, END # 1. 创建图构建器并指定状态结构 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(“router”, router) workflow.add_node(“search”, search_node) workflow.add_node(“general_chat”, general_chat_node) # 3. 设置入口点所有流程都从router开始 workflow.set_entry_point(“router”) # 4. 添加条件边根据router的结果决定下一步 def decide_next_step(state: AgentState) - str: if state.get(“needs_search”): return “search” # 需要搜索则前往search节点 else: return “general_chat” # 否则直接聊天 workflow.add_conditional_edges( “router”, # 从哪个节点出发 decide_next_step, # 决定下一个节点的函数 {“search”: “search”, “general_chat”: “general_chat”} # 可能的目的地映射 ) # 5. 添加普通边search节点执行完后应进入general_chat节点来总结回答 workflow.add_edge(“search”, “general_chat”) # 6. 添加普通边general_chat节点执行完后工作流结束 workflow.add_edge(“general_chat”, END) # 7. 编译图得到可执行对象 app workflow.compile()3.4 运行与测试现在我们可以运行这个图了。# 初始化状态 initial_state AgentState( messages[HumanMessage(content“LangGraph的最新版本是什么”)], needs_searchFalse, search_result“” ) # 执行图 final_state app.invoke(initial_state) # 查看最终结果 for msg in final_state[“messages”]: print(f”{msg.type}: {msg.content}”)预期输出[搜索节点] 正在搜索: LangGraph的最新版本是什么 ai: 我已搜索到信息关于‘LangGraph的最新版本是什么’的模拟搜索结果当前最新版本是v2.0发布于2023年底。 ai: 这里会是general_chat节点基于搜索结果生成的最终友好回答这个简单的例子展示了LangGraph的核心流程状态流转、节点执行和条件分支。router节点修改了needs_search状态条件边根据这个状态值决定路径最终所有结果都汇聚到messages中。4. 构建高级智能体ReAct模式实战ReActReasoning Acting是智能体的经典模式它让LLM循环地进行“思考-行动-观察”。用LangGraph实现ReAct智能体非常直观。4.1 设计状态与工具我们将创建一个能使用计算器和搜索工具的智能体。from typing import TypedDict, Annotated, List, Union import operator from langchain_core.messages import BaseMessage from langchain.tools import tool from langchain_community.tools.tavily_search import TavilySearchResults # 定义工具 tool def calculator(expression: str) - str: “”“计算一个数学表达式。例如‘3 * 7 5’“”“ try: # 警告实际项目中请使用更安全的eval替代方案如ast.literal_eval或数学库 result eval(expression) return f”计算结果: {result}” except Exception as e: return f”计算错误: {e}” # 假设我们有一个搜索工具需要API Key # search_tool TavilySearchResults() # 为演示我们创建一个模拟工具 tool def web_search(query: str) - str: “”“在网络上搜索信息。”“” return f”关于‘{query}’的模拟搜索结果这是一个模拟的搜索结果摘要。” # 可用工具列表 tools [calculator, web_search] # 定义状态 class ReActState(TypedDict): messages: Annotated[List[BaseMessage], operator.add] # 用于记录LLM的“思考”过程 scratchpad: Annotated[List[BaseMessage], operator.add] # 当前轮次防止无限循环 step: int4.2 创建核心节点Agent和Tools在ReAct图中通常有两个主要节点agent负责思考并决定使用哪个工具和tools执行工具。from langchain.agents import create_react_agent from langchain_core.prompts import ChatPromptTemplate # 1. 创建Agent节点 def agent_node(state: ReActState): # 构建Prompt包含指令、工具描述、历史scratchpad和当前问题 prompt ChatPromptTemplate.from_messages([ (“system”, “你是一个有帮助的助手可以使用工具。请严格按以下格式回应\n思考你的推理过程\n行动要使用的工具名\n行动输入工具的输入”), (“placeholder”, “{chat_history}”), # 这里会填入scratchpad (“human”, “{input}”) ]) # 将工具绑定到LLM llm_with_tools llm.bind_tools(tools) chain prompt | llm_with_tools # 调用LLM agent_response chain.invoke({ “chat_history”: state[“scratchpad”], “input”: state[“messages”][-1].content }) # 将LLM的响应记录到scratchpad中 return { “scratchpad”: [agent_response], “step”: state[“step”] 1 } # 2. 创建Tools节点 def tools_node(state: ReActState): # 从scratchpad中取出最新的LLM响应 last_message state[“scratchpad”][-1] # 解析LLM响应提取“行动”和“行动输入” # 这里简化处理实际应使用更健壮的解析器 content last_message.content action_line [l for l in content.split(‘\n’) if l.startswith(‘行动’)][0] action_input_line [l for l in content.split(‘\n’) if l.startswith(‘行动输入’)][0] tool_name action_line.replace(‘行动’, ‘’).strip() tool_input action_input_line.replace(‘行动输入’, ‘’).strip() # 找到对应的工具并执行 tool_to_use next((t for t in tools if t.name tool_name), None) if tool_to_use: tool_result tool_to_use.invoke(tool_input) result_msg AIMessage(contentf”观察{tool_result}”) else: result_msg AIMessage(contentf”观察工具‘{tool_name}’未找到。”) # 将工具执行结果观察记录到scratchpad return {“scratchpad”: [result_msg]}4.3 构建并运行ReAct图现在构建一个循环让Agent和Tools交替执行直到LLM决定结束。from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver # 创建图 workflow StateGraph(ReActState) workflow.add_node(“agent”, agent_node) workflow.add_node(“tools”, tools_node) workflow.set_entry_point(“agent”) # 定义边逻辑Agent之后总是去Tools workflow.add_edge(“agent”, “tools”) # 关键从Tools节点出来的条件边决定是继续循环还是结束 def should_continue(state: ReActState) - str: last_message state[“scratchpad”][-1] # 如果LLM在思考中说了“最终答案”或者步骤超过5步则结束 if “最终答案” in last_message.content or state[“step”] 5: return “end” else: return “continue” workflow.add_conditional_edges( “tools”, should_continue, {“continue”: “agent”, “end”: END} # 继续循环回Agent或结束 ) # 编译图 app workflow.compile()4.4 测试ReAct智能体# 使用检查点存储器便于调试和观察中间状态 memory MemorySaver() app_with_memory workflow.compile(checkpointermemory) config {“configurable”: {“thread_id”: “test_react_1”}} initial_state ReActState( messages[HumanMessage(content“请先计算15的平方然后搜索一下LangGraph的作者信息。”)], scratchpad[], step0 ) # 以流式方式执行可以看到每一步的输出 for event in app_with_memory.stream(initial_state, config, stream_mode“values”): event[“scratchpad”][-1].pretty_print()这个ReAct智能体会先调用计算器工具得到225然后调用搜索工具获取信息最后LLM综合两者给出最终答案。通过scratchpad我们可以清晰地看到LLM“思考-行动-观察”的完整链条。5. 高级特性与工程化实践掌握了基础构建后我们来看一些让LangGraph应用更健壮、更强大的高级特性。5.1 持久化与检查点Checkpoints对于长时间运行或需要中断恢复的智能体检查点功能至关重要。它能在每个节点执行后自动保存完整状态。from langgraph.checkpoint.sqlite import SqliteSaver import tempfile import os # 使用SQLite持久化检查点 db_file tempfile.NamedTemporaryFile(deleteFalse).name checkpointer SqliteSaver.from_conn_string(f”sqlite:///{db_file}”) # 在编译图时传入checkpointer persistent_app workflow.compile(checkpointercheckpointer) config {“configurable”: {“thread_id”: “user_session_123”}} try: # 第一次执行 state1 persistent_app.invoke(initial_state, config) print(“第一次执行完成。”) # 模拟应用重启后可以从检查点恢复继续执行 # 通过指定config中的相同thread_id可以加载历史状态 # 这对于实现多轮对话、异步任务恢复非常有用 finally: os.unlink(db_file) # 清理临时文件5.2 人工干预节点Human-in-the-Loop在某些关键业务流程中需要引入人工审核。LangGraph可以轻松地将“等待人工输入”作为一个节点。def human_review_node(state: AgentState): 模拟人工审核节点。在实际应用中这里会连接到一个Web界面或通知系统。 task_description state[“messages”][-1].content print(f”\n 需要人工审核 \n任务{task_description}\n”) # 在实际系统中这里会阻塞等待用户输入。 # 为演示我们模拟用户输入“批准” human_decision “批准” # 模拟输入 return {“human_decision”: human_decision} # 在图中加入人工审核节点并设置条件边只有审核通过才进入下一步。5.3 子图Subgraphs与模块化复杂的应用可以拆分成多个子图提高可维护性和复用性。子图本身也是一个StateGraph可以被主图当作一个节点来调用。# 定义一个处理“数据查询”的子图 subgraph_builder StateGraph(AgentState) # ... 构建子图内部的节点和边 ... subgraph subgraph_builder.compile() # 在主图中可以将子图作为一个节点添加 main_workflow.add_node(“data_query_subgraph”, subgraph)这种方式非常适合微服务架构每个团队可以负责开发一个独立的子图。6. 常见问题与调试技巧在开发LangGraph应用时你可能会遇到以下典型问题。6.1 状态State更新不符合预期这是最常见的问题。请牢记追加Add vs 覆盖Replace 使用Annotated[List[...], operator.add]的字段是追加否则是覆盖。确保你的节点返回的字典键值与State定义匹配且更新逻辑符合预期。调试方法 在每个节点函数开始和结束时打印State。或者使用LangGraph的stream模式观察状态变化。6.2 图陷入无限循环智能体可能无法自行终止。解决方案设置最大步数 在State中设置一个step计数器在条件边判断函数中检查是否超过阈值。优化Prompt 明确指示LLM在何时输出“最终答案”等终止标记。使用超时机制 在图编译或调用时设置超时。6.3 工具调用解析失败LLM可能不按你指定的格式返回工具调用信息。使用LangChain的bind_tools和with_structured_output 它们能强制LLM返回结构化的工具调用对象比解析文本更可靠。在Prompt中提供更清晰的示例 在系统提示词中给出2-3个完美的响应格式示例。6.4 性能优化对于复杂图性能可能成为瓶颈。异步执行 如果节点间没有严格的先后依赖可以考虑使用add_node的异步版本并利用asyncio。缓存 对于纯函数节点或昂贵的LLM调用可以考虑引入缓存机制。精简State 只保留必要的数据在State中避免传递过大的对象。7. 最佳实践与项目架构建议将LangGraph用于实际项目时遵循以下实践能避免很多坑。7.1 状态设计原则最小化 State只存储工作流真正需要流转和共享的数据。中间计算结果尽量放在节点局部变量中。明确类型 坚持使用TypedDict或Pydantic模型来定义State这能极大提高代码可读性和减少运行时错误。区分会话与流程 将会话ID、用户ID等元数据放在config中而不是State里。State应专注于业务流程数据。7.2 节点设计原则单一职责 每个节点只做一件事。一个节点要么调用LLM要么调用一个工具要么处理数据。幂等性 尽可能让节点函数是幂等的即相同输入产生相同输出这有利于调试和重试。错误处理 节点内部应做好异常捕获并选择是将错误信息放入State向下传递还是直接抛出异常终止图。7.3 图的版本管理与测试版本化 当图的结构节点、边发生变化时应视为一次版本升级。考虑如何平滑迁移正在运行中的旧图实例的状态。单元测试 为每个节点函数编写单元测试模拟输入State验证输出State。集成测试 为整个图编写集成测试使用典型输入验证最终输出和状态变化。7.4 生产环境部署可观测性 在每个节点的入口和出口记录日志包括State的快照。这对于排查生产问题至关重要。监控与告警 监控图的执行时长、错误率、循环次数等指标。配置化 将LLM的模型名称、API Key、工具开关等配置外置便于不同环境开发、测试、生产的切换。LangGraph代表了一种构建复杂AI应用的新范式它将流程控制与AI能力解耦通过声明式的“图”来管理复杂度。从简单的条件分支对话助手到拥有复杂推理循环的ReAct智能体其核心思想始终是状态驱动和流程可视化。学习LangGraph的最佳方式就是从一个具体需求开始画出你的工作流程图然后将其转化为节点和边。随着你对状态管理和条件流转越来越熟悉你会发现构建强大、可靠且可维护的AI应用不再是一件令人畏惧的事情。