1. 项目概述从 LangChain 到 LangGraph 的思维跃迁如果你已经用了一段时间 LangChain搭建过一些简单的 RAG 或者 Agent 应用可能会遇到一个瓶颈流程一旦复杂起来代码就变得像意大利面条一样各种回调、条件判断纠缠在一起状态管理更是让人头疼。这时候LangGraph 的出现就像给你递上了一把手术刀让你能清晰地解剖和构建复杂的、有状态的 AI 应用工作流。它不是要取代 LangChain而是 LangChain 生态中专门用于构建有状态、多参与者Agent工作流的框架。你可以把它理解为一个专门为 AI 智能体Agent设计的“流程图绘制与执行引擎”。“LangGraph 入门到精通”这个系列旨在带你从零开始深入这个框架的肌理。上一期我们可能聊了概念和核心思想而这一期“基础 API (一)”我们将真正开始动手聚焦于构建一个 LangGraph 应用最核心的基石State状态和Nodes节点。这是你理解后续所有高级特性如循环、分支、子图、持久化的前提。很多人在调用 API 时遇到的诸如‘type’ must be in [“enabled”, “disabled”, “auto”]或者关于上下文长度的报错其根源往往在于对状态的结构和流转没有清晰的认识。本文将用大量代码示例帮你夯实这个基础。2. 核心基石理解 StateGraph 的“状态”设计哲学在 LangGraph 中一切工作流都围绕StateGraph展开。与 LangChain 的Chain不同StateGraph显式地要求你定义一个共享的、类型化的状态对象。这个设计是 LangGraph 强大可控性的根源。2.1 为什么需要显式的状态想象一下你指挥一个团队完成项目。你不会把任务细节零散地记在各个成员的脑子里而是会有一个共享的项目看板状态上面清晰列出了需求文档输入、当前进度、已完成的模块、待解决的问题等。任何成员节点执行任务后都会去更新这个看板。这样每个成员都能基于最新的、统一的信息进行下一步工作。LangGraph 的State就是这个“项目看板”。它强制你将工作流中所有需要传递和更新的数据定义在一个地方。这样做有几个关键好处可预测性每个节点接收什么能修改什么一目了然。可调试性在任何步骤你都可以检查整个状态对象知道工作流进行到哪一步数据是什么。可持久化状态可以被轻松地保存到数据库或文件中实现工作流的暂停、恢复和回溯。易于并行与组合清晰的状态接口使得子图Subgraph和并行处理变得可能。2.2 定义状态从 TypedDict 到 Annotation在 Python 中LangGraph 主要使用TypedDict或 PydanticBaseModel来定义状态的结构。官方更推荐TypedDict因为它更轻量与 Python 类型提示系统集成得更好。我们先来看一个最简单的例子一个问答助手的状态可能只需要问题和答案from typing import TypedDict, List class AgentState(TypedDict): question: str answer: str但现实中的工作流往往更复杂。比如一个多工具调用的研究型 Agent它的状态可能包含from typing import TypedDict, List, Optional, Annotated from langgraph.graph.message import add_messages import operator class ResearchState(TypedDict): # 输入与核心目标 original_query: str refined_question: str # 对话历史使用 LangGraph 提供的注解实现自动累加 messages: Annotated[List, add_messages] # 研究过程数据 search_results: List[str] key_points: List[str] # 最终输出 report: Optional[str] # 控制流标志位 needs_deeper_research: bool iteration_count: int这里有几个关键点Annotated[List, add_messages]这是 LangGraph 的一个精髓。add_messages是一个归约器Reducer。它规定messages这个字段的更新规则不是简单的覆盖而是将节点返回的新消息列表List追加add到原有的messages列表之后。这对于维护对话历史至关重要。常见的归约器还有operator.add用于数值累加等。分离输入、中间过程和输出像original_query和report这样的字段清晰地划分了数据的生命周期。使用控制标志位needs_deeper_research这样的布尔值将用于后续图中控制流程的走向循环或分支。实操心得在项目初期多花时间设计好状态结构后期会省去大量重构的麻烦。思考每个节点需要读取哪些字段又需要更新哪些字段。对于列表类数据务必想清楚是覆盖还是追加并正确使用Annotated和归约器。2.3 状态流转的“宪法”StateGraph的构建定义好状态后我们就可以创建图的工作室——StateGraph。from langgraph.graph import StateGraph # 创建图并指定其状态类型 workflow StateGraph(ResearchState)这个workflow对象目前还是一张白纸。接下来我们要在上面添加节点Nodes和边Edges但在此之前我们必须深刻理解节点的运作机制。3. 工作流的执行单元Node节点的深度解析节点是 LangGraph 图中的基本执行单元。每个节点都是一个函数它接收当前整个状态作为输入然后返回一个字典这个字典包含了要更新的状态字段和值。3.1 节点函数的签名与契约一个节点函数看起来非常简单但内涵契约必须遵守def search_node(state: ResearchState) - dict: 研究节点根据精炼后的问题进行搜索。 # 1. 从状态中读取所需数据 query state[“refined_question”] # 2. 执行核心逻辑例如调用搜索API # 假设我们有一个模拟的搜索函数 results mock_web_search(query, top_k3) # 3. 返回要更新的状态部分 return {“search_results”: results}关键契约输入参数state是完整的、当前的状态对象如ResearchState。输出必须返回一个dict。这个字典的键必须是状态中定义的字段名值则是要为该字段设置的新值。更新逻辑LangGraph 引擎会根据你返回的字典结合该字段定义的归约器如果有去更新全局状态。例如如果search_results字段没有用Annotated指定归约器那么本次返回的results会直接覆盖掉该字段旧的值。3.2 复杂节点示例集成 LLM 与工具调用一个典型的 Agent 节点往往会调用 LLM 并决定使用哪个工具。下面是一个更贴近实战的节点它读取对话历史让 LLM 决定下一步行动from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from langchain_openai import ChatOpenAI llm ChatOpenAI(model“gpt-4-turbo”) def agent_decide_node(state: ResearchState) - dict: 智能体决策节点分析状态决定下一步是回答、搜索还是总结。 # 获取对话历史 messages state[“messages”] # 准备让 LLM 使用的工具假设我们定义了一个搜索工具 tools [search_tool] llm_with_tools llm.bind_tools(tools) # 调用 LLM传入历史消息 response llm_with_tools.invoke(messages) # 将 LLM 的响应消息添加到历史中归约器会自动处理 new_messages [response] # 检查 LLM 是否调用了工具 tool_calls response.tool_calls if tool_calls: # 如果有工具调用我们需要执行工具并生成 ToolMessage tool_messages [] for tc in tool_calls: tool_name tc[“name”] tool_args tc[“args”] if tool_name “web_search”: result perform_actual_search(**tool_args) tool_messages.append(ToolMessage(contentresult, tool_call_idtc[“id”])) new_messages.extend(tool_messages) # 可能还需要设置一个标志表示需要继续循环等待 Agent 下一步决策 return {“messages”: new_messages, “needs_deeper_research”: True} else: # LLM 直接给出了最终答案 final_answer response.content return {“messages”: new_messages, “answer”: final_answer, “needs_deeper_research”: False}这个节点展示了几个重要模式消息累加我们将response放入列表返回依赖messages字段的add_messages归约器自动完成历史追加。工具调用处理节点封装了工具执行的逻辑并将结果包装成ToolMessage返回这符合 LangChain 消息协议。状态更新除了更新messages还可能更新answer和控制标志needs_deeper_research。注意事项节点函数应尽量保持“纯”业务逻辑。避免在节点内部修改传入的state对象虽然它是可变的而是始终通过返回字典来声明更改。这符合函数式编程的思想使逻辑更清晰、可测试。3.3 将节点添加到图中定义好节点函数后使用add_node方法将其添加到图中workflow.add_node(“Search”, search_node) workflow.add_node(“Agent”, agent_decide_node) workflow.add_node(“Report”, write_report_node)add_node的第一个参数是节点的唯一标识符字符串第二个参数是节点函数。这个标识符将在定义边和编译图时被用到。4. 构建工作流连接节点与设置入口有了节点我们需要定义它们之间的执行顺序即添加边Edges。4.1 设置起点set_entry_point每个图需要一个开始的地方。workflow.set_entry_point(“Agent”)这表示工作流启动时第一个执行的节点是“Agent”。4.2 添加顺序边add_edge如果两个节点之间是简单的无条件顺序执行使用add_edge。workflow.add_edge(“Agent”, “Search”) workflow.add_edge(“Search”, “Report”)这定义了Agent - Search - Report的线性流程。4.3 添加条件边add_conditional_edges这才是 LangGraph 展现其智能的地方。节点执行完后下一个节点可以动态决定。from langgraph.graph import END def decide_next_step(state: ResearchState) - str: 根据状态决定下一个节点。 if state.get(“needs_deeper_research”): return “Search” # 需要进一步研究返回搜索节点 elif state.get(“report”): return END # 报告已生成结束工作流 else: return “Agent” # 默认返回 Agent 节点继续决策 # 将条件边添加到 Agent 节点之后 workflow.add_conditional_edges( “Agent”, decide_next_step, { “Search”: “Search”, “Agent”: “Agent”, END: END } ) # 搜索完成后固定返回 Report 节点 workflow.add_edge(“Search”, “Report”) # Report 节点完成后固定结束 workflow.add_edge(“Report”, END)add_conditional_edges详解第一个参数源节点“Agent”。第二个参数一个路由函数decide_next_step。这个函数接收更新后的状态作为输入返回一个字符串。这个字符串必须是第三个参数映射字典的键。第三个参数一个映射字典。键是路由函数可能返回的字符串值是对应的目标节点名。END是 LangGraph 内置的特殊节点表示工作流终止。在上面的例子中Agent节点执行后会调用decide_next_step函数。该函数检查状态中的needs_deeper_research和report字段决定下一步是去“Search”、回到“Agent”还是直接END。4.4 编译图从蓝图到可执行程序添加完所有节点和边后需要将图“编译”成一个可执行对象。app workflow.compile()这个app对象就是你的 AI 工作流应用程序。它有两个最核心的方法.invoke(input_state)和.stream(input_state)。5. 运行与调试invoke 与 stream 实战5.1 使用invoke同步执行invoke方法接收一个初始状态字典运行整个工作流直到结束遇到END并返回最终状态。# 准备初始状态必须符合 ResearchState 的结构 initial_state { “original_query”: “LangGraph 和 LangChain 的主要区别是什么”, “refined_question”: “LangGraph 和 LangChain 的主要区别是什么”, “messages”: [HumanMessage(content“LangGraph 和 LangChain 的主要区别是什么”)], “search_results”: [], “key_points”: [], “report”: None, “needs_deeper_research”: True, “iteration_count”: 0 } # 执行工作流 final_state app.invoke(initial_state) print(final_state[“report”])5.2 使用stream流式执行与调试stream方法对于调试和理解工作流执行过程至关重要。它返回一个生成器每执行完一个节点就产出一次当前状态。from langgraph.checkpoint import MemorySaver # 为了支持流式通常需要配置一个检查点存储器即使是内存的 app workflow.compile(checkpointerMemorySaver()) input_state {…} # 同上 print(“开始流式执行”) for step, state in app.stream(input_state, config{“configurable”: {“thread_id”: “test-1”}}): node_name step.tasks[0].name if step.tasks else “START” print(f“— 节点 [{node_name}] 执行完毕 —”) print(f“ 当前关键状态: question{state.get(‘refined_question’)}, research_needed{state.get(‘needs_deeper_research’)}”) print(f“ 最新消息: {state[‘messages’][-1] if state[‘messages’] else ‘None’}”) print()流式输出能让你清晰地看到每个节点执行后的状态快照。控制流是如何根据条件边进行跳转的。消息历史是如何一步步累积的。这是排查“为什么我的 Agent 陷入了死循环”或者“为什么工具调用结果没被正确传递”等问题的最有效手段。常见问题排查技巧如果你在使用stream时遇到api error: 400 ‘type’ must be in [“enabled”, “disabled”, “auto”]这类错误这通常不是 LangGraph 本身的错误而是你节点函数内部调用的底层 API如 OpenAI、DeepSeek 等返回的错误。你需要检查的是节点函数中调用 LLM 或外部服务的部分确保参数格式正确。使用stream可以帮你定位到具体是哪个节点抛出的错误。6. 状态持久化与检查点入门LangGraph 一个杀手级特性是内置的检查点Checkpoint机制它允许工作流在任何节点执行后暂停并将完整状态保存下来后续可以从该点恢复执行。这对于运行时间长、或需要人工干预的流程如审批非常有用。6.1 配置内存检查点最简单的检查点存储器是MemorySaver它将状态保存在进程内存中。from langgraph.checkpoint import MemorySaver memory MemorySaver() app workflow.compile(checkpointermemory)6.2 使用配置进行持久化执行现在当你调用invoke或stream时需要提供一个config其中包含thread_id来标识不同的会话线程。config {“configurable”: {“thread_id”: “user-session-123”}} # 第一次执行 result1 app.invoke(initial_state, configconfig) print(f“第一次执行后迭代次数: {result1[‘iteration_count’]}”) # 基于上一次的最终状态已自动保存为检查点继续执行 # 我们修改一下状态触发不同的路径 new_input {“needs_deeper_research”: False} result2 app.invoke(new_input, configconfig) # 注意这里传入的是增量更新 print(f“第二次执行后迭代次数: {result2[‘iteration_count’]}”)关键点第二次调用app.invoke时我们传入的new_input并不是完整状态而是增量更新。LangGraph 会首先从检查点加载thread_id对应的最新状态然后用new_input中的字段去更新它再从这个合并后的状态开始执行。这模拟了多轮交互的场景。6.3 检查点的工作原理每个节点执行后LangGraph 都会自动将当前状态保存为一个检查点。检查点不仅包含数据状态还记录了当前所在的节点位置。因此恢复时能精确地从下一个待执行的节点开始。实操心得在生产环境中你会使用SqliteSaver或自定义的数据库存储检查点。MemorySaver仅用于开发和测试。检查点机制是构建长周期、可恢复 AI 应用如客服对话、复杂工作流的基础务必理解其“加载旧状态 - 合并新输入 - 继续执行”的模式。7. 避坑指南与最佳实践结合网络热词中提到的常见错误这里总结一些初期高频问题api error: 400 ‘type’ must be in [“enabled”, “disabled”, “auto”]问题这通常是调用第三方大模型 API如 DeepSeek时传递了不正确或不被支持的参数。例如可能在ChatOpenAI或类似客户端中设置了一个无效的model_kwargs。排查检查你的节点函数中初始化 LLM 客户端的代码。确保参数名和值符合该 API 提供商的最新文档。使用stream方法定位到具体出错的节点。api error: 400 this model‘s maximum context length is ... tokens问题上下文长度超限。这在 LangGraph 中尤其常见因为messages历史会随着工作流执行不断累积add_messages。解决摘要化在特定节点如循环多次后添加一个“摘要节点”将冗长的历史消息总结成一段精简的文字然后替换或清空旧消息。滑动窗口只保留最近 N 轮对话的消息。优化状态设计并非所有中间信息都需要放在messages里。可以将详细的工具调用结果存放在其他字段如search_results而在messages中只保留精简的对话。工作流陷入无限循环问题条件边逻辑有误导致在几个节点间来回跳转无法到达END。排查使用stream调试这是最直观的方法观察状态变化和节点跳转路径。设置安全阀在状态中增加iteration_count字段每次循环递增。在路由函数中判断如果超过阈值如10次则强制返回END。状态更新不符合预期问题某个字段的值没有被正确更新或累加。排查检查归约器确认字段是否使用了Annotated及正确的归约器如add_messages用于消息列表operator.add用于整数。检查节点返回值确保节点返回的字典键名与状态字段名完全一致包括大小写。理解更新顺序多个节点对同一字段的更新会按照节点执行顺序依次应用归约逻辑。langgraph和langchain的区别混淆核心区别LangChain是一个用于构建 LLM 应用的全套工具链包含模型 I/O、提示模板、链、记忆、检索器等众多模块。LangGraph是LangChain生态中的一个专门库其核心焦点是构建有状态、多环节的循环工作流特别是 Agent 的编排。你可以只用LangChain的基础组件而用LangGraph来编排它们之间的复杂协作逻辑。掌握 State 和 Node 的设计是驾驭 LangGraph 的第一步。它们定义了数据的结构和流动的规则。在下一期中我们将深入更动态的部分如何构建循环、分支、以及利用 LangGraph 强大的MessagesState和ToolNode等高级抽象来简化通用模式。当你理解了这些基础 API再看官方文档中那些复杂的例子就会有一种豁然开朗的感觉。