LangGraph StateGraph实战:从零构建可调试、可重试的智能体工作流

发布时间:2026/7/8 18:56:20
LangGraph StateGraph实战:从零构建可调试、可重试的智能体工作流 1. 项目概述为什么“从零搭建智能体”这件事现在比任何时候都更值得你亲手做一遍LangGraph 这个词最近在技术社区里出现的频率已经快赶上当年 Docker 刚火起来时大家反复念叨“容器化”的劲头了。但和 Docker 不同的是LangGraph 并不解决一个具体、可见的运维问题它解决的是一个更底层、更抽象、也更让人挠头的问题怎么让大语言模型真正“动起来”而不是只当个高级聊天框我自己第一次用 LangChain 写完一个带记忆的问答机器人兴奋地发到群里结果被一位老哥一句话点醒“你这叫‘会说话的文档’不叫‘智能体’。它不会主动思考下一步该干什么也不会在出错时换条路走更不会把一个复杂任务拆成几步再串起来——它只是在等你喂问题。” 这句话让我琢磨了整整两天。后来我试了 LangGraph 的 StateGraph用三行代码定义了一个“搜索→总结→润色”的循环状态机看着它自己卡在摘要环节后主动调用另一个工具重试那一刻我才真正理解了标题里那个“智能体”的“智”字到底落在哪儿。这个项目标题里的“从零搭建”不是指从零开始造轮子而是从零开始建立对 Agent 构建范式的直觉。LangGraph 的核心价值恰恰在于它把过去散落在 LangChain、LlamaIndex、甚至手写 while 循环里的那些“流程控制逻辑”用一种可声明、可调试、可持久化的图结构给固化下来。StateGraph 就是这张图的画布而节点Node和边Edge就是它的笔和墨。它不替代 LLM而是给 LLM 装上方向盘和导航仪它不取代工具而是把工具调用变成图中一条有明确起点和终点的路径。所以当你看到热搜词里反复出现 “langgraph 教程”、“langgraph 和 langchain 的区别”本质上大家在问的是同一个问题我手里的 LLM 模型怎么才能从“被动应答者”升级为“主动执行者”这个项目就是给你一把最直接的钥匙。它适合所有已经能跑通一个简单 RAG 或 ChatBot但下一步想让模型真正“做事”的人——无论是想做个自动写周报的办公助手还是想搭个能自主调研竞品的市场分析 Agent甚至只是想搞懂面试官为什么总爱问“你如何设计一个带重试机制的 Agent 工作流”这篇实操记录都是你绕不开的第一课。2. 核心设计思路为什么不用 LangChain 的 RunnableSequence而要选 StateGraph2.1 传统链式调用的“天花板”在哪很多刚接触 LLM 应用开发的朋友第一反应是用 LangChain 的RunnableSequence或RunnableParallel把几个步骤串起来。比如写一个“用户提问→搜索资料→生成答案”的流程代码可能长这样from langchain_core.runnables import RunnableSequence from langchain_community.tools import DuckDuckGoSearchRun search DuckDuckGoSearchRun() prompt ChatPromptTemplate.from_template(根据以下搜索结果回答问题{results}。问题{question}) llm ChatOpenAI(modelgpt-4o) chain RunnableSequence( {results: search, question: lambda x: x[question]}, prompt, llm, StrOutputParser() )这段代码很干净运行起来也没问题。但它有一个致命的隐性缺陷整个流程是线性的、不可中断的、且状态完全黑盒。如果搜索环节返回了空结果prompt模板会拿到一个空字符串LLM 可能胡说一通而你根本没法在中间插手——既不能加个判断逻辑说“没搜到就换个关键词重试”也不能把这次失败记录下来供后续分析。它像一条单向高速公路车数据上了路你就只能祈祷它顺利抵达终点。我在一个客户项目里就吃过这个亏他们需要一个能自动核对合同条款的 Agent要求如果某条款在数据库里找不到对应解释就必须暂停并通知法务人工介入。用RunnableSequence实现我们最后不得不在每个节点里硬塞一堆if/else和异常捕获代码迅速变得臃肿难读调试时更是噩梦。2.2 StateGraph 的“状态机”思维让流程拥有“记忆”与“决策力”LangGraph 的破局点就是引入了有状态的图计算模型。StateGraph 的核心思想非常朴素任何复杂的任务都可以被分解为一系列“状态”State以及在这些状态之间切换的“规则”Transition Rule。它不关心你每个节点里具体跑什么模型或工具它只负责确保当系统处于某个状态时它知道下一步该去哪以及在什么条件下该去哪。我们来对比一下同一个“搜索-总结-润色”任务在两种范式下的表达差异特性RunnableSequence (链式)StateGraph (图式)流程控制固定顺序A→B→C无法跳转灵活跳转A→BB→CB→A重试C→END甚至 B→D分支状态管理无显式状态数据靠dict传递易丢失关键上下文显式定义State类所有节点共享同一份可读可写的上下文对象错误处理需在每个节点内手动try/except逻辑分散可在Edge上统一定义失败跳转规则如should_retry函数可调试性执行过程是黑盒只能看最终输出每次状态切换都会触发回调可清晰打印当前状态、进入节点、耗时等信息可扩展性新增一个分支逻辑往往要重构整个链新增一个节点和几条边即可主干流程完全不受影响这个差异决定了 LangGraph 不是一个“更好用的 LangChain”而是一个不同维度的抽象层。LangChain 像是帮你把螺丝刀、扳手、电钻都配齐了而 LangGraph 是给你一张带标注的施工蓝图告诉你什么时候该用哪把工具以及如果拧错了该回退到哪个工序重新来过。这也是为什么热搜词里“langgraph workflow” 和 “stategraph” 总是成对出现——Workflow 是目标StateGraph 是实现 Workflow 的唯一可靠路径。2.3 为什么“从零”必须从 StateGraph 开始——避开三个常见认知陷阱在带新人做 Agent 项目时我发现有三个特别容易踩的坑它们都源于对 LangGraph 定位的误解陷阱一“LangGraph 是 LangChain 的插件”错。LangGraph 是一个独立的、更底层的框架。它不依赖 LangChain 的Runnable接口虽然兼容它的Node可以是任意 Python 函数甚至可以是纯 NumPy 计算。把它当成 LangChain 的“增强包”会让你在设计时下意识地用 LangChain 的思维去套 LangGraph结果就是写出一堆“披着图皮的链”。陷阱二“State 就是传参的 dict”错。State是一个强类型、可验证、可序列化的数据契约。LangGraph 强制你用 Pydantic 的BaseModel来定义它这意味着你必须提前想清楚这个 Agent 在整个生命周期里哪些数据是必须全程携带的哪些是只在某个节点才需要的比如一个客服 Agent 的State至少要包含user_id,conversation_history,current_intent,last_search_results。这个定义过程本身就是一次深度的需求梳理。我见过太多项目因为 State 定义得过于随意比如全用dict[str, Any]导致后期加一个新功能就要翻遍所有节点去改参数签名痛苦不堪。陷阱三“图越复杂Agent 越智能”错。一个健康的 Agent 图应该遵循“奥卡姆剃刀”原则能用两个节点解决的绝不画三个。我自己的第一个生产级 Agent初始设计有 7 个节点包含了各种“预处理”、“后校验”、“兜底回复”等。上线后发现80% 的请求只经过其中 3 个节点。后来我把那 4 个“锦上添花”的节点全部砍掉性能提升 40%故障率反而下降。LangGraph 的威力不在于它能画多复杂的图而在于它让你能用最简的图表达最准的逻辑。所谓“从零搭建”第一步就是学会克制先画出那个最小、但能跑通闭环的三角形Input → Process → Output。3. 核心细节解析StateGraph 的四大支柱与实操要点3.1 支柱一State —— 你的 Agent 的“中央大脑”State是整个 LangGraph 的基石它不是一个简单的数据容器而是 Agent 的单一事实来源Single Source of Truth。LangGraph 要求你用 Pydantic v2 的BaseModel来定义它这绝非形式主义而是为了强制你在编码前就完成一次严谨的领域建模。我们以一个最典型的“研究助理 Agent”为例它的核心任务是根据用户问题搜索网络信息提炼关键点并用学术风格撰写摘要。它的State定义如下from typing import List, Optional, Dict, Any from pydantic import BaseModel, Field class ResearchState(BaseModel): # 用户原始输入不可变 question: str Field(..., description用户提出的原始研究问题) # 当前正在处理的子任务用于控制流程走向 current_step: str Field(defaultsearch, description当前执行步骤search, summarize, write) # 搜索阶段的结果List[Dict] 是标准格式便于后续节点处理 search_results: Optional[List[Dict[str, Any]]] Field( defaultNone, descriptionDuckDuckGo 搜索返回的原始结果列表 ) # 摘要阶段的输出是结构化数据而非纯文本 summary: Optional[Dict[str, str]] Field( defaultNone, descriptionLLM 提取的关键信息格式为 {main_claim: ..., evidence: [...]} ) # 最终输出也是 Agent 给用户的最终回复 final_answer: Optional[str] Field( defaultNone, description格式化后的最终学术摘要 ) # 全局错误计数器用于触发重试或降级 error_count: int Field(default0, ge0, le3, description累计错误次数超过3次则终止) # 用于调试的元信息不参与业务逻辑 debug_info: Dict[str, Any] Field(default_factorydict, description临时调试信息如耗时、节点名)这个定义里藏着几个关键实操要点必填字段Field(...)与默认值的哲学question是必填的因为它是整个流程的起点没有它一切无从谈起。而search_results和summary是Optional因为它们是在流程中逐步产生的。这种设计让State的初始化变得极其简单ResearchState(question量子计算在金融风控中的应用)即可启动。current_step字段的妙用这是 StateGraph 区别于其他工作流框架的灵魂所在。它不是一个装饰性的字段而是Edge边进行条件判断的直接依据。比如search节点执行完会把current_step改成summarize然后图引擎就会自动沿着current_step summarize这条边把状态流转到summarize节点。它让“流程控制”从代码逻辑下沉到了数据层面。error_count的防御性设计这是一个典型的“工程化思维”体现。它不解决具体错误但它为错误处理提供了统一的、可量化的接口。所有节点在捕获到异常时只需做一件事state.error_count 1。然后Edge可以基于这个数字决定是重试、跳转到备用节点还是直接结束。这比在每个节点里写if e.type TimeoutError: ...要优雅得多。实操心得我建议你在定义State时拿出一张白纸画出你设想的 Agent 流程图然后挨个问自己这个节点的输入是什么输出是什么哪些东西是下一个节点绝对需要的哪些是只有我调试时才要看的把这些问题的答案直接翻译成State的字段。这个过程比写一百行代码都重要。3.2 支柱二Node —— Agent 的“肌肉”与“器官”如果说State是大脑那么Node就是四肢百骸。在 LangGraph 中Node是一个纯函数Pure Function它接收一个State对象执行一些操作调用 LLM、查询数据库、运行 Python 代码然后返回一个新的State对象或者原地修改并返回它。一个合格的Node必须满足两个黄金法则输入唯一它只接收一个参数即State的实例。输出唯一它必须返回一个State的实例可以是新的也可以是修改后的。我们来实现上面ResearchState对应的三个核心Nodefrom langchain_community.tools import DuckDuckGoSearchRun from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser # Node 1: 搜索 def search_node(state: ResearchState) - ResearchState: 执行网络搜索 search DuckDuckGoSearchRun() try: results search.invoke(state.question) # DuckDuckGo 返回的是字符串我们将其标准化为 List[Dict] # 实际项目中这里会调用一个专门的解析器 standardized_results [{title: Web Result, snippet: results, link: N/A}] state.search_results standardized_results state.current_step summarize state.debug_info[search_time] 1.2 # 模拟耗时 except Exception as e: state.error_count 1 state.debug_info[search_error] str(e) return state # Node 2: 摘要 def summarize_node(state: ResearchState) - ResearchState: 从搜索结果中提取关键信息 if not state.search_results: state.error_count 1 return state # 构建提示词 prompt ChatPromptTemplate.from_messages([ (system, 你是一个专业的学术研究员。请从以下搜索结果中精准提取出关于{question}的三个核心论点及其支持证据。), (human, {search_text}) ]) llm ChatOpenAI(modelgpt-4o-mini, temperature0.2) chain prompt | llm | StrOutputParser() try: # 将搜索结果拼接成字符串 search_text \n.join([f{r[title]}: {r[snippet]} for r in state.search_results]) raw_summary chain.invoke({question: state.question, search_text: search_text}) # 这里应该用一个结构化解析器但为简化我们模拟一个 JSON 字符串 state.summary { main_claim: 量子计算能显著加速蒙特卡洛模拟从而提升金融风险预测精度。, evidence: [ 传统蒙特卡洛方法在大型投资组合上计算耗时过长。, 量子算法如HHL可在多项式时间内求解线性方程组。 ] } state.current_step write state.debug_info[summarize_time] 2.5 except Exception as e: state.error_count 1 state.debug_info[summarize_error] str(e) return state # Node 3: 撰写 def write_node(state: ResearchState) - ResearchState: 将结构化摘要转化为学术风格的最终答案 if not state.summary: state.error_count 1 return state prompt ChatPromptTemplate.from_messages([ (system, 你是一位资深的金融学教授。请将以下结构化信息用严谨、客观、符合学术规范的语言撰写成一段200字左右的摘要。), (human, 核心主张{main_claim}\n支持证据{evidence}) ]) llm ChatOpenAI(modelgpt-4o, temperature0.1) chain prompt | llm | StrOutputParser() try: evidence_str .join(state.summary[evidence]) final_answer chain.invoke({ main_claim: state.summary[main_claim], evidence: evidence_str }) state.final_answer final_answer state.current_step end # 流程结束标志 state.debug_info[write_time] 3.1 except Exception as e: state.error_count 1 state.debug_info[write_error] str(e) return state这段代码展示了Node的典型模式输入 State → 处理调用外部服务/模型→ 修改 State → 输出 State。注意几个细节错误处理的统一性每个Node在except块里都只做两件事state.error_count 1和state.debug_info[xxx] str(e)。这保证了错误处理逻辑的高度复用避免了重复代码。current_step的驱动作用每个成功的Node都会更新state.current_step这就像给流程装上了自动档图引擎会根据这个值自动选择下一条边。debug_info的价值它不参与业务但却是你调试时的救命稻草。上线后你可以轻松地把debug_info打印出来立刻知道是哪个环节慢了、哪个环节挂了。注意事项Node函数内部严禁有全局变量、类属性或任何外部状态依赖。它必须是“无副作用”的。如果你需要一个共享的数据库连接池正确的做法是把它作为Node的闭包变量或者通过add_node方法的kwargs注入。我曾经在一个项目里因为在一个Node里用了global cache_dict导致多用户并发时数据错乱排查了三天才定位到问题教训深刻。3.3 支柱三Edge —— Agent 的“神经突触”Edge是 StateGraph 的“决策中枢”它定义了状态在节点之间流转的条件与路径。如果说Node是“做什么”那么Edge就是“什么时候做”和“做完后去哪”。LangGraph 提供了两种定义Edge的方式add_edge(from_node, to_node)无条件边从 A 节点执行完必然走到 B 节点。add_conditional_edges(from_node, condition_func, path_map)条件边condition_func是一个函数它接收State返回一个字符串这个字符串作为path_map的 keypath_map的 value 就是目标节点。我们来为研究助理 Agent 设计完整的Edge网络from langgraph.graph import StateGraph, END # 初始化图 workflow StateGraph(ResearchState) # 添加节点 workflow.add_node(search, search_node) workflow.add_node(summarize, summarize_node) workflow.add_node(write, write_node) # 添加无条件边流程主干 workflow.add_edge(search, summarize) workflow.add_edge(summarize, write) # 添加条件边错误处理与流程终结 # 从 search 节点出发的条件边 def should_continue_after_search(state: ResearchState) - str: 搜索后判断下一步去哪 if state.error_count 3: return give_up # 放弃 elif state.search_results is None or len(state.search_results) 0: return search_retry # 重试 else: return summarize # 正常进入摘要 workflow.add_conditional_edges( search, should_continue_after_search, { give_up: END, # 直接结束 search_retry: search, # 自循环重试 summarize: summarize # 正常流转 } ) # 从 summarize 节点出发的条件边 def should_continue_after_summarize(state: ResearchState) - str: 摘要后判断下一步去哪 if state.error_count 3: return give_up elif state.summary is None: return summarize_retry else: return write workflow.add_conditional_edges( summarize, should_continue_after_summarize, { give_up: END, summarize_retry: summarize, write: write } ) # 从 write 节点出发的条件边 def should_continue_after_write(state: ResearchState) - str: 撰写后判断是否结束 if state.error_count 3: return give_up elif state.final_answer is None: return write_retry else: return END # 成功结束流程 workflow.add_conditional_edges( write, should_continue_after_write, { give_up: END, write_retry: write, END: END } ) # 设置入口点 workflow.set_entry_point(search)这个Edge网络构建了一个具备自愈能力的 Agent。它不再是“一步错步步错”的脆弱链条而是一个能感知自身健康状况并做出相应反应的有机体。should_continue_after_xxx函数就是它的“小脑”负责最基础的反射动作。实操心得condition_func的设计是 LangGraph 项目中最考验工程素养的地方。我建议你遵循一个简单原则一个condition_func只做一件事且这件事必须能用一句中文描述清楚。比如should_continue_after_search的职责就是“根据搜索结果和错误次数决定下一步”。如果你发现一个condition_func里写了七八个if/elif那说明你该把它拆分成多个更小的函数或者重新审视你的State设计——很可能State里缺了一个能简化判断逻辑的关键字段。3.4 支柱四Graph —— Agent 的“躯体”与“操作系统”StateGraph类本身就是整个 Agent 的“操作系统内核”。它负责加载Node、注册Edge、管理State的生命周期并提供最终的.compile()和.invoke()接口。compile()是一个关键步骤它会将你定义的图结构编译成一个可执行的CompiledGraph对象。这个过程LangGraph 会做几件重要的事拓扑排序检查图中是否存在死循环A→B→A或断连某个节点永远无法到达。类型验证确保每个Node的输入输出类型与State的定义完全匹配。优化准备为后续的异步执行、流式输出、检查点保存等高级特性做好底层准备。# 编译图 app workflow.compile() # 调用图同步 result app.invoke({question: 量子计算在金融风控中的应用}) print(result.final_answer) # 调用图流式适用于前端实时显示 for chunk in app.stream({question: 量子计算在金融风控中的应用}): print(chunk)invoke()和stream()是两个最常用的调用方式。invoke()是阻塞式调用适合后台批处理stream()是生成器它会按节点执行的顺序逐个yield出当前State的快照非常适合做前端的“打字机效果”或实时进度条。注意事项compile()是一个相对昂贵的操作它会进行图的静态分析。因此在生产环境中你应该只在应用启动时调用一次compile()然后将返回的app对象作为单例全局复用。我见过一个项目因为把app workflow.compile()写在了每次 API 请求的 handler 里导致 QPS 直接掉了 60%CPU 一直在做无意义的图分析。4. 实操过程从环境搭建到第一个可运行的 Agent4.1 环境准备用 Miniconda 创建纯净、可复现的开发环境标题里提到的“使用 miniconda 创建 langgraph”绝非一句空话。LangGraph 作为一个快速迭代的前沿框架其依赖项尤其是langchain-core,langgraph本身的版本兼容性是新手最容易栽跟头的地方。我强烈建议放弃pip install langgraph这种“一键安装”转而用 Conda 来管理环境。以下是我在 macOS 和 Ubuntu 上验证过的、最稳妥的步骤下载并安装 Miniconda访问 https://docs.conda.io/en/latest/miniconda.html 下载对应你系统的最新版 Miniconda 安装包推荐 Python 3.11 或 3.12。安装完成后重启终端确保conda命令可用。创建专用环境# 创建一个名为 langgraph-dev 的新环境指定 Python 版本 conda create -n langgraph-dev python3.11 # 激活环境 conda activate langgraph-dev # 升级 pip确保包管理器是最新的 pip install --upgrade pip安装核心依赖关键这里是重点。LangGraph 的官方文档有时会滞后而社区里流传的pip install langgraph命令可能会拉取到一个与你本地langchain-core版本不兼容的旧版。最保险的方式是指定一个已知稳定、且被广泛验证的版本组合。截至 2024 年 10 月我推荐的组合是langchain-core0.3.10langgraph0.2.50langchain-openai0.2.10执行安装命令pip install langchain-core0.3.10 langgraph0.2.50 langchain-openai0.2.10 langchain-community0.3.10为什么是这个组合langgraph0.2.50是第一个全面支持StateGraph的稳定版它修复了早期版本中add_conditional_edges的诸多 bug。langchain-core0.3.10是与之配套的、API 最稳定的版本。我曾试过langgraph0.1.x在add_conditional_edges的path_map参数上文档和实际行为严重不符浪费了整整一天。验证安装启动 Python 交互式环境尝试导入 from langgraph.graph import StateGraph from langgraph.checkpoint.memory import MemorySaver print(LangGraph 安装成功) LangGraph 安装成功如果没有报错恭喜你的地基已经打牢。4.2 编写第一个 Agent一个极简但功能完整的“天气播报员”理论讲再多不如亲手敲出第一行能跑的代码。我们来实现一个最简单的 Agent它接收一个城市名调用免费的 Open-Meteo API 获取当前天气然后用 LLM 生成一段口语化的播报。Step 1定义 Statefrom pydantic import BaseModel, Field from typing import Optional, Dict, Any class WeatherState(BaseModel): city: str Field(..., description用户询问的城市名) weather_data: Optional[Dict[str, Any]] Field(defaultNone, descriptionAPI 返回的原始天气数据) forecast: Optional[str] Field(defaultNone, descriptionLLM 生成的口语化播报) error_count: int Field(default0, ge0, le3)Step 2编写 Nodeimport requests from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser def fetch_weather_node(state: WeatherState) - WeatherState: 调用 Open-Meteo API 获取天气 # Open-Meteo 的地理编码 API先获取经纬度 geo_url fhttps://geocoding-api.open-meteo.com/v1/search?name{state.city}count1languageenformatjson try: geo_resp requests.get(geo_url, timeout5) geo_resp.raise_for_status() geo_data geo_resp.json() if not geo_data.get(results): raise ValueError(f未找到城市 {state.city} 的地理信息) lat geo_data[results][0][latitude] lon geo_data[results][0][longitude] # 调用天气 API weather_url fhttps://api.open-meteo.com/v1/forecast?latitude{lat}longitude{lon}currenttemperature_2m,weather_code,wind_speed_10mtimezoneauto weather_resp requests.get(weather_url, timeout5) weather_resp.raise_for_status() state.weather_data weather_resp.json() except Exception as e: state.error_count 1 return state def generate_forecast_node(state: WeatherState) - WeatherState: 用 LLM 生成播报 if not state.weather_data: state.error_count 1 return state current state.weather_data[current] temp current[temperature_2m] code current[weather_code] wind current[wind_speed_10m] # 天气代码映射简化版 weather_desc {0: 晴朗, 1: 晴间多云, 2: 多云, 3: 阴天, 45: 雾, 48: 冻雾, 51: 毛毛雨, 53: 持续毛毛雨, 55: 浓密毛毛雨, 56: 冻毛毛雨, 57: 冻浓密毛毛雨, 61: 小雨, 63: 中雨, 65: 大雨, 66: 冻雨, 67: 冻大雨, 71: 小雪, 73: 中雪, 75: 大雪, 77: 雪粒, 80: 小雨, 81: 中雨, 82: 大雨, 85: 湿雪, 86: 冻雪, 95: 雷暴, 96: 雷暴伴小冰雹, 99: 雷暴伴大冰雹} desc weather_desc.get(code, 未知天气) prompt ChatPromptTemplate.from_messages([ (system, 你是一个亲切的天气播报员。请用轻松、口语化、不超过100字的中文播报以下天气信息。不要使用专业术语。), (human, f城市{state.city}温度{temp}°C天气{desc}风速{wind} m/s。) ]) llm ChatOpenAI(modelgpt-4o-mini, temperature0.3) chain prompt | llm | StrOutputParser() try: state.forecast chain.invoke({}) except Exception as e: state.error_count 1 return stateStep 3构建 Graphfrom langgraph.graph import StateGraph, END workflow StateGraph(WeatherState) workflow.add_node(fetch_weather, fetch_weather_node) workflow.add_node(generate_forecast, generate_forecast_node) # 主干流程 workflow.add_edge(fetch_weather, generate_forecast) # 条件边错误处理 def should_continue(state: WeatherState) - str: if state.error_count 3: return END elif state.weather_data is None: return fetch_weather # 重试 else: return generate_forecast workflow.add_conditional_edges( fetch_weather, should_continue, { END: END, fetch_weather: fetch_weather, generate_forecast: generate_forecast } ) workflow.set_entry_point(fetch_weather) app workflow.compile()Step 4运行与测试# 测试 result app.invoke({city: Beijing}) if result.forecast: print(✅ 天气播报, result.forecast) else: print(❌ 执行失败错误次数, result.error_count)运行它你会看到类似这样的输出✅ 天气播报 北京今天天气不错气温18度阳光明媚微风轻拂出门记得带上薄外套哦这就是你的第一个 LangGraph Agent。它虽小却五脏俱全有状态、有节点、有边、有错误重试。它证明了LangGraph 的学习曲线远没有想象中陡峭。4.3 进阶技巧让 Agent 更“聪明”的三个实战方案方案一添加检查点Checkpointing—— 让 Agent “记得住事”默认情况下LangGraph 的app.invoke()是无状态的每次调用都是全新的开始。但现实中的 Agent往往需要记住上下文。比如用户问“北京的天气怎么样”然后紧接着问“那上海呢”一个优秀的 Agent 应该能记住“用户正在连续查询不同城市的天气”而不是每次都从零开始。LangGraph 提