最近在尝试把一些零散的 AI 工具串联起来做成一个能自动处理复杂任务的“智能体”时我遇到了一个典型问题流程跑通了但一旦任务步骤变多或者需要多个“智能体”协作代码就迅速变成了一团乱麻。状态管理、步骤跳转、错误处理、循环逻辑……这些原本在传统软件开发里很清晰的概念在构建 AI 驱动的应用时却常常需要自己从头“发明轮子”。这让我开始关注一个叫LangGraph的框架。它不像 LangChain 那样提供海量的工具链而是专注于一件事用图Graph的方式来定义和管理 AI 的工作流。听起来有点抽象但它的核心价值非常直接——把一次性的、脆弱的脚本变成可预测、可调试、可长期运行的“企业级”智能体系统。很多人一听到“图”和“架构”就觉得头大认为这是只有大厂架构师才需要关心的东西。这其实是个误解。LangGraph 解决的不是“高并发”或“分布式”那种宏观架构问题它解决的是每个开发者在构建 AI 应用时都会遇到的微观工作流失控问题。比如一个客服机器人到底是该直接回答还是去查知识库或者转人工这个决策逻辑和状态流转用 if-else 硬写会非常痛苦而 LangGraph 提供了一种声明式的、可视化的方式来定义它。所以这篇文章不会堆砌晦涩的论文术语而是从一个实践者的角度带你理解 LangGraph 到底改变了什么。我们会从“为什么需要它”开始一步步拆解它的核心概念并通过一个从单智能体到多智能体协作的实战案例让你看到如何用它把想法变成稳定、可维护的代码。你会发现它的学习曲线远比想象中平缓而带来的清晰度和可控性提升是立竿见影的。1. 先搞清楚LangGraph 解决的不是“智能”问题而是“流程”问题在深入代码之前我们必须先建立一个关键认知LangGraph 本身不生产“智能”。它不提供大语言模型LLM也不直接处理自然语言理解。它的核心定位是一个“编排框架”。你可以把它想象成一个超级增强版的if-else和while循环专门为 AI 应用设计。在传统的编程中流程控制是确定性的条件 A 成立就执行 B。但在 AI 应用中很多“条件”是 LLM 根据上下文“思考”后产生的非确定性结果。比如“用户这个问题我能否回答是否需要更多信息”。1.1 传统 AI 脚本的典型困境状态散落与逻辑胶水在没有专门框架之前我们如何构建一个多步骤的 AI 应用通常的写法是这样的# 一种常见的、脆弱的实现方式 def handle_user_query(query, conversation_history): # 步骤1判断意图 intent llm_classify_intent(query) if intent qa: # 步骤2检索知识 docs retrieve_knowledge(query) # 步骤3生成回答 answer llm_generate_answer(query, docs) return answer elif intent chitchat: # 步骤2闲聊处理 answer llm_chitchat(query, conversation_history) return answer elif intent need_more_info: # 步骤2追问澄清 clarifying_question llm_generate_clarification(query) # 步骤3等待用户回复... 状态怎么保存 # 我们需要把当前状态如等待 clarification和中间结果存到某处 save_state(user_id, {waiting_for: clarification, original_query: query}) return clarifying_question # ... 更多的 elif这段代码的问题显而易见状态管理混乱当流程需要暂停等待用户输入如追问时当前的“进度”和中间数据original_query必须手动保存到数据库或缓存中与业务逻辑高度耦合。逻辑嵌套深每个分支可能又衍生出新的分支代码可读性和可维护性急剧下降。难以调试和可视化你无法一眼看出整个应用有哪些节点、哪些路径。当流程复杂后理清逻辑依赖关系变得异常困难。缺乏通用模式错误重试、循环控制、并行执行、人工审核Human-in-the-loop等常见需求都需要重新发明一套实现。LangGraph 的出现正是为了系统性地解决这些“流程胶水”问题。它让你能用“画图”的方式定义应用图中的节点是功能单元调用 LLM、执行工具、条件判断边是控制流。1.2 LangGraph 的核心抽象图、状态、边理解 LangGraph只需要抓住三个核心概念State状态这是一个贯穿整个工作流的、共享的上下文对象。它通常是一个字典或 Pydantic 模型包含了所有节点需要读取和写入的数据。比如{messages: [...], user_query: ..., retrieved_docs: [...], next_step: ...}。LangGraph 负责在节点间传递和持久化这个状态。Node节点一个执行单元通常是一个函数。它接收当前的State执行一些操作如调用 LLM、运行工具然后返回一个对 State 的更新。例如一个“检索”节点会读取State[query]调用检索器然后将结果写入State[documents]。Edge边决定工作流下一步该走到哪个节点。边分为两种条件边Conditional Edge根据State中的某个值通常是 LLM 或函数判断的结果来决定下一个节点。这是实现分支逻辑的关键。普通边无条件地指向下一个节点。通过组合这些元素你可以定义出循环节点A - 条件边 - 根据条件返回节点A或结束、分支节点 - 条件边 - 节点B 或 节点C和并行高级特性等复杂流程。关键理解LangGraph 将你的应用逻辑从“代码的行文顺序”中解放出来变成了“对一张图的定义”。这张图就是你的应用架构图它清晰、可视、易于修改和推理。2. 从零开始用 LangGraph 构建你的第一个智能体工作流理论说得再多不如亲手搭建一个。让我们从一个最简单的例子开始一个能进行多轮对话并在必要时主动查询天气的聊天助手。这个智能体的逻辑是接收用户输入。判断用户是否在问天气意图识别。如果是调用天气查询工具并整合信息回复。如果不是直接让 LLM 生成回复。等待下一轮用户输入。2.1 环境准备与状态定义首先安装必要的包。这里我们使用 OpenAI 的模型你也可以替换为 Ollama 本地模型。pip install langgraph langchain-openai langchain-community接下来定义工作流的状态。我们使用TypedDict来获得更好的类型提示。from typing import TypedDict, List, Annotated import operator from langgraph.graph import StateGraph, END # 1. 定义状态结构 class AgentState(TypedDict): # 完整的对话历史 messages: Annotated[List[str], operator.add] # 用户的最新输入 latest_user_input: str # LLM 判断的意图 intent: str # 天气查询的结果如果有 weather_info: strAnnotated[List[str], operator.add]是 LangGraph 的一个精妙设计。它表示messages字段是一个列表当多个节点对它进行修改时默认使用operator.add即列表的extend操作来合并更新而不是覆盖。这非常适合记录对话历史。2.2 创建节点功能单元的实现节点就是普通的 Python 函数它接收State并返回一个包含更新字段的字典。from langchain_openai import ChatOpenAI import os os.environ[OPENAI_API_KEY] 你的密钥 llm ChatOpenAI(modelgpt-3.5-turbo) # 节点1接收用户输入 def receive_input(state: AgentState): # 在实际应用中这里可能从API接收数据 user_input input(用户: ) # 模拟用户输入 return {latest_user_input: user_input, messages: [f用户: {user_input}]} # 节点2判断意图 def classify_intent(state: AgentState): prompt f 请判断用户最新消息的意图。消息是{state[latest_user_input]} 可选意图[闲聊, 查询天气, 其他]。 只返回意图关键词。 response llm.invoke(prompt) intent response.content.strip() return {intent: intent} # 节点3查询天气模拟工具调用 def call_weather_tool(state: AgentState): # 这里模拟一个工具调用真实情况可能调用API city 北京 # 简单起见假设查询北京 weather 晴25摄氏度 return {weather_info: f{city}的天气是{weather}} # 节点4生成最终回复 def generate_response(state: AgentState): if state[intent] 查询天气: # 结合天气信息生成回复 prompt f用户问{state[latest_user_input]}。已知天气信息{state[weather_info]}。请生成友好回复。 else: # 普通闲聊回复 prompt f请根据对话历史回复用户最新消息{state[latest_user_input]}。对话历史{state[messages][-5:]} response llm.invoke(prompt) ai_message f助手: {response.content} return {messages: [ai_message]}2.3 组装成图定义工作流逻辑这是 LangGraph 最核心的部分我们将节点和边连接起来形成完整的工作流。# 初始化图构建器 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(接收输入, receive_input) workflow.add_node(意图识别, classify_intent) workflow.add_node(查询天气, call_weather_tool) workflow.add_node(生成回复, generate_response) # 3. 设置入口点 workflow.set_entry_point(接收输入) # 4. 添加边定义流程逻辑 workflow.add_edge(接收输入, 意图识别) # 接收输入后一定进行意图识别 # 从“意图识别”到下一个节点需要条件判断 def route_by_intent(state: AgentState): # 根据意图决定下一个节点 if state[intent] 查询天气: return 查询天气 else: # 闲聊或其他意图直接去生成回复 return 生成回复 workflow.add_conditional_edges( 意图识别, # 源节点 route_by_intent, # 路由判断函数 { 查询天气: 查询天气, # 如果函数返回“查询天气”则跳转到“查询天气”节点 生成回复: 生成回复, # 如果函数返回“生成回复”则跳转到“生成回复”节点 } ) # 5. 添加普通边 workflow.add_edge(查询天气, 生成回复) # 查询完天气后去生成回复 workflow.add_edge(生成回复, END) # 生成回复后工作流结束 # 6. 编译图 app workflow.compile()现在一张清晰的“图”就定义好了接收输入 - 意图识别 - (查询天气 - 生成回复 | 生成回复) - END。2.4 运行与可视化你可以像调用函数一样运行这个工作流初始状态是一个空字典。# 运行工作流 initial_state {messages: [], latest_user_input: , intent: , weather_info: } final_state app.invoke(initial_state) print(最终对话历史:, final_state[messages])LangGraph 的一个强大功能是可视化。你可以将图导出为 PNG 图片直观地看到整个流程。from IPython.display import Image, display try: display(Image(app.get_graph().draw_mermaid_png())) except: # 如果环境不支持可以输出Mermaid文本到支持的工具查看 print(app.get_graph().draw_mermaid())这张图就是你应用的活文档任何新成员都能快速理解业务逻辑的全貌。3. 进阶实战构建多智能体协作系统单智能体工作流只是开始。LangGraph 真正的威力在于优雅地编排多个智能体或称为“角色”协同工作解决更复杂的任务。例如一个内容创作系统可能包含“策划”、“撰稿”、“校对”三个智能体。让我们设计一个“多智能体审阅系统”Writer写手负责根据主题生成初稿。Critic批评家负责评审初稿提出修改意见。Editor编辑负责综合初稿和意见输出最终稿。 流程是Writer - Critic - Editor。Critic 可以要求 Writer 重写形成循环。3.1 定义多智能体状态与节点状态需要容纳多个智能体的输出。from typing import TypedDict, List, Annotated, Literal import operator class MultiAgentState(TypedDict): topic: str # 写作主题 draft: str # 当前草稿 critique: str # 批评意见 final_output: str # 最终输出 # 一个标志位控制流程走向 needs_revision: Annotated[Literal[yes, no], operator.add] # 记录所有步骤的日志 process_log: Annotated[List[str], operator.add] # 初始化LLM (可以使用不同配置的LLM代表不同角色) llm ChatOpenAI(modelgpt-4) def writer_node(state: MultiAgentState): 写手节点生成初稿 prompt f你是一位专业写手。请根据以下主题撰写一篇短文初稿{state[topic]} response llm.invoke(prompt) log_entry fWriter 生成了初稿。 return {draft: response.content, process_log: [log_entry]} def critic_node(state: MultiAgentState): 批评家节点评审初稿 prompt f你是一位严厉的批评家。请评审以下文章初稿指出其在逻辑、事实、文笔上的具体问题并提出修改建议。 初稿{state[draft]} 请直接输出批评意见。 response llm.invoke(prompt) # 简单判断如果批评意见超过一定长度或包含“重写”等关键词则要求修改 needs_revision yes if 重写 in response.content or len(response.content) 100 else no log_entry fCritic 给出了评审意见要求重写{needs_revision}。 return {critique: response.content, needs_revision: [needs_revision], process_log: [log_entry]} def editor_node(state: MultiAgentState): 编辑节点综合草稿和意见产出终稿 prompt f你是一位资深编辑。请综合以下初稿和批评意见输出一份修改后的最终稿。 初稿{state[draft]} 批评意见{state[critique]} 最终稿 response llm.invoke(prompt) log_entry fEditor 产出了最终稿。 return {final_output: response.content, process_log: [log_entry]}3.2 实现带循环的图逻辑这里的关键在于Critic 节点后需要一个条件边根据needs_revision的值决定是返回 Writer 重写还是继续到 Editor。from langgraph.graph import StateGraph, END # 构建图 workflow StateGraph(MultiAgentState) # 添加节点 workflow.add_node(Writer, writer_node) workflow.add_node(Critic, critic_node) workflow.add_node(Editor, editor_node) # 设置流程 workflow.set_entry_point(Writer) workflow.add_edge(Writer, Critic) # 定义条件路由函数 def decide_after_critique(state: MultiAgentState): # 取最近一次 needs_revision 的值 if state.get(needs_revision) and state[needs_revision][-1] yes: return rewrite # 需要重写 else: return finalize # 可以定稿 # 添加条件边 workflow.add_conditional_edges( Critic, decide_after_critique, { rewrite: Writer, # 返回 Writer 节点形成循环 finalize: Editor, } ) workflow.add_edge(Editor, END) # 编译 app workflow.compile()3.3 运行与观察循环现在运行这个多智能体系统。我们设置一个初始主题。initial_state { topic: 人工智能对未来工作的影响, draft: , critique: , final_output: , needs_revision: [], process_log: [] } # 运行并设置一个最大迭代次数防止无限循环 for i in range(5): # 最多循环5次 result app.invoke(initial_state) print(f\n 第{i1}轮运行 ) print(f流程日志: {result[process_log]}) print(f是否需要重写: {result.get(needs_revision, [])}) if result.get(final_output): print(f最终输出: {result[final_output][:200]}...) # 预览前200字符 break # 将本轮结果作为下一轮的初始状态模拟持续运行 initial_state result else: print(达到最大循环次数流程可能陷入循环。)通过输出日志你可以清晰地看到Writer - Critic - (Writer) - Critic - Editor的完整协作过程。这种循环和条件跳转逻辑如果用传统代码编写会非常复杂且难以维护而在 LangGraph 中它只是一张清晰的图。4. 从原型到生产LangGraph 的工程化考量将 LangGraph 工作流从实验脚本变为可投入生产环境的服务还需要考虑以下几个关键方面。这些是区分“玩具”和“工具”的核心。4.1 持久化与检查点让工作流“暂停”与“恢复”生产环境中的工作流往往需要运行很长时间如等待用户回复、等待外部 API 回调或者需要应对服务重启。LangGraph 的Checkpoint机制解决了这个问题。它允许你将工作流的完整状态包括执行到了哪个节点保存到数据库如 MySQL、PostgreSQL。当需要恢复时可以从上一个检查点继续执行。from langgraph.checkpoint.sqlite import SqliteSaver import sqlite3 # 1. 创建 SQLite 存储生产环境可用 PostgreSQL 等 conn sqlite3.connect(checkpoints.db) checkpointer SqliteSaver(conn) # 2. 在编译图时传入 checkpointer app workflow.compile(checkpointercheckpointer) # 3. 运行时会自动创建检查点 config {configurable: {thread_id: user_123_session_1}} initial_state {topic: ...} # 第一次调用创建流程 result1 app.invoke(initial_state, configconfig) print(f运行ID: {result1[metadata][run_id]}) # 模拟流程暂停... (例如等待用户输入) # 4. 根据 thread_id 和 run_id 恢复流程 # 我们可以获取当前状态或继续调用下一个节点 # app.get_state(config) 可以获取最新状态 # 再次调用 app.invoke() 会从上一个检查点继续这对于构建需要多轮交互的对话机器人或长耗时批处理任务至关重要。4.2 稳定性增强错误处理、超时与重试任何依赖外部服务LLM API、工具 API的节点都可能失败。LangGraph 允许你为节点配置错误处理策略。from langgraph.graph import StateGraph from langgraph.types import Command, interrupt def unreliable_external_api_node(state): import random if random.random() 0.3: # 模拟30%失败率 raise ConnectionError(API调用失败) return {result: success} def handle_api_error(state, error): 错误处理函数 # 可以记录日志、发送警报、修改状态以尝试其他方案等 print(f节点执行失败错误: {error}) # 返回一个指令例如重试、跳转到其他节点或结束 # return Command(retry5) # 重试5次需框架支持 # 或者跳转到降级处理节点 return {fallback_result: used cached data, next_node: fallback_node} # 在定义节点时可以关联错误处理器具体语法可能随版本变化此为概念示意 # workflow.add_node_with_fallback(api_node, unreliable_external_api_node, handle_api_error)此外对于 LLM 调用节点务必设置合理的超时timeout和重试逻辑可以使用 LangChain 的with_retry或with_fallbacks来包装 LLM 调用。4.3 可观测性与调试让黑盒变透明复杂的图工作流调试起来很困难。LangGraph 提供了强大的跟踪Tracing和日志功能。LangSmith 集成这是最佳实践。LangSmith 可以可视化记录每一次工作流执行的全过程包括每个节点的输入输出、耗时、Token 使用量等。这对于性能分析、错误排查和成本监控不可或缺。自定义日志如我们在状态中定义的process_log字段在每个节点记录关键动作是另一种有效的调试手段。图可视化如前所述app.get_graph().draw_mermaid()生成的图是理解和沟通架构的最佳工具。4.4 性能与扩展性思考节点粒度不要把所有逻辑塞进一个节点。将功能拆分为细粒度的节点如“意图识别”、“参数提取”、“工具调用”、“结果解析”有利于复用、测试和并行化。并发执行LangGraph 支持定义并行节点通过add_node和add_edge的特殊组合当多个节点间没有数据依赖时可以同时运行以提高效率。外部化状态对于非常大的状态如长文档可以考虑不将其完全放在工作流状态中而是只存储一个引用 ID如数据库记录 ID在节点内部按需加载。这可以避免状态对象过大带来的序列化开销。5. 核心价值与适用边界什么时候该用 LangGraph经过上面的实践我们可以更清晰地总结 LangGraph 的价值和它最适合的场景。5.1 LangGraph 带来的核心改变声明式编排你将“业务逻辑是什么”与“逻辑如何执行”分离开。你定义图框架负责调度、状态管理和持久化。这带来了极高的代码可读性和可维护性。复杂流程的直观建模循环、条件分支、多角色协作等模式用图来表示比用过程式代码直观得多。它迫使你思考应用的状态机模型这是一种更严谨的设计方式。内置的工程化支持检查点、可视化、与 LangSmith 的深度集成这些开箱即用的功能让你能快速搭建出具备生产就绪特性的应用而不是从零开始造轮子。与 LangChain 生态无缝衔接你可以轻松使用 LangChain 提供的数百个组件工具、检索器、链作为图中的节点享受两个生态的优势。5.2 最适合 LangGraph 的场景多步骤、有状态的对话系统客服机器人、游戏 NPC、教学助手等需要根据历史对话决定下一步行动。多智能体协作系统如我们演示的写作审阅流程或模拟辩论、竞拍、协同决策等场景。需要人工干预的流程内容审核、合同审批等流程可以在某个节点暂停等待人工审核Human-in-the-loop后再继续。复杂的批处理或数据处理管道其中某些步骤需要条件判断或循环例如数据清洗、验证、增强的流水线。5.3 可能不适用或需要简化的场景极其简单的线性链如果你的应用只是输入 - LLM - 输出没有分支和状态直接调用 LLM 或使用简单的 LangChainLCEL可能更轻量。对延迟极其敏感的实时应用图的调度本身有微小开销。对于要求毫秒级响应的场景需要仔细评估和性能测试。完全无状态的请求/响应每次请求都是独立的不需要记住之前任何交互。5.4 给实践者的最终建议从小图开始不要试图一开始就设计一个包含几十个节点的庞大系统。从一个核心的、3-5个节点的流程开始验证想法。状态设计是关键花时间设计好你的State结构。想清楚哪些数据是全局共享的哪些是节点局部使用的。良好的状态设计是清晰工作流的基础。拥抱可视化养成画图的习惯。在编码前先用白板或绘图工具画出工作流草图。这能帮你理清逻辑也便于团队沟通。优先考虑可观测性在早期就集成 LangSmith 或建立自己的日志规范。当流程出错时清晰的执行轨迹能帮你快速定位问题节点。理解它只是一个编排框架LangGraph 不解决如何调用 API、如何解析 PDF、如何优化提示词等问题。它解决的是如何将这些能力有序、可靠地组织起来。你的核心竞争力仍然在于对业务逻辑的深刻理解和对 AI 能力的恰当运用。LangGraph 的出现标志着 AI 应用开发从“脚本阶段”向“工程阶段”的演进。它提供的不是某个炫酷的新模型能力而是一套让复杂 AI 工作流变得可控、可维护、可扩展的工程学工具。当你下次再面对那些交织着判断、循环与协作的 AI 应用需求时不妨先画一张图或许你会发现最复杂的部分已经有人为你提供了优雅的解决方案。