1. 从LangChain到LangGraph为什么我们需要“图”来编排AI应用如果你和我一样是从LangChain开始接触大语言模型应用开发的那么你肯定对“链”Chain这个概念不陌生。LangChain通过将各种工具、提示词、模型和记忆模块像乐高积木一样串联起来极大地简化了早期AI应用的构建流程。我最初用它来搭建一些简单的问答机器人或者文档总结工具感觉非常顺手。但是当项目需求变得复杂不再是简单的“输入-处理-输出”单一路径时问题就来了。比如我想构建一个智能客服系统它需要根据用户问题的类型是咨询产品、投诉还是技术问题来决定调用哪个专业的工具链处理过程中可能还需要向用户追问更多细节或者根据历史对话记录调整回复策略。用传统的“链”来构建这种带有分支、循环和状态管理的应用代码很快就会变得像一团乱麻——各种if-else嵌套、回调函数满天飞状态在各个函数之间传来传去可读性和可维护性急剧下降。这感觉就像试图用一根直尺去画一幅复杂的电路图工具本身就成了瓶颈。这正是LangGraph诞生的背景。它不是要取代LangChain而是作为其生态系统内的一个更强大的补充专门用来解决复杂、有状态、多步骤的工作流编排问题。LangGraph的核心思想非常直观用“图”Graph来建模你的应用逻辑。在图中节点Node代表一个可执行的操作单元比如调用一次LLM、执行一个工具函数、做一个条件判断边Edge则定义了这些操作之间的流转路径。这让我们能够清晰地描述出“在什么情况下数据应该流向哪里”。从最近的热搜词也能看出大家的关注点langgraph和langchain的区别、langgraph 三要素、langgraph 原理。简单来说LangChain是工具箱和基础连接器而LangGraph是高级的流程设计器和执行引擎。当你需要处理需要记忆、循环、分支判断的复杂对话或任务时LangGraph就是那个更趁手的工具。所以这篇内容会聚焦在LangGraph最核心的基石——它的基础API上。我会带你从零开始理解构成LangGraph应用的几个基本要素State、Node和Edge并手把手教你如何用它们搭建第一个真正意义上的“图”应用。理解了这些你就能看懂官方文档并开始设计自己的复杂工作流了。2. 理解LangGraph的基石状态State与图Graph的定义在深入代码之前我们必须先建立两个核心的认知模型状态State和图Graph。这是理解LangGraph所有API的钥匙。2.1 状态State应用记忆的容器在LangGraph中State不是一个抽象概念它就是一个Python字典或者更准确地说是一个类似字典的对象。这个字典贯穿整个工作流的执行生命周期是所有节点共享的、唯一的数据存储中心。你可以把它想象成一个共享的白板每个步骤节点都可以在上面读取信息、写入新的信息或者修改已有的信息。为什么需要这样一个集中的状态回想一下用普通函数构建复杂流程的痛苦你需要在函数之间手动传递十几个参数或者维护一个全局变量既混乱又容易出错。LangGraph的State机制优雅地解决了这个问题。它强制你明确定义工作流需要关注哪些数据并且所有操作都围绕这个状态字典展开。定义一个状态通常意味着定义一个TypedDict类型化字典来明确状态中有哪些字段以及它们的类型。这能带来非常好的代码提示和类型检查的好处。例如一个简单的对话机器人状态可能长这样from typing import TypedDict, List, Annotated import operator class AgentState(TypedDict): # 用户的输入问题 input: str # 模型生成的回复 response: str # 对话历史记录 chat_history: List[str] # 一个计数器记录已经循环了几轮 loop_count: Annotated[int, operator.add]注意最后一个字段loop_count它使用了Annotated类型和operator.add。这是LangGraph的一个精妙设计用于声明如何更新这个字段。operator.add意味着当多个节点同时修改这个字段时它们的修改值会被相加。这是一种“归约器”Reducer用于解决并发修改冲突。常见的归约器还有operator.add: 求和用于计数器。list.append: 追加到列表用于历史记录。lambda old, new: new: 取最新值默认行为直接覆盖。这个设计让状态管理既灵活又安全。在后续的compiled_graph的invoke或stream方法中我们传入的初始参数就是对这个状态字典的第一次赋值。2.2 图Graph工作流的骨架有了状态容器我们需要一个骨架来组织操作流程这就是Graph。在LangGraph中你首先需要创建一个Graph对象。from langgraph.graph import StateGraph, END # 创建一个图并指定它要处理的状态类型 workflow StateGraph(AgentState)StateGraph是一个泛型类我们传入了之前定义的AgentState类型。这样后续所有对这个图的操作都会和这个状态类型关联起来提供类型安全。创建好的workflow对象目前还是一个空壳。它不知道有哪些步骤也不知道步骤之间如何连接。我们需要通过接下来的Node和Edge来为它赋予灵魂。这里先提一下END它是一个特殊的节点代表工作流的终点。当你把一条边指向END时就意味着工作流在此处可以合法结束。3. 核心API详解一节点Node与边Edge的构建现在我们来填充这个骨架的肌肉和神经——节点和边。3.1 节点Node执行具体工作的单元节点是图的基本执行单元。在LangGraph中一个节点就是一个普通的Python函数或可调用对象它必须接收一个参数即当前的状态字典并返回一个值这个值用于更新状态字典。关键规则节点函数返回的是一个字典这个字典的键必须是状态State中定义的字段名值则是要更新到该字段的新数据。LangGraph会自动将这个返回的字典与当前状态合并。让我们定义两个简单的节点def generate_response(state: AgentState) - dict: 节点A模拟LLM生成回复 # 从状态中获取用户输入和历史 user_input state[“input”] history state.get(“chat_history”, []) # 这里模拟一个简单的回复逻辑。真实场景中这里会调用LLM API。 # 比如可能会遇到热搜词里的错误api error: 400 this model‘s maximum context length is ... # 这就需要我们在节点函数里加入错误处理和上下文管理的逻辑。 simulated_response f“我已经收到你的消息{user_input}。这是我模拟的回复历史记录长度{len(history)}。” # 更新状态将新的回复存入response字段 # 同时我们也可以选择更新历史记录这里先不演示后面会用到归约器 return {“response”: simulated_response} def validate_input(state: AgentState) - dict: 节点B验证用户输入是否有效 user_input state[“input”] if not user_input or len(user_input.strip()) 2: # 如果输入无效我们更新一个标志位到状态中。 # 注意我们需要先在AgentState里定义这个字段这里为了演示假设我们定义了is_input_valid: bool # 实际上更好的做法是修改State定义或者使用动态字段需配置允许动态字段。 # 这里我们返回一个包含验证结果的字典。 return {“validation_passed”: False, “validation_message”: “输入太短或为空”} else: return {“validation_passed”: True, “validation_message”: “”}定义好函数后我们需要将它们“注册”到图中使其成为正式的节点# 将函数添加为节点并给节点起个名字 workflow.add_node(“generate”, generate_response) workflow.add_node(“validate”, validate_input)add_node方法接收两个参数节点名称字符串和节点函数。节点名称是后续连接边时使用的标识符。3.2 边Edge控制流程的方向边决定了执行完一个节点后接下来该去哪个节点。这是实现分支、循环的关键。LangGraph提供了几种添加边的方式。3.2.1 普通边固定流转这是最简单的情况无论节点执行结果如何都固定流向下一个节点。# 设置入口点工作流从哪个节点开始 workflow.set_entry_point(“validate”) # 添加一条从“validate”节点到“generate”节点的边 workflow.add_edge(“validate”, “generate”) # 添加一条从“generate”节点到终点的边 workflow.add_edge(“generate”, END)这样我们就定义了一个线性流程validate-generate-END。3.2.2 条件边Conditional Edge实现分支这是LangGraph最强大的特性之一。它允许根据当前状态的值动态决定下一步走向哪个节点。这通过一个“路由函数”来实现。假设我们在validate节点后需要根据验证结果决定是继续生成回复还是直接返回一个错误信息。我们可以先定义一个路由函数from langgraph.graph import StateGraph, END def route_after_validation(state: AgentState) - str: 根据验证结果路由到不同节点 if state.get(“validation_passed”, True): # 默认为True return “generate” # 验证通过去生成回复 else: return “__end__” # 验证失败直接结束。注意这里返回的是字符串“__end__”它等价于特殊的END节点。然后我们用add_conditional_edges方法来替换掉之前固定的add_edge# 移除之前的固定边 workflow.add_edge(“validate”, “generate”) # 添加条件边 workflow.add_conditional_edges( “validate”, # 源节点 route_after_validation, # 路由函数 # 指定路由函数可能返回的所有目标节点。 # 路由函数返回的字符串必须在这个映射里或者等于“__end__”。 {“generate”: “generate”, “__end__”: END} )现在流程就变成了从validate节点出来后执行route_after_validation函数该函数检查状态中的validation_passed字段如果为True则前往“generate”节点如果为False则直接结束工作流。关于__end__它是一个LangGraph预定义的字符串常量在条件边中用于表示结束。在add_conditional_edges的路径映射中我们需要将它映射到END对象。你也可以直接让路由函数返回END但使用“__end__”字符串是更常见的做法因为它更清晰地将逻辑控制路由函数和图的结构定义路径映射分开了。4. 编译与运行让图“活”起来定义好节点和边之后我们得到的workflow还是一个“蓝图”。需要将它编译Compile成一个可执行的对象。# 编译图 app workflow.compile()这个app对象通常我们叫它compiled_graph就是我们可以直接调用的“智能体”或“工作流”。它有两个最常用的方法invoke和stream。4.1 同步调用invokeinvoke方法接收一个初始状态字典同步执行整个工作流直到遇到END然后返回最终的状态。# 准备初始状态。必须包含State中定义的必需字段。 initial_state {“input”: “你好世界”, “chat_history”: []} # 执行工作流 final_state app.invoke(initial_state) print(final_state[“response”]) # 输出我已经收到你的消息你好世界。这是我模拟的回复历史记录长度0。如果我们的流程中有条件边并且验证失败了initial_state_invalid {“input”: “a”, “chat_history”: []} final_state_invalid app.invoke(initial_state_invalid) # 因为输入无效validate节点会将validation_passed设为False # 条件边会路由到__end__因此可能不会执行generate节点。 # final_state_invalid 中将包含validation_message但没有response。 print(final_state_invalid.get(“response”, “No response generated”)) print(final_state_invalid.get(“validation_message”))4.2 流式调用streamstream方法同样接收初始状态但它返回一个生成器Generator每执行完一个节点就yield一次结果。这对于需要实时展示执行过程、构建交互式应用如逐步显示的聊天界面非常有用。from langgraph.graph import MessagesState # 假设我们用了更常用的消息状态 # 为了演示stream我们用一个更简单的状态 class SimpleState(TypedDict): messages: Annotated[list, list.append] workflow2 StateGraph(SimpleState) def node1(state: SimpleState): return {“messages”: [“Hi from Node 1”]} def node2(state: SimpleState): return {“messages”: [“Hi from Node 2”]} workflow2.add_node(“n1”, node1) workflow2.add_node(“n2”, node2) workflow2.set_entry_point(“n1”) workflow2.add_edge(“n1”, “n2”) workflow2.add_edge(“n2”, END) app2 workflow2.compile() # 流式执行 initial_state {“messages”: []} for step in app2.stream(initial_state): node_name, output_state list(step.items())[0] # stream返回的是字典 {node_name: state} print(f“节点 [{node_name}] 执行完毕。当前消息: {output_state[‘messages’]}”) # 输出 # 节点 [n1] 执行完毕。当前消息: [‘Hi from Node 1’] # 节点 [n2] 执行完毕。当前消息: [‘Hi from Node 1’ ‘Hi from Node 2’]从stream的输出可以清晰地看到状态是如何随着每个节点的执行而逐步演变的。这对于调试复杂工作流至关重要。4.3 一个常见的陷阱与api error: 400的启示在热搜词中我们看到大量如api error: 400、maximum context length、insufficient balance等错误。这些错误不会由LangGraph本身抛出而是发生在你的节点函数内部比如调用DeepSeek、Claude等LLM API时。例如在generate_response节点中如果你直接调用client.chat.completions.create(...)而传入的对话历史太长就会触发maximum context length错误。LangGraph不会自动帮你处理这些错误。你必须在自己的节点函数中实现健壮的错误处理逻辑。def generate_response_safe(state: AgentState) - dict: try: # 调用LLM API response call_llm_api(state[“messages”]) return {“response”: response} except APIError as e: # 处理特定的API错误例如令牌不足、上下文过长 if “maximum context length” in str(e): # 策略截断最旧的历史消息或者返回一个要求用户简化问题的提示 truncated_messages truncate_messages(state[“messages”]) # 注意这里我们更新了messages需要重新调用或返回错误信息 return {“error”: “上下文过长已尝试简化”, “messages”: truncated_messages} elif “insufficient balance” in str(e): return {“error”: “API余额不足请检查账户”} else: # 其他未知错误 return {“error”: f“API调用失败{e}”} except Exception as e: # 处理其他异常 return {“error”: f“生成回复时发生未知错误{e}”}然后你可以在后续的节点或条件边中检查状态里是否有error字段从而决定是向用户展示错误信息还是进行重试等操作。这体现了用“图”编排的另一个优势错误处理可以变成一个显式的、可管理的流程节点而不是隐藏在try-catch块深处的难以维护的逻辑。5. 实战构建一个带循环和记忆的对话助手让我们综合运用以上知识构建一个稍微复杂点的例子一个能记住对话历史并且会持续追问直到获得满意答案的简单助手。第一步定义状态。这次我们需要更丰富的状态。from typing import TypedDict, List, Annotated, Literal import operator class ConversationState(TypedDict): # 用户的最新问题 latest_query: str # 完整的对话消息列表用于给LLM提供上下文 messages: Annotated[List[dict], list.append] # 使用list.append归约器 # LLM的回复 llm_response: str # 一个标志表示LLM是否认为需要进一步追问用户 needs_clarification: bool # 追问的问题如果需要的话 clarifying_question: str # 对话轮次计数 turn_count: Annotated[int, operator.add]第二步创建图和节点。from langgraph.graph import StateGraph, END workflow StateGraph(ConversationState) def process_query(state: ConversationState) - dict: 节点1处理用户查询准备给LLM的上下文 # 将用户的最新问题转换为AI消息格式并添加到历史中 # 注意messages字段使用了list.append归约器所以这里返回的列表会被追加到原有列表之后。 user_msg {“role”: “user”, “content”: state[“latest_query”]} # 我们返回它归约器会自动处理追加。 return {“messages”: [user_msg], “turn_count”: 1} # 每轮对话计数1 def call_llm(state: ConversationState) - dict: 节点2调用LLM并判断是否需要追问 # 模拟一个LLM调用。真实情况下这里会整合历史消息(messages)发送给API。 # 我们模拟一个简单的逻辑如果用户问题中包含“解释”LLM就假装需要更多信息。 history_text “ “.join([msg[“content”] for msg in state[“messages”][-5:]]) # 取最近5条 if “解释” in state[“latest_query”]: response_text “这个问题有点复杂。你能告诉我你具体对哪部分感兴趣吗” needs_clarify True clarify_q “请具体描述你需要解释的方面。” else: response_text f“根据你的问题‘{state[‘latest_query’]}’和近期对话我模拟了一个回答。对话历史摘要{history_text[-100:]}...” needs_clarify False clarify_q “” # 将AI的回复也添加到消息历史中 ai_msg {“role”: “assistant”, “content”: response_text} return { “llm_response”: response_text, “needs_clarification”: needs_clarify, “clarifying_question”: clarify_q, “messages”: [ai_msg] # 同样会被追加 } def ask_for_clarification(state: ConversationState) - dict: 节点3如果需要追问则生成追问提示 if state[“needs_clarification”]: # 在实际应用中这里可能会格式化一个更友好的问题。 # 我们直接使用LLM生成的追问问题。 question_to_ask state[“clarifying_question”] # 我们可以选择是否将这次追问也记录到messages中这里先不记录留给下一轮。 return {“response_to_user”: f“[助手需要澄清] {question_to_ask}”} else: # 如果不需要追问直接返回LLM的回复 return {“response_to_user”: state[“llm_response”]} def check_should_continue(state: ConversationState) - Literal[“ask”, “end”]: 路由函数判断是继续追问还是结束 if state[“needs_clarification”]: # 如果需要澄清我们进入一个“等待用户输入”的环节。 # 在真实图中这里可能会连接到一个等待外部输入的节点。 # 为了简化我们假设用户会立即提供澄清并循环回process_query。 # 但我们需要一个机制来避免无限循环。这里用turn_count简单限制。 if state.get(“turn_count”, 0) 5: # 最多进行5轮对话 return “wait_for_user_input” # 这是一个我们即将添加的“虚拟”节点 else: return “end” else: return “end” # 添加节点 workflow.add_node(“process”, process_query) workflow.add_node(“call_llm”, call_llm) workflow.add_node(“ask”, ask_for_clarification) # 添加一个“虚拟”节点模拟等待用户输入。在实际应用中这可能是一个中断点等待回调。 workflow.add_node(“wait_for_user_input”, lambda state: state) # 一个不改变状态的空节点 # 设置入口 workflow.set_entry_point(“process”) # 添加边 workflow.add_edge(“process”, “call_llm”) workflow.add_edge(“call_llm”, “ask”) # 从ask节点出来后根据条件决定下一步 workflow.add_conditional_edges( “ask”, check_should_continue, {“wait_for_user_input”: “wait_for_user_input”, “end”: END} ) # 如果路由到wait_for_user_input我们假设用户已经提供了新输入则循环回process节点。 # 这里需要一个机制来更新latest_query。在真实场景中这通常通过中断Interruption和外部更新状态实现。 # 为了演示循环我们假设wait_for_user_input节点之后自动将clarifying_question作为新的latest_query这显然不合理仅作演示。 def update_query_and_loop(state: ConversationState): # 模拟将追问的问题当作用户的新输入仅用于演示循环逻辑 new_query state[“clarifying_question”] or “请继续” return {“latest_query”: new_query, “needs_clarification”: False} # 重置标志 workflow.add_node(“prepare_next_round”, update_query_and_loop) workflow.add_edge(“wait_for_user_input”, “prepare_next_round”) workflow.add_edge(“prepare_next_round”, “process”) # 形成循环 # 编译图 app workflow.compile()第三步运行并观察循环。# 第一轮运行 initial_state {“latest_query”: “请解释一下量子计算”, “messages”: [], “turn_count”: 0} print(“ 第一轮 ) for step in app.stream(initial_state, subgraphsTrue): # subgraphsTrue 可以显示子图信息 node, state list(step.items())[0] print(f“节点 [{node}] - 最新回复: {state.get(‘response_to_user’ ‘N/A’)}”) if node “ask”: print(f“ 是否需要澄清: {state.get(‘needs_clarification’)}”) if state.get(“turn_count”, 0) 3: # 防止演示时无限循环手动打断 print(“演示手动中断循环”) break这个例子虽然简化了很多比如用模拟代替真实LLM调用循环逻辑比较生硬但它清晰地展示了LangGraph如何通过State、Node、Conditional Edge和循环来构建一个带有状态记忆和动态流程的对话助手。你可以看到turn_count在每个循环中自动累加messages历史被自动追加整个流程的控制逻辑都清晰地体现在图的结构中而不是隐藏在复杂的函数调用链里。当你需要修改逻辑时比如在超过3轮追问后自动给出一个总结而非继续追问你只需要修改check_should_continue路由函数和对应的边或者增加一个新的节点来处理“强制总结”的情况。这种模块化和声明式的编程方式对于维护和迭代复杂AI应用来说是至关重要的。