1. 从 LangChain 到 LangGraph为什么我们需要新的“大脑”如果你在过去一年里折腾过基于大语言模型的智能应用那么“LangChain”这个名字对你来说一定不陌生。它就像一个功能强大的“瑞士军刀”把提示词模板、记忆、工具调用、文档检索这些功能模块化让我们能像搭积木一样快速构建起一个AI应用的原型。我最初用它来做一个简单的文档问答机器人感觉确实方便几行代码就能跑起来。但当我试图构建一个更复杂的、需要多步骤决策和状态流转的“智能体”时问题就来了。比如我想做一个能自动处理用户工单的Agent。它需要先理解用户的问题然后根据问题类型决定是去查知识库、调用某个API获取数据还是需要向用户追问更多细节。在LangChain的框架里我可以用AgentExecutor配合一堆工具来实现。但很快代码就变成了一团乱麻状态管理分散在各个回调函数里步骤之间的依赖关系不清晰一旦流程需要循环比如用户回答不明确需要再次追问逻辑就变得异常复杂且难以调试。更头疼的是当我想给这个Agent加上“长期记忆”让它能记住和同一个用户的多次对话上下文时LangChain提供的方案总感觉像是后期打上的补丁不够优雅和可控。这其实就是从“链式思维”到“图式思维”的跃迁。LangChain的核心是“链”它定义了线性的、一步接一步的执行顺序适合确定性的流程。但真正的智能体行为更像一张“图”它有不同的节点代表思考、决策、执行等动作节点之间通过边连接而边的走向取决于当前的状态。我们需要一个能清晰描述这种有状态、可分支、可循环的计算过程的框架。这就是LangGraph出现的背景。它不是要取代LangChain而是LangChain生态的自然进化。你可以把LangGraph理解为专门为构建复杂、可控的智能体而设计的“大脑皮层”。它引入了“状态”作为一等公民用“图”来显式地定义智能体的工作流使得整个决策逻辑变得可视化、可调试、可持久化。从工程实践的角度看这意味着我们终于可以告别那些隐藏在if-else和回调地狱里的脆弱逻辑用一种更声明式、更健壮的方式来构建真正具备自主能力的AI智能体。接下来我会结合一个具体的工单处理Agent案例带你一步步拆解如何用LangGraph构建一个可控、可观测、可扩展的智能体系统。2. 核心概念拆解State、Node、Edge与Graph在动手写代码之前我们必须先吃透LangGraph的几个核心概念。这就像学开车先要明白方向盘、油门和刹车是干嘛的否则直接上路肯定手忙脚乱。2.1 State智能体的“记忆画布”State是LangGraph中最重要的概念它是一个贯穿整个图执行过程的、可变的字典。你可以把它想象成智能体的“短期工作记忆区”或一块共享的白板。所有节点Node都从这块白板上读取信息处理完后再把新的信息写回白板。定义一个State通常使用TypedDict类型化字典来明确其结构这能极大提升代码的可读性和类型安全性。在我们的工单处理Agent例子中State可能长这样from typing import TypedDict, Annotated from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): # 对话消息历史LangGraph提供了add_messages操作符来简化列表追加 messages: Annotated[list, add_messages] # 用户输入的原始问题 user_query: str # 经过分类的问题类型如“查询余额”、“故障申报”、“业务咨询” query_type: str # 从知识库或API获取到的中间结果 retrieved_info: str # 最终要返回给用户的答案 final_answer: str # 一个标志位用于控制图的走向比如“是否需要继续追问” needs_clarification: bool这里的关键是Annotated和add_messages。add_messages是一个“归约器”它定义了当多个节点同时尝试修改messages列表时如何合并这些修改这里是追加。这解决了并发操作状态时的冲突问题。其他字段如query_type我们使用operator.setitem作为归约器意味着后一个节点的写入会直接覆盖前一个节点的值。定义State时一定要想清楚每个字段的生命周期和更新规则这是构建稳定Agent的基础。2.2 Node与Edge智能体的“思考单元”与“决策路径”Node节点就是一个普通的Python函数或可调用对象它接收当前的State执行一些操作然后返回一个包含对State更新内容的字典。def classify_query(state: AgentState) - dict: 节点函数分类用户问题 query state[“user_query”] # 这里可以调用一个LLM或一个简单的分类器 # 假设我们有一个分类函数 q_type query_classifier(query) return {“query_type”: q_type}Edge边决定了执行完一个节点后下一步该去哪个节点。LangGraph提供了几种边条件边根据State中的某个值动态决定下一个节点。这是实现分支逻辑的核心。普通边固定指向下一个节点。入口边图的开始。出口边图的结束。条件边的定义非常直观它就是一个返回下一个节点名称的函数def route_after_classification(state: AgentState) - str: 条件边根据问题类型路由到不同处理节点 q_type state[“query_type”] if q_type “故障申报”: return “handle_ticket” elif q_type “查询余额”: return “query_balance” else: return “general_consultation”2.3 Graph将一切编织成工作流最后我们用StateGraph这个类把State、Node和Edge组装起来形成一个完整的、可执行的计算图。from langgraph.graph import StateGraph, END # 1. 创建图并指定State的类型 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(“classify”, classify_query) workflow.add_node(“handle_ticket”, handle_ticket_node) workflow.add_node(“query_balance”, query_balance_node) workflow.add_node(“general_consultation”, general_consultation_node) # 3. 设置入口点 workflow.set_entry_point(“classify”) # 4. 添加边包括条件边 workflow.add_conditional_edges( “classify”, route_after_classification, # 上面定义的条件路由函数 { “handle_ticket”: “handle_ticket”, “query_balance”: “query_balance”, “general_consultation”: “general_consultation” } ) # 5. 为其他节点添加普通边指向结束 workflow.add_edge(“handle_ticket”, END) workflow.add_edge(“query_balance”, END) workflow.add_edge(“general_consultation”, END) # 6. 编译图得到可执行对象 app workflow.compile()编译后的app就是一个可以调用的智能体了。你通过app.invoke()传入初始状态它就会按照你定义的图逻辑自动执行。这种声明式的编程方式将业务逻辑图结构和执行引擎LangGraph运行时彻底解耦使得代码的维护性和可测试性大大增强。你可以轻易地将这个图可视化出来一目了然地看到整个Agent的决策流程这是用传统代码难以实现的。3. 工程实践构建一个工单处理智能体现在我们把理论付诸实践构建一个相对完整的工单处理智能体。这个Agent需要完成以下功能1理解用户意图2根据意图采取不同行动查知识库、调API、反问3管理多轮对话4给出最终答复。3.1 定义状态与工具准备首先我们完善之前定义的State并准备一些工具。工具就是Agent可以调用的外部函数比如搜索、计算、调用API等。from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages import operator from langchain.tools import tool from langchain_community.utilities import SQLDatabase from some_api_client import TicketAPI class AgentState(TypedDict): messages: Annotated[List, add_messages] user_query: str query_type: str # “fault”, “query”, “consult” needs_clarification: bool clarification_question: str retrieved_data: str final_answer: str step_history: List[str] # 记录执行步骤用于调试 # 模拟工具定义 tool def search_knowledge_base(query: str) - str: 在内部知识库中搜索相关问题解决方案 # 这里可以是向量数据库检索 return f“根据知识库关于‘{query}’的解决方案是重启服务。” tool def query_user_account(user_id: str, query_type: str) - str: 查询用户账户信息余额、订单等 # 模拟调用内部账户API return f“用户{user_id}的{query_type}为100元。” tool def create_ticket(description: str, priority: str) - str: 在工单系统中创建一条新工单 # 模拟调用工单系统API ticket_id “TICKET-2024-001” return f“已创建工单 {ticket_id}优先级{priority}描述{description}” tools [search_knowledge_base, query_user_account, create_ticket]3.2 构建图节点分解智能体的思考过程一个复杂的节点内部可能包含LLM调用和工具使用。我们可以利用LangChain的create_react_agent或类似助手函数来构建单个节点的推理能力。但更LangGraph的方式是将“思考是否用工具”和“使用工具”也建模为图的一部分。这里为了清晰我们先在一个节点内完成小范围的推理。from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent from langchain.agents.format_scratchpad import format_log_to_str from langchain.tools.render import render_text_description llm ChatOpenAI(model“gpt-4”, temperature0) def classify_and_route_node(state: AgentState) - dict: 节点1分析用户输入并分类路由 step_history state.get(“step_history”, []) step_history.append(“进入 classify_and_route_node”) user_input state[“user_query”] # 使用LLM进行意图识别和分类 system_prompt “””你是一个工单分类助手。请分析用户输入判断其属于以下哪一类 1. fault: 故障申报或问题投诉。 2. query: 查询信息如余额、订单状态。 3. consult: 一般业务咨询。 4. unclear: 意图不明确需要进一步澄清。 只返回类别关键词fault/query/consult/unclear。输入{input}“”” response llm.invoke(system_prompt.format(inputuser_input)) q_type response.content.strip().lower() # 如果意图不明确设置澄清标志和问题 needs_clarify (q_type “unclear”) clarify_q “” if needs_clarify: clarify_q “您能具体描述一下您遇到的问题或想查询的信息吗” return { “query_type”: q_type, “needs_clarification”: needs_clarify, “clarification_question”: clarify_q, “step_history”: step_history } def handle_fault_node(state: AgentState) - dict: 节点2处理故障申报 step_history state.get(“step_history”, []) step_history.append(“进入 handle_fault_node”) user_input state[“user_query”] # 首先尝试从知识库找答案 kb_result search_knowledge_base.invoke(user_input) # 让LLM判断知识库答案是否足够解决用户问题 judge_prompt f“””用户问题{user_input} 知识库检索结果{kb_result} 请判断 1. 如果知识库结果能直接解决问题请直接总结一个友好的答复。 2. 如果不能请生成一个用于创建工单的详细描述。 只输出最终答复或工单描述。“”” llm_decision llm.invoke(judge_prompt) decision_text llm_decision.content final_answer decision_text # 如果LLM的判断看起来像工单描述这里用简单关键词判断则创建工单 if “工单” in decision_text or “ticket” in decision_text.lower(): ticket_result create_ticket.invoke(decision_text, “medium”) final_answer f“{decision_text}\n\n系统执行结果{ticket_result}” step_history.append(“故障处理完成”) return { “retrieved_data”: kb_result, “final_answer”: final_answer, “step_history”: step_history } def handle_query_node(state: AgentState) - dict: 节点3处理查询请求 step_history state.get(“step_history”, []) step_history.append(“进入 handle_query_node”) user_input state[“user_query”] # 简单从输入中提取用户ID和查询类型实际应用需更复杂的NLP # 这里仅为示例 assumed_user_id “user123” assumed_query “余额” api_result query_user_account.invoke(f“{assumed_user_id}, {assumed_query}”) final_answer f“查询结果{api_result}” step_history.append(“信息查询完成”) return { “retrieved_data”: api_result, “final_answer”: final_answer, “step_history”: step_history } def ask_clarification_node(state: AgentState) - dict: 节点4向用户提出澄清问题 step_history state.get(“step_history”, []) step_history.append(“进入 ask_clarification_node”) # 这个节点不修改最终答案只是将澄清问题放入消息历史并等待下一次输入 # 在实际流式应用中这里会暂停图执行等待外部输入。 # 为简化我们假设澄清问题就是输出。 clarify_q state.get(“clarification_question”, “请提供更多细节。”) return { “final_answer”: clarify_q, “step_history”: step_history }3.3 组装与执行让图运转起来现在我们用更复杂的条件边来组装整个图实现“澄清-回答”的循环。from langgraph.graph import StateGraph, END workflow StateGraph(AgentState) # 添加所有节点 workflow.add_node(“classify”, classify_and_route_node) workflow.add_node(“handle_fault”, handle_fault_node) workflow.add_node(“handle_query”, handle_query_node) workflow.add_node(“ask_clarify”, ask_clarification_node) workflow.add_node(“finalize”, lambda state: state) # 一个空节点作为汇聚点 # 设置入口 workflow.set_entry_point(“classify”) # 关键分类后的条件路由 def route_based_on_type_and_clarity(state: AgentState) - str: q_type state.get(“query_type”, “unclear”) needs_clarify state.get(“needs_clarification”, False) if needs_clarify: # 如果需要澄清则进入澄清节点 return “ask_clarify” else: # 否则根据分类结果路由 return { “fault”: “handle_fault”, “query”: “handle_query”, “consult”: “handle_fault” # 咨询也先走知识库流程 }.get(q_type, “ask_clarify”) # 默认情况也去澄清 workflow.add_conditional_edges( “classify”, route_based_on_type_and_clarity ) # 澄清节点之后我们需要重新分类。这里模拟用户已回答将新输入合并后重新开始。 # 一种方法是让ask_clarify节点不直接连接END而是连接一个“等待输入”的节点输入后重新进入classify。 # 为简化演示我们假设澄清后自动生成一个模拟回答并重新路由。 def after_clarification(state: AgentState) - str: # 在实际中这里state[‘messages’]里应该包含了用户的回复。 # 我们模拟用户回复后应重新进行分类。 return “classify” workflow.add_edge(“ask_clarify”, “classify”) # 形成一个澄清循环 # 处理节点完成后都汇聚到finalize节点然后结束 workflow.add_edge(“handle_fault”, “finalize”) workflow.add_edge(“handle_query”, “finalize”) workflow.add_edge(“finalize”, END) # 编译图 app workflow.compile()现在我们可以运行这个智能体了# 初始化状态 initial_state { “user_query”: “我的服务无法访问了怎么办”, “messages”: [], “step_history”: [] } # 执行图 final_state app.invoke(initial_state) print(“最终答案”, final_state[“final_answer”]) print(“执行步骤”, final_state[“step_history”])对于“我的服务无法访问了”这个问题图会沿着classify - handle_fault - finalize - END的路径执行最终可能输出一个从知识库找到的解决方案或者自动创建的工单信息。如果用户一开始输入的是“我想问一下”由于意图不明确图会走classify - ask_clarify - classify - ...的循环直到classify节点识别出明确意图为止。4. 高级模式与生产级考量上面的例子展示了LangGraph的核心用法但对于生产环境还有几个至关重要的高级模式和工程问题需要解决。4.1 人工接管与中断机制一个健壮的Agent不能一直“自动驾驶”。当它不确定或遇到无法处理的边界情况时应该能“举手”请求人工介入。这在LangGraph中可以通过“暂停”和“外部驱动”来实现。LangGraph的Checkpointer和Interrupt机制就是干这个的。简单来说你可以在图中定义一个特殊的“人工审核”节点。当流程到达这个节点时图的执行会被暂停并将当前状态持久化到数据库Checkpoint。然后你的系统可以发送一个通知如到管理后台等待人工输入。人工审核完成后系统再向这个被暂停的图实例注入新的状态如人工批示并恢复执行。from langgraph.checkpoint.aiosqlite import AsyncSqliteSaver from langgraph.graph import MessagesState # 使用支持暂停的检查点存储器 memory AsyncSqliteSaver.from_conn_string(“:memory:”) workflow StateGraph(MessagesState, config_schemaMyConfig) # ... 添加节点和边 ... # 定义一个发送通知并等待的节点 def human_review_node(state: MessagesState): # 1. 将当前问题和Agent建议保存到工单系统 ticket_id create_review_ticket(state[“messages”]) # 2. 抛出中断等待外部输入 raise PendingInterruption(“wait_for_human”, {“ticket_id”: ticket_id}) # 当从中断恢复时可以从state中读取人工输入的结果 human_feedback state.get(“human_feedback”) return {“messages”: human_feedback} workflow.add_node(“human_review”, human_review_node) # 在需要的地方如LLM置信度低时路由到这个节点在生产中你必须设计好这个“人机回环”的接口和状态恢复逻辑这是确保Agent可靠性的安全网。4.2 子图与模块化设计当智能体流程非常复杂时把所有的节点和边都放在一个主图里会变得难以维护。LangGraph支持子图允许你将一个功能模块例如一个完整的“文档检索与摘要”流程封装成一个独立的图然后在主图中将其作为一个节点来调用。这样做的好处是高内聚低耦合每个子图可以独立开发、测试和调试。复用性通用的处理流程如“信息核实”可以被多个主图复用。可视化清晰主图的结构会更简洁你可以通过折叠子图来查看不同层级的抽象。# 假设我们有一个封装好的“检索增强生成(RAG)”子图 rag_subgraph create_rag_graph().compile() # 在主图中可以把这个编译好的子图当作一个节点添加 def rag_node(state: AgentState): # 准备子图需要的输入状态 subgraph_state {“question”: state[“user_query”]} # 调用子图 result rag_subgraph.invoke(subgraph_state) # 处理子图输出更新主状态 return {“retrieved_data”: result[“answer”]} workflow.add_node(“rag_processor”, rag_node)4.3 可观测性与调试当Agent行为不符合预期时如何调试LangGraph内置了强大的追踪功能。你可以通过配置将每一步的执行详情输入State、输出State、节点名称、耗时等输出到控制台或集成到像LangSmith这样的APM平台。from langsmith import Client from langgraph.graph import StateGraph client Client() # 在编译时配置 app workflow.compile(checkpointermemory, debugTrue) # debugTrue会在控制台打印详细日志 # 或者使用LangSmith进行更细致的追踪 app workflow.compile( checkpointermemory, config{“configurable”: {“thread_id”: “user_123”}}, # LangChain/LangGraph通常与LangSmith自动集成只需设置环境变量 )我的经验是在开发阶段一定要把debugTrue打开并习惯查看每一步的状态流转。此外像我们在State中定义的step_history字段也是一个简单有效的自定义追踪手段可以帮助你快速定位问题发生在哪个业务节点。4.4 持久化与状态管理对于需要长时间运行、多次交互的Agent如一个客服对话机器人状态的持久化至关重要。你不能每次用户说话都从头开始。Checkpointer就是LangGraph官方推荐的状态持久化方案。它不仅能保存State还能保存图的结构和中断点确保会话的连续性。from langgraph.checkpoint.aiosqlite import AsyncSqliteSaver # 使用SQLite存储检查点生产环境可用Postgres checkpointer AsyncSqliteSaver.from_conn_string(“sqlite:///checkpoints.db”) app workflow.compile(checkpointercheckpointer) # 第一次调用传入thread_id标识同一个会话 config {“configurable”: {“thread_id”: “conversation_001”}} result1 app.invoke({“user_query”: “你好”}, configconfig) # 十分钟后用户再次发言使用相同的thread_id图会从上次中断的地方继续 result2 app.invoke({“user_query”: “我上次问的那个问题…”}, configconfig)选择Checkpointer时要考虑并发访问和性能。对于高并发场景基于内存的检查点可能不够用需要选择像PostgresSaver这样的分布式存储后端。5. 避坑指南与最佳实践在从LangChain迁移到LangGraph以及实际开发Agent的过程中我踩过不少坑也总结出一些让项目更稳健的经验。5.1 State设计的“纯度”与副作用坑在节点函数中直接修改传入的State字典或者执行一些不可预测的副作用如直接向数据库写入。解保持节点函数的“相对纯净”。它应该只读取State通过返回值来声明如何更新State。所有对外部系统的写操作数据库、API调用都应该封装在tool装饰的函数内并通过调用工具的方式来执行。这样做的目的是让图的计算过程更可预测、可回溯。如果需要在节点内写日志也最好通过State传递日志信息由一个专门的“日志节点”统一处理。5.2 条件边的路由函数要简单稳定坑在条件路由函数route_function中编写复杂的业务逻辑或调用LLM。解条件边函数应该尽可能简单、快速、确定。它最好只做基于State现有字段的简单判断。复杂的路由决策比如用LLM判断下一步该干嘛应该设计成一个独立的“路由决策”节点。这个节点负责更新State中的一个字段如next_node然后由一条固定的条件边读取这个字段来决定走向。这样逻辑更清晰也更容易调试。# 不推荐在条件边函数里做复杂操作 def complex_route(state): analysis llm.invoke(f“分析{state[‘query’]}”) # 避免在这里调用LLM if “复杂” in analysis: return “node_a” else: return “node_b” # 推荐使用一个专门的决策节点 def decision_node(state): analysis llm.invoke(f“分析{state[‘query’]}”) next_node “node_a” if “复杂” in analysis else “node_b” return {“next_node”: next_node} def simple_route(state): return state.get(“next_node”, “default_node”) # 条件边函数只做简单读取5.3 处理好循环与避免死循环坑图中存在循环如澄清循环但没有设置终止条件导致Agent和用户陷入无限问答。解一定要在循环路径上设置“安全阀”。可以在State中设置一个计数器如clarification_attempts每次循环递增。在条件路由函数中判断如果超过阈值比如3次则强制跳转到“转人工”或“默认回复”节点并结束循环。class AgentState(TypedDict): ... clarification_attempts: int 0 # 使用pydantic或默认值初始化 def route_after_clarify(state: AgentState) - str: if state[“clarification_attempts”] 3: return “escalate_to_human” # 跳转到人工节点 elif state[“needs_clarification”]: return “ask_clarify” else: return “process_query”5.4 工具调用与错误处理坑工具调用失败网络超时、API异常导致整个图执行崩溃。解在工具函数内部做好完善的异常捕获和容错处理返回结构化的错误信息而不是抛出异常。同时在调用工具的节点里也要检查工具返回的结果是否有效并能够根据错误类型更新State引导图走向错误处理或重试节点。tool def call_external_api(params): try: response requests.post(url, jsonparams, timeout10) response.raise_for_status() return {“success”: True, “data”: response.json()} except requests.exceptions.Timeout: return {“success”: False, “error”: “timeout”, “message”: “API请求超时”} except Exception as e: return {“success”: False, “error”: “unknown”, “message”: str(e)} def api_node(state): result call_external_api.invoke(state[“params”]) if not result[“success”]: # 更新State让下一个节点知道出错了 return {“api_status”: “failed”, “error_detail”: result} # 正常处理结果 return {“api_status”: “success”, “api_data”: result[“data”]}5.5 测试策略分层测试与模拟测试一个基于图的Agent比测试普通函数更复杂。我的策略是分层测试单元测试单独测试每个节点函数模拟输入State断言输出State。集成测试测试一个子图或简单的条件分支。可以使用app.invoke并注入特定的初始状态验证最终的输出状态和步骤历史是否符合预期。端到端测试模拟真实用户对话流测试整个编译好的图。这里需要大量使用Mock对象来模拟LLM和工具调用确保测试的稳定性和速度。可以使用unittest.mock库来patch掉ChatOpenAI.invoke和tool函数让它们返回预设的答案。从LangChain到LangGraph不仅仅是换了一个库更是思维模式从“链”到“图”的升级。它迫使你更清晰地思考智能体的状态空间和决策流程。刚开始可能会觉得繁琐但一旦你习惯了这种声明式的、可视化的编程方式再回去看那些面条式的回调代码就会觉得难以忍受。LangGraph带来的最大收益是可控性和可维护性这对于构建真正能在生产环境落地的AI智能体来说是至关重要的基石。