尧图建网站 尧图建网站 YAOTU WEB BUILD 免费咨询
ARTICLE DETAIL

资讯详情

深耕网站建设与建站编程的一线实战洞察。

LangGraph实战:从状态流转到条件路由,构建可落地的大模型工作流

LangGraph实战:从状态流转到条件路由,构建可落地的大模型工作流 先聊一个非常实际的感受学习 LangGraph 这类工具难点往往不在“某个函数怎么用”而是“为什么需要图结构”“状态怎么流转”“分支和循环该怎么设计”。网上很多教程要么只讲一条链式调用要么直接丢一串源码看完还是不知道怎么组合。这篇文章从零开始把 LangGraph 的核心概念、状态流转、条件路由、循环检测、并行分支、子图依次拆开配合完整可运行的示例代码帮你建立一套能直接落地的工作流设计思路。如果你正在用 LangChain 做 Agent 或复杂任务编排但发现传统 Chain 越来越难表达循环、分支和状态共享那么本文的 LangGraph 实战路线会比较适合你。文中代码默认使用 Python 环境示例以常见稳定版本为准具体版本号请根据你的项目实际调整。1. 背景与核心概念1.1 LangGraph 是什么LangGraph 是 LangChain 生态中的一个库它的核心定位是用“图”的方式来组织大模型应用的工作流。图由节点Node和边Edge组成节点负责执行具体逻辑边负责定义节点之间的流转关系。这个设计与我们熟悉的传统 Chain 执行方式有明显区别。传统 Chain 通常是线性执行A 完成后进入 BB 完成后进入 C。流程简单直接但遇到“需要循环重试”“根据模型输出走不同分支”“多个任务并行执行后再汇总”的场景时Chain 的线性结构会让代码变得比较别扭。LangGraph 则把执行流程抽象成一张有向图。你可以理解为节点是一个个处理单元每个节点接收当前状态执行逻辑然后返回状态更新。边决定下一步进入哪个节点。状态State是贯穿整张图的共享数据结构每个节点都能读取和修改它。这种设计让循环、条件分支、并行执行、人工介入等复杂流程都能用同一套“图”的方式来表达。后续我们会在实战部分看到一个带循环的 ReAct Agent 通过 LangGraph 实现时结构会非常清晰。1.2 LangGraph 解决什么问题先看一个具体的业务痛点。开发一个客服 Agent 时流程往往是接收用户问题。判断问题类型咨询、投诉、售后。根据类型调用不同工具。如果模型生成的答案不完整或调用工具失败需要重试。多步任务结束后统一汇总结果。如果使用传统的线性 Chain你需要手动维护上下文、写 if-else 控制流程还要考虑循环重试逻辑。代码会逐渐变成面条式结构排错也变得困难。LangGraph 把上述过程变成图定义“判断问题类型”是一个节点通过条件路由把下一步引到不同节点。“调用工具失败后重试”可以通过循环边实现。“多步任务汇总”可以通过多个节点指向汇总节点实现。状态集中管理流程可视化逻辑清晰排错也变得容易。1.3 LangChain 和 LangGraph 的区别这是不少初学者问得比较多的问题。两者关系可以用一句话概括LangChain 是一个大模型应用开发框架LangGraph 是这个框架中的一个编排库两者定位不同但可以配合使用。更具体的区别如下对比项LangChainLangGraph定位完整的应用开发框架包含模型接入、提示词管理、工具调用、记忆等专门负责工作流状态与流程编排的库核心抽象Chain、Agent、Tool、MemoryState、Node、Edge、StateGraph流程控制线性为主分支循环需要额外逻辑天然支持分支、循环、并行、子图State 管理链内消息传递共享状态需要额外设计全局状态对象节点间显式传递更新适用场景快速搭建单轮或多轮应用复杂 Agent、有状态任务流、需要精细控制的场景实际开发中两者经常混合使用。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 langgraph2.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 的常见方式是使用TypedDictfrom 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接收三个参数起始节点名称。路由函数接收 State返回一个字符串。路径映射表把路由函数返回值映射到目标节点名称。路由函数里可以做任何判断检查关键字、调用模型分类、判断某个字段是否为空等。这样可以非常灵活地控制流程方向。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.pyfrom typing import TypedDict from langgraph.graph import StateGraph, START, END class DemoState(TypedDict): question: str answer: str def parse_node(state: DemoState) - dict: print(fparse_node 收到问题: {state[question]}) return {answer: f已收到问题{state[question]}} def output_node(state: DemoState) - dict: print(foutput_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.pyfrom 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.pyfrom 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.pyfrom 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.pyfrom 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 的几种方式节点函数返回需要更新的字段字典这是最常用的方式。def my_node(state: MyState) - dict: return {field_a: 新值}返回空字典表示不修改任何字段。def my_node(state: MyState) - dict: return {}返回多个字段。def my_node(state: MyState) - dict: return {field_a: 新值, field_b: 123}使用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.py5.3 工作流代码文件路径langgraph_demo/workflow.pyfrom 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.pyfrom 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 常见报错表问题现象常见原因解决思路ImportError: cannot import name StateGraphlanggraph 版本过旧或未安装执行pip install -U langgraphValueError: Expected START or END node边定义中使用了不存在的节点名称检查节点名称是否拼写一致图执行时超时或无限循环路由函数持续返回同一节点检查递归条件设置recursion_limit并行节点执行顺序不符合预期对并行节点做了顺序假设并行节点之间不要有数据依赖状态字段被覆盖而不是追加字段没有使用Annotated需要使用累加逻辑时使用add_messages节点函数返回值不是 dictLangGraph 要求节点返回字典检查函数返回值类型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_node7.5 性能与并发控制并行节点过多时注意线程池大小的配置。如果节点调用了外部 API建议设置超时时间。在异步场景中可以查看官方文档中关于异步图的支持避免阻塞事件循环。对于长耗时任务考虑将任务拆成子图并设计中断与恢复机制。7.6 安全与合规边界不要在代码中把 API Key 写到仓库里。节点中执行动态代码或拼装 SQL 时必须有严格的输入校验。对用户输入做格式检查避免提示词注入。涉及用户数据时遵循最小权限原则不采集与任务无关的数据。生产环境改动前先在测试环境验证并备份相关配置和状态数据。8. 总结本篇文章从 LangGraph 是什么开始逐步拆解了 State、Node、Edge、条件路由、并行分支、循环检测、子图等核心概念并提供了多个可直接运行的示例代码。建议按照如下路线继续学习第一步跑通本文前三个最小示例理解 State 的更新机制。第二步使用conditional_edge实现自己的分支逻辑。第三步尝试在循环中使用真实模型构建一个简单的 Agent 工具调用流程。第四步用子图重构流程提升模块复用能力。第五步在实际项目中逐步加入日志、异常处理、配置管理和测试。LangGraph 的 API 迭代速度比较快不同版本的导入方式可能略有不同但图的思维模型是稳定不变的。把重心放在节点划分、状态流转和路由条件这些设计问题上你会发现不管是构建客服 Agent、数据分析流程还是更复杂的多智能体系统LangGraph 都能把项目的复杂度控制在一个清晰可控的范围内。代码写完只是开始多画图、多打日志、多思考状态变化才能真正用好它。
返回列表