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”, temperature=0) 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(input=user_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_schema=MyConfig) # ... 添加节点和边 ... # 定义一个发送通知并等待的节点 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支持子图，允许你将一个功能模块（例如，一个完整的“文档检索与摘要”流程）封装成一个独立的图，然后在主图中将其作为一个节点来调用。

这样做的好处是：

1. 高内聚低耦合：每个子图可以独立开发、测试和调试。
2. 复用性：通用的处理流程（如“信息核实”）可以被多个主图复用。
3. 可视化清晰：主图的结构会更简洁，你可以通过折叠子图来查看不同层级的抽象。

# 假设我们有一个封装好的“检索增强生成(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(checkpointer=memory, debug=True) # debug=True会在控制台打印详细日志 # 或者，使用LangSmith进行更细致的追踪 app = workflow.compile( checkpointer=memory, config={“configurable”: {“thread_id”: “user_123”}}, # LangChain/LangGraph通常与LangSmith自动集成，只需设置环境变量 )

我的经验是，在开发阶段一定要把debug=True打开，并习惯查看每一步的状态流转。此外，像我们在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(checkpointer=checkpointer) # 第一次调用，传入thread_id标识同一个会话 config = {“configurable”: {“thread_id”: “conversation_001”}} result1 = app.invoke({“user_query”: “你好”}, config=config) # 十分钟后，用户再次发言，使用相同的thread_id，图会从上次中断的地方继续 result2 = app.invoke({“user_query”: “我上次问的那个问题…”}, config=config)

选择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, json=params, timeout=10) 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比测试普通函数更复杂。我的策略是分层测试：

1. 单元测试：单独测试每个节点函数，模拟输入State，断言输出State。
2. 集成测试：测试一个子图或简单的条件分支。可以使用app.invoke并注入特定的初始状态，验证最终的输出状态和步骤历史是否符合预期。
3. 端到端测试：模拟真实用户对话流，测试整个编译好的图。这里需要大量使用Mock对象来模拟LLM和工具调用，确保测试的稳定性和速度。可以使用unittest.mock库来patch掉ChatOpenAI.invoke和@tool函数，让它们返回预设的答案。

从LangChain到LangGraph，不仅仅是换了一个库，更是思维模式从“链”到“图”的升级。它迫使你更清晰地思考智能体的状态空间和决策流程。刚开始可能会觉得繁琐，但一旦你习惯了这种声明式的、可视化的编程方式，再回去看那些面条式的回调代码就会觉得难以忍受。LangGraph带来的最大收益是可控性和可维护性，这对于构建真正能在生产环境落地的AI智能体来说，是至关重要的基石。