1. 先搞清楚 LangGraph 到底解决了什么核心问题如果你正在用 LangChain 构建复杂的 AI 应用比如需要多步骤决策、状态流转或者多个 AI 模型智能体协同工作的场景那你大概率会遇到一个头疼的问题流程控制。用 LangChain 的SequentialChain或者简单的函数调用处理一个简单的问答流程还行一旦任务流程变长、出现分支判断、需要循环执行或者需要让多个智能体比如一个负责检索、一个负责分析、一个负责生成接力或协作代码很快就会变得难以维护。这就是 LangGraph 要解决的核心问题。它不是一个替代品而是 LangChain 生态中的一个工作流编排框架。你可以把它理解为一个专门为 AI 应用设计的、可视化的流程图工具。它最核心的价值是引入了StateGraph状态图的概念让你能清晰地定义应用的状态State、节点Nodes代表一个执行步骤比如调用一次 LLM 或工具、边Edges决定下一步走向哪个节点以及循环Cycles用于多轮对话或迭代优化。所以这篇文章适合两类人看一是已经用过 LangChain但觉得流程控制太乱的开发者二是刚开始设计多智能体、复杂决策链应用想找一个更工程化方案的工程师。最值得关注的不是 LangGraph 的语法而是它如何通过StateGraph把一堆散乱的回调、条件判断和循环变成一个可定义、可调试、可持久化的“工作流蓝图”。这直接决定了你的 Agent 能否从玩具 demo 升级为工业级应用。2. 环境准备与核心概念对齐别急着写代码在动手之前先确保你的环境和认知对齐。LangGraph 不是独立运行的它构建在 LangChain 之上所以你的 Python 环境里需要同时安装它们。我建议先创建一个干净的虚拟环境避免依赖冲突。# 创建并激活虚拟环境以 conda 为例 conda create -n langgraph-demo python3.10 conda activate langgraph-demo # 安装核心依赖 pip install langchain langgraph langchain-openai这里注意langchain-openai是 LangChain 官方维护的 OpenAI 集成包比直接用openai库更方便。如果你要用其他模型比如智谱、DeepSeek就安装对应的langchain-*包。不要一上来就装一大堆先确保核心的能跑通。安装完后别急着跑复杂例子。先理解 LangGraph 里三个最重要的对象这能帮你省下大量后期调试的时间State状态这是一个贯穿整个工作流的共享数据字典。你所有节点读取和修改的都是它。比如一个 RAG 问答工作流的状态里可能包含question用户问题、retrieved_docs检索到的文档、answer最终答案。在 LangGraph 里State 通常用一个 TypedDict 或 Pydantic 模型来定义这能提供类型提示减少错误。Node节点一个普通的 Python 函数它接收当前 State执行一些操作比如调用 LLM、查询数据库然后返回一个包含对 State 修改的字典。一个节点只做一件事这是保持工作流清晰的关键。Edge边决定工作流从一个节点执行完后下一个该去哪个节点。边可以是固定的always也可以是根据 State 里的某个值动态决定的conditional。很多人一开始会混淆 Node 和 Chain。记住Node 更原子化它可能是 Chain 的一部分也可能就是一个工具调用。而 LangGraph 的工作流就是由这些 Node 和 Edge 编织成的网。3. 从零构建一个带状态管控的 RAG 问答工作流理论说再多不如跑一遍。我们从一个最经典的场景入手RAG检索增强生成问答。但我们要做得比简单调用RetrievalQA更精细、更可控。假设我们的需求是用户提问后先检索相关文档如果检索结果足够相关就直接生成答案如果相关性不够则让 LLM 先澄清问题再基于澄清后的问题重新检索。这个流程就涉及了状态流转和条件判断正好用 LangGraph 来建模。3.1 第一步定义 State 和初始化节点首先我们定义工作流的状态。这里我们需要记录问题、检索到的文档、答案以及一个判断是否需要澄清的标记。from typing import TypedDict, List, Annotated import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings from langchain.schema import Document import os # 1. 定义状态结构 class GraphState(TypedDict): question: str # 原始问题 clarified_question: str # 澄清后的问题可能和原始问题相同 retrieved_docs: List[Document] # 检索到的文档列表 answer: str # 最终答案 needs_clarification: bool # 是否需要澄清问题 # 2. 初始化一些组件这里用伪代码你需要替换成自己的 os.environ[“OPENAI_API_KEY”] “your-api-key” llm ChatOpenAI(model“gpt-4o-mini”, temperature0) embeddings OpenAIEmbeddings() # 假设你已经有一个加载好的向量数据库 vectorstore Chroma(persist_directory“./my_db”, embedding_functionembeddings) retriever vectorstore.as_retriever(search_kwargs{“k”: 4}) # 每次检索4条3.2 第二步实现各个功能节点接下来我们把流程中的每个步骤实现为一个节点函数。节点1检索文档这个节点接收 State用其中的问题可能是原始问题也可能是澄清后的问题去检索并将结果存回 State。def retrieve(state: GraphState): print(“---执行检索---”) question state.get(“clarified_question”) or state[“question”] docs retriever.invoke(question) return {“retrieved_docs”: docs}节点2判断相关性/是否需要澄清这是一个决策节点。它检查检索到的文档是否足够相关。这里用一个简单的规则如果检索到的文档中最相关的那个分数假设检索器返回分数低于某个阈值则认为需要澄清。def grade_documents(state: GraphState): print(“---评估文档相关性---”) docs state[“retrieved_docs”] # 假设我们使用的检索器返回的 Document 有 metadata 包含分数 # 这里简化处理如果没分数或者平均分数低就认为需要澄清 needs_clarification False if docs: # 这里只是一个示例逻辑实际中你可能需要更复杂的评估器甚至用另一个LLM # 例如检查第一条文档的元数据分数 if hasattr(docs[0], ‘metadata’) and docs[0].metadata.get(‘score’, 1) 0.7: needs_clarification True else: needs_clarification True # 没检索到任何文档更需要澄清 return {“needs_clarification”: needs_clarification}节点3澄清问题如果判定需要澄清这个节点会调用 LLM让它根据原始问题和检索到的文档可能不相关生成一个更清晰的问题。def clarify_question(state: GraphState): print(“---澄清问题---”) question state[“question”] docs_preview “\n”.join([doc.page_content[:200] for doc in state[“retrieved_docs”][:2]]) # 预览前两条 prompt f””” 用户的原问题是{question} 我们根据这个问题检索到了一些文档但它们可能不太相关 {docs_preview} 请你帮我将用户的问题重新表述一下使其更清晰、更具体更容易从知识库中找到答案。 请直接输出重新表述后的问题不要加任何解释。 “”” response llm.invoke(prompt) clarified_q response.content.strip() return {“clarified_question”: clarified_q}节点4生成最终答案当文档相关性足够时调用 LLM 基于检索到的文档生成答案。def generate(state: GraphState): print(“---生成答案---”) question state[“clarified_question”] or state[“question”] docs state[“retrieved_docs”] context “\n\n”.join([doc.page_content for doc in docs]) prompt f”””基于以下上下文信息回答用户的问题。如果上下文信息不足以回答问题请如实告知。 上下文 {context} 问题{question} 答案””” response llm.invoke(prompt) return {“answer”: response.content}3.3 第三步组装工作流图这是 LangGraph 最核心的部分。我们用StateGraph把上面的节点和边连起来。# 初始化一个状态图指定状态类型 workflow StateGraph(GraphState) # 添加节点 workflow.add_node(“retrieve”, retrieve) workflow.add_node(“grade_documents”, grade_documents) workflow.add_node(“clarify_question”, clarify_question) workflow.add_node(“generate”, generate) # 设置入口点从检索开始 workflow.set_entry_point(“retrieve”) # 添加边决定流程走向 # 1. 检索完成后进入“评估文档”节点 workflow.add_edge(“retrieve”, “grade_documents”) # 2. 根据评估结果决定下一步 def decide_next_step(state: GraphState): if state[“needs_clarification”]: return “clarify_question” # 需要澄清去澄清节点 else: return “generate” # 文档合格直接去生成答案 workflow.add_conditional_edges( “grade_documents”, # 从哪个节点出发 decide_next_step, # 决定下一个节点的函数 { “clarify_question”: “clarify_question”, # 函数返回的字符串映射到节点 “generate”: “generate”, } ) # 3. 澄清问题后应该带着新问题重新检索 workflow.add_edge(“clarify_question”, “retrieve”) # 注意这里形成了循环 # 4. 生成答案后工作流结束 workflow.add_edge(“generate”, END) # 编译图 app workflow.compile()这个图的结构是检索 - 评估 - (需要澄清 - 澄清 - 重新检索) 或 (直接 - 生成答案)。注意clarify_question到retrieve的边形成了一个循环这正是 LangGraph 强大之处可以轻松处理这种“迭代优化”的流程。3.4 第四步运行与调试现在我们可以像调用一个函数一样运行整个工作流。# 初始化输入状态 initial_state GraphState(question“LangGraph 和 LangChain 有什么区别”, retrieved_docs[], answer“”, needs_clarificationFalse) # 运行图 final_state app.invoke(initial_state) print(“\n 最终答案 ”) print(final_state[“answer”]) print(“\n 检索到的文档数 ) print(len(final_state[“retrieved_docs”]))运行后观察控制台打印的---执行检索---等日志你就能清晰地看到工作流的执行路径。如果needs_clarification被设为True你会看到流程走了澄清分支然后重新检索。注意在实际项目中grade_documents节点的逻辑可能更复杂比如用一个小型分类器或另一个 LLM 来评估相关性。这里为了演示用了简化逻辑。关键是展示条件边add_conditional_edges的用法。4. 进阶向多智能体与多模型协作演进单个智能体的工作流只是开始。工业级 Agent 架构往往需要多个各司其职的智能体协作。用 LangGraph 来协调多智能体比用一堆if-else或手动管理消息队列要清晰得多。假设我们有一个更复杂的场景一个分析用户需求一个负责检索一个负责安全检查一个负责生成报告。每个智能体可能使用不同的模型比如 GPT-4 做分析 Claude 做生成一个本地小模型做安全检查。4.1 设计多智能体 StateState 需要容纳更多信息。class MultiAgentState(TypedDict): user_input: str analysis_result: dict # 分析智能体的输出可能是结构化数据 retrieved_info: List safety_check_passed: bool safety_feedback: str final_report: str current_agent: str # 可选用于跟踪当前执行到哪个智能体4.2 实现不同的智能体节点每个节点代表一个智能体它们可以独立配置自己的 LLM 和工具。from langchain_anthropic import ChatAnthropic # 初始化不同的模型 gpt_analyzer ChatOpenAI(model“gpt-4”, temperature0) claude_writer ChatAnthropic(model“claude-3-sonnet-20240229”, temperature0.7) # 假设有一个本地安全模型或规则引擎 def local_safety_check(text: str) - tuple[bool, str]: # 实现你的安全检查逻辑 pass def analysis_agent(state: MultiAgentState): prompt f”分析用户需求{state[‘user_input’]}。请输出一个JSON包含‘意图’‘关键实体’‘所需信息类型’。” response gpt_analyzer.invoke(prompt) # 解析 response 为字典 import json result json.loads(response.content) return {“analysis_result”: result, “current_agent”: “analyzer”} def retrieval_agent(state: MultiAgentState): # 根据 analysis_result 里的‘所需信息类型’去不同的数据源检索 info_type state[‘analysis_result’].get(‘所需信息类型’, ‘general’) # 调用对应的检索器... retrieved_data do_retrieval(info_type, state[‘user_input’]) return {“retrieved_info”: retrieved_data, “current_agent”: “retriever”} def safety_agent(state: MultiAgentState): # 综合 user_input, analysis_result, retrieved_info 进行安全检查 combined_text f”{state[‘user_input’]} {str(state[‘retrieved_info’])}” passed, feedback local_safety_check(combined_text) return {“safety_check_passed”: passed, “safety_feedback”: feedback, “current_agent”: “safety”} def generation_agent(state: MultiAgentState): if not state[‘safety_check_passed’]: report f”请求无法完成。安全反馈{state[‘safety_feedback’]}” else: prompt f”基于以下分析结果和检索信息生成一份详细报告。分析{state[‘analysis_result’]}。信息{state[‘retrieved_info’]}。” response claude_writer.invoke(prompt) report response.content return {“final_report”: report, “current_agent”: “generator”}4.3 编排多智能体工作流工作流的编排逻辑就是你的“调度策略”。是串行、并行还是有条件的执行multi_agent_workflow StateGraph(MultiAgentState) # 添加节点 multi_agent_workflow.add_node(“analyzer”, analysis_agent) multi_agent_workflow.add_node(“retriever”, retrieval_agent) multi_agent_workflow.add_node(“safety_checker”, safety_agent) multi_agent_workflow.add_node(“report_generator”, generation_agent) # 设置流程分析 - 检索 - 安全检查 - 生成 multi_agent_workflow.set_entry_point(“analyzer”) multi_agent_workflow.add_edge(“analyzer”, “retriever”) multi_agent_workflow.add_edge(“retriever”, “safety_checker”) # 安全检查后根据结果决定是生成报告还是直接结束 def after_safety(state: MultiAgentState): if state[“safety_check_passed”]: return “report_generator” else: return END # 安全检查不通过直接结束 multi_agent_workflow.add_conditional_edges(“safety_checker”, after_safety, {“report_generator”: “report_generator”, END: END}) multi_agent_workflow.add_edge(“report_generator”, END) multi_agent_app multi_agent_workflow.compile()这个流程是线性的但你已经能看到框架的灵活性。你可以轻松地改成分析后同时启动检索和安全预检需要并行支持或者根据分析结果动态选择调用哪个专门的检索智能体。关键点在多智能体场景下State成为了智能体之间通信的唯一共享内存。每个智能体只读写 State 中自己负责的部分这极大地降低了耦合度使得单个智能体的替换、升级或测试变得非常容易。5. 生产落地持久化、可视化与监控一个能在本地跑通的图离“工业级”还差得远。接下来是让这个系统变得可靠、可维护、可观察。5.1 持久化与状态恢复LangGraph 的工作流可以持久化其状态这对于处理长时间运行或可能中断的任务至关重要比如一个需要用户多次交互的复杂对话 Agent。from langgraph.checkpoint import MemorySaver # 在编译图时加入检查点存储器 checkpointer MemorySaver() persistent_app workflow.compile(checkpointercheckpointer) # 第一次调用传入一个 config其中包含线程IDthread_id用于标识这个会话 config {“configurable”: {“thread_id”: “user_session_123”}} result1 persistent_app.invoke(initial_state, configconfig) # 假设流程在中途暂停了比如在等待用户澄清 # 我们可以根据 thread_id 加载之前的状态并继续执行 # 这里模拟从某个节点继续需要你知道中断时的节点ID实践中可能存于数据库 # LangGraph 内部会通过 checkpointer 自动管理状态快照。对于生产环境你需要将MemorySaver替换为支持数据库如 PostgreSQL、Redis的持久化存储。LangGraph 提供了BaseCheckpointSaver接口允许你自定义存储后端。5.2 可视化与调试LangGraph 自带可视化工具这是开发和调试的神器。# 将图导出为 PNG 图片 from IPython.display import Image, display try: display(Image(app.get_graph().draw_mermaid_png())) except: # 如果环境不支持可以输出 Mermaid 文本到 Mermaid Live Editor 查看 print(app.get_graph().draw_mermaid())在开发过程中频繁地可视化你的图能帮你快速发现流程设计上的逻辑错误比如死循环、无法到达的节点等。5.3 监控、日志与稳定性在生产中你需要知道每个请求走了哪条路径、每个节点耗时多久、是否出错。日志记录在每个节点函数的开头和结尾添加详细的日志记录输入 State 的摘要和输出结果。可以使用structlog或logging模块并集成到你的 ELK 或 Grafana 体系里。性能追踪使用像OpenTelemetry这样的工具对节点函数进行埋点追踪延迟和错误率。错误处理与重试节点函数内部应该有try-catch处理如 LLM API 调用失败、网络超时等异常。对于可重试的错误可以修改 State让工作流跳转到一个“重试”节点或直接重试当前节点这需要更精细的边控制。超时控制为整个图的执行或单个节点的执行设置超时防止某个环节卡死导致资源耗尽。6. 常见陷阱与性能优化要点踩过坑才知道哪里路滑。下面是一些在 LangGraph 项目落地时的高频问题。6.1 State 设计过载或混乱问题把太多不相关或生命周期不同的数据塞进同一个 State导致节点间依赖复杂难以理解。建议遵循单一职责原则。State 应该只包含工作流真正需要流转的数据。一些中间计算结果如果只在单个节点内使用就不要放进 State。可以使用嵌套的 TypedDict 或 Pydantic 模型来组织状态使其结构清晰。6.2 节点函数副作用过大问题节点函数除了修改 State还执行了不可逆的外部操作比如发送邮件、写入数据库。当工作流需要回滚或重试时会造成数据不一致。建议将副作用操作尽量后置。例如把“发送邮件”作为一个独立的最终节点只在所有计算和验证都通过后才执行。或者采用补偿事务Saga Pattern的思路来设计工作流。6.3 循环失控无限循环问题像我们 RAG 例子中的clarify - retrieve循环如果没有终止条件可能会无限进行下去。解决方案在 State 中设置一个计数器如iteration_count在决定是否进入循环的节点如grade_documents中检查它。def grade_documents_with_limit(state: GraphState): iteration state.get(“iteration_count”, 0) if iteration 3: # 最多循环3次 return {“needs_clarification”: False, “answer”: “经过多次尝试仍无法获得足够相关信息。”} # ... 原有的相关性判断逻辑 ... return {“needs_clarification”: needs_clarification, “iteration_count”: iteration 1}6.4 性能瓶颈同步调用与并行化问题默认情况下LangGraph 的节点是同步顺序执行的。如果图中有多个独立节点例如同时调用多个不同的 API 获取数据串行执行会拖慢整体速度。解决方案LangGraph 支持并发执行。你可以使用add_node定义节点然后通过add_edge或条件边让它们同时开始。关键在于这些并行节点的输入 State 应该是独立的或者 LangGraph 能处理 State 的合并。对于复杂的并行合并需要仔细设计 State 的结构和节点的输出。6.5 与现有 LangChain Chain 和 Agent 的集成问题我已经有很多现成的 LangChain Chain 或 Agent怎么用到 LangGraph 里方案非常简单。一个 LangGraph 的 Node 函数内部完全可以调用一个现有的LLMChain、SequentialChain甚至一个AgentExecutor。把 Chain 的输入从 State 里取出来执行 Chain再把结果放回 State 即可。这样你可以逐步地将一个庞大的单体 Chain 重构为由 LangGraph 编排的、模块化的小节点。def my_chain_node(state: MyState): # 假设 my_chain 是一个已经定义好的 LangChain Chain result my_chain.invoke({“input”: state[“some_key”]}) return {“output_key”: result[“output”]}6.6 版本管理与测试问题工作流逻辑修改后如何保证不影响已有的业务建议将工作流的定义图的结构和节点实现代码分开。对图的结构进行版本控制。为关键节点编写单元测试模拟输入 State验证输出 State 是否符合预期。对于整个图可以建立一套集成测试用例覆盖主要的分支路径。7. 总结何时该用 LangGraph经过上面的拆解你应该对 LangGraph 的能力和复杂度有了直观感受。它不是一个银弹而是解决特定问题的精密工具。强烈建议使用 LangGraph 的场景流程复杂包含条件分支和循环比如多轮对话、迭代式内容生成、带条件审核的流水线。多智能体协作需要清晰定义多个 AI 模型或工具之间的调用关系和数据流。需要持久化状态应用需要处理可能中断的长会话并能从中断点恢复。对可观测性要求高你需要清晰地监控一个请求到底走了哪条路径每个步骤的输入输出是什么。用简单 Chain 或脚本就够的场景简单的 QA一次检索一次生成。线性的、无分支的文档处理流水线。对开发速度要求极高且流程未来几乎不会变的原型。最后的建议不要一开始就追求设计一个完美的大图。从一个核心的小流程开始比如我们例子中的 RAG 判断循环把它跑通、可视化、加上检查点和日志。然后像搭积木一样逐步增加新的节点和分支。LangGraph 的价值在于它提供的约束和清晰度迫使你以“状态流转”的视角去思考 AI 应用这才是构建稳定、可维护的工业级 Agent 架构的真正起点。当你习惯了这种思维再回头看那些面条式的回调代码会感到前所未有的清爽。
LangGraph实战：构建可编排、可持久化的AI工作流与多智能体系统
1. 先搞清楚 LangGraph 到底解决了什么核心问题如果你正在用 LangChain 构建复杂的 AI 应用比如需要多步骤决策、状态流转或者多个 AI 模型智能体协同工作的场景那你大概率会遇到一个头疼的问题流程控制。用 LangChain 的SequentialChain或者简单的函数调用处理一个简单的问答流程还行一旦任务流程变长、出现分支判断、需要循环执行或者需要让多个智能体比如一个负责检索、一个负责分析、一个负责生成接力或协作代码很快就会变得难以维护。这就是 LangGraph 要解决的核心问题。它不是一个替代品而是 LangChain 生态中的一个工作流编排框架。你可以把它理解为一个专门为 AI 应用设计的、可视化的流程图工具。它最核心的价值是引入了StateGraph状态图的概念让你能清晰地定义应用的状态State、节点Nodes代表一个执行步骤比如调用一次 LLM 或工具、边Edges决定下一步走向哪个节点以及循环Cycles用于多轮对话或迭代优化。所以这篇文章适合两类人看一是已经用过 LangChain但觉得流程控制太乱的开发者二是刚开始设计多智能体、复杂决策链应用想找一个更工程化方案的工程师。最值得关注的不是 LangGraph 的语法而是它如何通过StateGraph把一堆散乱的回调、条件判断和循环变成一个可定义、可调试、可持久化的“工作流蓝图”。这直接决定了你的 Agent 能否从玩具 demo 升级为工业级应用。2. 环境准备与核心概念对齐别急着写代码在动手之前先确保你的环境和认知对齐。LangGraph 不是独立运行的它构建在 LangChain 之上所以你的 Python 环境里需要同时安装它们。我建议先创建一个干净的虚拟环境避免依赖冲突。# 创建并激活虚拟环境以 conda 为例 conda create -n langgraph-demo python3.10 conda activate langgraph-demo # 安装核心依赖 pip install langchain langgraph langchain-openai这里注意langchain-openai是 LangChain 官方维护的 OpenAI 集成包比直接用openai库更方便。如果你要用其他模型比如智谱、DeepSeek就安装对应的langchain-*包。不要一上来就装一大堆先确保核心的能跑通。安装完后别急着跑复杂例子。先理解 LangGraph 里三个最重要的对象这能帮你省下大量后期调试的时间State状态这是一个贯穿整个工作流的共享数据字典。你所有节点读取和修改的都是它。比如一个 RAG 问答工作流的状态里可能包含question用户问题、retrieved_docs检索到的文档、answer最终答案。在 LangGraph 里State 通常用一个 TypedDict 或 Pydantic 模型来定义这能提供类型提示减少错误。Node节点一个普通的 Python 函数它接收当前 State执行一些操作比如调用 LLM、查询数据库然后返回一个包含对 State 修改的字典。一个节点只做一件事这是保持工作流清晰的关键。Edge边决定工作流从一个节点执行完后下一个该去哪个节点。边可以是固定的always也可以是根据 State 里的某个值动态决定的conditional。很多人一开始会混淆 Node 和 Chain。记住Node 更原子化它可能是 Chain 的一部分也可能就是一个工具调用。而 LangGraph 的工作流就是由这些 Node 和 Edge 编织成的网。3. 从零构建一个带状态管控的 RAG 问答工作流理论说再多不如跑一遍。我们从一个最经典的场景入手RAG检索增强生成问答。但我们要做得比简单调用RetrievalQA更精细、更可控。假设我们的需求是用户提问后先检索相关文档如果检索结果足够相关就直接生成答案如果相关性不够则让 LLM 先澄清问题再基于澄清后的问题重新检索。这个流程就涉及了状态流转和条件判断正好用 LangGraph 来建模。3.1 第一步定义 State 和初始化节点首先我们定义工作流的状态。这里我们需要记录问题、检索到的文档、答案以及一个判断是否需要澄清的标记。from typing import TypedDict, List, Annotated import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings from langchain.schema import Document import os # 1. 定义状态结构 class GraphState(TypedDict): question: str # 原始问题 clarified_question: str # 澄清后的问题可能和原始问题相同 retrieved_docs: List[Document] # 检索到的文档列表 answer: str # 最终答案 needs_clarification: bool # 是否需要澄清问题 # 2. 初始化一些组件这里用伪代码你需要替换成自己的 os.environ[“OPENAI_API_KEY”] “your-api-key” llm ChatOpenAI(model“gpt-4o-mini”, temperature0) embeddings OpenAIEmbeddings() # 假设你已经有一个加载好的向量数据库 vectorstore Chroma(persist_directory“./my_db”, embedding_functionembeddings) retriever vectorstore.as_retriever(search_kwargs{“k”: 4}) # 每次检索4条3.2 第二步实现各个功能节点接下来我们把流程中的每个步骤实现为一个节点函数。节点1检索文档这个节点接收 State用其中的问题可能是原始问题也可能是澄清后的问题去检索并将结果存回 State。def retrieve(state: GraphState): print(“---执行检索---”) question state.get(“clarified_question”) or state[“question”] docs retriever.invoke(question) return {“retrieved_docs”: docs}节点2判断相关性/是否需要澄清这是一个决策节点。它检查检索到的文档是否足够相关。这里用一个简单的规则如果检索到的文档中最相关的那个分数假设检索器返回分数低于某个阈值则认为需要澄清。def grade_documents(state: GraphState): print(“---评估文档相关性---”) docs state[“retrieved_docs”] # 假设我们使用的检索器返回的 Document 有 metadata 包含分数 # 这里简化处理如果没分数或者平均分数低就认为需要澄清 needs_clarification False if docs: # 这里只是一个示例逻辑实际中你可能需要更复杂的评估器甚至用另一个LLM # 例如检查第一条文档的元数据分数 if hasattr(docs[0], ‘metadata’) and docs[0].metadata.get(‘score’, 1) 0.7: needs_clarification True else: needs_clarification True # 没检索到任何文档更需要澄清 return {“needs_clarification”: needs_clarification}节点3澄清问题如果判定需要澄清这个节点会调用 LLM让它根据原始问题和检索到的文档可能不相关生成一个更清晰的问题。def clarify_question(state: GraphState): print(“---澄清问题---”) question state[“question”] docs_preview “\n”.join([doc.page_content[:200] for doc in state[“retrieved_docs”][:2]]) # 预览前两条 prompt f””” 用户的原问题是{question} 我们根据这个问题检索到了一些文档但它们可能不太相关 {docs_preview} 请你帮我将用户的问题重新表述一下使其更清晰、更具体更容易从知识库中找到答案。 请直接输出重新表述后的问题不要加任何解释。 “”” response llm.invoke(prompt) clarified_q response.content.strip() return {“clarified_question”: clarified_q}节点4生成最终答案当文档相关性足够时调用 LLM 基于检索到的文档生成答案。def generate(state: GraphState): print(“---生成答案---”) question state[“clarified_question”] or state[“question”] docs state[“retrieved_docs”] context “\n\n”.join([doc.page_content for doc in docs]) prompt f”””基于以下上下文信息回答用户的问题。如果上下文信息不足以回答问题请如实告知。 上下文 {context} 问题{question} 答案””” response llm.invoke(prompt) return {“answer”: response.content}3.3 第三步组装工作流图这是 LangGraph 最核心的部分。我们用StateGraph把上面的节点和边连起来。# 初始化一个状态图指定状态类型 workflow StateGraph(GraphState) # 添加节点 workflow.add_node(“retrieve”, retrieve) workflow.add_node(“grade_documents”, grade_documents) workflow.add_node(“clarify_question”, clarify_question) workflow.add_node(“generate”, generate) # 设置入口点从检索开始 workflow.set_entry_point(“retrieve”) # 添加边决定流程走向 # 1. 检索完成后进入“评估文档”节点 workflow.add_edge(“retrieve”, “grade_documents”) # 2. 根据评估结果决定下一步 def decide_next_step(state: GraphState): if state[“needs_clarification”]: return “clarify_question” # 需要澄清去澄清节点 else: return “generate” # 文档合格直接去生成答案 workflow.add_conditional_edges( “grade_documents”, # 从哪个节点出发 decide_next_step, # 决定下一个节点的函数 { “clarify_question”: “clarify_question”, # 函数返回的字符串映射到节点 “generate”: “generate”, } ) # 3. 澄清问题后应该带着新问题重新检索 workflow.add_edge(“clarify_question”, “retrieve”) # 注意这里形成了循环 # 4. 生成答案后工作流结束 workflow.add_edge(“generate”, END) # 编译图 app workflow.compile()这个图的结构是检索 - 评估 - (需要澄清 - 澄清 - 重新检索) 或 (直接 - 生成答案)。注意clarify_question到retrieve的边形成了一个循环这正是 LangGraph 强大之处可以轻松处理这种“迭代优化”的流程。3.4 第四步运行与调试现在我们可以像调用一个函数一样运行整个工作流。# 初始化输入状态 initial_state GraphState(question“LangGraph 和 LangChain 有什么区别”, retrieved_docs[], answer“”, needs_clarificationFalse) # 运行图 final_state app.invoke(initial_state) print(“\n 最终答案 ”) print(final_state[“answer”]) print(“\n 检索到的文档数 ) print(len(final_state[“retrieved_docs”]))运行后观察控制台打印的---执行检索---等日志你就能清晰地看到工作流的执行路径。如果needs_clarification被设为True你会看到流程走了澄清分支然后重新检索。注意在实际项目中grade_documents节点的逻辑可能更复杂比如用一个小型分类器或另一个 LLM 来评估相关性。这里为了演示用了简化逻辑。关键是展示条件边add_conditional_edges的用法。4. 进阶向多智能体与多模型协作演进单个智能体的工作流只是开始。工业级 Agent 架构往往需要多个各司其职的智能体协作。用 LangGraph 来协调多智能体比用一堆if-else或手动管理消息队列要清晰得多。假设我们有一个更复杂的场景一个分析用户需求一个负责检索一个负责安全检查一个负责生成报告。每个智能体可能使用不同的模型比如 GPT-4 做分析 Claude 做生成一个本地小模型做安全检查。4.1 设计多智能体 StateState 需要容纳更多信息。class MultiAgentState(TypedDict): user_input: str analysis_result: dict # 分析智能体的输出可能是结构化数据 retrieved_info: List safety_check_passed: bool safety_feedback: str final_report: str current_agent: str # 可选用于跟踪当前执行到哪个智能体4.2 实现不同的智能体节点每个节点代表一个智能体它们可以独立配置自己的 LLM 和工具。from langchain_anthropic import ChatAnthropic # 初始化不同的模型 gpt_analyzer ChatOpenAI(model“gpt-4”, temperature0) claude_writer ChatAnthropic(model“claude-3-sonnet-20240229”, temperature0.7) # 假设有一个本地安全模型或规则引擎 def local_safety_check(text: str) - tuple[bool, str]: # 实现你的安全检查逻辑 pass def analysis_agent(state: MultiAgentState): prompt f”分析用户需求{state[‘user_input’]}。请输出一个JSON包含‘意图’‘关键实体’‘所需信息类型’。” response gpt_analyzer.invoke(prompt) # 解析 response 为字典 import json result json.loads(response.content) return {“analysis_result”: result, “current_agent”: “analyzer”} def retrieval_agent(state: MultiAgentState): # 根据 analysis_result 里的‘所需信息类型’去不同的数据源检索 info_type state[‘analysis_result’].get(‘所需信息类型’, ‘general’) # 调用对应的检索器... retrieved_data do_retrieval(info_type, state[‘user_input’]) return {“retrieved_info”: retrieved_data, “current_agent”: “retriever”} def safety_agent(state: MultiAgentState): # 综合 user_input, analysis_result, retrieved_info 进行安全检查 combined_text f”{state[‘user_input’]} {str(state[‘retrieved_info’])}” passed, feedback local_safety_check(combined_text) return {“safety_check_passed”: passed, “safety_feedback”: feedback, “current_agent”: “safety”} def generation_agent(state: MultiAgentState): if not state[‘safety_check_passed’]: report f”请求无法完成。安全反馈{state[‘safety_feedback’]}” else: prompt f”基于以下分析结果和检索信息生成一份详细报告。分析{state[‘analysis_result’]}。信息{state[‘retrieved_info’]}。” response claude_writer.invoke(prompt) report response.content return {“final_report”: report, “current_agent”: “generator”}4.3 编排多智能体工作流工作流的编排逻辑就是你的“调度策略”。是串行、并行还是有条件的执行multi_agent_workflow StateGraph(MultiAgentState) # 添加节点 multi_agent_workflow.add_node(“analyzer”, analysis_agent) multi_agent_workflow.add_node(“retriever”, retrieval_agent) multi_agent_workflow.add_node(“safety_checker”, safety_agent) multi_agent_workflow.add_node(“report_generator”, generation_agent) # 设置流程分析 - 检索 - 安全检查 - 生成 multi_agent_workflow.set_entry_point(“analyzer”) multi_agent_workflow.add_edge(“analyzer”, “retriever”) multi_agent_workflow.add_edge(“retriever”, “safety_checker”) # 安全检查后根据结果决定是生成报告还是直接结束 def after_safety(state: MultiAgentState): if state[“safety_check_passed”]: return “report_generator” else: return END # 安全检查不通过直接结束 multi_agent_workflow.add_conditional_edges(“safety_checker”, after_safety, {“report_generator”: “report_generator”, END: END}) multi_agent_workflow.add_edge(“report_generator”, END) multi_agent_app multi_agent_workflow.compile()这个流程是线性的但你已经能看到框架的灵活性。你可以轻松地改成分析后同时启动检索和安全预检需要并行支持或者根据分析结果动态选择调用哪个专门的检索智能体。关键点在多智能体场景下State成为了智能体之间通信的唯一共享内存。每个智能体只读写 State 中自己负责的部分这极大地降低了耦合度使得单个智能体的替换、升级或测试变得非常容易。5. 生产落地持久化、可视化与监控一个能在本地跑通的图离“工业级”还差得远。接下来是让这个系统变得可靠、可维护、可观察。5.1 持久化与状态恢复LangGraph 的工作流可以持久化其状态这对于处理长时间运行或可能中断的任务至关重要比如一个需要用户多次交互的复杂对话 Agent。from langgraph.checkpoint import MemorySaver # 在编译图时加入检查点存储器 checkpointer MemorySaver() persistent_app workflow.compile(checkpointercheckpointer) # 第一次调用传入一个 config其中包含线程IDthread_id用于标识这个会话 config {“configurable”: {“thread_id”: “user_session_123”}} result1 persistent_app.invoke(initial_state, configconfig) # 假设流程在中途暂停了比如在等待用户澄清 # 我们可以根据 thread_id 加载之前的状态并继续执行 # 这里模拟从某个节点继续需要你知道中断时的节点ID实践中可能存于数据库 # LangGraph 内部会通过 checkpointer 自动管理状态快照。对于生产环境你需要将MemorySaver替换为支持数据库如 PostgreSQL、Redis的持久化存储。LangGraph 提供了BaseCheckpointSaver接口允许你自定义存储后端。5.2 可视化与调试LangGraph 自带可视化工具这是开发和调试的神器。# 将图导出为 PNG 图片 from IPython.display import Image, display try: display(Image(app.get_graph().draw_mermaid_png())) except: # 如果环境不支持可以输出 Mermaid 文本到 Mermaid Live Editor 查看 print(app.get_graph().draw_mermaid())在开发过程中频繁地可视化你的图能帮你快速发现流程设计上的逻辑错误比如死循环、无法到达的节点等。5.3 监控、日志与稳定性在生产中你需要知道每个请求走了哪条路径、每个节点耗时多久、是否出错。日志记录在每个节点函数的开头和结尾添加详细的日志记录输入 State 的摘要和输出结果。可以使用structlog或logging模块并集成到你的 ELK 或 Grafana 体系里。性能追踪使用像OpenTelemetry这样的工具对节点函数进行埋点追踪延迟和错误率。错误处理与重试节点函数内部应该有try-catch处理如 LLM API 调用失败、网络超时等异常。对于可重试的错误可以修改 State让工作流跳转到一个“重试”节点或直接重试当前节点这需要更精细的边控制。超时控制为整个图的执行或单个节点的执行设置超时防止某个环节卡死导致资源耗尽。6. 常见陷阱与性能优化要点踩过坑才知道哪里路滑。下面是一些在 LangGraph 项目落地时的高频问题。6.1 State 设计过载或混乱问题把太多不相关或生命周期不同的数据塞进同一个 State导致节点间依赖复杂难以理解。建议遵循单一职责原则。State 应该只包含工作流真正需要流转的数据。一些中间计算结果如果只在单个节点内使用就不要放进 State。可以使用嵌套的 TypedDict 或 Pydantic 模型来组织状态使其结构清晰。6.2 节点函数副作用过大问题节点函数除了修改 State还执行了不可逆的外部操作比如发送邮件、写入数据库。当工作流需要回滚或重试时会造成数据不一致。建议将副作用操作尽量后置。例如把“发送邮件”作为一个独立的最终节点只在所有计算和验证都通过后才执行。或者采用补偿事务Saga Pattern的思路来设计工作流。6.3 循环失控无限循环问题像我们 RAG 例子中的clarify - retrieve循环如果没有终止条件可能会无限进行下去。解决方案在 State 中设置一个计数器如iteration_count在决定是否进入循环的节点如grade_documents中检查它。def grade_documents_with_limit(state: GraphState): iteration state.get(“iteration_count”, 0) if iteration 3: # 最多循环3次 return {“needs_clarification”: False, “answer”: “经过多次尝试仍无法获得足够相关信息。”} # ... 原有的相关性判断逻辑 ... return {“needs_clarification”: needs_clarification, “iteration_count”: iteration 1}6.4 性能瓶颈同步调用与并行化问题默认情况下LangGraph 的节点是同步顺序执行的。如果图中有多个独立节点例如同时调用多个不同的 API 获取数据串行执行会拖慢整体速度。解决方案LangGraph 支持并发执行。你可以使用add_node定义节点然后通过add_edge或条件边让它们同时开始。关键在于这些并行节点的输入 State 应该是独立的或者 LangGraph 能处理 State 的合并。对于复杂的并行合并需要仔细设计 State 的结构和节点的输出。6.5 与现有 LangChain Chain 和 Agent 的集成问题我已经有很多现成的 LangChain Chain 或 Agent怎么用到 LangGraph 里方案非常简单。一个 LangGraph 的 Node 函数内部完全可以调用一个现有的LLMChain、SequentialChain甚至一个AgentExecutor。把 Chain 的输入从 State 里取出来执行 Chain再把结果放回 State 即可。这样你可以逐步地将一个庞大的单体 Chain 重构为由 LangGraph 编排的、模块化的小节点。def my_chain_node(state: MyState): # 假设 my_chain 是一个已经定义好的 LangChain Chain result my_chain.invoke({“input”: state[“some_key”]}) return {“output_key”: result[“output”]}6.6 版本管理与测试问题工作流逻辑修改后如何保证不影响已有的业务建议将工作流的定义图的结构和节点实现代码分开。对图的结构进行版本控制。为关键节点编写单元测试模拟输入 State验证输出 State 是否符合预期。对于整个图可以建立一套集成测试用例覆盖主要的分支路径。7. 总结何时该用 LangGraph经过上面的拆解你应该对 LangGraph 的能力和复杂度有了直观感受。它不是一个银弹而是解决特定问题的精密工具。强烈建议使用 LangGraph 的场景流程复杂包含条件分支和循环比如多轮对话、迭代式内容生成、带条件审核的流水线。多智能体协作需要清晰定义多个 AI 模型或工具之间的调用关系和数据流。需要持久化状态应用需要处理可能中断的长会话并能从中断点恢复。对可观测性要求高你需要清晰地监控一个请求到底走了哪条路径每个步骤的输入输出是什么。用简单 Chain 或脚本就够的场景简单的 QA一次检索一次生成。线性的、无分支的文档处理流水线。对开发速度要求极高且流程未来几乎不会变的原型。最后的建议不要一开始就追求设计一个完美的大图。从一个核心的小流程开始比如我们例子中的 RAG 判断循环把它跑通、可视化、加上检查点和日志。然后像搭积木一样逐步增加新的节点和分支。LangGraph 的价值在于它提供的约束和清晰度迫使你以“状态流转”的视角去思考 AI 应用这才是构建稳定、可维护的工业级 Agent 架构的真正起点。当你习惯了这种思维再回头看那些面条式的回调代码会感到前所未有的清爽。