1. 从LangChain到LangGraph为什么我们需要一个新的范式如果你在过去一年里折腾过LLM应用开发那么“LangChain”这个名字对你来说一定不陌生。它像是一套乐高积木把大语言模型LLM、向量数据库、工具调用这些组件封装成标准化的“链”Chain让我们能快速拼装出一个可用的AI应用。但用久了尤其是当你试图构建一个需要多轮对话、状态保持、甚至能自主决策的复杂Agent时你可能会感到一丝掣肘。链式结构是线性的A做完做BB做完做C。但现实世界的问题尤其是需要与用户持续交互的智能体其逻辑往往是环状的、带分支的、甚至是可以循环的。这就是LangGraph诞生的背景。它不是要取代LangChain而是站在巨人的肩膀上提供了一个更强大的“编排”层。你可以把LangChain看作是提供了丰富的“零件”LLM调用、工具、记忆等而LangGraph则是一张“电路图”或“流程图”它定义了这些零件如何连接、在什么条件下触发、以及如何管理整个系统的“状态”。简单来说LangGraph的核心是让你用“图”Graph的思维来构建和运行基于LLM的、有状态的、可能循环的工作流。这直接对应了构建复杂Agent的核心需求记忆、规划、工具使用以及多步骤执行。最近“LangGraph”的热度持续攀升从“和LangChain的区别”到“长期记忆”、“子图”、“State详解”这些热搜词精准地反映了开发者们最关心的痛点如何超越简单的问答构建真正智能、可交互、能处理复杂任务的AI体。本文将从一个实践者的角度带你深入LangGraph的世界不仅告诉你它是什么更会通过具体的代码示例拆解其核心三要素并分享在集成与调试中那些官方文档不会写的“坑”。2. LangGraph核心三要素拆解State, Node, Edge理解LangGraph最关键的是吃透它的三个核心概念状态State、节点Node和边Edge。这构成了一个图工作流的基本骨架。2.1 状态State工作流的共享记忆与上下文在LangChain的链中信息通常是从一个组件“流”到下一个组件中间结果可能被丢弃或转换。而在LangGraph中State是一个贯穿整个图执行周期的、可变的共享数据容器。它定义了工作流中所有节点都能读取和写入的“全局变量”。State通常用一个Pydantic模型来定义这保证了类型安全和清晰的接口。例如我们要构建一个能分析用户需求并调用合适工具的助手其State可能包含from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages import operator class State(TypedDict): # 对话消息历史使用LangGraph提供的注解进行专用操作 messages: Annotated[List[dict], add_messages] # 用户当前输入的问题 user_query: str # 从用户问题中提取的关键信息如城市、日期 extracted_info: dict # 规划出的下一步动作如“调用天气工具” next_action: str # 工具调用的结果 tool_result: str # 最终给用户的回复 final_response: str这里的关键是Annotated的使用。add_messages是一个归约器Reducer这是LangGraph处理状态更新的一个精妙设计。当多个节点都可能向messages列表中添加消息时归约器定义了如何合并这些更新。add_messages会确保消息按顺序追加而不是覆盖。你也可以为其他字段定义自定义的归约器比如对于数字字段用operator.add来做累加。实操心得在设计State时务必保持精简和意图清晰。不要把所有可能用到的数据都塞进去。每个字段都应该有明确的职责并且要考虑其更新方式是覆盖、追加还是合并。过度复杂的State会让图的逻辑难以理解和调试。2.2 节点Node执行具体任务的函数节点是图中的工作单元它是一个普通的Python函数或可调用对象接收当前的State作为输入并返回一个包含对State所做更新的字典。def extract_user_intent(state: State) - dict: 节点函数分析用户意图提取关键信息 query state[“user_query”] # 这里可以调用一个LLM或规则引擎来解析query # 假设我们简单提取“天气”关键词和城市 extracted {“intent”: “unknown”, “city”: None} if “天气” in query: extracted[“intent”] “weather_query” # 简单演示实际应用需要用更复杂的NLP方法 if “北京” in query: extracted[“city”] “北京” elif “上海” in query: extracted[“city”] “上海” # 返回的字典中的键对应要更新的State字段 return {“extracted_info”: extracted}节点函数的返回值决定了State的哪些部分会被更新。LangGraph会将这些更新应用到共享的State上。节点可以执行任何操作调用LLM、查询数据库、运行计算、调用外部API等。2.3 边Edge控制流程的逻辑路由边决定了在某个节点执行完毕后接下来该执行哪个或哪些节点。这是LangGraph实现条件分支、循环等复杂逻辑的关键。边通常由一个路由函数Router来定义。有两种主要的边条件边Conditional Edge根据State中的值决定下一个节点。普通边直接指向下一个节点。LangGraph提供了一个简洁的方式来定义边使用langgraph.graph.StateGraph的add_conditional_edges方法。from langgraph.graph import StateGraph, END # 创建图 workflow StateGraph(State) # 先添加节点假设我们已经定义了多个节点函数 workflow.add_node(“extract_intent”, extract_user_intent) workflow.add_node(“plan_action”, plan_next_action) workflow.add_node(“call_tool”, call_weather_tool) workflow.add_node(“generate_response”, generate_final_response) # 设置入口点 workflow.set_entry_point(“extract_intent”) # 添加普通边提取意图后总是进入规划节点 workflow.add_edge(“extract_intent”, “plan_action”) # 添加条件边根据规划的结果决定下一步 def decide_next_step(state: State) - str: 路由函数根据‘next_action’字段决定下一步 action state.get(“next_action”) if action “need_weather”: return “call_tool” # 去调用工具节点 elif action “can_answer”: return “generate_response” # 直接生成回复 else: return “END” # 结束图执行 workflow.add_conditional_edges( “plan_action”, # 从哪个节点出发 decide_next_step, # 路由函数 {“call_tool”: “call_tool”, “generate_response”: “generate_response”, “END”: END} # 可能的目的地映射 ) # 添加更多边... workflow.add_edge(“call_tool”, “generate_response”) workflow.add_edge(“generate_response”, END)在这个例子中plan_action节点执行后会根据它写入State的next_action值由decide_next_step函数决定下一步是去调用工具还是直接生成回复或者结束。这种声明式的流程控制比用一堆if-else语句写在节点函数里要清晰和可维护得多。3. 构建你的第一个LangGraph智能体一个天气查询助手理论说得再多不如动手跑一遍。让我们构建一个简单的天气查询助手它会理解用户意图调用模拟的天气API并生成回复。这个例子将串联起State、Node和Edge。3.1 定义State与工具首先我们定义工作流的状态。为了简化我们聚焦核心字段。from typing import TypedDict, List, Annotated, Optional from langgraph.graph.message import add_messages class AssistantState(TypedDict): 智能体的状态定义 messages: Annotated[List[dict], add_messages] # 对话历史 user_input: str # 最新用户输入 detected_intent: Optional[str] # 检测到的意图如“weather”, “greeting” city: Optional[str] # 提取的城市名 tool_called: Optional[str] # 被调用的工具名 tool_result: Optional[str] # 工具调用结果 final_output: Optional[str] # 最终输出接下来我们定义一个模拟的天气工具。在LangGraph/LangChain生态中工具通常用tool装饰器来声明。from langchain.tools import tool tool def get_weather(city: str) - str: 根据城市名获取天气信息。 # 模拟一个API调用 weather_data { “北京”: “晴15-25°C微风”, “上海”: “多云18-28°C东南风3级”, “广州”: “阵雨22-30°C” } return weather_data.get(city, f“未找到{city}的天气信息”)3.2 实现各个节点函数我们将工作流分解为四个节点。节点1意图识别节点这个节点负责分析用户的最新输入判断其意图并提取关键实体如城市。def intent_recognition_node(state: AssistantState) - dict: 节点1识别用户意图 query state[“user_input”] # 在实际应用中这里应该调用一个LLM如通过ChatPromptTemplate和LLMChain # 为了演示我们使用简单的规则 intent None city None if any(word in query for word in [“天气”, “气温”, “下雨”]): intent “weather” # 非常简单的城市提取实际项目务必使用更健壮的方法如NER或LLM提取 for c in [“北京”, “上海”, “广州”]: if c in query: city c break elif any(word in query for word in [“你好”, “嗨”, “早上好”]): intent “greeting” else: intent “unknown” # 更新State return { “detected_intent”: intent, “city”: city }节点2规划与路由节点这个节点根据识别出的意图决定下一步该做什么。它不执行具体操作只做决策。def planning_node(state: AssistantState) - dict: 节点2规划下一步动作 intent state[“detected_intent”] next_step None tool_to_use None if intent “weather”: if state[“city”]: next_step “call_tool” tool_to_use “get_weather” else: # 如果意图是天气但没提供城市我们需要追问 next_step “ask_for_city” elif intent “greeting”: next_step “generate_response” # 直接生成问候回复 else: next_step “handle_unknown” # 处理未知意图 return { “next_step”: next_step, “tool_to_call”: tool_to_use }节点3工具调用节点这个节点负责执行具体的工具调用。def tool_call_node(state: AssistantState) - dict: 节点3调用工具 tool_name state[“tool_to_call”] result None if tool_name “get_weather”: # 调用我们之前定义的天气工具 result get_weather.invoke({“city”: state[“city”]}) # 注意get_weather.invoke()返回的是LangChain Tool的调用结果 return { “tool_called”: tool_name, “tool_result”: result }节点4响应生成节点这个节点综合所有信息生成最终返回给用户的自然语言回复。def response_generation_node(state: AssistantState) - dict: 节点4生成最终回复 intent state[“detected_intent”] output “” if intent “weather”: if state[“tool_result”]: output f“{state[‘city’]}的天气是{state[‘tool_result’]}” else: output “请问您想查询哪个城市的天气呢” elif intent “greeting”: output “你好我是天气助手有什么可以帮您” else: output “抱歉我没太明白您的意思。您可以问我关于天气的问题。” # 同时我们将本次交互的完整信息追加到消息历史中可选用于长期记忆 new_message {“role”: “assistant”, “content”: output} # 注意由于messages字段使用了add_messages归约器我们这样返回即可追加 return { “final_output”: output, “messages”: new_message # 这会被add_messages归约器处理追加到列表 }我们还需要补充两个简单的节点来处理特殊分支def ask_city_node(state: AssistantState) - dict: 追问城市节点 return {“final_output”: “您想查询哪个城市的天气呢”} def handle_unknown_node(state: AssistantState) - dict: 处理未知意图节点 return {“final_output”: “我目前主要擅长回答天气相关问题您可以试试问我‘北京天气怎么样’。”}3.3 组装图并运行现在我们将所有节点和边组装起来。from langgraph.graph import StateGraph, END # 1. 创建图并指定State类型 workflow StateGraph(AssistantState) # 2. 添加所有节点 workflow.add_node(“recognize_intent”, intent_recognition_node) workflow.add_node(“plan”, planning_node) workflow.add_node(“call_tool”, tool_call_node) workflow.add_node(“generate_resp”, response_generation_node) workflow.add_node(“ask_city”, ask_city_node) workflow.add_node(“handle_unknown”, handle_unknown_node) # 3. 设置入口点 workflow.set_entry_point(“recognize_intent”) # 4. 添加边和条件边 # 意图识别后总是进入规划节点 workflow.add_edge(“recognize_intent”, “plan”) # 规划节点后的条件路由 def router_after_plan(state: AssistantState) - str: next_step state.get(“next_step”) # 返回的值必须与add_conditional_edges中映射的键一致 return next_step workflow.add_conditional_edges( “plan”, router_after_plan, { “call_tool”: “call_tool”, “ask_for_city”: “ask_city”, “generate_response”: “generate_resp”, “handle_unknown”: “handle_unknown”, # 理论上这里也应该能处理END但我们的router目前不返回END } ) # 工具调用后进入生成响应节点 workflow.add_edge(“call_tool”, “generate_resp”) # 追问城市和处理未知意图后也进入生成响应节点或者可以直接结束这里我们让它生成回复 workflow.add_edge(“ask_city”, “generate_resp”) workflow.add_edge(“handle_unknown”, “generate_resp”) # 生成响应后工作流结束 workflow.add_edge(“generate_resp”, END) # 5. 编译图 app workflow.compile()现在我们可以运行这个智能体了。# 准备初始状态 initial_state: AssistantState { “messages”: [], # 初始对话历史为空 “user_input”: “北京今天天气怎么样”, “detected_intent”: None, “city”: None, “tool_called”: None, “tool_result”: None, “final_output”: None, } # 运行图 final_state app.invoke(initial_state) print(“最终回复”, final_state[“final_output”]) print(“完整状态”, final_state)执行上述代码你应该会看到输出类似于“最终回复北京的天气是晴15-25°C微风”。整个State对象里包含了每一步的执行结果。踩坑实录在定义条件边时路由函数router_after_plan返回的字符串必须与add_conditional_edges方法中path_map参数字典的键完全匹配。如果返回了一个字典中不存在的值LangGraph会抛出一个KeyError。这是初期调试时最常见的错误之一。一个好的实践是在路由函数里做好默认值处理比如return state.get(“next_step”, “handle_unknown”)。4. 进阶特性子图、长期记忆与流式输出掌握了基础构建后我们来看看那些让LangGraph真正强大的进阶特性这些也正是网络热搜词里大家关心的。4.1 子图Subgraph管理复杂性的利器当你的智能体逻辑变得非常复杂时把所有节点和边都放在一个主图里会难以维护。子图允许你将一部分功能模块化封装成一个独立的、可复用的图。例如我们可以把“意图识别”这个本身可能就很复杂的过程涉及LLM调用、实体提取、意图分类封装成一个子图。from langgraph.graph import StateGraph # 定义一个专用于意图识别的State class IntentState(TypedDict): query: str intent: Optional[str] entities: dict # 创建意图识别子图 intent_subgraph_builder StateGraph(IntentState) def llm_classify_node(state: IntentState): # 模拟LLM调用进行复杂分类 return {“intent”: “weather”, “entities”: {“city”: “北京”}} intent_subgraph_builder.add_node(“classify”, llm_classify_node) intent_subgraph_builder.set_entry_point(“classify”) intent_subgraph_builder.add_edge(“classify”, END) intent_subgraph intent_subgraph_builder.compile() # 在主图中我们可以像调用一个节点一样调用这个子图 def intent_node_with_subgraph(state: AssistantState): # 准备子图输入 subgraph_input IntentState(querystate[“user_input”], intentNone, entities{}) # 运行子图 subgraph_result intent_subgraph.invoke(subgraph_input) # 将子图结果映射回主State return { “detected_intent”: subgraph_result[“intent”], “city”: subgraph_result[“entities”].get(“city”) }然后在主图中add_node(“recognize_intent”, intent_node_with_subgraph)即可。子图让代码结构更清晰也便于团队协作和单元测试。4.2 长期记忆Long-term Memory的实现模式“长期记忆”是构建真正个性化Agent的关键。LangGraph本身不提供开箱即用的记忆存储但它通过State和**检查点Checkpointing**机制为实现记忆提供了完美的基础。长期记忆通常涉及两个层面对话记忆Conversation Memory保存在State的messages字段中使用add_messages归约器自动管理。但这仅限于单次图执行的生命周期。持久化记忆Persistent Memory需要将重要的State信息如用户偏好、历史摘要、事实知识保存到外部数据库如SQLite、PostgreSQL、Redis并在下次对话时加载。一个常见的模式是使用图编译时的checkpointer参数和**config配置**。from langgraph.checkpoint.sqlite import SqliteSaver import sqlite3 # 1. 创建一个SQLite检查点存储器 conn sqlite3.connect(“:memory:”) # 实际应用请用文件路径 checkpointer SqliteSaver(conn) # 2. 在编译图时传入checkpointer app_with_memory workflow.compile(checkpointercheckpointer) # 3. 运行时通过config指定线程ID通常用用户ID或会话ID config {“configurable”: {“thread_id”: “user_12345”}} initial_state {“user_input”: “你好” …} # 第一次调用会创建或加载这个thread_id对应的检查点 result1 app_with_memory.invoke(initial_state, configconfig) # 此时State包括messages会被自动持久化到SQLite # 第二次调用传入相同的configLangGraph会从检查点恢复上一次的State new_state_input {“user_input”: “我还想问问上海的天气” …} # 注意这里传入的初始状态会被合并到恢复的State中通常我们只传增量信息 result2 app_with_memory.invoke({“user_input”: “我还想问问上海的天气”}, configconfig) # 在result2的State里messages会包含上一次的对话历史通过这种方式智能体就拥有了跨越多次调用的“长期记忆”。你可以定制checkpointer来使用其他存储后端也可以选择只在特定节点执行后保存检查点通过checkpoint参数实现更精细的控制。4.3 流式输出Streaming与执行控制对于需要实时反馈的交互式应用流式输出至关重要。LangGraph的stream方法允许你逐步获取每个节点执行后的状态更新而不是等待整个图执行完毕。# 使用stream方法进行流式调用 inputs {“user_input”: “北京和上海的天气”} config {“configurable”: {“thread_id”: “stream_test”}} for event in app_with_memory.stream(inputs, configconfig, stream_mode“values”): # event 是一个元组 (node_name, state_update) node, state_update event print(f“节点 [{node}] 执行完毕”) if “final_output” in state_update and state_update[“final_output”]: print(“生成部分回复:”, state_update[“final_output”]) # 你可以在这里将state_update[“final_output”]发送给前端实现打字机效果关于热搜词中提到的“compiledstategraph.stream()如何终止”这是一个更高级的话题。流的终止通常由图的逻辑决定执行到END节点。如果你想从外部中断一个正在进行的流式执行这通常需要在异步上下文中处理通过取消任务如asyncio.Task.cancel()来实现而不是通过LangGraph的API直接终止。在设计图时可以考虑添加一个“取消”或“超时”节点通过条件边在某些情况下提前路由到END。5. LangGraph与LangChain、FastAPI及Dify的集成实践在实际项目中LangGraph很少单独使用它需要与Web框架、前端界面以及其他AI编排工具集成。5.1 与LangChain的深度融合LangGraph和LangChain是绝配。你的节点函数可以轻松使用LangChain的任何组件from langchain.prompts import ChatPromptTemplate from langchain.chat_models import ChatOpenAI # 假设使用OpenAI from langchain.schema.runnable import RunnablePassthrough # 在节点函数中使用LangChain Chain def llm_based_node(state: State): prompt ChatPromptTemplate.from_template(“你是一个助手。用户说{query}。请分析意图。”) model ChatOpenAI(model“gpt-3.5-turbo”) chain prompt | model # 运行Chain ai_message chain.invoke({“query”: state[“user_input”]}) # 处理结果并更新State... return {“intent”: “analyzed_by_llm”}你可以把LangChain Chain当作LangGraph Node的一个强大“执行引擎”。LangGraph负责流程和状态LangChain负责与LLM、工具、检索器等具体组件的交互。5.2 基于FastAPI构建LangGraph服务端将LangGraph智能体封装成API服务是常见的需求。FastAPI是一个高性能的现代框架。from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional app FastAPI(title“LangGraph智能体API”) # 定义请求/响应模型 class ChatRequest(BaseModel): message: str user_id: str # 用于区分不同用户的记忆 thread_id: Optional[str] None # 可选的特定会话线程 class ChatResponse(BaseModel): response: str thread_id: str status: str # 假设我们已经编译好了一个带检查点的图应用 ‘agent_app’ # app.post(“/chat”, response_modelChatResponse) async def chat_endpoint(request: ChatRequest): try: # 准备配置使用user_id或提供的thread_id thread_id request.thread_id or f“user_{request.user_id}” config {“configurable”: {“thread_id”: thread_id}} # 调用图。输入是增量消息检查点会加载历史状态。 # 注意这里我们只传入新的用户消息其他状态由检查点恢复。 inputs {“user_input”: request.message} result agent_app.invoke(inputs, configconfig) response_text result.get(“final_output”, “抱歉我没有生成回复。”) return ChatResponse( responseresponse_text, thread_idthread_id, status“success” ) except Exception as e: raise HTTPException(status_code500, detailf“智能体处理失败: {str(e)}”) # 可以添加其他端点如重置记忆 /reset?thread_idxxx这样前端应用就可以通过发送HTTP请求与拥有长期记忆的LangGraph智能体进行对话了。5.3 与Dify等平台的结合思考热搜词中出现了“Dify和LangGraph可以一起用吗”。Dify是一个低代码的LLM应用开发平台它提供了可视化的编排界面。从架构上看Dify更侧重于前端交互、知识库管理、插件市场和应用部署提供了一个开箱即用的全栈解决方案。LangGraph是一个纯后端的、代码优先的、用于构建复杂有状态工作流的Python框架。它们完全可以协同工作。一种典型的模式是使用Dify作为前端界面和基础设施如API网关、知识库检索而将最核心、最复杂的智能体逻辑用LangGraph实现并作为Dify的一个“自定义工具”或“API节点”来调用。Dify的工作流节点可以调用你部署的LangGraph智能体API并将返回结果集成到其对话流中。这样你既享受了Dify的便捷性又拥有了LangGraph带来的强大、灵活的编排能力。6. 调试、监控与性能优化实战指南开发复杂的图应用调试是一大挑战。以下是一些实战中总结出的经验。6.1 可视化与调试LangGraph提供了内置的可视化功能对于理解流程至关重要。# 方法1生成Mermaid图表字符串注意输出中禁止使用Mermaid代码块但你可以将其复制到支持Mermaid的编辑器中查看 graph_mermaid app.get_graph().draw_mermaid() print(graph_mermaid) # 会输出一个Mermaid格式的字符串 # 方法2打印图形结构 print(app.get_graph().print_ascii()) # 在控制台打印ASCII艺术图对于调试最有效的方法是在节点函数中增加详细的日志并观察State的变化。你可以使用Python的logging模块或者简单地在节点函数开始和结束时打印关键信息。def some_node(state: State): print(f“[some_node] 输入State: {state}”) # ... 业务逻辑 ... result {“key”: “value”} print(f“[some_node] 输出更新: {result}”) return result6.2 状态State管理的常见陷阱归约器冲突如果你为同一个State字段定义了多个归约器或者归约器的行为不符合预期会导致状态更新混乱。务必理解每个归约器的作用如add_messages是追加operator.setitem是覆盖。状态污染一个节点错误地修改了其他节点依赖的字段。设计State时要做到高内聚、低耦合明确每个节点的输入输出字段。使用Pydantic模型能帮助在开发早期发现类型错误。检查点膨胀如果每次调用都保存完整的State特别是包含长消息历史存储会快速增长。可以考虑定期对消息历史进行摘要Summarization只保存摘要和最近几条原始消息将摘要存入State从而压缩记忆。6.3 性能优化要点异步节点如果节点涉及网络I/O如调用LLM API、查询数据库应将其定义为异步函数async def并在图中使用异步调用ainvoke,astream可以显著提升并发性能。async def async_tool_call_node(state: State): result await some_async_api(state[“query”]) return {“result”: result}条件边优化条件边的路由函数应尽可能简单、快速。避免在路由函数中执行LLM调用等重型操作。路由决策应基于State中已有的、由前置节点计算好的结果。子图复用对于被频繁调用的复杂子图确保其编译后的对象compiledgraph被复用而不是每次调用都重新编译。LLM调用批处理如果多个节点都需要调用LLM可以考虑是否能够合并这些请求或者使用LangChain的批量调用功能来减少API往返次数。构建基于LangGraph的智能体是一个迭代过程。从简单的线性流程开始逐步引入条件逻辑、循环、子图和记忆。充分利用可视化工具来理解你的图结构用细致的日志来跟踪状态流变。当你习惯了用“图”的思维来设计AI应用时你会发现构建那些能够进行多轮决策、拥有记忆和规划能力的智能体不再是一件令人望而生畏的事情。它就像绘制一张清晰的地图让智能体在这张地图上自主而可靠地走向目标。
LangGraph核心三要素与智能体构建实战：从状态管理到复杂工作流编排
1. 从LangChain到LangGraph为什么我们需要一个新的范式如果你在过去一年里折腾过LLM应用开发那么“LangChain”这个名字对你来说一定不陌生。它像是一套乐高积木把大语言模型LLM、向量数据库、工具调用这些组件封装成标准化的“链”Chain让我们能快速拼装出一个可用的AI应用。但用久了尤其是当你试图构建一个需要多轮对话、状态保持、甚至能自主决策的复杂Agent时你可能会感到一丝掣肘。链式结构是线性的A做完做BB做完做C。但现实世界的问题尤其是需要与用户持续交互的智能体其逻辑往往是环状的、带分支的、甚至是可以循环的。这就是LangGraph诞生的背景。它不是要取代LangChain而是站在巨人的肩膀上提供了一个更强大的“编排”层。你可以把LangChain看作是提供了丰富的“零件”LLM调用、工具、记忆等而LangGraph则是一张“电路图”或“流程图”它定义了这些零件如何连接、在什么条件下触发、以及如何管理整个系统的“状态”。简单来说LangGraph的核心是让你用“图”Graph的思维来构建和运行基于LLM的、有状态的、可能循环的工作流。这直接对应了构建复杂Agent的核心需求记忆、规划、工具使用以及多步骤执行。最近“LangGraph”的热度持续攀升从“和LangChain的区别”到“长期记忆”、“子图”、“State详解”这些热搜词精准地反映了开发者们最关心的痛点如何超越简单的问答构建真正智能、可交互、能处理复杂任务的AI体。本文将从一个实践者的角度带你深入LangGraph的世界不仅告诉你它是什么更会通过具体的代码示例拆解其核心三要素并分享在集成与调试中那些官方文档不会写的“坑”。2. LangGraph核心三要素拆解State, Node, Edge理解LangGraph最关键的是吃透它的三个核心概念状态State、节点Node和边Edge。这构成了一个图工作流的基本骨架。2.1 状态State工作流的共享记忆与上下文在LangChain的链中信息通常是从一个组件“流”到下一个组件中间结果可能被丢弃或转换。而在LangGraph中State是一个贯穿整个图执行周期的、可变的共享数据容器。它定义了工作流中所有节点都能读取和写入的“全局变量”。State通常用一个Pydantic模型来定义这保证了类型安全和清晰的接口。例如我们要构建一个能分析用户需求并调用合适工具的助手其State可能包含from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages import operator class State(TypedDict): # 对话消息历史使用LangGraph提供的注解进行专用操作 messages: Annotated[List[dict], add_messages] # 用户当前输入的问题 user_query: str # 从用户问题中提取的关键信息如城市、日期 extracted_info: dict # 规划出的下一步动作如“调用天气工具” next_action: str # 工具调用的结果 tool_result: str # 最终给用户的回复 final_response: str这里的关键是Annotated的使用。add_messages是一个归约器Reducer这是LangGraph处理状态更新的一个精妙设计。当多个节点都可能向messages列表中添加消息时归约器定义了如何合并这些更新。add_messages会确保消息按顺序追加而不是覆盖。你也可以为其他字段定义自定义的归约器比如对于数字字段用operator.add来做累加。实操心得在设计State时务必保持精简和意图清晰。不要把所有可能用到的数据都塞进去。每个字段都应该有明确的职责并且要考虑其更新方式是覆盖、追加还是合并。过度复杂的State会让图的逻辑难以理解和调试。2.2 节点Node执行具体任务的函数节点是图中的工作单元它是一个普通的Python函数或可调用对象接收当前的State作为输入并返回一个包含对State所做更新的字典。def extract_user_intent(state: State) - dict: 节点函数分析用户意图提取关键信息 query state[“user_query”] # 这里可以调用一个LLM或规则引擎来解析query # 假设我们简单提取“天气”关键词和城市 extracted {“intent”: “unknown”, “city”: None} if “天气” in query: extracted[“intent”] “weather_query” # 简单演示实际应用需要用更复杂的NLP方法 if “北京” in query: extracted[“city”] “北京” elif “上海” in query: extracted[“city”] “上海” # 返回的字典中的键对应要更新的State字段 return {“extracted_info”: extracted}节点函数的返回值决定了State的哪些部分会被更新。LangGraph会将这些更新应用到共享的State上。节点可以执行任何操作调用LLM、查询数据库、运行计算、调用外部API等。2.3 边Edge控制流程的逻辑路由边决定了在某个节点执行完毕后接下来该执行哪个或哪些节点。这是LangGraph实现条件分支、循环等复杂逻辑的关键。边通常由一个路由函数Router来定义。有两种主要的边条件边Conditional Edge根据State中的值决定下一个节点。普通边直接指向下一个节点。LangGraph提供了一个简洁的方式来定义边使用langgraph.graph.StateGraph的add_conditional_edges方法。from langgraph.graph import StateGraph, END # 创建图 workflow StateGraph(State) # 先添加节点假设我们已经定义了多个节点函数 workflow.add_node(“extract_intent”, extract_user_intent) workflow.add_node(“plan_action”, plan_next_action) workflow.add_node(“call_tool”, call_weather_tool) workflow.add_node(“generate_response”, generate_final_response) # 设置入口点 workflow.set_entry_point(“extract_intent”) # 添加普通边提取意图后总是进入规划节点 workflow.add_edge(“extract_intent”, “plan_action”) # 添加条件边根据规划的结果决定下一步 def decide_next_step(state: State) - str: 路由函数根据‘next_action’字段决定下一步 action state.get(“next_action”) if action “need_weather”: return “call_tool” # 去调用工具节点 elif action “can_answer”: return “generate_response” # 直接生成回复 else: return “END” # 结束图执行 workflow.add_conditional_edges( “plan_action”, # 从哪个节点出发 decide_next_step, # 路由函数 {“call_tool”: “call_tool”, “generate_response”: “generate_response”, “END”: END} # 可能的目的地映射 ) # 添加更多边... workflow.add_edge(“call_tool”, “generate_response”) workflow.add_edge(“generate_response”, END)在这个例子中plan_action节点执行后会根据它写入State的next_action值由decide_next_step函数决定下一步是去调用工具还是直接生成回复或者结束。这种声明式的流程控制比用一堆if-else语句写在节点函数里要清晰和可维护得多。3. 构建你的第一个LangGraph智能体一个天气查询助手理论说得再多不如动手跑一遍。让我们构建一个简单的天气查询助手它会理解用户意图调用模拟的天气API并生成回复。这个例子将串联起State、Node和Edge。3.1 定义State与工具首先我们定义工作流的状态。为了简化我们聚焦核心字段。from typing import TypedDict, List, Annotated, Optional from langgraph.graph.message import add_messages class AssistantState(TypedDict): 智能体的状态定义 messages: Annotated[List[dict], add_messages] # 对话历史 user_input: str # 最新用户输入 detected_intent: Optional[str] # 检测到的意图如“weather”, “greeting” city: Optional[str] # 提取的城市名 tool_called: Optional[str] # 被调用的工具名 tool_result: Optional[str] # 工具调用结果 final_output: Optional[str] # 最终输出接下来我们定义一个模拟的天气工具。在LangGraph/LangChain生态中工具通常用tool装饰器来声明。from langchain.tools import tool tool def get_weather(city: str) - str: 根据城市名获取天气信息。 # 模拟一个API调用 weather_data { “北京”: “晴15-25°C微风”, “上海”: “多云18-28°C东南风3级”, “广州”: “阵雨22-30°C” } return weather_data.get(city, f“未找到{city}的天气信息”)3.2 实现各个节点函数我们将工作流分解为四个节点。节点1意图识别节点这个节点负责分析用户的最新输入判断其意图并提取关键实体如城市。def intent_recognition_node(state: AssistantState) - dict: 节点1识别用户意图 query state[“user_input”] # 在实际应用中这里应该调用一个LLM如通过ChatPromptTemplate和LLMChain # 为了演示我们使用简单的规则 intent None city None if any(word in query for word in [“天气”, “气温”, “下雨”]): intent “weather” # 非常简单的城市提取实际项目务必使用更健壮的方法如NER或LLM提取 for c in [“北京”, “上海”, “广州”]: if c in query: city c break elif any(word in query for word in [“你好”, “嗨”, “早上好”]): intent “greeting” else: intent “unknown” # 更新State return { “detected_intent”: intent, “city”: city }节点2规划与路由节点这个节点根据识别出的意图决定下一步该做什么。它不执行具体操作只做决策。def planning_node(state: AssistantState) - dict: 节点2规划下一步动作 intent state[“detected_intent”] next_step None tool_to_use None if intent “weather”: if state[“city”]: next_step “call_tool” tool_to_use “get_weather” else: # 如果意图是天气但没提供城市我们需要追问 next_step “ask_for_city” elif intent “greeting”: next_step “generate_response” # 直接生成问候回复 else: next_step “handle_unknown” # 处理未知意图 return { “next_step”: next_step, “tool_to_call”: tool_to_use }节点3工具调用节点这个节点负责执行具体的工具调用。def tool_call_node(state: AssistantState) - dict: 节点3调用工具 tool_name state[“tool_to_call”] result None if tool_name “get_weather”: # 调用我们之前定义的天气工具 result get_weather.invoke({“city”: state[“city”]}) # 注意get_weather.invoke()返回的是LangChain Tool的调用结果 return { “tool_called”: tool_name, “tool_result”: result }节点4响应生成节点这个节点综合所有信息生成最终返回给用户的自然语言回复。def response_generation_node(state: AssistantState) - dict: 节点4生成最终回复 intent state[“detected_intent”] output “” if intent “weather”: if state[“tool_result”]: output f“{state[‘city’]}的天气是{state[‘tool_result’]}” else: output “请问您想查询哪个城市的天气呢” elif intent “greeting”: output “你好我是天气助手有什么可以帮您” else: output “抱歉我没太明白您的意思。您可以问我关于天气的问题。” # 同时我们将本次交互的完整信息追加到消息历史中可选用于长期记忆 new_message {“role”: “assistant”, “content”: output} # 注意由于messages字段使用了add_messages归约器我们这样返回即可追加 return { “final_output”: output, “messages”: new_message # 这会被add_messages归约器处理追加到列表 }我们还需要补充两个简单的节点来处理特殊分支def ask_city_node(state: AssistantState) - dict: 追问城市节点 return {“final_output”: “您想查询哪个城市的天气呢”} def handle_unknown_node(state: AssistantState) - dict: 处理未知意图节点 return {“final_output”: “我目前主要擅长回答天气相关问题您可以试试问我‘北京天气怎么样’。”}3.3 组装图并运行现在我们将所有节点和边组装起来。from langgraph.graph import StateGraph, END # 1. 创建图并指定State类型 workflow StateGraph(AssistantState) # 2. 添加所有节点 workflow.add_node(“recognize_intent”, intent_recognition_node) workflow.add_node(“plan”, planning_node) workflow.add_node(“call_tool”, tool_call_node) workflow.add_node(“generate_resp”, response_generation_node) workflow.add_node(“ask_city”, ask_city_node) workflow.add_node(“handle_unknown”, handle_unknown_node) # 3. 设置入口点 workflow.set_entry_point(“recognize_intent”) # 4. 添加边和条件边 # 意图识别后总是进入规划节点 workflow.add_edge(“recognize_intent”, “plan”) # 规划节点后的条件路由 def router_after_plan(state: AssistantState) - str: next_step state.get(“next_step”) # 返回的值必须与add_conditional_edges中映射的键一致 return next_step workflow.add_conditional_edges( “plan”, router_after_plan, { “call_tool”: “call_tool”, “ask_for_city”: “ask_city”, “generate_response”: “generate_resp”, “handle_unknown”: “handle_unknown”, # 理论上这里也应该能处理END但我们的router目前不返回END } ) # 工具调用后进入生成响应节点 workflow.add_edge(“call_tool”, “generate_resp”) # 追问城市和处理未知意图后也进入生成响应节点或者可以直接结束这里我们让它生成回复 workflow.add_edge(“ask_city”, “generate_resp”) workflow.add_edge(“handle_unknown”, “generate_resp”) # 生成响应后工作流结束 workflow.add_edge(“generate_resp”, END) # 5. 编译图 app workflow.compile()现在我们可以运行这个智能体了。# 准备初始状态 initial_state: AssistantState { “messages”: [], # 初始对话历史为空 “user_input”: “北京今天天气怎么样”, “detected_intent”: None, “city”: None, “tool_called”: None, “tool_result”: None, “final_output”: None, } # 运行图 final_state app.invoke(initial_state) print(“最终回复”, final_state[“final_output”]) print(“完整状态”, final_state)执行上述代码你应该会看到输出类似于“最终回复北京的天气是晴15-25°C微风”。整个State对象里包含了每一步的执行结果。踩坑实录在定义条件边时路由函数router_after_plan返回的字符串必须与add_conditional_edges方法中path_map参数字典的键完全匹配。如果返回了一个字典中不存在的值LangGraph会抛出一个KeyError。这是初期调试时最常见的错误之一。一个好的实践是在路由函数里做好默认值处理比如return state.get(“next_step”, “handle_unknown”)。4. 进阶特性子图、长期记忆与流式输出掌握了基础构建后我们来看看那些让LangGraph真正强大的进阶特性这些也正是网络热搜词里大家关心的。4.1 子图Subgraph管理复杂性的利器当你的智能体逻辑变得非常复杂时把所有节点和边都放在一个主图里会难以维护。子图允许你将一部分功能模块化封装成一个独立的、可复用的图。例如我们可以把“意图识别”这个本身可能就很复杂的过程涉及LLM调用、实体提取、意图分类封装成一个子图。from langgraph.graph import StateGraph # 定义一个专用于意图识别的State class IntentState(TypedDict): query: str intent: Optional[str] entities: dict # 创建意图识别子图 intent_subgraph_builder StateGraph(IntentState) def llm_classify_node(state: IntentState): # 模拟LLM调用进行复杂分类 return {“intent”: “weather”, “entities”: {“city”: “北京”}} intent_subgraph_builder.add_node(“classify”, llm_classify_node) intent_subgraph_builder.set_entry_point(“classify”) intent_subgraph_builder.add_edge(“classify”, END) intent_subgraph intent_subgraph_builder.compile() # 在主图中我们可以像调用一个节点一样调用这个子图 def intent_node_with_subgraph(state: AssistantState): # 准备子图输入 subgraph_input IntentState(querystate[“user_input”], intentNone, entities{}) # 运行子图 subgraph_result intent_subgraph.invoke(subgraph_input) # 将子图结果映射回主State return { “detected_intent”: subgraph_result[“intent”], “city”: subgraph_result[“entities”].get(“city”) }然后在主图中add_node(“recognize_intent”, intent_node_with_subgraph)即可。子图让代码结构更清晰也便于团队协作和单元测试。4.2 长期记忆Long-term Memory的实现模式“长期记忆”是构建真正个性化Agent的关键。LangGraph本身不提供开箱即用的记忆存储但它通过State和**检查点Checkpointing**机制为实现记忆提供了完美的基础。长期记忆通常涉及两个层面对话记忆Conversation Memory保存在State的messages字段中使用add_messages归约器自动管理。但这仅限于单次图执行的生命周期。持久化记忆Persistent Memory需要将重要的State信息如用户偏好、历史摘要、事实知识保存到外部数据库如SQLite、PostgreSQL、Redis并在下次对话时加载。一个常见的模式是使用图编译时的checkpointer参数和**config配置**。from langgraph.checkpoint.sqlite import SqliteSaver import sqlite3 # 1. 创建一个SQLite检查点存储器 conn sqlite3.connect(“:memory:”) # 实际应用请用文件路径 checkpointer SqliteSaver(conn) # 2. 在编译图时传入checkpointer app_with_memory workflow.compile(checkpointercheckpointer) # 3. 运行时通过config指定线程ID通常用用户ID或会话ID config {“configurable”: {“thread_id”: “user_12345”}} initial_state {“user_input”: “你好” …} # 第一次调用会创建或加载这个thread_id对应的检查点 result1 app_with_memory.invoke(initial_state, configconfig) # 此时State包括messages会被自动持久化到SQLite # 第二次调用传入相同的configLangGraph会从检查点恢复上一次的State new_state_input {“user_input”: “我还想问问上海的天气” …} # 注意这里传入的初始状态会被合并到恢复的State中通常我们只传增量信息 result2 app_with_memory.invoke({“user_input”: “我还想问问上海的天气”}, configconfig) # 在result2的State里messages会包含上一次的对话历史通过这种方式智能体就拥有了跨越多次调用的“长期记忆”。你可以定制checkpointer来使用其他存储后端也可以选择只在特定节点执行后保存检查点通过checkpoint参数实现更精细的控制。4.3 流式输出Streaming与执行控制对于需要实时反馈的交互式应用流式输出至关重要。LangGraph的stream方法允许你逐步获取每个节点执行后的状态更新而不是等待整个图执行完毕。# 使用stream方法进行流式调用 inputs {“user_input”: “北京和上海的天气”} config {“configurable”: {“thread_id”: “stream_test”}} for event in app_with_memory.stream(inputs, configconfig, stream_mode“values”): # event 是一个元组 (node_name, state_update) node, state_update event print(f“节点 [{node}] 执行完毕”) if “final_output” in state_update and state_update[“final_output”]: print(“生成部分回复:”, state_update[“final_output”]) # 你可以在这里将state_update[“final_output”]发送给前端实现打字机效果关于热搜词中提到的“compiledstategraph.stream()如何终止”这是一个更高级的话题。流的终止通常由图的逻辑决定执行到END节点。如果你想从外部中断一个正在进行的流式执行这通常需要在异步上下文中处理通过取消任务如asyncio.Task.cancel()来实现而不是通过LangGraph的API直接终止。在设计图时可以考虑添加一个“取消”或“超时”节点通过条件边在某些情况下提前路由到END。5. LangGraph与LangChain、FastAPI及Dify的集成实践在实际项目中LangGraph很少单独使用它需要与Web框架、前端界面以及其他AI编排工具集成。5.1 与LangChain的深度融合LangGraph和LangChain是绝配。你的节点函数可以轻松使用LangChain的任何组件from langchain.prompts import ChatPromptTemplate from langchain.chat_models import ChatOpenAI # 假设使用OpenAI from langchain.schema.runnable import RunnablePassthrough # 在节点函数中使用LangChain Chain def llm_based_node(state: State): prompt ChatPromptTemplate.from_template(“你是一个助手。用户说{query}。请分析意图。”) model ChatOpenAI(model“gpt-3.5-turbo”) chain prompt | model # 运行Chain ai_message chain.invoke({“query”: state[“user_input”]}) # 处理结果并更新State... return {“intent”: “analyzed_by_llm”}你可以把LangChain Chain当作LangGraph Node的一个强大“执行引擎”。LangGraph负责流程和状态LangChain负责与LLM、工具、检索器等具体组件的交互。5.2 基于FastAPI构建LangGraph服务端将LangGraph智能体封装成API服务是常见的需求。FastAPI是一个高性能的现代框架。from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional app FastAPI(title“LangGraph智能体API”) # 定义请求/响应模型 class ChatRequest(BaseModel): message: str user_id: str # 用于区分不同用户的记忆 thread_id: Optional[str] None # 可选的特定会话线程 class ChatResponse(BaseModel): response: str thread_id: str status: str # 假设我们已经编译好了一个带检查点的图应用 ‘agent_app’ # app.post(“/chat”, response_modelChatResponse) async def chat_endpoint(request: ChatRequest): try: # 准备配置使用user_id或提供的thread_id thread_id request.thread_id or f“user_{request.user_id}” config {“configurable”: {“thread_id”: thread_id}} # 调用图。输入是增量消息检查点会加载历史状态。 # 注意这里我们只传入新的用户消息其他状态由检查点恢复。 inputs {“user_input”: request.message} result agent_app.invoke(inputs, configconfig) response_text result.get(“final_output”, “抱歉我没有生成回复。”) return ChatResponse( responseresponse_text, thread_idthread_id, status“success” ) except Exception as e: raise HTTPException(status_code500, detailf“智能体处理失败: {str(e)}”) # 可以添加其他端点如重置记忆 /reset?thread_idxxx这样前端应用就可以通过发送HTTP请求与拥有长期记忆的LangGraph智能体进行对话了。5.3 与Dify等平台的结合思考热搜词中出现了“Dify和LangGraph可以一起用吗”。Dify是一个低代码的LLM应用开发平台它提供了可视化的编排界面。从架构上看Dify更侧重于前端交互、知识库管理、插件市场和应用部署提供了一个开箱即用的全栈解决方案。LangGraph是一个纯后端的、代码优先的、用于构建复杂有状态工作流的Python框架。它们完全可以协同工作。一种典型的模式是使用Dify作为前端界面和基础设施如API网关、知识库检索而将最核心、最复杂的智能体逻辑用LangGraph实现并作为Dify的一个“自定义工具”或“API节点”来调用。Dify的工作流节点可以调用你部署的LangGraph智能体API并将返回结果集成到其对话流中。这样你既享受了Dify的便捷性又拥有了LangGraph带来的强大、灵活的编排能力。6. 调试、监控与性能优化实战指南开发复杂的图应用调试是一大挑战。以下是一些实战中总结出的经验。6.1 可视化与调试LangGraph提供了内置的可视化功能对于理解流程至关重要。# 方法1生成Mermaid图表字符串注意输出中禁止使用Mermaid代码块但你可以将其复制到支持Mermaid的编辑器中查看 graph_mermaid app.get_graph().draw_mermaid() print(graph_mermaid) # 会输出一个Mermaid格式的字符串 # 方法2打印图形结构 print(app.get_graph().print_ascii()) # 在控制台打印ASCII艺术图对于调试最有效的方法是在节点函数中增加详细的日志并观察State的变化。你可以使用Python的logging模块或者简单地在节点函数开始和结束时打印关键信息。def some_node(state: State): print(f“[some_node] 输入State: {state}”) # ... 业务逻辑 ... result {“key”: “value”} print(f“[some_node] 输出更新: {result}”) return result6.2 状态State管理的常见陷阱归约器冲突如果你为同一个State字段定义了多个归约器或者归约器的行为不符合预期会导致状态更新混乱。务必理解每个归约器的作用如add_messages是追加operator.setitem是覆盖。状态污染一个节点错误地修改了其他节点依赖的字段。设计State时要做到高内聚、低耦合明确每个节点的输入输出字段。使用Pydantic模型能帮助在开发早期发现类型错误。检查点膨胀如果每次调用都保存完整的State特别是包含长消息历史存储会快速增长。可以考虑定期对消息历史进行摘要Summarization只保存摘要和最近几条原始消息将摘要存入State从而压缩记忆。6.3 性能优化要点异步节点如果节点涉及网络I/O如调用LLM API、查询数据库应将其定义为异步函数async def并在图中使用异步调用ainvoke,astream可以显著提升并发性能。async def async_tool_call_node(state: State): result await some_async_api(state[“query”]) return {“result”: result}条件边优化条件边的路由函数应尽可能简单、快速。避免在路由函数中执行LLM调用等重型操作。路由决策应基于State中已有的、由前置节点计算好的结果。子图复用对于被频繁调用的复杂子图确保其编译后的对象compiledgraph被复用而不是每次调用都重新编译。LLM调用批处理如果多个节点都需要调用LLM可以考虑是否能够合并这些请求或者使用LangChain的批量调用功能来减少API往返次数。构建基于LangGraph的智能体是一个迭代过程。从简单的线性流程开始逐步引入条件逻辑、循环、子图和记忆。充分利用可视化工具来理解你的图结构用细致的日志来跟踪状态流变。当你习惯了用“图”的思维来设计AI应用时你会发现构建那些能够进行多轮决策、拥有记忆和规划能力的智能体不再是一件令人望而生畏的事情。它就像绘制一张清晰的地图让智能体在这张地图上自主而可靠地走向目标。