你有没有试过用大模型写个能自己跑流程的智能体结果发现它要么卡在某个步骤不动要么逻辑混乱最后只能手动介入或者你看到 LangGraph、智能体这些概念很火但打开官方文档面对 StateGraph、Nodes、Edges 这些抽象概念感觉像在看天书不知道从何下手这正是很多开发者从“知道智能体”到“用好智能体”之间最大的鸿沟。智能体不是简单的“大模型API调用”它需要一个清晰的架构来管理状态、控制流程、处理分支和循环。而 LangGraph正是为解决这个问题而生的框架。它不是一个替代品而是一个“流程编排引擎”把大模型的单次问答能力组织成可预测、可调试的复杂工作流。网上很多教程要么停留在概念要么代码片段零散很难串联成一个完整的、可运行的实战项目。更重要的是很多人忽略了智能体开发中比写代码更关键的东西对状态State的精准设计、对工具Tools的边界定义以及对异常和循环的稳健处理。这些才是决定一个智能体能否从Demo走向实际应用的核心。这篇文章我们就彻底拆解 LangGraph。我不会只给你一堆代码而是带你理解它背后的设计哲学然后通过一个从零开始的实战项目——构建一个“技术博客选题与大纲生成智能体”——来掌握其核心组件和开发心法。你会发现一旦理解了它的“图”思维很多复杂的智能体场景都会变得清晰可控。1. 为什么是 LangGraph理解“图”思维与智能体架构的困境在深入代码之前我们必须先回答一个问题当我们在说“智能体”时我们到底在构建什么以及为什么需要 LangGraph 这样的框架1.1 智能体的本质超越单次问答的状态机一个最简单的智能体可能是“用户提问 - 大模型思考 - 调用工具 - 返回结果”。但这只解决了一次性任务。真实的智能体比如一个客服机器人、一个数据分析助手或一个内容创作助手其工作流往往是这样的理解用户复杂意图可能涉及多轮澄清。规划一系列步骤搜索信息、查询数据库、执行计算、生成报告。根据每一步的结果动态决定下一步是继续、分支、循环还是终止。在整个过程中维护一个共享的“上下文”或“状态”比如已收集的信息、已执行的操作、用户的偏好等。这本质上是一个有状态的工作流或者说一个状态机。传统的链式调用如 LangChain 的 LCEL擅长线性管道但对于包含条件分支、循环、并行等复杂逻辑的流程就显得力不从心代码会迅速变得难以维护。LangGraph 的核心价值就是把智能体的工作流明确定义为一张“图”Graph。在这张图里节点Nodes代表一个具体的执行单元比如“调用大模型”、“执行工具”、“更新状态”。边Edges代表节点之间的流转条件比如“如果工具执行成功则前往节点A如果失败则前往节点B”。状态State是一个贯穿整个图、所有节点都能读写的数据结构它是智能体的“记忆”和“工作区”。这种“图”的思维让复杂的、非线性的智能体逻辑变得可视化、可编排、可调试。它把智能体从“黑盒”变成了一个你可以清晰看到数据流和控制流的“白盒”系统。1.2 LangGraph 与 LangChain 的关系不是替代是补充很多人会混淆 LangGraph 和 LangChain。简单来说LangChain是一个庞大的生态提供了连接大模型、工具、向量数据库等组件的各种“积木”Chains, Agents, Tools, Retrievers等。它的早期 Agent 实现如 ReAct逻辑相对固化复杂流程的定制成本高。LangGraph是 LangChain 生态系统内一个专注于编排复杂、有状态工作流的框架。它继承了 LangChain 的组件如LLM、Tools但提供了更强大、更灵活的流程控制能力。你可以把它看作是 LangChain 生态中专门解决“智能体流程引擎”问题的利器。所以你的技术栈很可能是LangGraph流程编排 LangChain 组件LLM、Tools等 任意大模型 API。1.3 实战项目的目标一个能闭环工作的内容智能体为了不让学习停留在理论我们设定一个贯穿全文的实战目标构建一个“技术博客选题与大纲生成智能体”。它的工作流程是输入一个模糊的技术主题如“如何学习LangGraph”。过程智能体首先分析主题生成几个更具体的候选选题方向。让用户或模拟用户选择一个方向。针对选定的方向智能体进行网络搜索调用工具收集最新的资料和趋势。基于搜索到的信息生成一份详细的、结构化的博客大纲。最后可以将大纲保存为文件。输出一份可直接用于写作的、信息充实的博客大纲。这个项目涵盖了智能体的典型要素多步骤规划、工具调用、条件判断用户选择、状态维护。接下来我们就用 LangGraph 一步步实现它。2. 环境搭建与核心组件初识从零构造你的第一张图在开始画“图”之前先把画布和颜料准备好。2.1 基础环境配置假设你使用 Python推荐使用虚拟环境。核心依赖如下pip install langgraph langchain-openai langchain-community duckduckgo-searchlanggraph: 核心框架。langchain-openai: 用于调用 OpenAI 系列模型如 GPT-4o, GPT-3.5-Turbo。如果你用其他模型需安装对应的 LangChain 集成包。langchain-community: 包含许多社区贡献的工具和组件我们用它里面的 DuckDuckGo 搜索工具。duckduckgo-search: 搜索工具的后端。关键点你需要一个可用的 OpenAI API Key或其他兼容 OpenAI 接口的大模型 API Key并将其设置为环境变量。export OPENAI_API_KEYyour-api-key-here # 或者在代码中设置 import os os.environ[OPENAI_API_KEY] your-api-key-here2.2 理解 LangGraph 的四大核心抽象这是理解 LangGraph 的基石务必厘清。State状态是什么一个类似字典TypedDict的结构定义了在整个工作流中流转和共享的所有数据。为什么重要状态是智能体的“记忆”。节点从状态中读取输入将输出写回状态。良好的状态设计是流程清晰的关键。在我们的项目中状态可能包含topic原始主题、refined_topics细化后的选题列表、selected_topic用户选择的主题、search_results搜索到的资料、outline生成的大纲。Node节点是什么一个执行具体任务的函数。它接收当前State作为输入执行操作如调用LLM、运行工具然后返回一个更新后的State或包含更新部分的字典。关键节点应该是纯函数其输出只依赖于输入的状态这有利于测试和调试。Edge边是什么定义了从一个节点到下一个节点的流转规则。分为两种普通边Linear Edge无条件地从一个节点指向下一个节点。条件边Conditional Edge根据State中的某个值或某个函数的返回值决定下一步走向哪个节点。这是实现分支和循环的核心。Graph图是什么由Nodes和Edges组成的网络。LangGraph 提供了一个StateGraph类来帮你构建和管理这个网络。最终产物通过编译compile图你会得到一个可执行的Graph对象它像一台状态机你输入初始状态它就会按照你定义的图逻辑运行。2.3 构建第一张简单的图感受流程在实现复杂智能体前我们先建一个超简单的图来建立直觉。from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END import operator # 1. 定义状态我们的“工作区”里有什么 class MyState(TypedDict): message: Annotated[str, operator.add] # 使用注解表示这个字段会被追加append counter: int # 2. 定义节点函数 def node_hello(state: MyState) - dict: 节点A打招呼并计数 new_message fHello! Counter is {state[counter]}. return {message: new_message, counter: state[counter] 1} def node_world(state: MyState) - dict: 节点B说世界 return {message: World! } # 3. 构建图 builder StateGraph(MyState) # 添加节点 builder.add_node(hello_node, node_hello) builder.add_node(world_node, node_world) # 设置入口点 builder.set_entry_point(hello_node) # 添加边hello_node 执行完后无条件前往 world_node builder.add_edge(hello_node, world_node) # world_node 执行完后图结束 builder.add_edge(world_node, END) # 4. 编译图得到可执行对象 graph builder.compile() # 5. 运行图 initial_state MyState(message, counter0) final_state graph.invoke(initial_state) print(final_state) # 输出{message: Hello! Counter is 0. World! , counter: 1}这段代码展示了完整流程定义状态 - 定义节点 - 构建图添加节点和边- 编译 - 运行。注意Annotated[str, operator.add]的用法它告诉 LangGraphmessage字段是追加模式而不是覆盖。这是管理状态更新的重要机制。3. 实战构建博客选题与大纲生成智能体现在我们将核心组件应用于实战项目。我们将构建一个包含多个节点和条件分支的复杂图。3.1 第一步设计智能体的状态State状态设计是蓝图。我们要想清楚智能体在整个流程中需要记住什么。from typing import TypedDict, List, Optional, Annotated import operator class BlogAgentState(TypedDict): 博客智能体的完整状态定义 # 输入 original_topic: str # 用户原始输入的主题 # 第一阶段选题细化 refined_topics: Optional[List[str]] # 模型生成的细化选题列表 selected_topic_index: Optional[int] # 用户模拟选择的索引 # 第二阶段资料收集 search_query: Optional[str] # 基于选定主题生成的搜索词 search_results: Optional[List[str]] # 搜索到的资料摘要列表 # 第三阶段大纲生成 outline: Optional[str] # 最终生成的博客大纲 # 辅助与日志 messages: Annotated[List[str], operator.add] # 记录智能体的思考和执行日志 error: Optional[str] # 记录运行中的错误信息设计解析original_topic是起点不可变。refined_topics,selected_topic_index,search_query,search_results,outline代表了流程中不同阶段的产出用Optional表示可能为空。messages使用operator.add用于追加日志方便调试和追溯。error用于异常处理。3.2 第二步实现各个节点Nodes每个节点负责一个具体的、原子性的任务。from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_community.tools import DuckDuckGoSearchRun import json # 初始化LLM和工具 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.7) search_tool DuckDuckGoSearchRun() # 节点1细化选题 def node_refine_topic(state: BlogAgentState) - dict: 接收原始主题生成3个更具体的候选选题。 prompt ChatPromptTemplate.from_messages([ (system, 你是一个资深技术博客策划人。请将用户给出的宽泛技术主题细化成3个具体、有吸引力、适合单篇博客展开的选题。以JSON列表格式返回例如[\选题1\, \选题2\, \选题3\]), (human, 原始主题{topic}) ]) chain prompt | llm response chain.invoke({topic: state[original_topic]}) try: # 尝试解析LLM返回的JSON topics json.loads(response.content) if isinstance(topics, list) and len(topics) 0: log_msg f已生成细化选题{topics} return { refined_topics: topics, messages: [log_msg] } else: raise ValueError(解析结果不是有效列表) except (json.JSONDecodeError, ValueError) as e: # 如果解析失败提供备选方案 log_msg fLLM返回格式异常使用备用选题。错误{e} backup_topics [ f{state[original_topic]}核心概念与快速入门, f深入理解{state[original_topic]}的工作原理, f{state[original_topic]}实战从零构建一个完整项目 ] return { refined_topics: backup_topics, messages: [log_msg], error: str(e) } # 节点2模拟用户选择实际应用中这里可连接真实UI def node_select_topic(state: BlogAgentState) - dict: 模拟用户从候选选题中选择一个。这里我们简单选择第一个。 if not state.get(refined_topics): return {error: 没有可选的细化选题, messages: [错误选题列表为空]} # 模拟选择索引0。真实场景可通过参数传入或交互获取。 selected_index 0 selected_topic state[refined_topics][selected_index] log_msg f模拟用户选择索引[{selected_index}] - {selected_topic} return { selected_topic_index: selected_index, search_query: selected_topic, # 直接将选题作为搜索词 messages: [log_msg] } # 节点3执行网络搜索 def node_web_search(state: BlogAgentState) - dict: 使用选定的主题进行网络搜索获取最新资料。 query state.get(search_query) if not query: return {error: 搜索词为空, messages: [错误无法执行搜索搜索词缺失]} try: search_result search_tool.run(query) # 对搜索结果进行简单处理例如截取前2000字符 processed_result search_result[:2000] if search_result else 未找到相关信息 log_msg f对『{query}』完成搜索获取信息长度{len(processed_result)} return { search_results: [processed_result], # 存为列表方便后续扩展多条结果 messages: [log_msg] } except Exception as e: error_msg f搜索工具执行失败{e} return { error: error_msg, search_results: [搜索过程发生错误], messages: [error_msg] } # 节点4生成博客大纲 def node_generate_outline(state: BlogAgentState) - dict: 基于原始主题、选定选题和搜索资料生成详细博客大纲。 selected_topic state[refined_topics][state[selected_topic_index]] search_info state.get(search_results, [无额外资料])[0] prompt ChatPromptTemplate.from_messages([ (system, 你是一位优秀的科技文章作者。请根据以下信息撰写一份详细、结构清晰、适合技术博客的Markdown格式大纲。 大纲要求 1. 包含引言、核心内容至少3个大节每节2-4个子点、总结与展望。 2. 核心内容要结合提供的搜索资料确保信息时效性和准确性。 3. 在适当位置标注【需补充案例】或【需核实数据】。 4. 输出纯Markdown内容。), (human, 博客核心主题{selected_topic} 原始背景主题{original_topic} 相关搜索资料摘要 {search_info} 请生成大纲 ) ]) chain prompt | llm response chain.invoke({ selected_topic: selected_topic, original_topic: state[original_topic], search_info: search_info }) log_msg f已基于主题『{selected_topic}』生成博客大纲。 return { outline: response.content, messages: [log_msg] } # 节点5处理错误一个兜底节点 def node_handle_error(state: BlogAgentState) - dict: 当其他节点设置error状态时此节点被触发记录错误并终止流程。 error_msg state.get(error, 未知错误) log_msg f流程因错误终止{error_msg} # 可以在这里添加错误通知逻辑如发送警报 return { messages: [log_msg] }节点设计要点原子性每个节点只做一件事。健壮性都有基本的错误处理如try-except提供默认值。状态更新返回的字典只包含需要更新的State字段LangGraph 会自动合并。工具使用node_web_search展示了如何集成 LangChain Tool。3.3 第三步编排图结构Graph与条件边这是 LangGraph 最精妙的部分我们用图来定义智能体的“决策逻辑”。from langgraph.graph import StateGraph, END # 1. 创建图构建器并指定状态类型 builder StateGraph(BlogAgentState) # 2. 添加所有节点 builder.add_node(refine_topic, node_refine_topic) builder.add_node(select_topic, node_select_topic) builder.add_node(web_search, node_web_search) builder.add_node(generate_outline, node_generate_outline) builder.add_node(handle_error, node_handle_error) # 3. 设置入口点从“细化选题”开始 builder.set_entry_point(refine_topic) # 4. 添加普通边无条件流转 # 细化选题后进入选择节点 builder.add_edge(refine_topic, select_topic) # 选择选题后进入搜索节点 builder.add_edge(select_topic, web_search) # 搜索完成后进入生成大纲节点 builder.add_edge(web_search, generate_outline) # 生成大纲后流程结束 builder.add_edge(generate_outline, END) # 5. 添加条件边实现错误处理分支 # 定义一个路由函数决定下一步去哪 def route_after_error(state: BlogAgentState) - str: 根据状态中是否有error决定下一步。 如果有error去错误处理节点否则继续正常流程。 if state.get(error): return handle_error # 否则返回一个特殊值让 LangGraph 继续执行预设的普通边 # 这里我们返回 None并依赖下面的 add_conditional_edges 配置 return __continue__ # 为关键节点添加条件边。 # 这意味着在执行完 refine_topic, select_topic, web_search 节点后 # 都会先调用 route_after_error 函数判断。 # 如果函数返回 “handle_error”则跳转到错误处理节点。 # 如果返回 “__continue__”则沿着之前用 add_edge 定义的普通边继续走。 builder.add_conditional_edges( refine_topic, route_after_error, {handle_error: handle_error, __continue__: select_topic} # 映射关系 ) builder.add_conditional_edges( select_topic, route_after_error, {handle_error: handle_error, __continue__: web_search} ) builder.add_conditional_edges( web_search, route_after_error, {handle_error: handle_error, __continue__: generate_outline} ) # 错误处理节点执行后直接结束流程 builder.add_edge(handle_error, END) # 6. 编译图生成最终的可执行智能体 blog_agent_graph builder.compile()图结构解析主流程refine_topic-select_topic-web_search-generate_outline-END。这是一个线性主干。错误处理分支在refine_topic,select_topic,web_search这三个可能出错的节点后都添加了条件边。route_after_error函数检查state[‘error’]。如果发现错误流程立即跳转到handle_error节点然后结束。这实现了短路错误处理。灵活性你可以轻松修改这个图。例如在select_topic后根据用户选择的不同跳转到不同的搜索策略节点。这就是“图”的威力。3.4 第四步运行与调试智能体现在让我们运行这个智能体并观察其状态变化。# 定义初始状态 initial_state: BlogAgentState { original_topic: 如何学习LangGraph, refined_topics: None, selected_topic_index: None, search_query: None, search_results: None, outline: None, messages: [], error: None } print( 开始执行博客智能体 ) # 使用 stream 方法可以观察每一步的输出非常适合调试 for step, output in blog_agent_graph.stream(initial_state, stream_modevalues): node_name list(step.keys())[0] if step else Unknown print(f\n--- 节点 [{node_name}] 执行完毕 ---) # 打印当前状态的关键信息 if output.get(refined_topics): print(f细化选题: {output[refined_topics]}) if output.get(selected_topic_index) is not None: idx output[selected_topic_index] print(f已选择选题索引: {idx} - {output.get(refined_topics, [])[idx]}) if output.get(search_results): print(f搜索完成结果摘要长度: {len(output[search_results][0])}) if output.get(outline): print(大纲生成成功预览前500字符) print(output[outline][:500] ...) if output.get(messages): print(f日志: {output[messages][-1]}) # 打印最新一条日志 if output.get(error): print(f⚠️ 错误: {output[error]}) print(\n 智能体执行结束 ) # 你也可以用 invoke 一次性获取最终状态 # final_state blog_agent_graph.invoke(initial_state) # print(final_state[outline])使用stream模式你可以清晰地看到智能体是如何一步步“思考”和“行动”的每个节点如何修改状态。这是调试复杂工作流不可或缺的功能。4. 进阶记忆、工具与生产级考量一个能跑的 Demo 和一個能在生产环境使用的智能体之间还隔着几条街的距离。LangGraph 提供了更强大的特性来弥合这个差距。4.1 为智能体添加记忆Memory上面的例子中状态只在单次运行中有效。真实的智能体如聊天机器人需要记住跨对话的历史。LangGraph 通过Checkpointer机制实现。from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.graph import StateGraph, START, END import tempfile import os # 使用临时SQLite数据库作为检查点存储 with tempfile.NamedTemporaryFile(suffix.sqlite, deleteFalse) as tmp: db_path tmp.name checkpointer SqliteSaver.from_conn_string(fsqlite:///{db_path}) # 重新构建图但这次在编译时传入 checkpointer builder_with_memory StateGraph(BlogAgentState) # ... (添加所有节点和边的代码与之前相同) ... builder_with_memory.set_entry_point(refine_topic) # ... (添加边和条件边的代码) ... # 关键编译时配置 checkpointer blog_agent_with_memory builder_with_memory.compile(checkpointercheckpointer) # 运行带有记忆的智能体需要指定一个 config其中包含 configurable 键来标识会话。 config {configurable: {thread_id: user_123}} # thread_id 代表一个会话线程 print( 第一次运行记忆为空) state1 blog_agent_with_memory.invoke(initial_state, configconfig) print(f第一次运行后大纲已生成: {state1[outline][:100]}...) # 模拟第二次运行基于同样的会话thread_id状态会被保存和加载 # 我们可以修改输入观察智能体如何利用或覆盖之前的状态 new_initial_state: BlogAgentState { original_topic: LangGraph的高级特性, # 新主题 refined_topics: None, # ... 其他字段重置或为None ... messages: [], error: None } print(\n 第二次运行相同thread_id新主题) # 注意由于我们定义了 original_topic 会覆盖旧状态所以流程会重新开始。 # 但 messages 字段因为用了 operator.add可能会累积。 state2 blog_agent_with_memory.invoke(new_initial_state, configconfig) print(f第二次运行的日志条数: {len(state2[messages])}) # 清理临时文件 os.unlink(db_path)记忆机制的核心Checkpointer将每次图运行后的完整状态持久化到数据库如 SQLite。configurable中的thread_id是会话的唯一标识。相同thread_id的调用会加载上一次的状态作为起点。这对于构建多轮对话、长期任务如分步骤填写表格的智能体至关重要。4.2 更优雅地集成工具MCPModel Context Protocol的启示在我们的例子中我们直接使用了 LangChain 的DuckDuckGoSearchRun工具。但在更复杂的场景你可能需要集成代码解释器、数据库、文件系统等众多工具。手动管理这些工具的配置和调用会很繁琐。MCPModel Context Protocol是一个新兴的开放协议旨在标准化大模型与外部工具/数据源称为“上下文”的交互方式。它定义了工具发现、调用和结果返回的标准格式。虽然 LangGraph 不强制使用 MCP但它的思想值得借鉴将工具抽象为统一的、可发现的接口。在实践中你可以使用 LangChain 的 Tool 抽象就像我们例子中做的这是最直接的方式。创建自定义工具节点对于复杂操作可以将其封装成一个独立的 Node 函数而不是简单的 Tool。利用 LangGraph 的ToolNodeLangGraph 提供了ToolNode类可以自动处理 LangChain Tool 的调用和结果解析简化代码。from langgraph.prebuilt import ToolNode # 假设我们有多个工具 tools [search_tool, calculator_tool, file_read_tool] # 创建一个工具节点它能自动路由到正确的工具 tool_node ToolNode(tools) # 然后将 tool_node 作为一个节点添加到图中 builder.add_node(use_tool, tool_node) # 在状态中可能需要一个字段如 next_tool_to_use 来指示该节点使用哪个工具4.3 生产级智能体的关键考量当你准备将 LangGraph 智能体部署出去时请关注以下几点稳定性与错误处理我们例子中的错误处理还很基础。生产环境需要更精细的错误分类网络错误、工具错误、LLM格式错误、业务逻辑错误等和恢复策略重试、降级、人工接管。使用add_conditional_edges可以构建复杂的错误处理子图。可观测性State中的messages字段是一个简单的日志。生产环境需要集成像 OpenTelemetry 这样的标准记录每个节点的输入输出、耗时、LLM Token 使用量等方便监控和调试。流式输出对于耗时较长的智能体如生成长篇内容需要向用户实时返回部分结果。这需要结合 LLM 的流式输出和 LangGraph 的stream接口将中间状态如“正在搜索…”“正在生成大纲第一部分…”推送到前端。权限与安全智能体能调用工具这意味着它拥有工具的权限。必须严格限制工具的能力例如文件工具只能访问特定目录数据库工具只有查询权限。在节点函数中实现权限校验。成本控制智能体可能多次调用 LLM 和付费 API。需要在状态中跟踪成本并设置预算和中断机制。可以在一个专门的“成本监控”节点中实现并通过条件边在成本超限时提前终止流程。5. 从项目到思维LangGraph 带来的范式转变通过这个完整的项目你应该已经感受到LangGraph 不仅仅是一个库它更是一种构建可靠 AI 应用的思维方式。它迫使你从“ prompt 工程”转向“流程工程”。你不再只是绞尽脑汁设计一个万能 prompt而是像架构师一样设计一个由可靠组件节点和明确规则边组成的系统。大模型LLM在这个系统中更像是一个强大的、但需要被妥善管理的“决策者”或“内容生成器”而不是唯一的执行引擎。这种转变带来了几个显著优势可调试性你可以精确追踪到是哪个节点的输入输出出了问题。可维护性修改流程就是修改图结构而不是重写混乱的 if-else 代码。可复用性设计良好的节点如“搜索节点”、“格式化节点”可以像乐高一样在不同的智能体中复用。可控性你可以通过条件边轻松实现“如果搜索无结果则转向本地知识库”这样的复杂逻辑。回到我们最初的困惑为什么自己写的智能体总是容易“失控”答案往往是缺乏一个清晰的、显式的流程控制层。LangGraph 提供的正是这一层。它不保证你的智能体一定聪明但它能保证你的智能体行为是符合你设计的、可预测的。所以下次当你再面对一个复杂的、多步骤的 AI 应用需求时不妨先拿出一张纸画一画它的状态图从哪里开始经过哪些步骤在何处判断分支最终到哪里结束。这张图就是你的 LangGraph 智能体的蓝图。剩下的就是用代码把这张图实现出来。