1. 项目概述为什么我们需要 LangGraph 来编排复杂工作流如果你尝试过用 LangChain 来构建一个稍微复杂点的 AI 应用比如一个需要多轮对话、条件分支、外部工具调用的智能体你很可能遇到过这样的困境代码很快变成了一团乱麻。各种if-else判断、状态维护、回调函数交织在一起逻辑变得难以追踪和调试。这就像试图用一堆零散的乐高积木搭建一座精密的大厦虽然每块积木LangChain 的链、代理、工具都很棒但缺少一个清晰的“施工图纸”和“骨架”来定义它们如何协同工作。这正是 LangGraph 要解决的核心问题。它不是一个替代 LangChain 的新框架而是 LangChain 生态系统中的一个专门用于编排Orchestration的库。你可以把它想象成乐高套装里的那张步骤说明书或者软件开发中的流程图工具。它提供了一种声明式的方式来定义和运行由多个步骤组成、可能包含循环和条件分支的工作流。我最近在重构一个智能客服项目时将原本基于 LangChain Agent 的、充斥着大量胶水代码的逻辑迁移到了一个由 LangGraph 驱动的 12 步工作流中。整个过程下来代码的可读性、可维护性和可观测性得到了质的提升。简单来说LangGraph 让你能用“画图”的思维来写代码。你定义好状态State的流转路径Graph它来负责执行和调度。这对于构建需要长期记忆、复杂决策路径或多人协作的 AI 应用来说几乎是目前最优雅的解决方案。接下来我就以这个“搭骨架”的过程为例拆解如何用 LangGraph 一步步构建一个健壮的工作流。2. 核心概念拆解LangGraph 的三要素与心智模型在动手写代码之前必须理解 LangGraph 的三个核心概念状态State、节点Node和边Edge。这构成了你设计工作流的心智模型。2.1 状态State工作流的共享记忆空间状态是一个类似字典Dict的对象它在工作流的所有节点之间共享和传递。你可以把它理解为整个工作流的“全局变量”或“上下文白板”。在 LangGraph 中状态通常使用 TypedDict 或 Pydantic BaseModel 来定义这能提供良好的类型提示和验证。例如在我们的客服工作流中状态可能包含这些字段from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages import operator class State(TypedDict): # 对话历史 messages: Annotated[List[dict], add_messages] # 特殊注解用于自动追加消息 # 用户当前查询 user_query: str # 从知识库检索到的上下文 retrieved_context: List[str] # 生成的初步答案 draft_answer: str # 是否需要转接人工 need_human: bool # 调用的工具名称用于日志和调试 last_tool_called: str这里有个关键点messages字段使用了Annotated[List[dict], add_messages]。add_messages是一个归约器Reducer它定义了当多个节点同时修改这个字段时如何合并这些修改这里是追加消息。这是 LangGraph 处理并发或分支合并时状态同步的优雅机制。实操心得在设计状态时要遵循“最小必要”原则。只把需要在节点间共享的数据放入状态。过度设计的状态结构会让工作流变得难以理解。同时善用Annotated和归约器来处理列表、计数器等需要聚合操作的字段这能避免很多状态覆盖的坑。2.2 节点Node工作流中的原子操作单元节点是一个普通的 Python 函数或可调用对象它接收当前状态作为输入并返回一个包含对状态所做更新的字典。节点应该职责单一比如“检索知识库”、“调用 LLM 生成回答”、“判断意图”。def retrieve_node(state: State) - dict: 知识库检索节点 query state[“user_query”] # 假设我们有一个检索函数 contexts vectorstore.similarity_search(query, k3) return {“retrieved_context”: [doc.page_content for doc in contexts]} def llm_generate_node(state: State) - dict: LLM生成节点 from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI prompt ChatPromptTemplate.from_messages([ (“system”, “你是一个客服助手请根据以下上下文回答问题。上下文{context}”), (“user”, “{query}”) ]) chain prompt | ChatOpenAI(model“gpt-4”) context “\n\n”.join(state[“retrieved_context”]) response chain.invoke({“context”: context, “query”: state[“user_query”]}) return {“draft_answer”: response.content}节点函数返回的字典中的键必须与状态中定义的字段名对应其值将用于更新状态。如果返回{“need_human”: True}那么工作流状态中的need_human字段就会被设置为True。2.3 边Edge控制流程的方向盘边决定了工作流的走向。它连接节点并根据条件决定下一个执行哪个节点。边主要有两种类型普通边Linear Edge无条件地从一个节点指向下一个节点。条件边Conditional Edge根据一个条件函数lambda或普通函数的返回值决定下一步走哪条分支。这个函数接收状态作为输入返回一个字符串该字符串对应下一个目标节点的名称。条件边是实现“循环”和“分支”的关键。例如你可以设置一个条件检查答案是否满足要求如果不满足就跳回“重写”节点或者检查用户是否表达了不满如果是则跳转到“安抚”或“转人工”节点。理解了这三个概念我们就可以像搭积木一样用节点和边来描绘出整个工作流的骨架图。3. 12步工作流骨架搭建实战下面我将构建一个简化的但覆盖核心模式的智能客服工作流。这个工作流包含12个关键节点展示了从接收到用户问题到最终回复或转人工的完整逻辑。我们会一步步实现它。3.1 步骤一定义工作流状态与工具首先我们定义完整的状态和需要用到的工具例如搜索、数据库查询。from typing import TypedDict, List, Annotated, Literal, Optional from langgraph.graph.message import add_messages from pydantic import BaseModel, Field import json # 1. 定义状态 class GraphState(TypedDict): # 核心对话流 messages: Annotated[List[dict], add_messages] user_input: str # 处理过程数据 parsed_intent: Optional[str] # 解析出的用户意图 requires_clarification: bool # 是否需要澄清问题 clarification_question: Optional[str] # 澄清问题内容 retrieved_faqs: List[str] # 检索到的FAQ答案 retrieved_kb_docs: List[str] # 检索到的知识库文档 # 生成与评估 generated_response: str response_confidence: float # 回答置信度 safety_check_passed: bool # 安全检查是否通过 # 流程控制标志 should_escalate: bool # 是否应升级转人工 escalation_reason: Optional[str] # 转人工原因 current_step: str # 当前步骤名用于调试 # 2. 定义一些模拟工具 class ToolSet: staticmethod def intent_classifier(query: str) - dict: 简陋的意图分类器 intents [“产品咨询”, “故障报修”, “账单问题”, “投诉建议”, “闲聊”] # 模拟分类逻辑 if “怎么用” in query or “功能” in query: return {“intent”: “产品咨询”, “confidence”: 0.9} elif “坏了” in query or “用不了” in query: return {“intent”: “故障报修”, “confidence”: 0.85} elif “钱” in query or “扣费” in query: return {“intent”: “账单问题”, “confidence”: 0.8} else: return {“intent”: “闲聊”, “confidence”: 0.5} staticmethod def retrieve_faq(query: str, intent: str) - List[str]: 模拟FAQ检索 faq_db { “产品咨询”: [“产品A支持7天无理由退货。”, “产品B需要安装专用驱动。”], “故障报修”: [“请尝试重启设备。”, “检查网络连接是否正常。”], “账单问题”: [“账单明细可在‘我的账户’中查看。”, “扣费问题请联系支付渠道。”], } return faq_db.get(intent, [“抱歉暂时没有相关信息。”]) staticmethod def retrieve_knowledge_base(query: str) - List[str]: 模拟知识库检索更深入的内容 # 这里可以集成真实的向量数据库如Chroma、Pinecone return [f“关于‘{query}’的深度技术文档摘要。”] staticmethod def safety_check(response: str) - bool: 简单的内容安全检查 blacklist [“暴力”, “违禁词”] return not any(word in response for word in blacklist) staticmethod def human_escalation_protocol(reason: str) - str: 模拟转人工流程 return f“问题已记录转人工原因{reason}。坐席将很快接入。”3.2 步骤二实现12个核心节点每个节点都是一个纯函数负责一项具体任务。# 节点1输入解析与标准化 def parse_input(state: GraphState) - dict: print(f“【节点1】解析输入: {state[‘user_input’]}”) # 简单清理和标准化输入 cleaned_input state[“user_input”].strip() return {“user_input”: cleaned_input, “current_step”: “parse_input”} # 节点2用户意图识别 def classify_intent(state: GraphState) - dict: print(f“【节点2】识别意图”) result ToolSet.intent_classifier(state[“user_input”]) return { “parsed_intent”: result[“intent”], “response_confidence”: result[“confidence”], # 意图置信度作为初始置信度 “current_step”: “classify_intent” } # 节点3判断是否需要澄清意图置信度低时 def decide_clarification(state: GraphState) - dict: print(f“【节点3】判断是否需要澄清置信度: {state[‘response_confidence’]}”) needs_clarify state[“response_confidence”] 0.7 clarification_q None if needs_clarify: clarification_q f“您的问题是关于‘{state[‘parsed_intent’]}’吗还是其他方面” return { “requires_clarification”: needs_clarify, “clarification_question”: clarification_q, “current_step”: “decide_clarification” } # 节点4生成澄清问题如果需要 def generate_clarification(state: GraphState) - dict: print(f“【节点4】生成澄清问题”) # 这里可以直接将澄清问题作为回复并等待用户下一轮输入。 # 在LangGraph中这通常通过更新messages并引导工作流进入一个等待用户输入的“暂停”状态来实现。 # 为简化我们假设澄清问题已在上一步生成本节点只是确认。 return { “generated_response”: state[“clarification_question”], “current_step”: “generate_clarification” } # 节点5检索FAQ针对高频简单问题 def retrieve_faq(state: GraphState) - dict: print(f“【节点5】检索FAQ意图: {state[‘parsed_intent’]}”) if state[“requires_clarification”]: # 如果需要澄清先不检索FAQ return {“retrieved_faqs”: [], “current_step”: “retrieve_faq”} faqs ToolSet.retrieve_faq(state[“user_input”], state[“parsed_intent”]) return {“retrieved_faqs”: faqs, “current_step”: “retrieve_faq”} # 节点6判断FAQ是否足够回答 def evaluate_faq_sufficiency(state: GraphState) - dict: print(f“【节点6】评估FAQ充分性”) # 简单逻辑如果检索到FAQ且意图明确则认为FAQ足够 is_sufficient bool(state[“retrieved_faqs”]) and state[“response_confidence”] 0.8 and not state[“requires_clarification”] return {“current_step”: “evaluate_faq_sufficiency”} # 注意这个判断结果不直接更新状态而是通过条件边来影响流程。 # 节点7深度知识库检索针对复杂问题 def retrieve_knowledge_base(state: GraphState) - dict: print(f“【节点7】深度知识库检索”) docs ToolSet.retrieve_knowledge_base(state[“user_input”]) return {“retrieved_kb_docs”: docs, “current_step”: “retrieve_knowledge_base”} # 节点8合成答案生成 def generate_answer(state: GraphState) - dict: print(f“【节点8】合成生成答案”) from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate # 组装上下文 context_parts [] if state[“retrieved_faqs”]: context_parts.append(“常见问题解答\n” “\n”.join(state[“retrieved_faqs”])) if state[“retrieved_kb_docs”]: context_parts.append(“相关知识文档\n” “\n”.join(state[“retrieved_kb_docs”])) context “\n\n”.join(context_parts) if context_parts else “无相关上下文。” prompt ChatPromptTemplate.from_messages([ (“system”, “你是一个专业的客服助手。请严格根据以下提供的信息来回答用户的问题。如果信息不足请礼貌告知。\n信息{context}”), (“user”, “{query}”) ]) llm ChatOpenAI(model“gpt-3.5-turbo”, temperature0.1) chain prompt | llm response chain.invoke({“context”: context, “query”: state[“user_input”]}) return { “generated_response”: response.content, “current_step”: “generate_answer” } # 节点9答案安全检查 def safety_check_answer(state: GraphState) - dict: print(f“【节点9】答案安全检查”) is_safe ToolSet.safety_check(state[“generated_response”]) return {“safety_check_passed”: is_safe, “current_step”: “safety_check_answer”} # 节点10答案质量与置信度评估 def evaluate_answer_quality(state: GraphState) - dict: print(f“【节点10】评估答案质量”) # 模拟一个简单的评估检查答案长度和是否包含“不确定”词汇 answer state[“generated_response”] length_ok len(answer) 10 is_confident “不确定” not in answer and “抱歉” not in answer final_confidence 0.8 if (length_ok and is_confident) else 0.4 return { “response_confidence”: final_confidence, “current_step”: “evaluate_answer_quality” } # 节点11判断是否需要转接人工 def decide_escalation(state: GraphState) - dict: print(f“【节点11】判断是否转人工”) should_escalate False reason None # 触发转人工的条件 if not state[“safety_check_passed”]: should_escalate True reason “回答内容安全性检查未通过” elif state[“response_confidence”] 0.5: should_escalate True reason “答案置信度过低” elif state[“parsed_intent”] “投诉建议”: # 假设投诉类直接转人工 should_escalate True reason “用户意图为投诉建议” elif “找人工” in state[“user_input”].lower(): should_escalate True reason “用户明确要求转人工” return { “should_escalate”: should_escalate, “escalation_reason”: reason, “current_step”: “decide_escalation” } # 节点12执行转人工或最终回复 def finalize_response(state: GraphState) - dict: print(f“【节点12】最终响应处理”) final_response “” if state[“should_escalate”]: final_response ToolSet.human_escalation_protocol(state[“escalation_reason”]) else: final_response state[“generated_response”] # 将最终响应添加到消息历史中模拟 new_message {“role”: “assistant”, “content”: final_response} return { “messages”: [new_message], # add_messages归约器会处理追加 “current_step”: “finalize_response” }3.3 步骤三构图与条件路由定义这是 LangGraph 最核心的部分我们将节点用边连接起来并定义路由逻辑。from langgraph.graph import StateGraph, END # 初始化图 workflow StateGraph(GraphState) # 添加所有节点 workflow.add_node(“parse_input”, parse_input) workflow.add_node(“classify_intent”, classify_intent) workflow.add_node(“decide_clarification”, decide_clarification) workflow.add_node(“generate_clarification”, generate_clarification) workflow.add_node(“retrieve_faq”, retrieve_faq) workflow.add_node(“evaluate_faq_sufficiency”, evaluate_faq_sufficiency) # 这是一个“路由节点” workflow.add_node(“retrieve_knowledge_base”, retrieve_knowledge_base) workflow.add_node(“generate_answer”, generate_answer) workflow.add_node(“safety_check_answer”, safety_check_answer) workflow.add_node(“evaluate_answer_quality”, evaluate_answer_quality) workflow.add_node(“decide_escalation”, decide_escalation) workflow.add_node(“finalize_response”, finalize_response) # 设置入口点 workflow.set_entry_point(“parse_input”) # 添加普通边线性流程 workflow.add_edge(“parse_input”, “classify_intent”) workflow.add_edge(“classify_intent”, “decide_clarification”) workflow.add_edge(“generate_clarification”, “retrieve_faq”) # 生成澄清后继续流程 workflow.add_edge(“retrieve_faq”, “evaluate_faq_sufficiency”) workflow.add_edge(“retrieve_knowledge_base”, “generate_answer”) workflow.add_edge(“generate_answer”, “safety_check_answer”) workflow.add_edge(“safety_check_answer”, “evaluate_answer_quality”) workflow.add_edge(“evaluate_answer_quality”, “decide_escalation”) workflow.add_edge(“decide_escalation”, “finalize_response”) workflow.add_edge(“finalize_response”, END) # 添加条件边分支流程 # 条件1是否需要澄清 def route_after_clarification(state: GraphState) - str: if state[“requires_clarification”]: return “generate_clarification” # 需要澄清跳转到生成澄清节点 else: return “retrieve_faq” # 不需要澄清直接进行FAQ检索 workflow.add_conditional_edges( “decide_clarification”, # 来源节点 route_after_clarification, # 条件函数 { “generate_clarification”: “generate_clarification”, “retrieve_faq”: “retrieve_faq” } ) # 条件2FAQ是否足够回答 def route_after_faq_evaluation(state: GraphState) - str: # 这里我们需要一个判断逻辑。由于evaluate_faq_sufficiency节点没有更新状态 # 我们需要在条件函数里重新计算或采用其他方式。 # 更佳实践让evaluate_faq_sufficiency节点返回一个结果标志。 # 为了演示我们简化如果检索到FAQ且不是澄清状态就认为足够。 if state[“retrieved_faqs”] and not state[“requires_clarification”]: return “generate_answer” # FAQ足够直接去生成答案可以基于FAQ else: return “retrieve_knowledge_base” # FAQ不足需要深度检索 workflow.add_conditional_edges( “evaluate_faq_sufficiency”, route_after_faq_evaluation, { “generate_answer”: “generate_answer”, “retrieve_knowledge_base”: “retrieve_knowledge_base” } ) # 条件3安全检查是否通过 def route_after_safety_check(state: GraphState) - str: if state[“safety_check_passed”]: return “evaluate_answer_quality” # 安全继续评估质量 else: return “decide_escalation” # 不安全直接触发转人工判断 workflow.add_conditional_edges( “safety_check_answer”, route_after_safety_check, { “evaluate_answer_quality”: “evaluate_answer_quality”, “decide_escalation”: “decide_escalation” } )3.4 步骤四编译与运行工作流将图编译成可执行对象并运行测试。# 编译图 app workflow.compile() # 为了可视化我们可以打印图的结构需要安装pygraphviz或者使用内置的mermaid输出 try: # 显示Mermaid格式的图定义可复制到Mermaid在线编辑器中查看 print(app.get_graph().draw_mermaid()) except: print(“无法生成Mermaid图将显示文本结构。”) print(app.get_graph().print_ascii()) # 运行工作流 initial_state: GraphState { “messages”: [], “user_input”: “我的设备突然无法开机了怎么办”, “parsed_intent”: None, “requires_clarification”: False, “clarification_question”: None, “retrieved_faqs”: [], “retrieved_kb_docs”: [], “generated_response”: “”, “response_confidence”: 0.0, “safety_check_passed”: True, “should_escalate”: False, “escalation_reason”: None, “current_step”: “” } print(“\n 开始执行工作流 ”) final_state None # app.invoke 会执行整个图直到结束 for step in app.stream(initial_state, stream_mode“values”): step_name step[“current_step”] print(f“流程状态更新 - 当前步骤: {step_name}”) if step_name “finalize_response”: final_state step print(f“最终回复: {step[‘messages’][-1][‘content’] if step.get(‘messages’) else ‘无’}”) print(“\n 工作流执行完毕 ”) if final_state: print(f“是否转人工: {final_state.get(‘should_escalate’)}”) print(f“最终置信度: {final_state.get(‘response_confidence’)}”)运行上述代码你会看到工作流按照我们设计的路径一步步执行打印出每个节点的日志并最终给出回答或转人工的决定。这个骨架清晰地分离了逻辑节点函数和流程图结构修改业务流程只需调整图的连接关系而无需深入修改每个节点的内部代码。4. 高级特性与实战技巧搭建起基础骨架后LangGraph 还有一些高级特性能让你的工作流更强大、更健壮。4.1 中断Interruption与人工介入在真实场景中工作流可能需要暂停以等待用户输入比如我们生成的澄清问题或者允许人工坐席中途接管。LangGraph 通过interrupt机制支持这一点。核心思路是在需要暂停的节点如generate_clarification不直接连接下一个节点而是抛出一个特定的中断键interrupt。编译图时通过interrupt_before或interrupt_after参数来指定哪些节点可以中断。from langgraph.graph import StateGraph, START, END from langgraph.checkpoint import MemorySaver from langgraph.prebuilt import ToolNode import asyncio # 使用检查点存储器这是支持中断和长会话的基础 memory MemorySaver() workflow_with_interrupt StateGraph(GraphState, config_schemadict) # ... 添加节点同上... # 配置中断在 generate_clarification 节点之后中断 workflow_with_interrupt.add_edge(“generate_clarification”, “__interrupt__”) # 指向一个特殊的中断边 app_with_interrupt workflow_with_interrupt.compile( checkpointermemory, interrupt_before[“retrieve_faq”] # 也可以在 retrieve_faq 之前中断但这里我们在澄清后中断 ) # 更精细的控制可以使用 interrupt_after[“generate_clarification”]运行时当流程到达中断点app.stream()会暂停并返回一个包含”__interrupt__”键的结果。你的外部程序如Web服务器可以捕获这个中断将澄清问题发送给用户等待用户回复后再携带新的用户输入和相同的会话IDconfigurable继续执行工作流。4.2 并行执行与归约如果工作流中有多个可以并行执行的任务例如同时检索FAQ和知识库或者同时进行安全检查和情感分析LangGraph 支持通过add_node和条件边模拟并行但更优雅的方式是使用Pregel的并发特性。不过在基础的StateGraph中我们可以设计一个“并行节点”它内部调用多个函数然后合并结果。更常见的模式是利用状态的“归约器”如之前add_messages使用的add_messages。当两个分支同时修改同一个列表字段时归约器定义了如何合并。例如你可以定义一个operator.add归约器来对数字字段求和。from typing import Annotated import operator class ParallelState(TypedDict): score_a: Annotated[int, operator.add] # 使用加法归约 score_b: Annotated[int, operator.add] results: Annotated[list, lambda x, y: x y] # 自定义列表合并归约 def node_a(state: ParallelState): return {“score_a”: 5, “results”: [“result_a”]} def node_b(state: ParallelState): return {“score_b”: 3, “results”: [“result_b”]} # 在图中如果node_a和node_b在同一个“步骤”中被执行通过特定配置 # 最终状态会是score_a5, score_b3, results[“result_a”, “result_b”]4.3 子图Subgraph封装复杂逻辑当一个工作流变得非常庞大时你可以将其中功能相关的节点组封装成一个子图。子图本身也是一个StateGraph可以被主图当作一个节点来调用。这极大地提升了模块化和复用性。例如我们可以把“检索模块”FAQ检索知识库检索封装成一个子图。from langgraph.graph import StateGraph as SubStateGraph # 1. 定义子图的状态通常是主状态的一个子集 class RetrievalState(TypedDict): user_input: str parsed_intent: str retrieved_faqs: List[str] retrieved_kb_docs: List[str] # 2. 创建子图 retrieval_subgraph SubStateGraph(RetrievalState) retrieval_subgraph.add_node(“get_faq”, lambda s: {“retrieved_faqs”: ToolSet.retrieve_faq(s[“user_input”], s[“parsed_intent”])}) retrieval_subgraph.add_node(“get_kb”, lambda s: {“retrieved_kb_docs”: ToolSet.retrieve_knowledge_base(s[“user_input”])}) # 假设子图内是并行检索 retrieval_subgraph.add_edge(“get_faq”, END) retrieval_subgraph.add_edge(“get_kb”, END) retrieval_subgraph.set_entry_point(“get_faq”) # 可以设置多个入口这里简化 compiled_retrieval retrieval_subgraph.compile() # 3. 在主图中将子图作为一个节点添加 def retrieval_node(state: GraphState): # 准备子图输入 sub_input { “user_input”: state[“user_input”], “parsed_intent”: state[“parsed_intent”], “retrieved_faqs”: [], “retrieved_kb_docs”: [] } # 运行子图 sub_result compiled_retrieval.invoke(sub_input) # 将子图结果映射回主状态 return { “retrieved_faqs”: sub_result[“retrieved_faqs”], “retrieved_kb_docs”: sub_result[“retrieved_kb_docs”] } main_workflow.add_node(“retrieval”, retrieval_node) # 用这个节点替代原来的 retrieve_faq 和 retrieve_knowledge_base 节点5. 调试、监控与性能优化一个复杂的工作流上线后可观测性至关重要。你需要知道每个请求走了哪条路径、在每个节点耗时多少、状态如何变化。5.1 利用stream_mode进行调试在开发时使用app.stream(..., stream_mode“values”)可以逐步获取每个节点执行后的完整状态方便你跟踪状态变化。使用stream_mode“updates”则只获取状态中发生变化的字段更轻量。for step in app.stream(initial_state, stream_mode“values”): print(json.dumps(step, indent2, ensure_asciiFalse)) # 打印完整状态 # 或者只关注特定字段 if “generated_response” in step: print(f“生成的答案: {step[‘generated_response’][:100]}...”)5.2 集成日志与追踪LangGraph 与 LangSmith 深度集成。设置环境变量LANGCHAIN_TRACING_V2true和LANGCHAIN_API_KEY后工作流的每一次运行都会被自动记录到 LangSmith。你可以在其界面上清晰地看到整个有向无环图DAG的执行轨迹、每个节点的输入输出、耗时和任何错误这是调试复杂流程的神器。对于自定义日志可以在每个节点函数内部加入详细的logging。import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) def some_node(state): logger.info(f“进入节点 some_node当前意图: {state.get(‘parsed_intent’)}”) # ... 业务逻辑 ... logger.info(f“离开节点 some_node更新字段: {list(result.keys())}”) return result5.3 性能优化要点缓存LLM调用对于确定性高的节点如意图分类可以使用langchain.cache如InMemoryCache,SQLiteCache来缓存LLM的响应避免重复计算。异步节点如果节点涉及网络I/O如API调用、数据库查询将其定义为异步函数async def并使用langgraph的异步接口astream来运行可以显著提升吞吐量。精简状态避免在状态中存储过大的对象如完整的文档内容。只存储必要的引用或摘要。条件边优化条件边的判断函数应尽可能简单快速。避免在条件函数中进行昂贵的计算或IO操作。超时与重试对于可能失败的外部服务调用在节点函数内部实现重试逻辑或超时机制避免单个节点卡住整个工作流。6. 常见问题与避坑指南在实际使用 LangGraph 搭建工作流时我踩过不少坑这里总结几个最常见的问题一状态更新不生效或覆盖现象节点返回了更新字典但后续节点读取的状态值没变。原因最常见的是忘记了状态字段需要使用Annotated和归约器。对于列表、字典等可变类型如果没有归约器后一个节点的更新会直接覆盖前一个节点的更新而不是合并。解决仔细设计状态结构。对于需要追加的列表使用Annotated[List, add_messages]或自定义归约函数。对于数值累加使用operator.add。问题二条件边路由错误或进入死循环现象工作流没有按预期分支或者在两个节点间无限循环。原因条件函数的返回值与为边配置的目标节点名称不匹配。或者图的逻辑设计存在循环出口缺失。解决打印条件函数的返回值确保它是你定义的边映射中的一个键。在图中务必为所有可能的分支设置终点END。可以使用一个“兜底”条件边指向END或一个错误处理节点。使用app.get_graph().print_ascii()可视化你的图检查连接关系。问题三节点函数过于臃肿现象单个节点函数做了太多事情代码难以维护和测试。解决严格遵守“单一职责原则”。如果一个节点函数超过50行或者做了两件以上的事如“检索并加工数据”考虑将其拆分成多个节点或者封装成子图。节点应该像乐高积木小而专。问题四错误处理缺失现象工作流中某个节点抛出异常如API调用失败导致整个流程崩溃。解决在每个可能出错的节点内部进行try-catch并将错误信息作为状态的一部分传递下去引导流程进入一个“错误处理”或“降级”节点。LangGraph 本身也支持在编译时设置retry_strategy。问题五对“流”模式理解不透现象想实现“流式”输出即LLM生成一个字就返回一个字但发现app.stream返回的是节点级别的状态快照。解决LangGraph 的stream是节点流的流不是Token流。要实现Token流式输出需要在生成答案的节点内部使用支持流式响应的LLM如ChatOpenAI(streamingTrue)并在该节点中通过yield或其他回调机制将Token逐块输出。这需要将工作流与你的服务框架如FastAPI的流式响应机制结合。问题六忘记配置检查点Checkpointer现象无法实现多轮对话的记忆或者无法使用中断功能。解决对于需要维持会话状态的应用在编译图时务必传入一个checkpointer参数如MemorySaver()或SqliteSaver(“checkpoints.db”)。并在调用invoke/stream时通过configurable参数指定唯一的thread_id来关联同一会话。搭建 LangGraph 工作流就像绘制一张精密的电路图。初期需要花时间理解状态流和路由的概念但一旦掌握其带来的清晰度、可维护性和灵活性是传统线性代码无法比拟的。从这12步的骨架开始你可以逐步扩展出处理成百上千个节点的复杂AI智能体系统而代码结构依然清晰可控。