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，而要选 StateGraph？

2.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(model="gpt-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）。它不关心你每个节点里具体跑什么模型或工具，它只负责确保：当系统处于某个状态时，它知道下一步该去哪，以及在什么条件下该去哪。

我们来对比一下同一个“搜索-总结-润色”任务，在两种范式下的表达差异：

这个差异，决定了 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(default="search", description="当前执行步骤：search, summarize, write") # 搜索阶段的结果，List[Dict] 是标准格式，便于后续节点处理 search_results: Optional[List[Dict[str, Any]]] = Field( default=None, description="DuckDuckGo 搜索返回的原始结果列表" ) # 摘要阶段的输出，是结构化数据，而非纯文本 summary: Optional[Dict[str, str]] = Field( default=None, description="LLM 提取的关键信息，格式为 {'main_claim': '...', 'evidence': ['...']}" ) # 最终输出，也是 Agent 给用户的最终回复 final_answer: Optional[str] = Field( default=None, description="格式化后的最终学术摘要" ) # 全局错误计数器，用于触发重试或降级 error_count: int = Field(default=0, ge=0, le=3, description="累计错误次数，超过3次则终止") # 用于调试的元信息，不参与业务逻辑 debug_info: Dict[str, Any] = Field(default_factory=dict, description="临时调试信息，如耗时、节点名")

这个定义里藏着几个关键实操要点：

1. 必填字段（Field(...)）与默认值的哲学：question是必填的，因为它是整个流程的起点，没有它，一切无从谈起。而search_results和summary是Optional，因为它们是在流程中逐步产生的。这种设计，让State的初始化变得极其简单：ResearchState(question="量子计算在金融风控中的应用")即可启动。

2. current_step字段的妙用：这是 StateGraph 区别于其他工作流框架的灵魂所在。它不是一个装饰性的字段，而是Edge（边）进行条件判断的直接依据。比如，search节点执行完，会把current_step改成"summarize"，然后图引擎就会自动沿着current_step == "summarize"这条边，把状态流转到summarize节点。它让“流程控制”从代码逻辑，下沉到了数据层面。

3. 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对应的三个核心Node：

from 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(model="gpt-4o-mini", temperature=0.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(model="gpt-4o", temperature=0.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的 key，path_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. 实操过程：从环境搭建到第一个可运行的 Agent

4.1 环境准备：用 Miniconda 创建纯净、可复现的开发环境

标题里提到的“使用 miniconda 创建 langgraph”，绝非一句空话。LangGraph 作为一个快速迭代的前沿框架，其依赖项（尤其是langchain-core,langgraph本身）的版本兼容性，是新手最容易栽跟头的地方。我强烈建议，放弃pip install langgraph这种“一键安装”，转而用 Conda 来管理环境。

以下是我在 macOS 和 Ubuntu 上验证过的、最稳妥的步骤：

1. 下载并安装 Miniconda
   访问 https://docs.conda.io/en/latest/miniconda.html ，下载对应你系统的最新版 Miniconda 安装包（推荐 Python 3.11 或 3.12）。安装完成后，重启终端，确保conda命令可用。

2. 创建专用环境

   # 创建一个名为 langgraph-dev 的新环境，指定 Python 版本 conda create -n langgraph-dev python=3.11 # 激活环境 conda activate langgraph-dev # 升级 pip，确保包管理器是最新的 pip install --upgrade pip

3. 安装核心依赖（关键！）
   这里是重点。LangGraph 的官方文档有时会滞后，而社区里流传的pip install langgraph命令，可能会拉取到一个与你本地langchain-core版本不兼容的旧版。最保险的方式，是指定一个已知稳定、且被广泛验证的版本组合。截至 2024 年 10 月，我推荐的组合是：

   • langchain-core==0.3.10
   • langgraph==0.2.50
   • langchain-openai==0.2.10

   执行安装命令：

   pip install "langchain-core==0.3.10" "langgraph==0.2.50" "langchain-openai==0.2.10" "langchain-community==0.3.10"

   为什么是这个组合？
   langgraph==0.2.50是第一个全面支持StateGraph的稳定版，它修复了早期版本中add_conditional_edges的诸多 bug。langchain-core==0.3.10是与之配套的、API 最稳定的版本。我曾试过langgraph==0.1.x，在add_conditional_edges的path_map参数上，文档和实际行为严重不符，浪费了整整一天。

4. 验证安装
   启动 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：定义 State

from pydantic import BaseModel, Field from typing import Optional, Dict, Any class WeatherState(BaseModel): city: str = Field(..., description="用户询问的城市名") weather_data: Optional[Dict[str, Any]] = Field(default=None, description="API 返回的原始天气数据") forecast: Optional[str] = Field(default=None, description="LLM 生成的口语化播报") error_count: int = Field(default=0, ge=0, le=3)

Step 2：编写 Node

import 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 = f"https://geocoding-api.open-meteo.com/v1/search?name={state.city}&count=1&language=en&format=json" try: geo_resp = requests.get(geo_url, timeout=5) 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 = f"https://api.open-meteo.com/v1/forecast?latitude={lat}&longitude={lon}&current=temperature_2m,weather_code,wind_speed_10m&timezone=auto" weather_resp = requests.get(weather_url, timeout=5) 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(model="gpt-4o-mini", temperature=0.3) chain = prompt | llm | StrOutputParser() try: state.forecast = chain.invoke({}) except Exception as e: state.error_count += 1 return state

Step 3：构建 Graph

from 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 提