1. 项目概述：为什么我们需要 LangGraph？

如果你最近在折腾大模型应用，尤其是想搞点能自主决策、能串联多个步骤的智能体（Agent），那你大概率已经听过 LangChain 的大名，也可能被它早期版本里那些略显繁琐的 Agent 实现方式搞得有点头疼。任务一复杂，状态管理、循环控制、错误处理这些脏活累活就全冒出来了，代码写着写着就成了“面条”。这正是 LangGraph 诞生的背景，它不是来取代 LangChain 的，而是 LangChain 官方推出的一个专门用于构建有状态、多环节的 Agent 应用的新框架。你可以把它理解为给 Agent 开发装上了一套精密的“流程引擎”和“中央控制器”。

这一卷我们要聊的，就是如何从“会用 Agent”到“能设计并实现一个健壮的 Agent 系统”。LangGraph 的核心思想很直观：把你的 Agent 工作流建模成一个图（Graph）。图中的节点（Node）是一个个执行单元（比如调用一次大模型、执行一个工具函数、做一个条件判断），边（Edge）则定义了节点之间的流转逻辑。这听起来是不是有点像画流程图？没错，它的设计哲学就是让复杂的工作流变得可视、可控、可调试。我们将彻底拆解这个框架，不光是讲怎么调用 API，更要弄明白它背后的设计理念、解决的核心痛点，以及在实际项目中如何避开那些新手常踩的“坑”。

2. LangGraph 核心概念与设计哲学拆解

在一头扎进代码之前，我们必须先理解 LangGraph 的几个基石性概念。这能帮助我们在设计工作流时，做出更合理的选择。

2.1 状态管理：一切的中心

LangGraph 的核心是一个持续更新的状态（State）对象。这个状态对象在图的整个执行生命周期中流转，每个节点读取它、修改它，并决定下一步去哪里。这解决了传统链式调用中状态传递混乱的问题。

状态的定义：通常，我们会用一个 Pydantic 的BaseModel来定义状态，明确其中每个字段的类型和含义。例如，一个简单的问答 Agent 状态可能包含：

from typing import List, Annotated from typing_extensions import TypedDict import operator class AgentState(TypedDict): question: str # 用户输入的问题 context: List[str] # 检索到的相关文档片段 reasoning: str # 模型的思考链 answer: str # 最终生成的答案 iterations: Annotated[int, operator.add] # 已执行的循环次数

注意iterations字段的Annotated[int, operator.add]注解，这是 LangGraph 的一个精妙设计。它声明这个字段是一个“可归约的”（reducible）值。当多个节点并发修改它时（比如两个并行工具调用都增加了迭代次数），LangGraph 知道应该用operator.add这个函数来合并（归约）这些修改，而不是简单地覆盖。这对于计数、收集列表等场景至关重要。

状态更新的机制：LangGraph 采用了函数式编程中“更新器”（updater）的概念。节点函数不直接修改传入的状态对象，而是返回一个字典，其中包含要更新的字段和值。框架会自动将这些更新应用到全局状态上。这种方式保证了状态变更的可预测性和可追溯性。

2.2 节点与边：构建工作流的积木

节点（Node）是工作流的基本执行单元。一个节点就是一个普通的 Python 函数（或可调用对象），它接收当前状态作为参数，并返回对状态的更新。一个节点可以做的事情非常灵活：

• 调用一个大模型。
• 执行一个预定义的工具（如网络搜索、数据库查询）。
• 运行一段业务逻辑代码。
• 只是一个简单的条件判断或计算。

边（Edge）定义了工作流的控制逻辑。它决定了在当前节点执行完毕后，下一步应该去哪个节点。LangGraph 提供了几种类型的边：

1. 起始边（Start Edge）：定义工作流从哪个节点开始。
2. 条件边（Conditional Edge）：根据当前状态的某些值，动态决定下一个节点。这是实现分支和循环的关键。
3. 普通边（Normal Edge）：无条件地指向下一个节点。

通过组合节点和边，你可以构建出顺序、分支、循环乃至更复杂的拓扑结构，比如支持自我修正的 Agent，或者需要多轮工具调用的复杂任务规划器。

2.3 图编译与执行：从定义到运行

定义好节点和边之后，你需要创建一个StateGraph对象，将节点和边添加进去，最后调用compile()方法。这个编译过程会进行验证，并生成一个可执行的、优化的计算图。编译后的图对象有一个invoke(input_state)方法，这是启动工作流的入口。

编译时优化：compile()方法不只是做个简单的打包。它会检查图的连通性，优化执行路径，并为可视化做好准备。编译后的图对象还包含了所有必要的元信息，使得序列化、持久化和在不同环境间迁移成为可能。

执行模式：invoke是同步执行。对于长时间运行的任务，LangGraph 还支持异步执行（ainvoke）以及流式响应（stream），后者可以让你实时看到每个节点执行后的状态快照，对于调试和构建交互式 UI 体验非常有用。

3. 核心组件深度解析与实操要点

理解了基本概念，我们来深入看看构成 LangGraph 工作流的几个关键部分，以及在实际编码时需要注意什么。

3.1 状态模式设计：TypedDict vs BaseModel

如上所述，定义状态有两种主流方式：使用typing.TypedDict或pydantic.BaseModel。它们各有优劣：

• TypedDict：更轻量，是 Python 的类型提示标准库的一部分，无需额外依赖。它在运行时没有验证，性能开销极小。适合状态结构简单、确定，且你信任各个节点会正确更新字段的场景。

  from typing import TypedDict, List class SimpleState(TypedDict): messages: List[dict] step: int

• Pydantic BaseModel：功能强大，提供运行时数据验证、序列化/反序列化（如 JSON）、以及更丰富的字段类型（如自定义校验器）。这能极大增强工作流的健壮性，在节点返回意外数据时能尽早报错，而不是让错误状态流传下去。

  from pydantic import BaseModel, Field, validator from typing import List class RobustState(BaseModel): messages: List[dict] = Field(default_factory=list) step: int = Field(ge=0, description="Non-negative step counter") @validator('messages') def messages_must_have_role(cls, v): for msg in v: if 'role' not in msg or 'content' not in msg: raise ValueError('Message must have "role" and "content"') return v

实操心得：在项目初期快速原型阶段，可以使用TypedDict。一旦工作流逻辑稳定，尤其是需要与外部系统（如数据库、API）交换数据时，强烈建议切换到 Pydantic BaseModel。它带来的验证和文档化收益，远超一点点性能开销，能避免很多隐蔽的 bug。

3.2 条件边与路由逻辑：实现智能决策

条件边是让 Agent 拥有“判断力”的核心。它通常与一个“路由函数”（Router）配合使用。这个函数查看当前状态，返回下一个要执行的节点的名称。

from langgraph.graph import StateGraph, END import operator class State(TypedDict): question: str analysis: str needs_search: bool answer: str def agent_node(state: State): # 模拟大模型分析问题 if "天气" in state["question"]: return {"analysis": "这是一个天气查询问题", "needs_search": True} else: return {"analysis": "这是一个通用知识问题", "needs_search": False, "answer": "我可以直接回答。"} def search_node(state: State): # 模拟搜索工具 return {"answer": f"根据搜索，{state['question']}的答案是晴天。"} def route_after_analysis(state: State): # 路由函数：根据 `needs_search` 字段决定下一步 if state.get("needs_search"): return "search_tool" else: return END # 直接结束 # 构建图 graph = StateGraph(State) graph.add_node("analyze", agent_node) graph.add_node("search_tool", search_node) graph.add_edge("analyze", route_after_analysis) # 将路由函数作为边 graph.add_edge("search_tool", END) graph.set_entry_point("analyze") app = graph.compile()

注意事项：路由函数应该保持纯净，只读状态，不修改状态。它的唯一职责就是根据当前状态做决策。复杂的业务逻辑应该放在节点里。此外，确保路由函数返回的节点名称，一定是你已经在图中添加过的，否则会在编译或运行时出错。

3.3 工具集成与结构化输出：扩展 Agent 能力

单独的 LLM 能力有限，Agent 的强大在于能使用工具。LangGraph 与 LangChain 的工具生态无缝集成。

集成 LangChain Tools：

from langchain.tools import TavilySearchResults from langgraph.prebuilt import ToolNode # 创建工具 search_tool = TavilySearchResults(max_results=2) # 创建工具节点，可以绑定多个工具 tool_node = ToolNode(tools=[search_tool]) # 将 tool_node 添加到你的图中 graph.add_node("web_search", tool_node)

ToolNode是一个 LangGraph 预置的节点，它能自动处理工具调用格式：读取状态中类似{"tools": [...], "tool_calls": [...]}的结构，执行对应的工具，并将结果以标准化格式写回状态。

处理结构化输出：很多时候，我们需要模型输出结构化的数据（如 JSON）以便后续节点处理。这通常需要配合 Pydantic 和 LangChain 的with_structured_output方法。在 LangGraph 中，你可以创建一个专门的“结构化解析节点”：

from langchain_core.pydantic_v1 import BaseModel, Field from langchain_openai import ChatOpenAI class Decision(BaseModel): reasoning: str = Field(description="思考过程") next_step: str = Field(description="下一步动作", enum=["search", "answer", "clarify"]) llm = ChatOpenAI(model="gpt-4") structured_llm = llm.with_structured_output(Decision) def decision_node(state: State): # 假设 state[‘messages’] 包含对话历史 decision = structured_llm.invoke(state["messages"]) # 将结构化的决策结果更新到状态 return {"decision": decision.dict()}

这样，decision_node的输出就是一个清晰的Decision对象，后续的路由函数可以直接根据state[‘decision’][‘next_step’]来做出判断，代码非常清晰。

4. 构建一个具备自我修正能力的问答 Agent：完整实操

让我们综合以上知识，构建一个相对复杂的 Agent。这个 Agent 能回答问题，如果对自己生成的答案信心不足，它会主动调用搜索工具获取信息，然后重新生成答案，最多尝试两次。

4.1 步骤一：定义状态与工具

首先，我们定义工作流的状态。它需要记录问题、对话历史、模型生成的答案、置信度、检索到的上下文以及重试次数。

from typing import TypedDict, List, Optional, Literal from typing_extensions import Annotated import operator class SelfCorrectingState(TypedDict): """自我修正Agent的状态定义""" question: str # 用户原始问题 messages: Annotated[List[dict], operator.add] # 对话历史，使用归约操作符 initial_answer: Optional[str] # 模型首次生成的答案 confidence: Optional[float] # 模型对答案的置信度 (0-1) search_context: Optional[List[str]] # 搜索工具返回的上下文 final_answer: Optional[str] # 最终答案 attempt: Annotated[int, operator.add] # 尝试次数计数器 phase: Literal["initial_answer", "check_confidence", "search", "synthesize", "done"] # 当前阶段

我们使用Literal来定义phase字段，这将在路由逻辑中发挥巨大作用。Annotated用于messages和attempt，确保在并发或循环场景下更新正确。

接着，准备工具。这里我们使用一个模拟的搜索工具和一个置信度评估工具。

# 模拟一个网络搜索工具 def web_search_tool(query: str) -> List[str]: print(f"[工具调用] 搜索查询: {query}") # 模拟返回一些“搜到”的文档片段 return [ f"关于‘{query}’的权威资料片段A。", f"根据最新研究，‘{query}’的相关信息B。" ] # 模拟一个置信度评估工具（实际中可能用另一个LLM调用来实现） def evaluate_confidence(answer: str, question: str) -> float: """简单模拟置信度评估。实际项目需要更复杂的逻辑。""" keywords = ["可能", "也许", "据我所知", "不确定"] if any(kw in answer for kw in keywords): return 0.3 # 低置信度 elif len(answer) < 10: return 0.5 # 中等置信度 else: return 0.8 # 高置信度

4.2 步骤二：实现各个功能节点

我们将工作流分解为以下几个节点：

1. 生成初始答案节点：调用 LLM 直接回答问题。
2. 评估置信度节点：评估初始答案的置信度。
3. 搜索节点：如果置信度低，调用搜索工具。
4. 综合生成节点：结合搜索上下文，重新生成最终答案。
5. 完成节点：整理最终输出。

from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-3.5-turbo") def generate_initial_answer_node(state: SelfCorrectingState): """节点1：生成初始答案""" print(f"[节点1] 正在生成初始答案...") # 构建给LLM的提示 prompt = f"请直接回答以下问题：{state['question']}。请给出简洁明确的答案。" response = llm.invoke(prompt) initial_answer = response.content # 更新状态 new_messages = [{"role": "assistant", "content": f"初始答案: {initial_answer}"}] return { "initial_answer": initial_answer, "messages": new_messages, "phase": "check_confidence" # 进入下一阶段 } def check_confidence_node(state: SelfCorrectingState): """节点2：评估答案置信度""" print(f"[节点2] 正在评估答案置信度...") confidence = evaluate_confidence(state["initial_answer"], state["question"]) print(f" 置信度评分: {confidence}") next_phase = "search" if confidence < 0.6 else "synthesize" return { "confidence": confidence, "phase": next_phase # 根据置信度决定下一阶段 } def search_node(state: SelfCorrectingState): """节点3：执行搜索""" print(f"[节点3] 置信度不足，正在搜索更多信息...") search_results = web_search_tool(state["question"]) return { "search_context": search_results, "phase": "synthesize" # 搜索完成后进入综合阶段 } def synthesize_final_answer_node(state: SelfCorrectingState): """节点4：综合信息生成最终答案""" print(f"[节点4] 正在生成最终答案...") attempt = state.get("attempt", 0) + 1 base_info = f"用户问题：{state['question']}\n" if state.get("search_context"): # 如果有搜索上下文，结合上下文生成 context_str = "\n".join(state["search_context"]) prompt = f"{base_info}参考以下信息：\n{context_str}\n请生成一个准确、全面的最终答案。" else: # 如果置信度高，直接优化初始答案 prompt = f"{base_info}基于以下初始回答：'{state['initial_answer']}'，请将其润色为更完整、专业的最终答案。" response = llm.invoke(prompt) final_answer = response.content new_messages = [{"role": "assistant", "content": f"最终答案 (尝试{attempt}): {final_answer}"}] return { "final_answer": final_answer, "messages": new_messages, "attempt": 1, # 增加尝试计数 "phase": "done" } def finalize_node(state: SelfCorrectingState): """节点5：完成处理，整理输出""" print(f"[节点5] 流程结束。") # 这里可以做一些清理或格式化工作，比如确保final_answer不为空 if not state.get("final_answer"): return {"final_answer": state.get("initial_answer", "未能生成答案。")} # 状态已更新，无需返回新内容也可 return {}

4.3 步骤三：构建图并定义路由逻辑

现在，我们将节点组装起来，并定义它们之间的流转逻辑。

from langgraph.graph import StateGraph, END # 1. 创建图 workflow = StateGraph(SelfCorrectingState) # 2. 添加所有节点 workflow.add_node("generate", generate_initial_answer_node) workflow.add_node("check_conf", check_confidence_node) workflow.add_node("search", search_node) workflow.add_node("synthesize", synthesize_final_answer_node) workflow.add_node("finalize", finalize_node) # 3. 设置入口点 workflow.set_entry_point("generate") # 4. 添加边（包括条件边） workflow.add_edge("generate", "check_conf") # 生成答案后必然去检查置信度 # 从 check_conf 出来的条件路由 def route_after_check(state: SelfCorrectingState): if state["phase"] == "search": return "search" else: # phase == "synthesize" return "synthesize" workflow.add_conditional_edges( "check_conf", route_after_check, {"search": "search", "synthesize": "synthesize"} ) # 搜索完成后去综合 workflow.add_edge("search", "synthesize") # 综合完成后去最终化 workflow.add_edge("synthesize", "finalize") # 最终化后结束 workflow.add_edge("finalize", END) # 5. 编译图 app = workflow.compile()

4.4 步骤四：执行与验证

让我们用两个不同的问题来测试这个 Agent。

# 测试1：一个可能置信度较低的问题 print("=== 测试1: 复杂问题 ===") state_input1 = {"question": "量子计算的主要技术挑战是什么？", "messages": [], "attempt": 0, "phase": "initial_answer"} result1 = app.invoke(state_input1) print(f"\n最终答案：{result1['final_answer'][:200]}...") # 截断显示 print(f"总尝试次数：{result1['attempt']}") # 测试2：一个简单直接的问题 print("\n\n=== 测试2: 简单问题 ===") state_input2 = {"question": "法国的首都是哪里？", "messages": [], "attempt": 0, "phase": "initial_answer"} result2 = app.invoke(state_input2) print(f"\n最终答案：{result2['final_answer']}") print(f"总尝试次数：{result2['attempt']}")

执行上述代码，你会在控制台看到工作流一步步执行的日志。对于第一个复杂问题，由于模拟的置信度评估较低（答案中可能包含“可能”等词），Agent 会走generate -> check_conf -> search -> synthesize -> finalize的路径。对于第二个简单问题，置信度评估较高，则会走generate -> check_conf -> synthesize -> finalize的路径，跳过了搜索环节。

这个例子展示了 LangGraph 如何清晰地管理一个包含条件判断和循环（通过attempt计数可以扩展为重试逻辑）的复杂工作流。所有的状态流转和业务逻辑都通过图和节点直观地表达了出来。

5. 高级特性与生产级考量

当你掌握了基础构建方法后，以下这些高级特性和考量能帮助你将 LangGraph 应用到更严肃的生产环境中。

5.1 持久化检查点与人类干预

LangGraph 支持**检查点（Checkpoint）**机制，这可以说是其最强大的特性之一。它允许你将工作流的完整状态（包括所有变量、执行历史）保存下来，之后可以从这个精确的点恢复执行。这对于以下场景至关重要：

• 长时间运行的任务：任务可以暂停、恢复，甚至迁移到另一台服务器。
• 等待人工审核/输入：Agent 可以在某个节点暂停，生成一个待办事项发送给人类，等人类回复后再继续。
• 错误恢复与重试：如果某个节点因临时错误失败，可以从上一个检查点重试，而不是从头开始。

使用检查点：通常需要配置一个持久化存储后端（如内存、数据库、Redis）。LangGraph 提供了接口，你需要实现CheckpointSaver逻辑。

from langgraph.checkpoint import MemorySaver from langgraph.graph import StateGraph # 使用内存存储检查点（生产环境应用数据库） memory = MemorySaver() workflow = StateGraph(..., checkpointer=memory) # 在调用时，可以指定一个线程ID（thread_id）来关联同一会话 config = {"configurable": {"thread_id": "user_123_session_1"}} # 首次调用会创建检查点 result1 = app.invoke({"question": "..."}, config=config) # 假设在这里任务被暂停了... # 之后，可以从同一个thread_id恢复，并传入新的输入或继续执行 # 例如，模拟人类输入后继续 result2 = app.invoke({"human_feedback": "请再解释一下第一步。"}, config=config)

通过检查点，你可以构建出真正能与人类协作、支持断点续传的复杂 Agent 应用。

5.2 子图与模块化设计

对于非常复杂的工作流，你可以使用子图（Subgraph）来模块化你的设计。子图允许你将一个完整的图作为一个节点嵌入到另一个更大的图中。这带来了几个好处：

• 代码复用：将通用的功能（如“检索增强生成RAG模块”、“代码审查模块”）封装成子图，在不同主图中重复使用。
• 逻辑清晰：将大问题分解为小问题，每个子图负责一个相对独立的子任务。
• 简化主图：主图只需要关注高层的流程控制，细节隐藏在子图中。

创建与使用子图：

from langgraph.graph import StateGraph # 1. 定义一个子图（例如，一个专门的RAG检索子图） def build_rag_subgraph(): rag_graph = StateGraph(...) # ... 添加检索、重排、生成等节点 rag_graph.add_node(...) rag_graph.add_edge(...) return rag_graph.compile() # 2. 在主图中，将这个编译好的子图作为一个节点添加 main_graph = StateGraph(...) rag_app = build_rag_subgraph() main_graph.add_node("rag_module", rag_app) # 将子图作为节点添加 # 在主图的其他节点中，可以调用这个子图节点，它会接收和返回符合子图状态定义的数据。

5.3 并发执行与优化

LangGraph 支持在条件允许的情况下并发执行多个节点。这通过State定义中的Annotated归约操作符来实现。当框架检测到两个节点不依赖于对方的输出（即它们修改的状态字段是独立的），它可能会尝试并发执行它们以提升效率。

设计并发友好的状态：为了最大化并发潜力，在设计状态 Schema 时，尽量让不同的任务修改不同的字段。例如，一个节点负责更新search_results，另一个节点负责更新user_profile，它们之间没有依赖，就可以并发。

注意事项：并发不是银弹。如果节点之间有严格的先后顺序依赖（比如必须先检索才能生成），强行并发会导致错误。你需要仔细设计工作流的依赖关系。可视化工具（后面会提到）可以帮助你分析节点间的依赖。

5.4 监控、日志与可视化

在生产环境中，对 Agent 工作流进行监控和调试是必须的。

结构化日志：在每个节点的函数中，使用结构化的日志记录（如 Python 的logging模块，并输出 JSON 格式）。记录关键信息，如节点开始/结束时间、输入状态摘要、输出状态摘要、工具调用详情、LLM 的请求与响应（注意脱敏）等。这有助于后续的问题追踪和性能分析。

LangGraph 可视化：编译后的图对象有一个get_graph()方法，可以输出图的 Mermaid 格式定义。你可以将其复制到 Mermaid Live Editor 中，立即生成可视化的流程图。这是理解和沟通工作流设计的绝佳工具。

# 生成图的Mermaid文本表示 graph_diagram = app.get_graph().draw_mermaid() print(graph_diagram)

将输出的文本粘贴到支持 Mermaid 的 Markdown 编辑器或在线工具中，就能看到一张清晰的工作流图。

追踪与可观测性：考虑集成像 LangSmith 这样的 LLM 应用追踪平台。LangSmith 与 LangGraph 有很好的集成，可以记录每一次invoke的完整执行轨迹，包括每个节点的输入输出、耗时、LLM 调用成本等，是进行调试、优化和评估的利器。

6. 常见问题、调试技巧与避坑指南

在实际开发中，你一定会遇到各种问题。下面是一些常见坑点和解决思路。

6.1 状态更新不生效或覆盖

问题现象：节点返回了更新字典，但状态没有变化，或者后一个节点的更新覆盖了前一个节点的。

排查与解决：

1. 检查状态 Schema 定义：确保你返回的字典键名与状态 Schema 中定义的字段名完全一致。Python 是大小写敏感的。
2. 理解归约操作符：对于使用Annotated声明了归约操作符的字段（如列表、计数器），节点返回的应该是增量值，而不是完整值。例如，对于Annotated[List, operator.add]，你应该返回{"messages": [new_message]}，而不是{"messages": all_messages}。框架会自动帮你合并。
3. 检查节点执行顺序：如果两个节点并发执行且修改了同一个非归约字段，后完成的一个会覆盖前一个。你需要通过设计工作流（如添加边）来避免这种竞争条件，或者将该字段改为归约字段（如果业务逻辑允许）。

6.2 条件边路由错误或进入死循环

问题现象：工作流没有按预期分支，或者一直在某几个节点间循环，无法结束。

排查与解决：

1. 打印调试路由函数：在路由函数内部添加print语句，输出当前状态的关键字段和它决定返回的节点名称。确保你的逻辑条件 (if/else) 覆盖了所有可能的情况，并且最终一定能返回一个有效的节点名或END。
2. 检查phase或标志字段：像我们例子中使用phase字段来显式控制流程是一个好习惯。确保每个节点在更新状态时，都正确设置了下一个phase。
3. 避免无限循环：对于可能循环的路径（比如重试逻辑），一定要设置一个“逃生阀”。在我们的例子中，attempt计数器可以用于限制重试次数。可以在路由函数中检查if state[‘attempt’] > MAX_ATTEMPTS: return “finalize”。
4. 使用可视化工具：将图可视化，检查边的连接是否正确，是否存在形成闭环（循环）但缺少退出条件的情况。

6.3 工具调用失败或格式错误

问题现象：集成 LangChain Tools 时，节点报错，提示工具调用格式不正确。

排查与解决：

1. 确保状态格式匹配：ToolNode期望状态中包含特定的键来获取工具调用指令，通常是"tools"和"tool_calls"。你需要确保上一个节点（通常是 LLM 节点）的输出格式符合这个要求。使用bind_tools方法可以帮助 LLM 格式化输出。

   from langchain_core.messages import HumanMessage llm_with_tools = llm.bind_tools([search_tool]) # 当调用 llm_with_tools 时，如果模型决定调用工具，其返回的消息会包含标准的 tool_calls 结构。

2. 检查工具 Schema：确保传递给ToolNode的工具列表与 LLM 绑定的工具列表一致。工具名称、描述、参数 Schema 必须匹配，否则 LLM 可能生成无法解析的调用。
3. 处理工具错误：工具执行可能会失败（网络超时、API 错误等）。考虑在ToolNode外围包裹一层错误处理逻辑，或者使用自定义节点来更精细地控制工具调用和错误回退。

6.4 性能优化与成本控制

问题：复杂工作流调用 LLM 次数多，响应慢，成本高。

优化策略：

1. 缓存：对频繁且结果不变的 LLM 调用或工具调用（如某些知识查询）实施缓存。可以使用langchain.cache或外部缓存如 Redis。
2. 精简上下文：传递给 LLM 的messages或context要尽可能相关和精简。在 RAG 场景中，做好检索结果的重排和过滤，只保留最相关的片段。
3. 模型选型：并非所有节点都需要最强大的模型。对于路由判断、简单分类等任务，可以使用更小、更快的模型（如gpt-3.5-turbo），只在需要深度推理或生成的节点使用大模型（如gpt-4）。这被称为“混合模型”策略。
4. 异步与流式：如果前端允许，使用astream进行流式响应，可以提升用户体验。对于批量处理任务，使用异步调用 (ainvoke) 来并发执行多个独立的工作流实例。
5. 设置超时与重试：为 LLM 调用和工具调用配置合理的超时时间和重试策略，避免单个故障阻塞整个工作流。

6.5 测试策略

测试 Agent 工作流比测试普通函数更复杂，因为涉及状态流转和外部调用。

1. 单元测试节点函数：将每个节点函数当作纯函数（尽可能）进行测试。使用模拟（Mock）对象来替代 LLM 和工具调用，验证给定输入状态时，函数是否返回了预期的状态更新字典。
2. 集成测试子图：对编译好的子图进行测试，使用模拟的 LLM 和工具，验证整个子流程的输入输出是否符合预期。
3. 端到端测试：针对关键用户旅程，进行少量的端到端测试，使用真实的 LLM 和工具（但可能使用沙箱环境或测试密钥），确保整个工作流能跑通。这类测试速度慢、成本高，应作为验收测试。
4. 利用 LangSmith：LangSmith 的追踪功能不仅可以用于调试，也可以用于测试。你可以录制一次成功的运行轨迹作为“黄金标准”，在后续的测试中，将新的运行轨迹与它进行对比，检查关键节点的输入输出是否有重大偏离。

LangGraph 将一个复杂的 Agent 系统设计问题，转化为了一个相对直观的“构图”问题。它通过清晰的状态管理和显式的控制流，让多步骤、有状态、带条件的 LLM 应用变得可维护、可调试。从简单的线性链到复杂的、带循环和人工干预的工作流，它都能提供良好的支持。掌握它的核心概念——状态、节点、边、条件路由，并善用检查点、子图等高级特性，你就能设计出强大而稳健的智能体应用。