LangChain与LangGraph：LLM智能体开发的双引擎架构解析

1. 项目概述：为什么说 LangChain 和 LangGraph 是 LLM 应用开发的“双引擎”

你刚写完一个调用大模型 API 的脚本，输入“帮我总结这篇PDF”，它真就吐出了一段文字——但下一次你让它“先查文档目录，再定位到第三章第二节，提取其中所有带‘风险’二字的句子，最后按严重程度排序”，它大概率会卡壳、幻觉、漏项，甚至直接编造一个不存在的章节。这不是模型能力不够，而是你缺了一套能指挥模型“分步思考、自主决策、调用工具、记住上下文、处理失败”的操作系统。LangChain 和 LangGraph 就是为此而生的两把利剑：LangChain 是 LLM 应用的“基础建设队”，它把模型、数据、工具、记忆这些零散砖块标准化成可插拔的模块；LangGraph 则是“作战指挥中心”，它让这些模块不再线性排队，而是能像真实人类一样，根据当前状态动态决定下一步走哪条路、调哪个工具、回退还是重试。它们不是替代大模型的“新模型”，而是让大模型真正变成可用、可靠、可扩展的“智能体（Agent）”的底层框架。如果你正在做 RAG 检索增强、多步骤数据分析、自动化客服流程、或者需要模型自己调用数据库或 API 的项目，LangChain 提供了开箱即用的链（Chain）、代理（Agent）和记忆（Memory）抽象；而当你发现业务逻辑开始变得复杂——比如用户问“查订单”要先鉴权、再查库、再判断是否超时、超时则触发通知——这时候 LangChain 的简单 Agent 就力不从心了，LangGraph 的有向图状态机就成了唯一解。我去年重构一个金融合规报告生成系统时，就是先用 LangChain 快速搭出原型，等业务规则膨胀到 17 个分支判断、5 类异常处理路径后，果断切到 LangGraph，整个系统的可读性、可测试性和运维稳定性直接翻倍。这二者的关系，不是“谁取代谁”，而是“LangChain 解决‘能不能做’，LangGraph 解决‘能不能稳、能不能扩、能不能查’”。

2. 核心设计思路拆解：从“线性流水线”到“状态驱动图”的范式跃迁

2.1 LangChain 的核心思想：抽象层封装与组件化组装

LangChain 的设计哲学非常务实：它不碰模型训练，也不重写推理引擎，而是专注解决 LLM 工程落地中最琐碎、最重复的“胶水代码”。它的核心抽象有四个支柱：Model I/O、Retrieval、Chain、Agent。Model I/O 封装了所有大模型调用细节——不管是 OpenAI、Anthropic 还是本地部署的 Llama 3，你只需换一个llm = ChatOpenAI()或llm = Ollama(model="llama3")，后面所有 prompt 渲染、流式响应、token 计数逻辑全由框架接管。Retrieval 抽象则把“向量数据库怎么连、相似度阈值设多少、召回后怎么重排”这些细节打包成VectorStoreRetriever，你只管传入retriever.get_relevant_documents(query)。Chain 是 LangChain 最标志性的概念，它把多个步骤串成一条可复用、可配置的流水线。比如一个标准 RAG Chain：prompt | llm | output_parser，其中prompt是一个带变量的模板，llm是模型实例，output_parser是一个把原始 JSON 字符串转成 Python 字典的解析器。这个 Chain 可以被保存、加载、调试，甚至嵌套进另一个更大的 Chain 里。Agent 更进一步，它让模型自己决定“下一步该做什么”。LangChain 内置了 ReAct Agent，其底层逻辑是：模型输出一段包含Thought:、Action:、Action Input:、Observation:的文本，框架负责解析Action名称（比如 “search_api”），调用对应工具，把结果塞回Observation，再喂给模型继续推理。这种设计极大降低了 Agent 开发门槛，但它的致命局限在于——所有决策都依赖模型单次输出的文本解析，无法显式定义状态、无法处理循环、无法在失败时跳转到备用路径。我曾用它做一个电商比价 Agent，当某平台 API 返回 503 错误时，模型有时会忽略错误信息继续生成，有时又会陷入“重试-失败-重试”的死循环，根本没法加熔断或降级逻辑。

2.2 LangGraph 的破局点：将 Agent 从“黑盒文本解析”升级为“白盒状态机”

LangGraph 的诞生，本质上是对 LangChain Agent 范式的反思与重构。它抛弃了“靠模型输出文本指令来驱动流程”的脆弱模式，转而引入一个清晰、可编程、可观察的状态机模型。它的核心是三个概念：State、Node、Edge。State 是一个字典（dict），承载整个 Agent 运行过程中的所有数据——用户原始输入、中间计算结果、工具调用返回值、错误标记、甚至自定义的计数器。Node 是一个纯函数，接收 State 作为输入，执行具体操作（比如调用 LLM、查询数据库、做数学计算），并返回一个更新后的 State。Edge 是连接 Node 的有向边，但它不是静态的，而是由一个Condition Function动态决定的。这个函数接收当前 State，返回下一个要执行的 Node 名称。这才是 LangGraph 的灵魂所在：流程控制权从模型手中，交还给了开发者编写的确定性逻辑。举个实际例子：一个贷款审批 Agent。LangChain Agent 可能这样写：模型输出Action: check_credit_score, Action Input: {"user_id": "123"}→ 框架调用信用分接口 → 模型再输出Action: approve_loan或Action: reject_loan。但现实中，信用分只是第一步，后面还有收入验证、负债率计算、反欺诈扫描、人工复核触发等多个环节，每个环节都有成功、失败、需人工介入三种状态。用 LangGraph，你可以明确定义：

• check_creditNode：调用接口，成功则 State 加"credit_ok": True，失败则加"error": "credit_unavailable"；
• should_routeCondition Function：检查 State 中是否有"error"，有则跳转到handle_errorNode；否则检查"credit_ok"是否为 True，是则去verify_income，否则直接reject；
• handle_errorNode：记录日志、发送告警、设置重试计数器，然后根据计数器决定是retry_check_credit还是escalate_to_human。

这种设计带来的好处是颠覆性的：可测试性——你可以完全绕过 LLM，用 mock 数据直接调用check_credit({"user_id": "123"})测试逻辑；可观测性——每一步 State 变化、Node 执行耗时、Edge 路由选择都被完整记录，排查问题时不再对着一长串模型输出猜谜；可扩展性——新增一个“环保资质校验”环节，只需写一个新 Node 和对应的 Edge 条件，不影响其他 16 个节点。我团队在部署一个跨境物流跟踪 Agent 时，最初用 LangChain，上线两周内因网络抖动导致的路由错误率高达 8%，切换 LangGraph 后，通过在should_route中加入if "network_error" in state and state.get("retry_count", 0) < 3的判断，错误率直接压到 0.2% 以下，且所有重试行为都可审计。

2.3 LCEL：LangChain 的“函数式编程语法糖”，为 LangGraph 铺路

LCEL（LangChain Expression Language）常被误解为一个独立工具，其实它是 LangChain 为了拥抱函数式编程范式而设计的一套声明式语法。它的核心理念是：一切皆可组合（Everything is Composable）。一个 LCEL 表达式prompt | llm | output_parser看似简单，背后是三个对象实现了Runnable接口，|操作符重载了__or__方法，自动构建出一个RunnableSequence。这种设计让 Chain 的定义变得极其简洁，更重要的是，它为 LangGraph 的集成扫清了障碍。因为 LangGraph 的 Node 本质上就是一个Runnable，所以你可以直接把一个复杂的 LCEL Chain 当作一个 Node 使用。比如，你有一个用 LCEL 写的、融合了 RAG 和规则校验的“合同关键条款提取器”，它本身就是一个Runnable，在 LangGraph 里，你只需workflow.add_node("extract_clauses", contract_extractor_chain)，它就变成了图中的一个标准节点。LCEL 不是 LangGraph 的替代品，而是 LangChain 生态内部的“高级 DSL”，它让 LangChain 的组件能无缝融入 LangGraph 的图结构中，形成“小任务用 LCEL 快速组装，大流程用 LangGraph 精细编排”的黄金组合。很多新手纠结“该学 LCEL 还是 LangGraph”，答案很直白：LCEL 是 LangChain 的现代写法，必须掌握；LangGraph 是当你的 LCEL Chain 开始变得臃肿、分支变多、错误处理变复杂时，自然演进的下一步。

3. 核心实操要点与环境搭建：从零开始跑通第一个 LangGraph Agent

3.1 环境准备：版本对齐与依赖管理是稳定运行的第一道关卡

LangChain 和 LangGraph 的迭代速度极快，版本不匹配是新手踩坑的头号原因。截至 2024 年底，生产环境推荐的组合是：LangChain 0.1.20+、LangGraph 0.1.25+、Python 3.10 或 3.11。绝对不要用pip install langchain这种模糊命令，它会拉取最新版，而最新版可能已移除旧 API。我的标准做法是：新建一个干净的 Conda 环境，明确指定版本。以下是我在 Ubuntu 22.04 上的实操步骤：

# 创建专用环境，避免污染全局Python conda create -n langgraph-env python=3.11 conda activate langgraph-env # 安装核心框架（注意：langgraph 依赖 langchain-core，但不依赖 langchain） pip install "langchain-core==0.1.58" "langchain==0.1.20" "langgraph==0.1.25" # 安装一个轻量级 LLM 用于本地测试（避免每次调 API 花钱和等网络） pip install "ollama==0.1.9" # 启动 Ollama 服务（需提前下载模型） ollama run llama3:8b-instruct

提示：Ollama 是目前最友好的本地 LLM 运行时，llama3:8b-instruct在 16GB 内存的笔记本上也能流畅运行，且支持流式响应，完美模拟真实 API 行为。不要一上来就折腾 vLLM 或 Text Generation Inference，那会把精力全耗在 GPU 驱动和 CUDA 版本上。

安装完成后，务必验证版本：

import langchain import langgraph print(f"LangChain version: {langchain.__version__}") print(f"LangGraph version: {langgraph.__version__}") # 输出应为：LangChain version: 0.1.20，LangGraph version: 0.1.25

如果版本不符，立刻用pip install "langchain==0.1.20"强制降级。我见过太多人因为langgraph==0.2.0用了add_edge新 API，却搭配langchain==0.0.300的旧Runnable，结果workflow.compile()直接报AttributeError: 'RunnableLambda' object has no attribute 'invoke'，查了三天才发现是版本错配。

3.2 第一个 LangGraph Agent：一个带重试机制的天气查询器

我们不从“Hello World”开始，而是直接实现一个有真实价值的小 Agent：它接收用户城市名，调用免费天气 API 获取当前温度，如果 API 失败（网络超时或返回非 200），自动重试最多 2 次，超过则返回友好提示。这个例子覆盖了 State 管理、Node 编写、Condition Function、Edge 路由、错误处理等全部核心要素。

第一步：定义 State 结构LangGraph 要求你显式定义 State 的 Schema，这强迫你思考数据流。我们用 Pydantic V2（LangGraph 原生支持）：

from typing import TypedDict, Annotated, Optional from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver import operator class WeatherState(TypedDict): city: str # 用户输入的城市 temperature: Optional[float] # 查询到的温度 error: Optional[str] # 错误信息 retry_count: Annotated[int, operator.add] = 0 # 重试计数器，Annotated[int, operator.add] 表示此字段支持 += 操作

注意：retry_count的Annotated[int, operator.add]是关键。它告诉 LangGraph，当多个 Node 都修改retry_count时，框架会自动用operator.add合并值，避免竞态。这是 LangGraph 状态合并的精妙设计，不用手动锁。

第二步：编写 Nodes（纯函数）

import requests import time def fetch_weather(state: WeatherState) -> WeatherState: """Node：调用天气API""" city = state["city"] try: # 使用免费的 wttr.in API，无需密钥 response = requests.get(f"http://wttr.in/{city}?format=%t", timeout=5) if response.status_code == 200: # 响应如 "+22°C"，提取数字 temp_str = response.text.strip().replace("°C", "").replace("+", "") return {"temperature": float(temp_str)} else: raise Exception(f"API returned {response.status_code}") except Exception as e: # 记录错误，并增加重试计数 return { "error": str(e), "retry_count": 1 # 注意：这里不是 +=1，而是返回增量值，LangGraph 会自动累加 } def should_retry(state: WeatherState) -> str: """Condition Function：决定下一步路由""" if state.get("error") and state.get("retry_count", 0) < 2: return "fetch_weather" # 重试 fetch_weather Node elif state.get("error"): return "handle_failure" # 重试次数用尽，去失败处理 else: return END # 成功，结束流程 def handle_failure(state: WeatherState) -> WeatherState: """Node：处理最终失败""" return { "temperature": None, "error": f"查询失败，已重试{state['retry_count']}次：{state['error']}" }

第三步：构建图（Graph）并编译

# 初始化图，传入 State 类型 workflow = StateGraph(WeatherState) # 添加 Nodes workflow.add_node("fetch_weather", fetch_weather) workflow.add_node("handle_failure", handle_failure) # 设置入口点（Entry Point） workflow.set_entry_point("fetch_weather") # 添加 Edges：从 fetch_weather 的输出，根据 should_retry 函数决定去哪 workflow.add_conditional_edges( "fetch_weather", should_retry, { "fetch_weather": "fetch_weather", # 重试，回到自身 "handle_failure": "handle_failure", # 去失败处理 END: END # 成功，结束 } ) # 添加从 handle_failure 到 END 的边（失败处理完就结束） workflow.add_edge("handle_failure", END) # 编译图，获得可执行的 App app = workflow.compile(checkpointer=MemorySaver())

第四步：运行与调试

# 运行！传入初始 State initial_state = {"city": "Beijing"} result = app.invoke(initial_state) print(f"最终结果: {result}") # 输出类似：{'city': 'Beijing', 'temperature': 25.0, 'error': None, 'retry_count': 0} # 故意触发失败（用一个不存在的城市名） initial_state_fail = {"city": "NonExistentCity123"} result_fail = app.invoke(initial_state_fail) print(f"失败结果: {result_fail}") # 输出类似：{'city': 'NonExistentCity123', 'temperature': None, 'error': '查询失败，已重试2次：...'}

这个 50 行的代码，已经是一个工业级可用的、带重试的 Agent。它的强大在于：每一行代码都对应一个可解释、可测试、可监控的行为。你可以单独测试fetch_weather({"city": "Beijing"})，可以打印app.get_graph().draw_mermaid_png()（虽然我们禁用 mermaid，但draw_mermaid()文本输出可看结构），可以在app.invoke()时传入config={"recursion_limit": 10}防止无限循环。这正是 LangGraph 区别于“写一堆 if-else 调模型”的本质——它把业务逻辑从模型的不可控输出中解放出来，交还给确定性的代码。

3.3 LangChain LCEL 实战：用一行代码组装一个 RAG Chain

在 LangGraph 的大图中，Node 往往是复杂的业务逻辑，而 LCEL 则擅长快速组装“小而美”的数据处理单元。下面是一个生产环境中高频使用的 RAG Chain，它接收用户问题，从向量库检索相关文档，再用 LLM 生成答案：

from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_community.vectorstores import Chroma from langchain_community.embeddings import OllamaEmbeddings from langchain_community.llms import Ollama # 1. 定义 Prompt（使用 ChatPromptTemplate，支持 system/user/assistant 角色） prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个专业的客服助手。请严格基于以下提供的上下文信息回答用户问题。如果上下文没有相关信息，请回答'根据现有资料，我无法回答这个问题。'"), ("human", "问题：{question}\n\n上下文：{context}") ]) # 2. 初始化 LLM 和 Embedding llm = Ollama(model="llama3:8b-instruct", temperature=0.1) embedding = OllamaEmbeddings(model="llama3:8b-instruct") # 3. 加载向量库（假设已有 chroma_db 目录） vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embedding) retriever = vectorstore.as_retriever(search_kwargs={"k": 3}) # 检索 top3 # 4. 组装 LCEL Chain（核心就这一行！） rag_chain = ( {"context": retriever, "question": lambda x: x["question"]} # 输入映射：retriever 作用于 question | prompt # 格式化 Prompt | llm # 调用 LLM | StrOutputParser() # 解析为字符串 ) # 5. 调用 result = rag_chain.invoke({"question": "我们的退货政策是什么？"}) print(result)

这段代码的威力在于其组合性与可替换性。如果你想换一个更强大的模型，只需改llm = ChatOpenAI(model="gpt-4-turbo")；想换向量库，把Chroma换成PGVector；想加一个“答案置信度打分”环节，插入一个| confidence_scorerNode。LCEL 的|操作符，让这种替换变得像搭积木一样直观。在 LangGraph 中，你可以把这个rag_chain当作一个 Node：

workflow.add_node("answer_customer_question", rag_chain)

它就自动获得了invoke、stream、batch等所有Runnable接口，无缝融入图中。这就是 LCEL 和 LangGraph 协同工作的典型场景：LCEL 负责“怎么做”（How），LangGraph 负责“什么时候做、做几次、失败了怎么办”（When/How Many/What If）。

4. 深度对比与选型指南：LangChain vs LangGraph，何时该用哪一个？

4.1 功能维度对比：一张表看清核心差异

这张表不是为了贬低 LangChain，而是为了精准定位。LangChain 的AgentExecutor在 2023 年让无数开发者第一次体验到“模型自己调工具”的震撼，它的历史功绩不可磨灭。但工程实践告诉我们：易用性（Ease of Use）和可靠性（Reliability）往往是一对矛盾体。LangChain 选择了前者，LangGraph 则勇敢地承担起后者的重任。

4.2 选型决策树：5 个关键问题帮你一锤定音

在项目启动前，花 2 分钟回答以下 5 个问题，就能 90% 确定该用哪个：

1. 你的业务流程是否超过 3 个逻辑步骤？
   例如：“用户提问 → 检索文档 → 提取关键句 → 与知识库比对 → 生成答案 → 发送邮件通知” —— 这是 6 步。LangChain 的 Chain 会变得冗长难维护，LangGraph 的图结构则天然适合。

2. 是否存在需要根据中间结果动态分支的判断？
   例如：“如果信用分 > 700，走快速审批；否则，触发人工审核；如果人工审核超时，则自动拒绝”。这种 if-elif-else，在 LangGraph 里是should_route函数里的几行if/elif；在 LangChain Agent 里，你得指望模型每次都正确输出Action: fast_approve或Action: manual_review，风险极高。

3. 是否要求对每一次失败进行精确的、可配置的重试或降级？
   例如：“API 超时重试 2 次，503 错误降级到缓存数据，404 错误记录并告警”。LangGraph 的retry_count和error字段让你能写出任意复杂的策略；LangChain 的ToolException只能捕获，无法优雅降级。

4. 是否需要完整的执行链路追踪，用于线上问题排查或客户投诉溯源？
   例如，客户投诉“为什么我的贷款被拒？”，你能立刻拿出一份包含“信用分查询结果：680”、“人工审核未响应”、“最终决策：拒绝”的完整 trace 吗？LangGraph 的get_state_history()一键生成；LangChain 的日志需要你手动拼凑。

5. 团队是否有能力维护一个状态机模型？
   这是最现实的问题。LangGraph 要求开发者理解状态、副作用、并发合并。如果你的团队全是 Python 新手，强行上 LangGraph 可能适得其反。我的建议是：用 LangChain 快速 MVP，当第一个客户提出“你们的流程能不能加个风控校验环节？”时，就是切换 LangGraph 的最佳时机。我们有个项目，就是这么从 LangChain 的 3 个 Chain，平滑演进到 LangGraph 的 12 个 Node，全程无停机。

4.3 混合架构实战：LangChain + LangGraph 的黄金搭档

在真实项目中，LangChain 和 LangGraph 绝非“二选一”，而是“主从协作”。一个典型的混合架构如下：

• 顶层是 LangGraph 图：它定义整个 Agent 的宏观骨架。Entry Point 是用户输入，出口是最终响应。图中包含几个核心 Node：

  • preprocess_input：用 LangChain 的DocumentLoader和TextSplitter清洗用户上传的 PDF/Word。
  • route_to_module：一个 Condition Function，根据用户问题关键词（如“账单”、“物流”、“退款”）决定进入哪个子模块。
  • billing_module：一个 LangGraph 子图，专门处理账单查询，内部可能包含auth_node、db_query_node、calculate_node。
  • log_and_monitor：一个通用 Node，记录所有 State 变化到 Prometheus。

• 子模块是 LangChain LCEL Chain：比如billing_module里的db_query_node，它本身就是一个 LCEL Chain：{"query": generate_sql_prompt} | llm | sql_parser | execute_sql。这个 Chain 被封装成一个 Node，对外只暴露invoke(input)接口。

这种架构的优势是“各司其职”：LangGraph 负责流程治理（Orchestration），LangChain 负责数据处理（Transformation）。它既享受了 LangGraph 的可靠性和可观测性，又保留了 LangChain 的开发效率和生态丰富性。我们一个政务热线 AI 项目，就采用此架构，支撑了日均 5000+ 次多步骤咨询（查政策→比对资格→生成材料清单→预约窗口），上线半年，流程变更 17 次，从未因框架问题导致故障。

5. 常见问题与避坑指南：那些只有踩过才懂的实战经验

5.1 “State 更新不生效”：Pydantic 模型的深拷贝陷阱

现象：你在 Node 函数里修改了 State 字典，比如state["temperature"] = 25.0，但下一个 Node 收到的 State 里temperature还是None。

原因：LangGraph 默认会对 State 进行深拷贝（deep copy），以保证 Node 之间的隔离性。如果你的 State 是一个 Pydantic 模型实例（而非普通 dict），而你又没正确实现model_dump()或model_copy()，深拷贝就会失败或产生意外行为。

解决方案：永远用普通 Python dict 定义 State。这是 LangGraph 官方强烈推荐的做法，也是我们团队的铁律。不要写：

# ❌ 错误：用 Pydantic BaseSettings（已弃用）或复杂模型 class State(BaseModel): city: str # ... 其他字段

而要写：

# ✅ 正确：用 TypedDict 或 plain dict class WeatherState(TypedDict): city: str temperature: Optional[float] # ... # 或者更简单，直接用 dict，不定义类 # initial_state = {"city": "Beijing", "temperature": None}

实操心得：TypedDict 是最佳选择，它提供 IDE 类型提示，又保持 dict 的轻量和兼容性。我曾在一个项目里为了“类型安全”硬上了 Pydantic V2 模型，结果因为model_config = ConfigDict(arbitrary_types_allowed=True)配置不当，导致 State 在跨线程时丢失，debug 了整整两天。从此，State 就是 dict，简单、粗暴、可靠。

5.2 “无限循环”：Condition Function 的逻辑漏洞

现象：app.invoke()卡住，CPU 占用 100%，日志里反复打印同一个 Node 的执行日志。

原因：should_retry这类 Condition Function 返回了错误的 Node 名称，或者其逻辑存在死循环。最常见的错误是：if state.get("error"): return "fetch_weather"，但fetch_weatherNode 本身又可能设置error，导致永远在fetch_weather和should_retry之间打转。

解决方案：Condition Function 必须是幂等的，且必须有明确的终止条件。修复上面的例子：

def should_retry(state: WeatherState) -> str: # ✅ 正确：检查 error AND retry_count，确保有上限 if state.get("error") and state.get("retry_count", 0) < 2: return "fetch_weather" elif state.get("error"): return "handle_failure" else: return END # ✅ 明确的终止分支

实操心得：在写 Condition Function 时，养成习惯：打开编辑器，把所有可能的state字段组合列出来（如{"error": None, "retry_count": 0}、{"error": "timeout", "retry_count": 1}、{"error": "timeout", "retry_count": 2}），然后逐个检查你的if/elif/else是否覆盖了所有情况，并且每个分支都指向一个有效的、不会导致循环的 Node 或END。这比写代码还重要。

5.3 “性能瓶颈”：同步阻塞调用拖垮整个图

现象：一个 Node 里调用了慢速的外部 API（如老系统 SOAP 接口），导致整个 Graph 执行缓慢，甚至超时。

原因：LangGraph 默认是同步执行的。fetch_weatherNode 里的requests.get()是阻塞的，它会卡住整个事件循环。

解决方案：将阻塞调用包装为异步函数，并用asyncio.to_thread在线程池中执行。LangGraph 原生支持异步 Node：

import asyncio import requests async def fetch_weather_async(state: WeatherState) -> WeatherState: city = state["city"] try: # 在线程池中执行阻塞的 requests loop = asyncio.get_event_loop() response = await loop.run_in_executor( None, # 使用默认线程池 lambda: requests.get(f"http://wttr.in/{city}?format=%t", timeout=5) ) if response.status_code == 200: temp_str = response.text.strip().replace("°C", "").replace("+", "") return {"temperature": float(temp_str)} else: raise Exception(f"API returned {response.status_code}") except Exception as e: return {"error": str(e), "retry_count": 1} # 在 workflow.add_node 时，直接传入这个 async 函数 workflow.add_node("fetch_weather", fetch_weather_async)

实操心得：不要试图在 Node 里用time.sleep()模拟延迟，那只会让问题更隐蔽。真正的性能问题，永远来自外部依赖。我的经验是：所有涉及网络、文件 IO、数据库查询的 Node，第一反应就应该是异步化。LangGraph 的异步支持非常成熟，app.ainvoke()、app.astream()都能完美工作。

5.4 “调试困难”：如何像调试普通 Python 一样调试 LangGraph

现象：app.invoke()报错，堆栈信息指向 LangGraph 内部，找不到自己的代码在哪一行出了问题。

解决方案：利用 LangGraph 的get_graph()和get_state_history()，以及 Python 的breakpoint()。

• get_graph()：获取图的结构描述，可打印workflow.get_graph().draw_mermaid()查看文本版流程图，确认 Node 连接是否符合预期。
• get_state_history()：这是神器。在app.invoke()后，立即调用：

  history = app.get_state_history(config) for state in history: print(f"Step {state.config['checkpoint_id']}: {state.values}")

  你会看到每一步 State 的完整快照，清晰看到数据是如何一步步被修改的。
• breakpoint()：在你的 Node 函数第一行加breakpoint()，运行时会进入 pdb 调试器，你可以pp state查看当前状态，n下一步，c继续。

实操心得：我从不依赖 IDE 的图形化调试器来调试 LangGraph。get_state_history()是我每天必用的命令，它让我对整个流程的掌控感，远超任何可视化工具。记住，LangGraph 的本质是 Python 代码，调试它，就用最原始、最可靠的 Python 调试方法。

5.5 “部署难题”：如何将 LangGraph App 部署为生产 API

现象：本地app.invoke()跑得好好的，但用 FastAPI 封装后，app.invoke()报RecursionError或Checkpointer not found。

原因：LangGraph 的 `check