LangGraph实战：构建有状态AI工作流与复杂Agent编排-北京尧图网络

这类工具最值得先看的不是功能列表而是能不能在普通环境里稳定跑起来。LangGraph 本质上是一个用于构建有状态、多步骤 AI 应用的工作流框架它把复杂的 AI 代理Agent逻辑比如对话、决策、工具调用变成了可视化的“图”结构。如果你正在用 LangChain 做 AI 应用感觉链条太长、状态管理混乱或者想实现带循环、分支的复杂 Agent那 LangGraph 就是解决这些问题的关键工具。它最核心的价值不是替代 LangChain而是补上了 LangChain 在编排复杂、有状态工作流时的短板。很多人一上来就纠结两者的区别其实更该关心的是自己的项目需不需要“记忆”、需不需要“循环判断”、需不需要清晰的“状态流转”。如果需要那从 LangGraph 入门会比直接硬啃 LangChain 的复杂链条更直观。我建议先从最小样例开始跑通一个带状态循环的简单 Agent再去看官方那些复杂的案例。下面按实际落地顺序拆一遍。1. 先理解 LangGraph 到底解决了什么编排问题在接触具体代码之前如果没搞清楚 LangGraph 的定位很容易把它当成另一个 LangChain 来学结果越学越迷糊。它的出现是为了应对 AI 应用开发中的一个具体痛点如何清晰、可维护地编排那些带有状态、循环、分支和工具调用的多步骤 AI 逻辑。1.1 从 LangChain 的链条到 LangGraph 的图用 LangChain 时我们习惯把流程写成一条“链”Chain输入 - A 处理 - B 处理 - 输出。这对于线性任务很友好。但当你需要做一个能自主思考、根据中间结果决定下一步做什么的 AI 代理时问题就来了。比如一个客服 Agent它需要1. 理解用户问题2. 决定查知识库还是问用户澄清3. 根据查询结果决定是直接回答还是继续追问。这个过程可能有循环状态比如已经问过的问题、查询到的信息需要在多个步骤间传递。用传统的链式思维来写代码会变得充满if-else和状态管理难以阅读和维护。而 LangGraph 引入了“图”Graph的概念把每个步骤节点和步骤之间的流转条件边定义清楚。整个工作流变成一个可以清晰可视化的流程图状态沿着图的边在节点间流动。1.2 “状态”是 LangGraph 的灵魂这是理解 LangGraph 最关键的一点。它的核心是一个State对象这个对象是一个字典在整个图的执行过程中流转和更新。每个节点一个处理函数读取当前状态进行计算或调用 AI然后修改状态中的某些字段。图的下一个节点基于更新后的状态决定是否执行、如何执行。例如你的状态可能包含messages: 对话历史列表。next: 指示下一步该执行哪个节点。knowledge: 从外部工具获取的知识。count: 循环计数器防止无限循环。这种显式的状态管理让复杂逻辑的数据流变得一目了然调试时也能清楚地看到每一步之后状态变成了什么样。1.3 它和 LangChain 不是二选一而是配合使用这是最常见的误解。LangGraph 不是 LangChain 的替代品它们是互补的。你可以使用 LangChain 来构建底层的、功能单一的“链”比如一个专门做摘要的链一个专门调用某 API 的工具。然后使用 LangGraph 作为“总指挥”把这些 LangChain 构建好的链和工具作为“节点”组装起来管理它们之间的调用顺序、循环和状态传递。所以学习 LangGraph 的前提通常是你已经对 LangChain 的基本概念如 LLM、Prompt、Chain、Tool有了解。如果你完全是新手我建议先花一点时间理解 LangChain 的核心抽象再来看 LangGraph会顺畅很多。2. 搭建环境与跑通第一个“循环对话”Agent理论说再多不如跑一行代码。LangGraph 的环境搭建很简单但有几个版本依赖的坑需要注意。2.1 基础环境准备与安装首先确保你有一个 Python 环境3.8 以上。我强烈建议使用虚拟环境venv 或 conda来管理依赖避免包冲突。# 创建并激活虚拟环境以 venv 为例 python -m venv langgraph-env source langgraph-env/bin/activate # Linux/macOS # 或 langgraph-env\Scripts\activate # Windows # 核心安装 pip install langgraph langchain-openai这里注意langgraph是核心框架。langchain-openai包含了 LangChain 对 OpenAI API 的集成。我们用它来提供最基础的 LLM 调用能力。你也可以安装langchain-anthropic等对应其他模型的包。如果你需要用到 LangChain 的更多功能比如文本加载器、向量库可以安装langchain元包但通常langchain-core和langchain-openai就够了。安装完成后别忘了设置你的 OpenAI API Key或其他模型的 Key作为环境变量export OPENAI_API_KEYyour-api-key-here # Linux/macOS # 或在代码中设置 os.environ[“OPENAI_API_KEY”] ‘your-key’2.2 构建一个最简单的“循环问答”图我们来创建一个 Agent它模拟一个有点“固执”的助手你问它问题它总是先尝试回答如果你说“不对”它就会尝试换一种方式再答一次直到你说“对了”或达到最大循环次数。from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI # 1. 定义状态结构 class AgentState(TypedDict): # 对话消息历史 messages: Annotated[list, operator.add] # 用户最后一次输入 last_user_input: str # 尝试次数 attempt_count: int # 2. 初始化 LLM llm ChatOpenAI(model“gpt-3.5-turbo”) # 3. 定义节点函数 def generate_response(state: AgentState): 节点根据历史消息生成助手的回复 history state[“messages”] # 构造一个简单的提示 prompt f“基于之前的对话请给出一个友好的回答。当前用户最后说{state[‘last_user_input’]}” if state[“attempt_count”] 0: prompt f“ (这是第{state[‘attempt_count’]}次尝试)” response llm.invoke(prompt) # 将助手回复添加到消息历史 new_message {“role”: “assistant”, “content”: response.content} return {“messages”: [new_message], “last_user_input”: state[“last_user_input”]} def check_feedback(state: AgentState): 节点检查用户反馈决定下一步 last_user_msg state[“last_user_input”].lower() # 简单的关键词判断 if “对了” in last_user_msg or “谢谢” in last_user_msg: # 用户满意结束 return “end” elif state[“attempt_count”] 3: # 尝试次数太多结束 return “end” else: # 用户可能说“不对”、“再想想”需要重新生成 # 更新尝试次数 new_state {“attempt_count”: state[“attempt_count”] 1} # 返回下一个节点名和要更新的状态 return (“generate_response”, new_state) # 4. 构建图 builder StateGraph(AgentState) # 添加节点 builder.add_node(“generate_response”, generate_response) builder.add_node(“check_feedback”, check_feedback) # 设置入口点 builder.set_entry_point(“generate_response”) # 添加边生成回复后去检查反馈 builder.add_edge(“generate_response”, “check_feedback”) # 添加条件边检查反馈后根据返回值决定去哪 builder.add_conditional_edges( “check_feedback”, check_feedback, # 这个函数返回下一个节点名或 “end” {“generate_response”: “generate_response”, “end”: END} ) # 5. 编译图 graph builder.compile() # 6. 运行图 initial_state { “messages”: [], “last_user_input”: “什么是机器学习”, “attempt_count”: 0 } result graph.invoke(initial_state) print(“最终消息历史:”, result[“messages”])这个例子包含了 LangGraph 的核心要素状态定义使用TypedDict明确状态有哪些字段。Annotated[list, operator.add]是一个魔法表示对messages列表执行追加操作而不是覆盖。节点每个节点是一个普通函数接收状态返回要更新的状态部分。边定义了节点执行的顺序。条件边add_conditional_edges允许根据一个函数的返回值动态决定下一个节点这是实现循环和分支的关键。编译与执行图需要先编译然后通过invoke传入初始状态来运行。跑通这个例子你就完成了从 0 到 1 最关键的一步理解状态如何流动以及图如何被驱动执行。2.3 使用 LangGraph Studio 可视化你的图LangGraph 提供了一个强大的可视化工具——LangGraph Studio。它能将你编译好的图自动生成一个交互式界面让你能看到状态流、执行路径并且能逐步调试。安装和启动非常简单pip install langgraph-cli langgraph dev然后在浏览器打开http://localhost:2024。你可以将上面代码中编译得到的graph对象提供给 Studio它会以流程图形式展示出来。这对于调试复杂工作流、向他人解释你的 Agent 逻辑有巨大帮助。我建议在开发任何稍复杂的图时都打开 Studio 边改边看能极大提升效率避免在代码里迷失方向。3. 深入核心概念状态、节点、边与工具集成跑通 Hello World 后需要系统理解这几个核心概念才能设计出实用的 Agent。3.1 状态State的设计模式状态设计是 LangGraph 应用好坏的关键。除了上面例子里的简单字典还有几种常见模式1. 共享状态与私有状态有些信息是所有节点都需要读写的如messages对话历史这适合放在顶层状态。有些信息是某个节点计算产生的中间结果只给后续少数节点使用可以考虑设计成子状态或者通过明确的字段名来区分。2. 状态缩减器Reducer上面例子中Annotated[list, operator.add]里的operator.add就是一个“缩减器”。它定义了当多个节点都返回对同一个字段如messages的更新时如何合并这些更新。除了add追加还有operator.setitem: 替换字典中的项。自定义函数实现更复杂的合并逻辑。3. 状态验证与初始化可以在状态类中使用__init__或property来设置默认值或进行简单验证确保状态结构一致。3.2 节点Node函数的最佳实践节点函数应该保持“单一职责”。一个节点最好只做一件事要么调用一次 LLM要么调用一个工具要么做一次逻辑判断。输入与输出输入总是接收一个完整的State对象。输出返回一个字典这个字典的键必须是State中定义的字段的子集值是将要用于更新该字段的数据。LangGraph 会根据你定义的状态缩减器来合并更新。错误处理节点函数内部应该有try...except来处理可能出现的异常如网络超时、API 限额。一种模式是捕获异常后在状态中设置一个error字段并让条件边路由到一个专门的“错误处理”节点。3.3 边Edges与条件路由边决定了工作流的走向。固定边add_edge(“node_a”, “node_b”)表示node_a执行完后无条件执行node_b。条件边add_conditional_edges是 LangGraph 的精华。它需要一个“路由函数”这个函数接收当前状态返回下一个节点的名称字符串或者返回一个包含(next_node, state_updates)的元组。这让你能实现if-else、switch-case甚至循环。入口与终点set_entry_point指定开始节点。END是一个特殊节点表示图执行结束。3.4 集成 LangChain Tools真正的 Agent 能力在于使用工具。LangGraph 可以无缝集成 LangChain 的Tool对象。from langchain.tools import Tool from langchain_community.utilities import SerpAPIWrapper # 定义一个搜索工具 search SerpAPIWrapper() search_tool Tool( name“Search”, funcsearch.run, description“useful for when you need to answer questions about current events” ) # 在节点函数中使用工具 def search_node(state: State): query state[“query”] # 调用工具 result search_tool.run(query) return {“search_result”: result, “query”: query} # 将节点添加到图中 builder.add_node(“search”, search_node)你可以设计一个“工具调用”节点它根据状态中的intent字段决定调用哪个工具然后将工具执行结果写回状态。另一个“结果合成”节点再根据原始问题和工具结果调用 LLM 生成最终回答。这就是一个典型 ReAct 模式 Agent 的 LangGraph 实现。4. 构建实战 AI Agent带记忆与工具调用的客服助手现在我们把所有概念组合起来构建一个更接近真实场景的客服助手 Agent。这个 Agent 需要1. 理解用户意图2. 决定使用内部知识库还是联网搜索3. 维护对话历史记忆4. 在无法回答时礼貌地请求澄清。4.1 定义完整状态与工具from typing import TypedDict, Annotated, Literal import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain.tools import Tool from langchain_community.utilities import SerpAPIWrapper from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.document_loaders import TextLoader # 假设我们有一个简单的知识库文本文件 “knowledge.txt” loader TextLoader(“knowledge.txt”) documents loader.load() text_splitter RecursiveCharacterTextSplitter(chunk_size500, chunk_overlap50) docs text_splitter.split_documents(documents) embeddings OpenAIEmbeddings() vectorstore Chroma.from_documents(docs, embeddings) retriever vectorstore.as_retriever() # 定义工具 search SerpAPIWrapper() tools [ Tool(name“KnowledgeBase”, funcretriever.get_relevant_documents, description“查询内部知识库以获得产品、政策等信息。”), Tool(name“WebSearch”, funcsearch.run, description“搜索互联网获取最新、实时的信息。”) ] # 定义状态 class CustomerSupportState(TypedDict): # 对话历史 messages: Annotated[list, operator.add] # 当前用户查询 current_query: str # 从工具获取的信息 retrieved_info: list # Agent 的下一步决策 next_action: Literal[“answer”, “clarify”, “search_web”, “search_kb”] # 需要向用户澄清的内容 clarification_needed: str4.2 实现节点函数llm ChatOpenAI(model“gpt-4”, temperature0) def classify_intent(state: CustomerSupportState): 节点1分类用户意图决定下一步行动 history_text “\n”.join([f“{m[‘role’]}: {m[‘content’]}” for m in state[“messages”][-5:]]) # 取最近5轮历史 prompt f“”” 你是一个客服助手。根据对话历史和当前问题决定下一步做什么。历史对话 {history_text} 当前用户问题{state[‘current_query’]} 请从以下选项中选择 - ‘answer’: 如果你已经有足够信息直接回答。 - ‘clarify’: 如果问题模糊需要用户澄清。 - ‘search_kb’: 如果问题关于产品、政策等内部知识需要查询知识库。 - ‘search_web’: 如果问题需要最新、实时的外部信息。只输出选择项不要输出其他文字。 “”” decision llm.invoke(prompt).content.strip().lower() # 简单清理输出确保是四个选项之一 if decision not in [“answer”, “clarify”, “search_kb”, “search_web”]: decision “clarify” # 默认请求澄清 return {“next_action”: decision} def retrieve_from_kb(state: CustomerSupportState): 节点2查询知识库 docs tools[0].func(state[“current_query”]) info [doc.page_content for doc in docs][:3] # 取前3条 return {“retrieved_info”: info} def search_the_web(state: CustomerSupportState): 节点3联网搜索 result tools[1].func(state[“current_query”]) return {“retrieved_info”: [result]} def ask_for_clarification(state: CustomerSupportState): 节点4生成澄清问题 prompt f“用户的问题 ‘{state[‘current_query’]}’ 比较模糊。请生成一个礼貌、具体的问题来请求用户澄清。” clarification llm.invoke(prompt).content return {“clarification_needed”: clarification, “next_action”: “end_for_now”} def generate_final_answer(state: CustomerSupportState): 节点5合成最终答案 context “” if state[“retrieved_info”]: context “\n”.join([“参考信息”] state[“retrieved_info”]) history_text “\n”.join([f“{m[‘role’]}: {m[‘content’]}” for m in state[“messages”][-3:]]) prompt f“”” 你是一个专业的客服助手。对话历史 {history_text} 当前问题{state[‘current_query’]} {context} 请生成友好、准确、简洁的最终回答。 “”” answer llm.invoke(prompt).content new_message {“role”: “assistant”, “content”: answer} return {“messages”: [new_message], “retrieved_info”: [], “next_action”: “end”}4.3 组装工作流图builder StateGraph(CustomerSupportState) # 添加所有节点 builder.add_node(“classify_intent”, classify_intent) builder.add_node(“retrieve_kb”, retrieve_from_kb) builder.add_node(“search_web”, search_the_web) builder.add_node(“ask_clarify”, ask_for_clarification) builder.add_node(“generate_answer”, generate_final_answer) # 设置入口点 builder.set_entry_point(“classify_intent”) # 定义条件路由函数 def route_after_intent(state: CustomerSupportState): action state[“next_action”] if action “search_kb”: return “retrieve_kb” elif action “search_web”: return “search_web” elif action “clarify”: return “ask_clarify” elif action “answer”: return “generate_answer” else: return “ask_clarify” # 兜底 # 从意图分类节点出发根据决策路由 builder.add_conditional_edges( “classify_intent”, route_after_intent ) # 定义工具调用后的路由都去生成答案 builder.add_edge(“retrieve_kb”, “generate_answer”) builder.add_edge(“search_web”, “generate_answer”) # 澄清节点执行后本轮结束等待用户下次输入 builder.add_edge(“ask_clarify”, END) # 生成答案后本轮结束 builder.add_edge(“generate_answer”, END) # 编译图 support_agent builder.compile()4.4 运行与迭代现在你可以运行这个 Agent 了# 初始化状态 init_state { “messages”: [{“role”: “user”, “content”: “你们的退货政策是什么”}], “current_query”: “你们的退货政策是什么”, “retrieved_info”: [], “next_action”: “”, “clarification_needed”: “” } # 执行一轮 result support_agent.invoke(init_state) print(“助手回复:”, result[“messages”][-1][“content”] if result[“messages”] else “等待澄清”)这个 Agent 已经具备了基本的决策、工具调用和记忆能力。你可以通过 LangGraph Studio 可视化它看到用户问题如何触发classify_intent然后根据决策路由到不同的工具节点最后合成回答。5. 生产化考量持久化、并发、监控与调试当你的 LangGraph Agent 从 Demo 走向实际应用时有几个关键点必须提前规划。5.1 状态的持久化与加载在 Web 服务或长时间运行的进程中你需要将会话状态State保存到数据库如 Redis、PostgreSQL或文件系统中以便在服务重启或处理后续请求时能够恢复。LangGraph 的State本质上是字典可以方便地序列化为 JSON。你需要做的是为每个会话创建唯一 ID。在图的checkpoint处保存状态。你可以利用条件边或特定节点在状态更新后将其保存。在请求开始时加载状态。一种简单模式是创建一个“持久化节点”和一个“加载节点”或者使用 LangGraph 提供的Pregel底层 API 来配置持久化后端。5.2 处理并发请求LangGraph 图本身是静态的、可重入的。这意味着多个请求可以共享同一个编译好的graph对象它们是线程安全的。瓶颈通常在于LLM API 的速率限制你需要为你的 LLM 提供商如 OpenAI设置合理的重试、退避和并发控制策略。可以考虑使用langchain的RunnableConfig进行配置或使用像tenacity这样的重试库。工具调用的 I/O如果工具涉及网络请求如搜索、数据库查询确保它们有超时设置并且考虑使用异步asyncio来提高吞吐量。对于高并发场景建议将 LangGraph Agent 封装为异步服务如使用 FastAPI并利用异步版本的 LLM 和工具客户端。5.3 日志、监控与追踪调试一个复杂的、有状态的图完善的日志至关重要。在每个节点函数的开始和结束记录日志包括节点名、输入状态的关键字段、输出状态的关键字段、耗时、任何错误。使用 LangSmith这是 LangChain 官方的追踪平台。它与 LangGraph 深度集成。只需设置环境变量LANGCHAIN_TRACING_V2true和LANGCHAIN_API_KEY你的图执行过程会被自动记录。你可以在 LangSmith UI 上看到每个节点的输入输出、耗时、token 使用量这对于性能分析和调试是无价之宝。自定义监控指标记录每个请求的总体耗时、每个节点的平均耗时、工具调用成功率、最终用户满意度如果可获取等。5.4 图的版本管理与测试随着业务逻辑变化你的图可能会迭代。你需要管理不同版本的图。将图的构建代码纳入版本控制如 Git。考虑图的序列化graph对象可以通过pickle序列化但要注意其中包含的函数和对象引用。更稳健的做法是保存生成图的源代码和配置在部署时重新编译。编写单元测试为每个节点函数编写测试模拟不同的输入状态验证输出状态是否符合预期。为整个图编写集成测试模拟完整的用户会话。进行“冒烟测试”部署新版本后用一组固定的测试用例快速验证核心功能是否正常。6. 常见陷阱、性能优化与进阶模式即使理解了基本概念在实际开发中还是会遇到一些坑。这里总结几个高频问题和优化思路。6.1 状态设计过重或过轻陷阱把所有数据都塞进顶层状态导致状态对象庞大在每个节点间传递效率低且逻辑耦合度高。优化遵循“最小化共享状态”原则。只将真正需要在多个节点间共享的数据放在顶层状态。对于中间计算结果可以考虑设计嵌套状态子字典。让节点函数返回计算结果并通过边传递给下一个需要它的节点作为参数或通过状态中的特定字段。6.2 条件路由函数过于复杂陷阱route_after_intent这样的路由函数里写了大量逻辑甚至在里面调用 LLM使得路由逻辑难以测试和理解。优化路由函数应该尽量简单、纯粹。它最好只做基于当前状态的简单判断。如果决策需要复杂推理如调用 LLM应该将其设计成一个独立的“决策节点”该节点更新状态中的一个字段如next_action然后由简单的路由函数读取这个字段来决定下一跳。6.3 无限循环或僵死陷阱Agent 陷入“思考-行动-再思考”的无限循环或者因为某个条件永远无法满足而卡住。解决方案设置最大步数在状态中引入step_count字段每次循环递增。在路由函数中检查如果超过阈值则强制路由到END或一个“失败处理”节点。超时控制在图级别或调用级别设置超时。看门狗Watchdog可以设计一个全局的监控机制如果发现同一状态在短时间内被重复处理则介入中断。6.4 LLM 调用成本与延迟优化缓存对相同的提示词进行缓存。可以使用langchain的CacheBacked或集成外部缓存如 Redis。批处理如果多个节点都需要调用 LLM且输入相互独立可以考虑将这些调用合并为一个批处理请求如果 LLM API 支持。模型选择用小模型如 GPT-3.5-turbo处理简单的分类、路由任务用大模型如 GPT-4处理需要深度推理或合成的任务。流式输出对于需要长时间运行的图考虑支持流式输出让用户能尽早看到部分结果。6.5 进阶模式子图与分层设计对于极其复杂的工作流可以考虑使用子图Subgraph。将一个大的图分解成几个逻辑上独立的子图每个子图解决一个子问题。主图通过调用子图来组织整体流程。这有助于代码的模块化和复用。LangGraph 支持将编译好的图作为一个节点添加到另一个图中。这为构建分层、可复用的 Agent 系统提供了强大的支持。我个人更建议先把单任务跑稳再考虑批量和接口。这个方案真正落地时最该盯住的不是功能列表而是输入格式、资源占用和失败重试。踩过几次之后我发现很多问题不是工具能力不够而是前置环境和输入材料没有处理干净。从 LangGraph 的图思维出发把业务逻辑拆解成清晰的状态和节点本身就是一个极好的设计练习它能迫使你思考数据流和决策点这比直接写一堆杂乱无章的if-else要有价值得多。

LangGraph实战：构建有状态AI工作流与复杂Agent编排

相关推荐

利用CRC实现单比特错误纠正：原理、实现与嵌入式应用

Python招聘数据爬虫可视化系统开发全解析

零基础用pkNX改宝可梦Switch版：从解包ROM到生成补丁的完整上手路线

更多资讯

解决Windows下Hyper-V与VirtualBox冲突的完整指南

强化学习与最优控制：一本书讲透序列决策的来龙去脉

Octop不是WorkBuddy替代品，而是AI Agent工业化施工套件

Ubuntu 24.04 x11vnc远程桌面：Wayland切换与稳定托管

Redis大key与热key排查优化实战指南

AI音乐创作：多模态生成式AI技术实践与优化

今日更新

基于YOLOv11的石油泄漏智能检测系统开发实践

AI行业变局：合规、Mod化与上下文管理的实战启示

计算机视觉与AI原生应用融合的技术实践

本周精选

防幻觉智能体架构：电源硬件设计中的大模型编排与工具化实践

OpenGL面试核心考点：渲染管线、矩阵变换与性能排查全解析

全国计算机二级C语言真题PDF高效刷题指南

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？