1. 项目概述为什么我们需要LangGraph如果你正在用LangChain构建AI应用大概率遇到过这样的场景你的Agent需要先调用一个工具去查天气然后根据天气结果决定是推荐室内活动还是户外运动最后再调用另一个工具去搜索具体的活动推荐。这个“先A后B根据结果决定C”的逻辑在代码里可能就是一堆if-else和函数调用随着业务复杂代码很快会变成难以维护的“面条代码”。更头疼的是你想跟同事或者产品经理解释这个Agent到底是怎么运行的光靠口述或者看代码效率极低。这就是LangGraph要解决的核心问题。它不是一个全新的框架而是构建在LangChain之上的一个库专门用于创建有状态、多步骤的AI工作流。你可以把它想象成给AI Agent的“大脑”画一张清晰的“电路图”。这张图定义了Agent思考和行为的所有可能路径、决策点以及状态如何在不同步骤间流转。可视化编排是它的杀手锏让你能直观地设计、调试和理解复杂的Agent逻辑而不是在代码的迷宫里打转。简单说LangGraph让AI Agent从“一锤子买卖”的简单问答进化成了能处理复杂任务、拥有记忆和决策能力的“智能流程”。它适合所有正在或计划构建复杂AI应用的开发者无论你是想做一个能自动处理多轮对话和工具调用的客服助手还是一个能自主分析数据、撰写报告并发送邮件的自动化分析AgentLangGraph都能提供清晰、可维护的架构支持。2. LangGraph核心概念与架构拆解要玩转LangGraph得先吃透它的几个核心“零件”。理解了这些你再看它的可视化界面就会觉得一目了然。2.1 三要素State、Node、Edge这是LangGraph模型的基石几乎所有的编排都围绕它们展开。State状态这是工作流的“记忆中枢”和“数据总线”。它是一个字典Dict或Pydantic模型在整个工作流的生命周期中流转和更新。比如一个客服Agent的State里可能包含user_query用户问题、conversation_history对话历史、retrieved_docs检索到的知识、final_answer最终回复等字段。每个节点Node读取State的一部分处理然后把结果写回State的对应字段。State的设计决定了工作流的数据流是架构的第一步。注意State的设计要遵循“最小化”和“清晰化”原则。不要把所有东西都塞进一个巨大的State里。按模块划分比如input_state,processing_state,output_state或者直接用Pydantic模型定义字段和类型这样在编写节点函数时IDE的自动补全和类型检查会帮你大忙减少运行时错误。Node节点节点就是工作流中的一个个“处理单元”。每个节点是一个普通的Python函数或可调用对象它接收当前的State执行一些操作比如调用LLM、运行工具、处理数据然后返回一个包含对State更新内容的字典。例如一个“检索”节点它的函数会从State里读取user_query调用向量数据库检索然后把结果写入State的retrieved_docs字段。Edge边边定义了工作流的“控制流”即节点之间的跳转逻辑。边决定了“上一个节点执行完后接下来该去哪个节点”。LangGraph提供了几种类型的边条件边Conditional Edge这是实现分支逻辑的关键。它根据State中的某个条件比如LLM的输出里是否包含特定关键词或者某个工具调用的结果是否成功来决定下一步走向哪个节点。这模拟了人类的“判断-决策”过程。普通边无条件地指向下一个节点。入口边Entry Point定义工作流的开始节点。通过组合这三种元素你就能构建出从简单的线性流程到复杂的、带循环和条件分支的任意工作流图。2.2 与LangChain的关系不是替代而是增强很多人会困惑LangGraph和LangChain到底是什么关系。这里必须澄清LangGraph不是LangChain的替代品而是它的一个专业化扩展。LangChain是一个全面的框架提供了构建LLM应用所需的各种组件模型封装LLMs、提示模板Prompts、链Chains、记忆Memory、检索器Retrievers和代理Agents。它的AgentExecutor已经能够处理简单的工具调用循环。LangGraph则聚焦于一点构建复杂、有状态、多参与者Multi-Actor的工作流。它把LangChain的组件如LLM、Tools、Memory当作“乐高积木”然后用“图”这个更强大、更直观的方式来组装和协调这些积木。你可以这样类比LangChain给了你发动机LLM、轮胎Tools、方向盘Prompt而LangGraph给了你整辆车的设计蓝图和控制系统。用LangChain的Agent你写的是“脚本”用LangGraph你设计的是“流程图”。当你的业务逻辑超过3个步骤或者需要复杂分支时LangGraph在可维护性和可调试性上的优势就非常明显了。2.3 可视化编排的价值从“黑盒”到“白盒”可视化是LangGraph最吸引人的特性之一。通过几行代码将图编译后你可以生成一个交互式的可视化界面。这对开发流程是革命性的设计阶段产品经理、算法工程师和开发工程师可以围着一张图讨论业务逻辑确保理解一致避免后期返工。调试阶段当Agent行为不符合预期时你可以沿着可视化的路径回溯精确看到是哪个节点的输入/输出出了问题State在每一步的变化也清晰可见。这比在日志里大海捞针高效无数倍。协作与文档生成的图本身就是最好的技术文档新成员 onboarding 时看一眼图就能对系统架构有个七八分理解。3. 从零构建你的第一个LangGraph智能工作流理论说得再多不如亲手搭一个。我们来构建一个经典的“研究助手”Agent用户提出一个复杂问题Agent先决定是否需要联网搜索需要则搜索并总结最后生成一份结构化的回答报告。3.1 环境准备与依赖安装首先确保你的Python环境建议3.10然后安装必要的包。这里我们主要需要langgraph和langchain的相关组件以及一个LLM这里用OpenAI的GPT模型为例和一个搜索工具用Tavily搜索API。pip install langgraph langchain langchain-openai tavily-python安装完成后记得设置你的API密钥。通常我会在项目根目录创建一个.env文件来管理密钥并使用python-dotenv加载。# .env 文件 OPENAI_API_KEYsk-你的openai密钥 TAVILY_API_KEY你的tavily密钥# 在代码开头加载环境变量 import os from dotenv import load_dotenv load_dotenv() from langchain_openai import ChatOpenAI from langchain_community.tools.tavily_search import TavilySearchResults # 初始化LLM和工具 llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 使用一个轻量且稳定的模型 search_tool TavilySearchResults(max_results3) # 限制搜索结果为3条避免信息过载3.2 定义工作流状态State我们使用Pydantic来定义State这是最规范和安全的方式能利用类型提示。from typing import TypedDict, List, Optional, Annotated from langgraph.graph.message import add_messages import operator class State(TypedDict): # 输入 question: str # 决策与处理过程 needs_search: Optional[bool] None # 是否需要搜索 search_results: Optional[str] None # 搜索结果 analysis: Optional[str] None # 对问题的分析或对搜索结果的总结 # 输出 final_answer: Optional[str] None # 最终答案 # LangGraph内置的消息历史用于支持多轮对话 messages: Annotated[list, add_messages]这里我们定义了几个关键字段。Annotated类型和add_messages是LangGraph提供的语法糖用于自动管理对话消息列表非常方便。needs_search将作为我们条件分支的判断依据。3.3 创建节点Nodes接下来我们创建三个核心节点函数route_question路由判断、web_search网络搜索、generate_answer生成答案。节点1路由判断节点这个节点负责分析用户问题决定是否需要联网搜索。from langchain_core.prompts import ChatPromptTemplate def route_question(state: State): 判断问题是否需要联网搜索获取最新信息。 question state[question] # 构建一个提示词让LLM做判断 route_prompt ChatPromptTemplate.from_messages([ (system, 你是一个问题分类助手。请根据用户问题判断是否需要联网搜索最新信息来回答。仅当问题涉及实时信息、新闻、最新事件或非常具体的当前数据时才需要搜索。对于常识、概念解释或历史事实无需搜索。只回复SEARCH或NO_SEARCH。), (human, 用户问题{question}) ]) route_chain route_prompt | llm decision route_chain.invoke({question: question}).content.strip() # 更新State return {needs_search: decision SEARCH}节点2网络搜索节点只有当route_question节点判定需要搜索时才会执行此节点。def web_search(state: State): 执行网络搜索并格式化结果。 if not state.get(needs_search): # 如果不需要搜索直接返回空结果避免浪费API调用 return {search_results: 未执行搜索。} question state[question] try: # 调用搜索工具 results search_tool.invoke(question) # 将搜索结果列表格式化为一个字符串 formatted_results \n\n.join([f来源 {i1}: {r[content]} for i, r in enumerate(results)]) return {search_results: formatted_results} except Exception as e: # 搜索失败时的容错处理 return {search_results: f搜索过程中出现错误{str(e)}}节点3生成最终答案节点这是工作流的终点综合所有信息生成最终回答。def generate_answer(state: State): 综合所有信息生成最终答案。 question state[question] search_info state.get(search_results) # 根据是否有搜索信息构建不同的提示词 if search_info and search_info ! 未执行搜索。: prompt_text f 请基于以下用户问题和搜索到的信息生成一个全面、结构清晰的回答。 用户问题{question} 搜索到的信息 {search_info} 请确保回答 1. 直接回应用户问题的核心。 2. 整合搜索信息并注明信息来源于网络搜索。 3. 如果搜索信息不足或矛盾请明确指出。 4. 使用友好的语气。 else: prompt_text f 请基于你的知识回答以下用户问题。如果问题涉及你知识截止日期后的最新事件请诚实说明。 用户问题{question} 请给出一个结构清晰、准确的回答。 answer_prompt ChatPromptTemplate.from_messages([ (system, 你是一个有帮助的研究助手。), (human, prompt_text) ]) answer_chain answer_prompt | llm final_answer answer_chain.invoke({}).content return {final_answer: final_answer, analysis: 已回答用户问题。}3.4 编排图结构连接节点与边现在我们把上面散落的“零件”组装成一张能运转的“图”。from langgraph.graph import StateGraph, END # 1. 创建一个图并指定State的类型 workflow StateGraph(State) # 2. 将我们定义的函数添加为节点 workflow.add_node(router, route_question) # 路由节点 workflow.add_node(search, web_search) # 搜索节点 workflow.add_node(answer, generate_answer) # 回答节点 # 3. 设置入口点工作流从router节点开始 workflow.set_entry_point(router) # 4. 添加条件边从router出来后根据needs_search的值决定去向 workflow.add_conditional_edges( router, # 这是一个判断函数它检查State并返回下一个节点的名称 lambda state: search if state.get(needs_search) else answer, # 指定可能的下一个节点 {search: search, answer: answer} ) # 5. 添加普通边从search节点无条件指向answer节点 workflow.add_edge(search, answer) # 6. 设置answer节点为终点 workflow.add_edge(answer, END) # 7. 编译图得到可执行的对象 app workflow.compile()至此一个具备分支判断能力的智能工作流就构建完成了。你可以通过app.invoke()来运行它。3.5 运行与可视化运行工作流非常简单只需传入初始状态。# 定义初始输入 initial_state {question: 2024年巴黎奥运会中国代表团获得了多少枚金牌, messages: []} # 执行工作流 result app.invoke(initial_state) print(result[final_answer])可视化你的工作流这是LangGraph的精华所在# 将图导出为PNG图片 from IPython.display import Image, display try: display(Image(app.get_graph().draw_mermaid_png())) except: # 如果无法直接显示可以保存到文件 app.get_graph().draw_mermaid_png().save(my_first_agent_workflow.png) print(流程图已保存为 my_first_agent_workflow.png)生成的图会清晰地显示三个节点以及从router出发根据条件分别指向search或answer的两条路径。这张图就是你Agent逻辑的活文档。4. 高级特性与实战技巧掌握了基础工作流后我们可以探索一些更强大的特性让Agent变得更智能、更健壮。4.1 实现长期记忆与多轮对话上面的例子是单次交互。要让Agent记住对话历史实现真正的多轮对话我们需要利用State里的messages字段和LangGraph的add_messages注解。关键在于每个节点在处理时不仅要更新业务字段还要把LLM的输入输出作为消息添加到messages列表中。通常我们会创建一个“代理节点”它内部封装了与LLM的交互和消息管理。from langgraph.graph import MessagesState from langchain_core.messages import HumanMessage, AIMessage # 使用预定义的MessagesState它已经包含了messages字段 class ChatState(MessagesState): question: str needs_search: Optional[bool] None search_results: Optional[str] None final_answer: Optional[str] None def call_llm(state: ChatState): 一个集成了消息管理的LLM调用节点。 # 1. 从历史消息和当前问题构建对话上下文 # 假设最新的一条Human消息是当前问题 conversation_history state[messages] # 构建给LLM的提示包含历史对话 prompt ChatPromptTemplate.from_messages([ (system, 你是一个友好的助手。), *conversation_history, # 将历史消息作为上下文传入 (human, {question}) ]) chain prompt | llm response chain.invoke({question: state[question]}) # 2. 关键将本轮对话的HumanMessage和AIMessage添加到State中 # LangGraph的add_messages注解会自动处理这个列表的更新 # 我们返回一个包含新消息的字典State中的messages字段会自动合并 new_messages [ HumanMessage(contentstate[question]), AIMessage(contentresponse.content) ] return {messages: new_messages, final_answer: response.content}这样每次调用call_llm节点对话历史都会自动累积后续节点就能基于完整的上下文进行决策和生成。4.2 子图Subgraph与模块化设计当工作流变得非常庞大时把所有逻辑塞进一张大图会难以管理。LangGraph支持子图允许你将一部分功能例如一个完整的“检索增强生成RAG流程”封装成一个独立的子图然后在主图中像调用单个节点一样调用它。这极大地提升了代码的模块化和复用性。你可以为不同的功能模块如“数据验证”、“内容生成”、“审核过滤”分别构建子图然后像搭积木一样组合它们。from langgraph.graph import StateGraph as SubStateGraph # 1. 定义一个RAG子图 def rag_retrieval(state): # ... 实现检索逻辑 ... return {retrieved_docs: docs} def rag_synthesis(state): # ... 实现生成逻辑 ... return {draft_answer: draft} rag_workflow SubStateGraph(State) rag_workflow.add_node(retrieve, rag_retrieval) rag_workflow.add_node(synthesize, rag_synthesis) rag_workflow.add_edge(retrieve, synthesize) rag_workflow.set_entry_point(retrieve) rag_subgraph rag_workflow.compile() # 2. 在主图中将子图作为一个节点添加 main_workflow StateGraph(State) # rag_subgraph可以像普通函数一样被调用 main_workflow.add_node(rag_module, rag_subgraph) # ... 添加其他节点和边 ...4.3 错误处理与持久化在生产环境中工作流可能因网络、API限制或意外输入而失败。LangGraph提供了interrupt和checkpoint机制来增强鲁棒性。中断Interrupt你可以在图中预设“中断点”。当运行到该节点时工作流会暂停将当前状态持久化到数据库如Redis、SQLite。之后可以从这个中断点恢复执行。这非常适合处理需要人工审核或等待外部异步响应的长流程。检查点Checkpoint类似于游戏存档在关键步骤后自动保存状态。如果后续步骤失败可以回滚到上一个检查点重试而不是从头开始。实现这些功能通常需要配置一个Checkpointer对象并在编译图时传入。from langgraph.checkpoint.sqlite import SqliteSaver # 使用SQLite存储检查点 checkpointer SqliteSaver.from_conn_string(:memory:) # 生产环境换成实际数据库路径 app workflow.compile(checkpointercheckpointer) # 调用时传入一个config其中包含线程ID用于标识这次会话 config {configurable: {thread_id: user_123_session_1}} result app.invoke(initial_state, configconfig) # 如果中断状态会被保存。下次可以用相同的thread_id恢复。5. 常见问题、调试技巧与性能优化在实际开发和部署中你会遇到各种问题。下面是我踩过坑后总结的一些实战经验。5.1 调试与问题排查1. 状态State追踪不清晰问题不知道某个节点执行后State到底变成了什么样。解决在每个节点的函数内部关键步骤前后打印State。或者使用LangGraph的内置日志。在调用app.invoke()时设置debugTrue它会在控制台输出每个节点执行前后的State快照一目了然。result app.invoke(initial_state, debugTrue)2. 条件边Conditional Edge不按预期跳转问题工作流总是走错分支。解决首先检查条件判断函数lambda state: ...。确保它读取的State字段名完全正确且值的类型是你预期的是布尔值True/False还是字符串SEARCH。最稳妥的方法是在route_question节点里把决定needs_search值的逻辑和日志打印清楚。3. 可视化图与代码逻辑不符问题画出来的图少了边或者节点连接错误。解决确保add_edge和add_conditional_edges的调用顺序和参数正确。边的添加必须在所有节点添加之后。编译图compile()后立即生成可视化图进行比对。5.2 性能优化与最佳实践1. 避免在State中存储过大对象State会在每个节点间被序列化/反序列化尤其是在使用持久化检查点时。不要在State里存巨大的列表、字典或二进制数据如图片。只存储必要的引用如文件路径、数据库ID或文本摘要。2. 并行化节点执行如果两个节点间没有数据依赖即它们不需要对方的输出理论上可以并行执行以加快速度。LangGraph本身是线性执行的但你可以通过设计将可并行任务放在同一个节点内用asyncio或线程池实现并发或者探索社区中关于LangGraph并行执行的研究注意这属于高级用法可能破坏状态流的一致性。3. LLM调用优化缓存对相同的提示词进行缓存可以节省大量成本和时间。可以使用langchain.cache如InMemoryCache,SQLiteCache。批量处理如果工作流需要处理多个相似但独立的任务如分析10份文档不要创建10个独立的工作流实例。可以设计一个“批处理”节点在该节点内循环调用LLM并将结果汇总更新到State。4. 工具Tools使用规范权限与安全暴露给Agent的工具要有严格的权限控制。特别是涉及写操作发邮件、改数据库或敏感信息查询的工具必须在工具内部或调用前增加权限校验逻辑。工具描述清晰给工具的函数写清晰、准确的description这直接影响到LLM能否正确理解和使用该工具。描述中应包含输入参数的明确说明。5.3 与其他工作流工具的对比你可能会听到n8n、Dify、Flowable等工具。它们和LangGraph定位有何不同工具核心定位与LangGraph对比LangGraphAI智能体Agent工作流编排。核心是协调LLM、工具和记忆实现自主决策和复杂推理。专为AI设计深度集成LLM生态LangChain状态管理和条件分支为AI场景高度优化。n8n通用自动化工作流。连接各种SaaS应用、API和数据库实现数据同步、通知等业务流程自动化。更偏向于“无代码/低代码”的IT自动化有丰富的预制应用连接器但不擅长处理基于LLM的复杂逻辑判断。DifyAI应用开发平台。提供可视化界面构建基于LLM的应用包含RAG、Agent工作流等功能。Dify是一个更高层的平台其工作流功能可能底层就使用了LangGraph。LangGraph是代码库提供更底层的灵活性和控制力。FlowableBPMN业务流程管理引擎。用于企业级复杂业务流程的建模、执行和监控如审批流。处理严格定义、规则驱动的业务流程注重合规、审计和人工任务。LangGraph处理的是非确定性的、由AI驱动的动态流程。选择建议如果你的核心是构建一个能思考、能调用工具、能处理开放域任务的AI智能体LangGraph是目前最专业、最灵活的选择。如果你只是想将ChatGPT连接到Slack和Google Sheetsn8n可能更快。如果你想快速搭建一个带界面的AI应用而不想写太多后端代码Dify是优秀选择。