掌握这 6 个 LangGraph 核心概念，真正理解 AI Agent 工作流

大多数人复制粘贴第一个教程让它跑起来然后一旦尝试修改任何东西就彻底卡住。原因就在这六个概念。大多数人可以在十分钟内构建出自己的第一个 LangGraph agent。它能运行。然后他们改了一个地方。他们添加了一个分支。graph 永远不会停止。或者它跳过了一个他们确信会运行的 node。或者它在 session 之间忘掉了一切。调试一个小时后他们得出同样的结论“LangGraph 很复杂。”其实不是。缺失的并不是另一个教程也不是更大的代码示例。缺失的是一种 mental model用来解释 graph 为什么会以这种方式运行。一旦你理解了六个核心概念——state 如何流动、node 如何通信、edge 如何做决策以及 memory 实际上如何工作——LangGraph 就会变得出奇地可预测。本文将从第一性原理拆解这六个概念。掌握它们之后你将不再靠试错来调试 graph而是能够自信地构建它们。为什么选择 LangGraphLangChain 让串联 prompt 变得很容易prompt | llm | parser。简洁、可读适合简单任务。但真实的 AI agent 很少沿着一条直线前进。一个客户支持 agent 会读取消息决定是搜索知识库还是调用 tool在失败时重试并且需要记住完整对话。线性 chain 做不到这些。LangGraph 通过显式 primitives 处理 branching、looping、retry、persistence 和 human-in-the-loop checkpoint让 agent 的行为在每一步都可见。LangChain 的 2026 State of Agent Engineering 报告发现超过 70% 的 production agents 采用 graph structure而不是简单的 linear chain。真实的业务流程很少一路直达终点。下面这六个概念解释了这种 graph structure 实际上是如何工作的。概念 1StateState 是工作流中每一步都可以读取和写入的共享笔记本。在 LangGraph 之前agent state 是分散的有些在变量里有些在 memory 里有些在 conversation history 里。你永远无法确定某个步骤到底知道什么。LangGraph 通过提前让 state 显式化并带有类型来解决这个问题。from typing import TypedDict, Annotated from operator import add class AgentState(TypedDict): question: str # the users input answer: str # what the agent produces messages: Annotated[list, add] # conversation history, grows over timegraph 中的每个 node 都可以查看并修改 state 中的任何字段。只从你实际需要的字段开始。大多数指南会跳过这一点不要一开始就设计一个包含 20 个字段的 state。让需求自然浮现。Annotated[list, add]值得理解。默认情况下当两个 node 更新同一个字段时第二个会覆盖第一个。添加add作为 annotation 会告诉 LangGraph 合并 list而不是替换它们。它适用于 messages 和累积结果。对于当前状态字段例如你正处于哪个步骤使用普通类型即可。初学者常见错误 将带有 usage metadata 的完整 LLM response 存入 state。一个构建文档处理 agent 的团队把原始 LLM response 存在 state 中。到 50 个文档时每个 checkpoint 的 state object 达到了 180KB。Postgres 写入时间上升到 400ms并开始影响响应时间。修复方式是把 state 精简到下游 node 实际需要的内容。概念 2NodesNode 只是一个 Python function。它接收当前 state执行一些工作然后返回它想要更新的字段。就是这样。如果你会写 Python function就能构建 node。调用 LLM、访问 database、修改文本、发起 API call。Python 能做什么它就能做什么。只有一条规则接收 state返回 state updates。from langchain_openai import ChatOpenAI from langchain.schema import HumanMessage llm ChatOpenAI(modelgpt-4o-mini) def answer_node(state: AgentState) - dict: # Reads from state, returns only the fields that changed response llm.invoke([HumanMessage(contentstate[question])]) return {answer: response.content} def refine_node(state: AgentState) - dict: prompt fMake this clearer: {state[answer]} response llm.invoke([HumanMessage(contentprompt)]) return {answer: response.content}初学者常见错误 从 node 返回完整 state。你只需要返回被你修改的字段。返回所有内容会在多个 node 更新重叠字段时导致隐蔽的覆盖 bug。LangGraph nodes 只是 Python functions。这个 framework 比看起来更简单。概念 3EdgesEdges 是连接 nodes 的线路。它们告诉 LangGraph 接下来运行哪个 node。有两种类型。Direct edges 总是走同一条路径当 node A 完成后总是运行 node B。Conditional edges 会根据当前 state 选择去哪里当 node A 完成后检查 state 并做出决定。from langgraph.graph import StateGraph, END graph StateGraph(AgentState) # Add nodes graph.add_node(answer, answer_node) graph.add_node(refine, refine_node) # Direct edge: answer always goes to refine graph.add_edge(answer, refine) # Direct edge: refine goes to END graph.add_edge(refine, END) graph.set_entry_point(answer) app graph.compile()初学者常见错误 忘记END。如果你没有把最后一个 node 连接到ENDgraph 会永远运行等待一个永远不会到来的下一步。这是早期 LangGraph 代码中导致 infinite loops 的最常见原因。概念 4Conditional Edges这就是 LangGraph 真正强大的地方。conditional edge 不是总是前往同一个下一个 node而是检查 state并返回接下来要运行的 node 名称。可以把它想象成铁路道岔。火车是 state。道岔检查 state并把火车送到两条轨道之一。def route_based_on_quality(state: AgentState) - str: # Check the current answer quality if len(state[answer]) 50: return refine # too short, needs more work return done # good enough, finish graph.add_conditional_edges( answer, # from this node route_based_on_quality, # use this function to decide { refine: refine, # if function returns refine, go to refine node done: END # if function returns done, end the graph } )Conditional edges 是 agent 的决策机制。一个 function 检查 state 并返回下一个 node 名称。这就是“我应该使用另一个 tool 还是停止”的实现方式。初学者常见错误 返回 graph 中不存在的 node 名称。错误信息很晦涩排查拼写错误所花的时间会比预期更久。始终确保返回值与你的 edge mapping dictionary 中的 key 完全匹配。概念 5CheckpointingCheckpointing 是 LangGraph 让你的 agent 拥有 persistent memory 的方式。没有 checkpointer 时每次调用app.invoke()都会从头开始。agent 不会记得过去的 sessions。添加 checkpointer 后agent 会在每次 node transition 后保存其 state并按 thread ID 作为 key。下一次使用相同 thread ID 调用时会从上次离开的地方精确继续。一个 production crash 的修复原本可能需要一周时间来编写 custom serialization、Redis state cache 和 session reconstruction function而使用 LangGraph checkpointing 只花了 45 分钟。from langgraph.checkpoint.memory import MemorySaver # dev only # from langgraph.checkpoint.sqlite import SqliteSaver # single-server prod # from langgraph.checkpoint.postgres import PostgresSaver # multi-instance prod checkpointer MemorySaver() app graph.compile(checkpointercheckpointer) # thread_id groups all interactions for one session config {configurable: {thread_id: user-session-42}} # First call: agent runs and saves state app.invoke({question: What is LangGraph?}, config) # Second call with same thread_id: picks up where it left off app.invoke({question: Show me a code example}, config)开发环境使用MemorySaver。单服务器 production 使用SqliteSaver。当你需要多台服务器共享同一份 state 时使用PostgresSaver。初学者常见错误 在 production 中使用MemorySaver。它把所有内容都存储在 RAM 中。服务器一重启所有 agent state 都会消失。概念 6Human-in-the-LoopInterrupts60% 的 production agent systems 添加了人工干预点。它们不是完全自治的 agents而是在关键决策点暂停等待人工确认然后继续运行。LangGraph 通过interrupt_before实现这一点。你指定哪个 node 应该触发暂停。graph 会在进入该 node 之前停止等待人工审核并可选地更新 state然后再恢复。# Compile with interrupt_before to pause before the risky node app graph.compile( checkpointercheckpointer, interrupt_before[send_email] # pause before this node ) config {configurable: {thread_id: task-99}} # Graph runs until it hits send_email, then pauses app.invoke({task: Draft and send a refund email}, config) # A human reviews the draft here, optionally updates state # graph.update_state(config, {draft: Updated email text}) # Resume from where it paused, with human-reviewed state app.invoke(None, config)初学者常见错误 试图用一个 conversation turn 来实现人工审批而不是使用 interrupt。询问 model “我应该继续吗”并信任它的回答并不是 human-in-the-loop。这是在让 agent 批准自己的行动。一个批准自己高风险决策的 agent 并没有受到监督。它只是在表演。这六个概念如何连接在一起完整图景如下决策框架关键要点State 是一个 typed dictionary每个 node 都可以读取和写入。只定义你需要的内容。保持精简。Nodes 是 Python functions。它们接收 state执行工作并只返回被它们修改的字段。Direct edges 总是前往同一个下一个 node。始终把最后一个 node 连接到END。Conditional edges 运行一个 function检查 state并返回下一个 node 的名称。决策就是这样发生的。Checkpointing 会在每个 node 之后保存 state。开发环境使用MemorySaver。production 使用SqliteSaver或PostgresSaver。Interrupts 会在指定 node 之前暂停 graph并等待人工介入。使用app.invoke(None, config)恢复。接下来学什么state、nodes 和 edges 这三部分骨架可以在不改变结构的情况下扩展到 production。为 retrieval 添加更多 nodes你就得到一个 RAG pipeline。为 intent classification 添加一个 routing edge你就得到一个 support router。添加一个 checkpointer你就得到 persistent memory。所有这些都不需要重新思考基本原理。一旦这六个概念变得自然下一层值得理解的是 reducers如何控制多个 node 更新同一个字段时会发生什么、sub-graphs在另一个 graph 内运行一个 graph用于复杂的 multi-agent systems和 streaming在完整 graph 完成之前向用户发送中间结果。但这些都是第二层问题。先熟练构建一个使用上述全部六个概念的 graph。运行它。破坏它。修复它。这个动手循环才是让 LangGraph 的其余部分真正融会贯通的关键。这里给大家精心整理了一份全面的AI大模型学习资源包括AI大模型全套学习路线图从入门到实战、精品AI大模型学习书籍手册、视频教程、实战学习、面试题等资料免费分享扫码免费领取全部内容1. 成长路线图学习规划要学习一门新的技术作为新手一定要先学习成长路线图方向不对努力白费。这里我们为新手和想要进一步提升的专业人士准备了一份详细的学习成长路线图和规划。可以说是最科学最系统的学习成长路线。2. 大模型经典PDF书籍书籍和学习文档资料是学习大模型过程中必不可少的我们精选了一系列深入探讨大模型技术的书籍和学习文档它们由领域内的顶尖专家撰写内容全面、深入、详尽为你学习大模型提供坚实的理论基础。书籍含电子版PDF3. 大模型视频教程对于很多自学或者没有基础的同学来说书籍这些纯文字类的学习教材会觉得比较晦涩难以理解因此我们提供了丰富的大模型视频教程以动态、形象的方式展示技术概念帮助你更快、更轻松地掌握核心知识。4. 2026行业报告行业分析主要包括对不同行业的现状、趋势、问题、机会等进行系统地调研和评估以了解哪些行业更适合引入大模型的技术和应用以及在哪些方面可以发挥大模型的优势。5. 大模型项目实战学以致用当你的理论知识积累到一定程度就需要通过项目实战在实际操作中检验和巩固你所学到的知识同时为你找工作和职业发展打下坚实的基础。6. 大模型面试题面试不仅是技术的较量更需要充分的准备。在你已经掌握了大模型技术之后就需要开始准备面试我们将提供精心整理的大模型面试题库涵盖当前面试中可能遇到的各种技术问题让你在面试中游刃有余。7. 资料领取全套内容免费抱走学 AI 不用再找第二份不管你是 0 基础想入门 AI 大模型还是有基础想冲刺大厂、了解行业趋势这份资料都能满足你现在只需按照提示操作就能免费领取扫码免费领取全部内容