企业官网建设流程全解析

先聊一个非常实际的感受：学习 LangGraph 这类工具，难点往往不在“某个函数怎么用”，而是“为什么需要图结构”“状态怎么流转”“分支和循环该怎么设计”。网上很多教程要么只讲一条链式调用，要么直接丢一串源码，看完还是不知道怎么组合。这篇文章从零开始，把 LangGraph 的核心概念、状态流转、条件路由、循环检测、并行分支、子图依次拆开，配合完整可运行的示例代码，帮你建立一套能直接落地的工作流设计思路。

如果你正在用 LangChain 做 Agent 或复杂任务编排，但发现传统 Chain 越来越难表达循环、分支和状态共享，那么本文的 LangGraph 实战路线会比较适合你。文中代码默认使用 Python 环境，示例以常见稳定版本为准，具体版本号请根据你的项目实际调整。

1. 背景与核心概念

1.1 LangGraph 是什么

LangGraph 是 LangChain 生态中的一个库，它的核心定位是：用“图”的方式来组织大模型应用的工作流。图由节点（Node）和边（Edge）组成，节点负责执行具体逻辑，边负责定义节点之间的流转关系。

这个设计与我们熟悉的传统 Chain 执行方式有明显区别。传统 Chain 通常是线性执行：A 完成后进入 B，B 完成后进入 C。流程简单直接，但遇到“需要循环重试”“根据模型输出走不同分支”“多个任务并行执行后再汇总”的场景时，Chain 的线性结构会让代码变得比较别扭。

LangGraph 则把执行流程抽象成一张有向图。你可以理解为：

• 节点是一个个处理单元，每个节点接收当前状态，执行逻辑，然后返回状态更新。
• 边决定下一步进入哪个节点。
• 状态（State）是贯穿整张图的共享数据结构，每个节点都能读取和修改它。

这种设计让循环、条件分支、并行执行、人工介入等复杂流程，都能用同一套“图”的方式来表达。后续我们会在实战部分看到，一个带循环的 ReAct Agent 通过 LangGraph 实现时，结构会非常清晰。

1.2 LangGraph 解决什么问题

先看一个具体的业务痛点。开发一个客服 Agent 时，流程往往是：

1. 接收用户问题。
2. 判断问题类型（咨询、投诉、售后）。
3. 根据类型调用不同工具。
4. 如果模型生成的答案不完整或调用工具失败，需要重试。
5. 多步任务结束后统一汇总结果。

如果使用传统的线性 Chain，你需要手动维护上下文、写 if-else 控制流程，还要考虑循环重试逻辑。代码会逐渐变成面条式结构，排错也变得困难。

LangGraph 把上述过程变成图定义：

• “判断问题类型”是一个节点，通过条件路由把下一步引到不同节点。
• “调用工具失败后重试”可以通过循环边实现。
• “多步任务汇总”可以通过多个节点指向汇总节点实现。

状态集中管理，流程可视化，逻辑清晰，排错也变得容易。

1.3 LangChain 和 LangGraph 的区别

这是不少初学者问得比较多的问题。两者关系可以用一句话概括：LangChain 是一个大模型应用开发框架，LangGraph 是这个框架中的一个编排库，两者定位不同，但可以配合使用。

更具体的区别如下：

实际开发中，两者经常混合使用。LangChain 提供模型封装、工具定义、提示词模板等基础能力，LangGraph 负责把这些能力编排成一张可执行的图。后面示例中也能看到，我们依然会从 LangChain 引入模型和工具相关的封装。

1.4 LangGraph 的常见应用场景

• 客服 Agent：根据用户意图路由到不同处理节点。
• 多步骤任务编排：例如数据分析流程，先取数、再清洗、再建模、再生成报告。
• Agent 工具调用循环：模型自主决定是否调用工具，收到结果后继续推理，直到生成最终答案。
• 并行子任务处理：对多条输入分别处理，最后汇总结果。
• 人工审核介入：关键节点暂停执行，等待人工确认后再继续。

接下来进入环境准备环节。

2. 环境准备与版本说明

2.1 运行环境

本文示例基于 Python 3.9 及以上版本。操作系统方面，Windows、macOS、Linux 都可以运行，命令差异不大。建议在项目中使用虚拟环境，避免依赖冲突。

建议提前安装 Python 和 pip，并确认命令行可以正常执行python与pip。

2.2 安装依赖

LangGraph 目前主要通过 pip 安装。以下命令会安装 LangGraph 和 LangChain 核心库：

pip install langgraph pip install langchain-core

如果后续示例中需要调用具体模型，还需要根据模型服务商安装对应包。例如使用 OpenAI 兼容接口时：

pip install langchain-openai

需要注意，LangGraph 的 API 仍在快速迭代中，不同版本的导入路径和函数签名可能存在差异。本文示例以常见稳定版本为准，如果遇到导入报错，可以优先检查版本：

pip show langgraph

2.3 版本说明

为了不让文章内容随着版本变化而失效，这里不写死具体版本号。你只需要理解核心思路，然后映射到当前安装的版本即可。常见的导入方式如下：

from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages

其中START和END是 LangGraph 内置的特殊节点，分别表示流程起点和终点。部分旧版本中可能使用START的替代写法，以官方文档为准。

3. 核心概念与原理拆解

3.1 State：整个工作流的“共享内存”

State 是 LangGraph 中最重要的概念之一。你可以把它理解为一个跨节点传递的数据容器。每个节点接收当前 State，执行完逻辑后返回 State 的更新部分，LangGraph 会把这些更新合并到全局 State 中。

定义 State 的常见方式是使用TypedDict：

from typing import TypedDict class AgentState(TypedDict): question: str answer: str steps: list[str]

如果某些字段需要支持累加或特殊合并逻辑，可以使用Annotated，配合add_messages等函数。例如消息列表通常需要不断追加：

from typing import Annotated from typing_extensions import TypedDict from langgraph.graph.message import add_messages class ChatState(TypedDict): messages: Annotated[list, add_messages]

这里Annotated的作用是告诉 LangGraph：当多个节点都返回messages字段时，不要把旧值直接覆盖，而是用add_messages函数把新消息追加进去。

3.2 Node：图中的一个执行单元

Node 就是一个普通的 Python 函数，签名通常为(state) -> dict。它接收当前的 State，返回一个字典，表示要更新的字段。

示例：

def node_a(state: AgentState) -> dict: return {"answer": f"处理问题：{state['question']}"}

节点内部可以调用模型、调用工具、读写文件、访问数据库，几乎什么都能做。LangGraph 不限制节点内部逻辑，只要最终返回一个包含更新字段的字典即可。

3.3 Edge：连接节点的“道路”

Edge 定义节点之间的流转顺序。LangGraph 提供几种常见边：

• 普通边：A 执行完后无条件进入 B。
• 条件边：根据函数返回值或 State 内容，动态决定下一个节点。
• 起始边：从 START 出发，进入第一个节点。
• 结束边：节点进入 END，流程结束。

普通边示例：

from langgraph.graph import StateGraph, START, END builder = StateGraph(AgentState) builder.add_node("node_a", node_a) builder.add_node("node_b", node_b) builder.add_edge(START, "node_a") builder.add_edge("node_a", "node_b") builder.add_edge("node_b", END)

3.4 conditional_edge：条件路由与分支控制

条件边是 LangGraph 分支能力的核心。它的作用是：当前节点执行完毕后，根据某个判断函数的结果，决定下一步进入哪个节点。

基本写法：

def route_by_type(state: AgentState) -> str: if "投诉" in state["question"]: return "complaint_node" return "normal_node" builder.add_conditional_edge( "classify_node", route_by_type, { "complaint_node": "complaint_node", "normal_node": "normal_node", }, )

add_conditional_edge接收三个参数：

1. 起始节点名称。
2. 路由函数，接收 State，返回一个字符串。
3. 路径映射表，把路由函数返回值映射到目标节点名称。

路由函数里可以做任何判断：检查关键字、调用模型分类、判断某个字段是否为空等。这样可以非常灵活地控制流程方向。

3.5 并行分支

LangGraph 支持从一个节点同时连接到多个节点，多个节点都执行完毕后，再汇聚到同一个节点。这种模式通常叫 fan-out / fan-in。

用普通边就能实现并行：

builder.add_edge("start_node", "task_a") builder.add_edge("start_node", "task_b") builder.add_edge("start_node", "task_c") builder.add_edge("task_a", "merge_node") builder.add_edge("task_b", "merge_node") builder.add_edge("task_c", "merge_node")

默认情况下，并行节点是在一个线程池中并发调度的。如果需要控制最大并发数，可以配置StateGraph的编译参数，例如interrupt_before、recursion_limit等，具体可以查阅官方文档。

3.6 循环检测

LangGraph 允许边指回之前的节点，形成循环。这是实现 Agent 工具调用重试、模型自我反思等能力的关键。

示例：模型节点生成答案后，判断是否包含错误标记，如果包含则回到生成节点重新执行。

def should_retry(state: AgentState) -> str: if state.get("need_retry"): return "agent_node" return END

循环带来了一个风险：如果路由函数有 bug，流程可能无限循环。LangGraph 提供了recursion_limit限制递归深度。编译后运行图时，可以通过配置传入：

graph.invoke(initial_state, config={"recursion_limit": 10})

当执行步数超过限制时，会抛异常终止，避免死循环拖垮服务。

3.7 子图（Subgraph）

子图就是把一个已经编译好的图，当作另一个图的节点来使用。适合场景：

• 将复杂的流程拆成多个模块，每个模块独立开发和测试。
• 多个父流程复用同一个子流程。
• 在子流程内部使用独立的状态结构，与父流程解耦。

使用子图时，父图的节点函数可以直接调用已编译子图的invoke方法，也可以通过“将子图作为节点添加”的方式接入。

4. 实战：从单节点到完整工作流

下面从一个最小示例开始，逐步加入条件路由、并行分支和子图，最终构建一个可运行的文档处理工作流。

4.1 首个示例：最小状态图

先创建一个最简单的图：接收用户问题，经过两个节点处理后输出结果。

文件路径：example_01_basic.py

from typing import TypedDict from langgraph.graph import StateGraph, START, END class DemoState(TypedDict): question: str answer: str def parse_node(state: DemoState) -> dict: print(f"parse_node 收到问题: {state['question']}") return {"answer": f"已收到问题：{state['question']}"} def output_node(state: DemoState) -> dict: print(f"output_node 输出答案: {state['answer']}") return {} builder = StateGraph(DemoState) builder.add_node("parse_node", parse_node) builder.add_node("output_node", output_node) builder.add_edge(START, "parse_node") builder.add_edge("parse_node", "output_node") builder.add_edge("output_node", END) graph = builder.compile() result = graph.invoke({"question": "LangGraph 怎么学习？"}) print(result)

运行方式：

python example_01_basic.py

预期输出：

parse_node 收到问题: LangGraph 怎么学习？ output_node 输出答案: 已收到问题：LangGraph 怎么学习？ {'question': 'LangGraph 怎么学习？', 'answer': '已收到问题：LangGraph 怎么学习？'}

这个例子虽然简单，但完整展示了 StateGraph 的基本要素：定义 State、创建 Builder、添加节点、添加边、编译、调用。

4.2 理解 State 的更新机制

在上面的例子中，parse_node返回了answer字段，LangGraph 自动把它合并到全局 State 中。output_node读取state["answer"]时能看到该值。

这里有个细节：节点函数返回的并不是“完整的新 State”，而是“需要更新的字段”。LangGraph 会把返回值与旧 State 做合并。如果没有返回某个字段，该字段会保持不变。

4.3 加入条件路由

现在扩展一下：根据用户问题中是否包含“投诉”，走不同的处理节点。

文件路径：example_02_conditional.py

from typing import TypedDict from langgraph.graph import StateGraph, START, END class RouteState(TypedDict): question: str response: str def classify_node(state: RouteState) -> dict: print(f"分类节点：{state['question']}") return {} def complaint_node(state: RouteState) -> dict: return {"response": "这是投诉处理节点：我们会尽快联系售后专员。"} def normal_node(state: RouteState) -> dict: return {"response": "这是普通咨询节点：这里是标准回答。"} def route_question(state: RouteState) -> str: if "投诉" in state["question"]: return "complaint_node" return "normal_node" builder = StateGraph(RouteState) builder.add_node("classify_node", classify_node) builder.add_node("complaint_node", complaint_node) builder.add_node("normal_node", normal_node) builder.add_edge(START, "classify_node") builder.add_conditional_edge( "classify_node", route_question, { "complaint_node": "complaint_node", "normal_node": "normal_node", }, ) builder.add_edge("complaint_node", END) builder.add_edge("normal_node", END) graph = builder.compile() result = graph.invoke({"question": "我要投诉物流太慢"}) print(result["response"])

运行后输出：

分类节点：我要投诉物流太慢 这是投诉处理节点：我们会尽快联系售后专员。

如果把question换成普通问题，流程会进入normal_node。这就是条件路由的完整工作方式。

4.4 加入循环：实现重试机制

在真实 Agent 中，模型可能需要多次调用工具才能得到最终答案。下面用一个简化的循环示例说明：如果结果中包含“需要重试”的标志，就回到处理节点再次执行。

文件路径：example_03_loop.py

from typing import TypedDict from langgraph.graph import StateGraph, START, END class LoopState(TypedDict): input_text: str attempt: int result: str def process_node(state: LoopState) -> dict: attempt = state.get("attempt", 1) print(f"第 {attempt} 次处理，输入：{state['input_text']}") if attempt < 3: return { "attempt": attempt + 1, "result": f"第 {attempt} 次结果不完整，需要重试", } return { "attempt": attempt + 1, "result": "第 3 次结果：处理完成", } def after_process(state: LoopState) -> str: if "需要重试" in state["result"]: return "process_node" return END builder = StateGraph(LoopState) builder.add_node("process_node", process_node) builder.add_edge(START, "process_node") builder.add_conditional_edge( "process_node", after_process, { "process_node": "process_node", END: END, }, ) graph = builder.compile() result = graph.invoke( {"input_text": "hello", "attempt": 1}, config={"recursion_limit": 10}, ) print(result)

运行输出：

第 1 次处理，输入：hello 第 2 次处理，输入：hello 第 3 次处理，输入：hello {'input_text': 'hello', 'attempt': 4, 'result': '第 3 次结果：处理完成'}

这里用attempt字段控制循环次数，防止无限循环。实际项目中使用模型判断是否重试时，务必设置recursion_limit，并且要设计明确的中止条件。

4.5 并行分支与结果汇总

并行分支适合对多条独立输入分别处理，然后统一汇聚。下面示例同时处理两个文本片段，最后汇总。

文件路径：example_04_parallel.py

from typing import TypedDict from langgraph.graph import StateGraph, START, END class ParallelState(TypedDict): text1: str text2: str processed1: str processed2: str summary: str def process_text1(state: ParallelState) -> dict: return {"processed1": f"[T1]{state['text1']}"} def process_text2(state: ParallelState) -> dict: return {"processed2": f"[T2]{state['text2']}"} def merge_node(state: ParallelState) -> dict: summary = f"{state['processed1']} + {state['processed2']}" return {"summary": summary} builder = StateGraph(ParallelState) builder.add_node("process_text1", process_text1) builder.add_node("process_text2", process_text2) builder.add_node("merge_node", merge_node) builder.add_edge(START, "process_text1") builder.add_edge(START, "process_text2") builder.add_edge("process_text1", "merge_node") builder.add_edge("process_text2", "merge_node") builder.add_edge("merge_node", END) graph = builder.compile() result = graph.invoke({"text1": "上午", "text2": "下午"}) print(result["summary"])

输出：

[T1]上午 + [T2]下午

从这个例子可以看出，并行分支本质上只是从一个节点添加多条出边，LangGraph 会自动调度执行。但要注意：并行节点之间不能有数据依赖，因为执行顺序不固定。如果节点 B 依赖节点 A 的结果，就必须用边把顺序固定下来。

4.6 使用子图拆分复杂流程

当一个图变得复杂时，可以把某个阶段封装成子图。下面示例定义了一个子图，用于统一处理文本规范化，然后父图调用它。

文件路径：example_05_subgraph.py

from typing import TypedDict from langgraph.graph import StateGraph, START, END class NormalizeState(TypedDict): raw: str normalized: str def normalize_step(state: NormalizeState) -> dict: return {"normalized": state["raw"].strip().lower()} sub_builder = StateGraph(NormalizeState) sub_builder.add_node("normalize_step", normalize_step) sub_builder.add_edge(START, "normalize_step") sub_builder.add_edge("normalize_step", END) sub_graph = sub_builder.compile() class ParentState(TypedDict): raw_text: str final_text: str def process_with_subgraph(state: ParentState) -> dict: sub_result = sub_graph.invoke({"raw": state["raw_text"]}) return {"final_text": f"处理结果：{sub_result['normalized']}"} parent_builder = StateGraph(ParentState) parent_builder.add_node("process_with_subgraph", process_with_subgraph) parent_builder.add_edge(START, "process_with_subgraph") parent_builder.add_edge("process_with_subgraph", END) parent_graph = parent_builder.compile() result = parent_graph.invoke({"raw_text": " Hello LangGraph "}) print(result["final_text"])

输出：

处理结果：hello langgraph

子图的最大好处是模块化：子图内部可以拥有独立的 State 和节点，如果其他父流程也需要文本规范化，直接复用同一个子图即可。

4.7 节点函数中修改 State 值的完整说明

结合前面几个示例，可以总结出修改 State 的几种方式：

1. 节点函数返回需要更新的字段字典，这是最常用的方式。

def my_node(state: MyState) -> dict: return {"field_a": "新值"}

2. 返回空字典表示不修改任何字段。

def my_node(state: MyState) -> dict: return {}

3. 返回多个字段。

def my_node(state: MyState) -> dict: return {"field_a": "新值", "field_b": 123}

4. 使用Annotated和消息累加器时，返回的消息会自动追加到列表中。

def chat_node(state: ChatState) -> dict: return {"messages": [{"role": "assistant", "content": "你好"}]}

这里的逻辑是：节点返回值会覆盖或者合并，具体取决于 State 字段的类型定义。简单类型直接覆盖，带Annotated的类型使用合并函数。

5. 完整实战：构建一个选题分类与处理工作流

现在把前面学的知识整合起来，实现一个更完整的示例：输入一段用户问题，先做意图分类，分类后进入不同处理分支，每个分支内部执行两步，最后汇总并结束。

这个例子中会包含：

• 条件路由。
• 并行分支。
• 子图思想。
• recursion_limit配置。

5.1 功能需求

输入一段文本，例如：

• “我想投诉快递延时”
• “你们有什么课程推荐”
• “我要退款”

输出：

• 意图判断结果。
• 对应处理流程后的回复。
• 必要的日志信息。

5.2 文件结构

langgraph_demo/ ├── workflow.py └── main.py

5.3 工作流代码

文件路径：langgraph_demo/workflow.py

from typing import TypedDict from langgraph.graph import StateGraph, START, END class WorkflowState(TypedDict): user_input: str intent: str response: str detail: str # ---------- 节点定义 ---------- def classify_node(state: WorkflowState) -> dict: text = state["user_input"] if "投诉" in text or "退款" in text: intent = "after_sale" elif "推荐" in text or "课程" in text: intent = "consult" else: intent = "other" return {"intent": intent} def after_sale_detail_node(state: WorkflowState) -> dict: return {"detail": "售后处理中，已记录工单"} def after_sale_reply_node(state: WorkflowState) -> dict: return {"response": f"【售后】{state['detail']}，预计 24 小时内回复。"} def consult_detail_node(state: WorkflowState) -> dict: return {"detail": "顾问正在整理课程资料"} def consult_reply_node(state: WorkflowState) -> dict: return {"response": f"【咨询】{state['detail']}，请查看最新课程大纲。"} def other_node(state: WorkflowState) -> dict: return {"intent": "other", "response": "【其他】请稍后，人工客服将为您服务。"} # ---------- 条件路由 ---------- def route_by_intent(state: WorkflowState) -> str: if state["intent"] == "after_sale": return "after_sale_node" if state["intent"] == "consult": return "consult_node" return "other_node" # ---------- 构建图 ---------- builder = StateGraph(WorkflowState) builder.add_node("classify_node", classify_node) builder.add_node("after_sale_node", after_sale_detail_node) builder.add_node("after_sale_reply_node", after_sale_reply_node) builder.add_node("consult_node", consult_detail_node) builder.add_node("consult_reply_node", consult_reply_node) builder.add_node("other_node", other_node) builder.add_edge(START, "classify_node") builder.add_conditional_edge( "classify_node", route_by_intent, { "after_sale_node": "after_sale_node", "consult_node": "consult_node", "other_node": "other_node", }, ) builder.add_edge("after_sale_node", "after_sale_reply_node") builder.add_edge("after_sale_reply_node", END) builder.add_edge("consult_node", "consult_reply_node") builder.add_edge("consult_reply_node", END) builder.add_edge("other_node", END) graph = builder.compile()

文件路径：langgraph_demo/main.py

from workflow import graph if __name__ == "__main__": test_inputs = [ "我想投诉快递延时", "你们有什么课程推荐", "我要退款", "随便聊聊", ] for text in test_inputs: result = graph.invoke( {"user_input": text}, config={"recursion_limit": 20}, ) print(f"输入：{text}") print(f"意图：{result['intent']}") print(f"回复：{result['response']}") print("----------")

5.4 运行与输出

运行命令：

cd langgraph_demo python main.py

预期输出：

输入：我想投诉快递延时 意图：after_sale 回复：【售后】售后处理中，已记录工单，预计 24 小时内回复。 ---------- 输入：你们有什么课程推荐 意图：consult 回复：【咨询】顾问正在整理课程资料，请查看最新课程大纲。 ---------- 输入：我要退款 意图：after_sale 回复：【售后】售后处理中，已记录工单，预计 24 小时内回复。 ---------- 输入：随便聊聊 意图：other 回复：【其他】请稍后，人工客服将为您服务。 ----------

这个示例把条件路由、多节点流转集中在一个图中体现。你可以在此基础上增加模型调用、工具调用和外部接口，形成一个真正可用于生产的流程原型。

6. 常见问题与排查思路

6.1 常见报错表

6.2 如何排查图执行路径

一个非常实用的技巧是打印节点执行日志。你可以在每个节点函数入口添加print，也可以给节点加一个统一的日志装饰器：

def log_node(func): def wrapper(state): print(f"进入节点：{func.__name__}") result = func(state) print(f"离开节点：{func.__name__}，返回字段：{list(result.keys())}") return result return wrapper

然后在节点定义上添加@log_node，即可跟踪整张图的执行顺序。

6.3 避免循环陷阱

循环是 LangGraph 中风险最高的一环。建议：

• 循环体必须包含“接近退出条件”的逻辑，例如计数器加一。
• 条件边返回的节点应该是“有限集合”。
• 在生产环境中设置合理的recursion_limit。
• 为循环次数设置上限，达到上限后走降级分支。

7. 最佳实践与工程建议

7.1 图上做设计，代码里做实现

建议先用画图工具画出节点和边，再写代码。节点命名统一采用“动词_名词”的风格，例如parse_input、call_tool、generate_answer。图的拓扑结构是系统架构的一部分，应该像数据库表一样被重视。

7.2 State 设计原则

State 是整张图的共享数据结构，设计时要注意：

• 字段尽量少而清晰。
• 避免在 State 中存放重量级对象，例如大段文本结果可以直接存字符串或列表，而不是自定义对象。
• 需要持久化的数据与临时数据分开管理。
• 对于消息列表，使用Annotated[list, add_messages]来保持追加语义。
• 谨慎使用可变对象作为 State 默认值，建议在初始化时显式传入。

7.3 异常处理与日志

每个节点都可能调用外部模型、API 或数据库，异常处理不能省略。

建议在节点函数内使用 try-except，并将错误信息写入 State 的error字段：

def tool_node(state: MyState) -> dict: try: result = call_external_api(state["query"]) return {"result": result} except Exception as e: return {"error": str(e), "should_retry": True}

然后通过条件边检查should_retry，决定是重试还是进入兜底节点。这样异常处理也变成了图的一部分，流程可控。

7.4 配置管理与测试

• 模型名、API Key、数据库连接等配置不要硬编码在节点函数中。
• 使用 environment variables 或配置中心管理。
• 推荐为每个图编写单元测试，尤其是条件路由函数。路由函数是纯函数，测试很方便：

def test_route_by_intent(): state = {"intent": "after_sale"} assert route_by_intent(state) == "after_sale_node"

7.5 性能与并发控制

• 并行节点过多时，注意线程池大小的配置。
• 如果节点调用了外部 API，建议设置超时时间。
• 在异步场景中，可以查看官方文档中关于异步图的支持，避免阻塞事件循环。
• 对于长耗时任务，考虑将任务拆成子图，并设计中断与恢复机制。

7.6 安全与合规边界

• 不要在代码中把 API Key 写到仓库里。
• 节点中执行动态代码或拼装 SQL 时，必须有严格的输入校验。
• 对用户输入做格式检查，避免提示词注入。
• 涉及用户数据时，遵循最小权限原则，不采集与任务无关的数据。
• 生产环境改动前先在测试环境验证，并备份相关配置和状态数据。

8. 总结

本篇文章从 LangGraph 是什么开始，逐步拆解了 State、Node、Edge、条件路由、并行分支、循环检测、子图等核心概念，并提供了多个可直接运行的示例代码。建议按照如下路线继续学习：

• 第一步：跑通本文前三个最小示例，理解 State 的更新机制。
• 第二步：使用conditional_edge实现自己的分支逻辑。
• 第三步：尝试在循环中使用真实模型，构建一个简单的 Agent 工具调用流程。
• 第四步：用子图重构流程，提升模块复用能力。
• 第五步：在实际项目中逐步加入日志、异常处理、配置管理和测试。

LangGraph 的 API 迭代速度比较快，不同版本的导入方式可能略有不同，但图的思维模型是稳定不变的。把重心放在节点划分、状态流转和路由条件这些设计问题上，你会发现不管是构建客服 Agent、数据分析流程，还是更复杂的多智能体系统，LangGraph 都能把项目的复杂度控制在一个清晰可控的范围内。代码写完只是开始，多画图、多打日志、多思考状态变化，才能真正用好它。