企业官网建设流程全解析

LangGraph 这个库这两年讨论度很高，但我在社区里看过不少人的练习路径有个共同问题：一上来就追着多代理架构跑，结果图是搭出来了，State 却改得乱七八糟，一加分支就崩。我自己练 LangGraph 的顺序和别人不太一样，最先啃的就是状态代码——State 的定义、节点返回值怎么合并、Reducer 怎么控制更新、条件分支里状态怎么传。这套练扎实之后，再看那些多节点、带分支、带记忆的流程，基本就通了。这篇文章就把我完整的状态代码练习过程记录下来，环境基于我常用的较新版 LangGraph，适合刚接触这个库、以及已经会搭图但对状态更新机制一知半解的同学。

1. 先把状态模型这层窗户纸捅破

1.1 状态到底是什么、为什么值得专门练

LangGraph 的核心是一个状态图：节点（Node）负责干活，边（Edge）决定顺序，而贯穿所有节点之间、让每个节点都能读到上下文的东西，就是 State。State 本身就是一个字典结构的对象，你用 TypedDict 或 dataclass 提前定义好它长什么样，然后每次节点执行完后，把“想改的字段”做成一个字典返回，图的运行时负责把这些改动合并回总状态里。

我说的“练状态代码”，练的就是这三件事：第一，怎么定义一份合理的状态结构；第二，节点返回什么样的数据、以什么方式合并进总状态；第三，在分支、循环、并行节点里，状态怎么传递不丢不串。这三件事任何一个没搞明白，图都能跑，但跑出来的结果就是不对，而且特别难排查。

为什么说这个值得单独练？因为很多传统开发者的直觉是“函数输入输出”。一个函数处理完数据，把结果返回给下一个函数，参数一颗颗传下去。这套思路在写普通后端接口时没问题，但一旦涉及多步骤的智能体流程——比如先规划、再检索、再生成、中途还要根据结果决定走哪条分支——函数式的传参会让代码变成一团乱麻。State 本质上是一块“公共白板”，所有节点都能往上写、都能往里读，你不需要在调用时手动把每一步的临时变量传来传去。这个模型想通了，LangGraph 就算入门了。

1.2 从“接力棒”到“公共白板”的思维切换

先看一个典型的旧思路代码：

def step1(raw_text: str) -> str: cleaned = raw_text.strip().lower() return cleaned def step2(cleaned: str) -> list[str]: tokens = cleaned.split() return tokens result = step2(step1(" Hello LangGraph "))

每一步函数都只接收上一个函数的返回值，如果你想在最后一步同时拿到 raw_text、cleaned、tokens，就得用元组或者类把这些都包着往下传。代码一多，参数列表就失控了。

换成 LangGraph 的状态模型，写法变成了这样：

from typing import TypedDict from langgraph.graph import StateGraph, START, END class MyState(TypedDict): raw_text: str cleaned: str tokens: list[str] def step1(state: MyState) -> dict: return {"cleaned": state["raw_text"].strip().lower()} def step2(state: MyState) -> dict: return {"tokens": state["cleaned"].split()} builder = StateGraph(MyState) builder.add_node("step1", step1) builder.add_node("step2", step2) builder.add_edge(START, "step1") builder.add_edge("step1", "step2") builder.add_edge("step2", END) graph = builder.compile() print(graph.invoke({"raw_text": " Hello LangGraph "}))

注意 step2 里读的state["cleaned"]，并不是 step1 直接传给你的参数，而是 step1 写好、图运行时合并进 State 之后的公共数据。这就像一个团队共用的项目白板，每个人干完活把自己负责的那部分更新上去，下一个同事需要什么直接看白板，不需要等某个人把结果递到手里。

你可能觉得这个例子太简单，看不出差别。但你要是在这个图里再加一个“日志收集节点”、加一个“异常处理分支”，State 的优势马上就出来了——日志往logs字段里累积，分支判断读state["status"]，每个节点只需要关心自己读写的那几个字段，不要碰自己不关心的东西。这就是状态驱动设计的核心思想。

2. 第一个状态练习：最小可运行图

2.1 把环境准备好

LangGraph 是纯 Python 库，安装很简单：

pip install langgraph

建议 Python 版本 3.9 以上，另外如果你计划跑带持久化的例子，还需要确认langgraph-checkpoint已经作为依赖装好。我练习时习惯在普通 Python 脚本里写，因为脚本跑完能直接看到 print 输出，比在交互式环境里更容易保留完整记录。

注意：LangGraph 的 API 在 0.1 到 0.2 之间有过调整，比较明显的变化是入口图用START而不是"__start__"。我这篇文章的代码基于 0.2.x 及以上版本，如果你还在用很老的版本，建议先升级到最新稳定版再照着练。

2.2 用 TypedDict 把状态结构先钉死

状态结构是一张图的“数据库表结构”，先想清楚再写节点。我第一次练习时直接上来写节点函数，结果反复改字段名，图一跑就 KeyError。后来我学乖了：任何练习的第一步，永远是定义状态。

from typing import TypedDict class CounterState(TypedDict): count: int logs: list[str]

这个状态有两个字段：count用来记录数值，logs用来记录每一步的处理日志。就这么简单，但注意一个细节：字段的类型要写具体，int、list[str]都明确标注，因为后面写节点函数时 IDE 会帮你做提示，字段名拼错时也能更快发现。

为什么用 TypedDict 而不是普通字典？因为普通字典太自由了，节点里写state["cnt"]和state["count"]都不会报错，运行时才会炸；TypedDict 至少能在编辑器里给你一层提示和约束。LangGraph 本身也支持 pydantic 模型或 dataclass 作为状态定义，但初学者从 TypedDict 起步最合适，它足够轻量，心智负担最小。

2.3 用三个节点跑通完整链路

定义好状态之后，我加三个节点：一个加一、一个翻倍、一个收尾记录。重点看每个节点怎么返回“更新补丁”。

from typing import TypedDict from langgraph.graph import StateGraph, START, END class CounterState(TypedDict): count: int logs: list[str] def add_one(state: CounterState) -> dict: new_count = state["count"] + 1 return {"count": new_count, "logs": state["logs"] + [f"add_one -> {new_count}"]} def double(state: CounterState) -> dict: new_count = state["count"] * 2 return {"count": new_count, "logs": state["logs"] + [f"double -> {new_count}"]} def final_note(state: CounterState) -> dict: return {"logs": state["logs"] + [f"final count is {state['count']}"]} builder = StateGraph(CounterState) builder.add_node("add_one", add_one) builder.add_node("double", double) builder.add_node("final", final_note) builder.add_edge(START, "add_one") builder.add_edge("add_one", "double") builder.add_edge("double", "final") builder.add_edge("final", END) graph = builder.compile() result = graph.invoke({"count": 1, "logs": []}) print(result)

运行结果：

{'count': 4, 'logs': ['add_one -> 2', 'double -> 4', 'final count is 4']}

我建议你亲手跑一遍这个例子，重点观察两件事。第一，invoke的入参是“初始状态”，返回值是“最终状态”，中间每个节点产生的新状态你是看不见的，除非你在节点里打印。第二，节点返回的字典不需要包含所有字段，比如double节点完全可以不返回logs（原样保留），也可以像我这样手动拼接后返回。这里有个初学者最容易踩的坑：以为节点返回什么，状态就“变成”什么；实际上对于没有配置特殊合并逻辑的字段，返回的 value 会直接覆盖旧值。

2.4 在节点里埋点观察状态流转

光看最终结果，你可能还是不知道状态到底怎么变的。我的练习习惯是在每个节点开头打印当前状态：

def double(state: CounterState) -> dict: print(f"[double] 进入时状态: {state}") new_count = state["count"] * 2 return {"count": new_count, "logs": state["logs"] + [f"double -> {new_count}"]}

这样跑一次就能看到每条边触发时，状态里已经累积了什么、当前节点准备改什么。等你以后排查“哪个节点把状态改坏了”时，这个习惯能帮你省下一大半时间。

提示：compile()之后得到的是一个可调用的编译对象。你可以反复调用invoke，每次传入新的初始状态，互不干扰。这跟“同一个图的多次运行之间默认没有记忆”这一特性有关，后面讲 Checkpointer 时再展开。

3. 状态更新的核心：Reducer 机制一定要亲手试

3.1 默认覆盖行为：先知道“什么都不写”时是什么效果

在 LangGraph 里，如果你在状态定义里对某个字段没做任何标注，那它的默认更新方式就是“整体覆盖”。两个节点不管谁先谁后，只要都返回同一个字段名，后执行的节点就会把前一个节点的值整个替换掉。

这个行为本身没问题，count这种标量字段用覆盖就很合理——最后一步的值就是我想要的值。但一旦字段是列表、字典这种“需要累积”的数据，覆盖行为就会让你抓狂。比如上面例子里如果我不手动把state["logs"] + [新日志]拼好再返回，而是直接返回{"logs": ["新日志一"]}，那跑完后logs里只剩最后一个节点的记录，前面的全丢了。

所以你需要理解 Reducer：它决定了“多个更新同时作用到同一个字段上时，怎么合并”。没有 Reducer，就是后写覆盖；有 Reducer，就能做到追加、拼接、去重等自定义合并。

3.2 用 add_messages 维护对话消息列表

聊到 LangGraph 的红区，最常用的就是add_messages。它原本是给聊天消息设计的 Reducer，但理解它的行为能帮你更好地理解所有 Reducer 的运作方式。

from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class ChatState(TypedDict): messages: Annotated[list, add_messages] intent: str def user_turn(state: ChatState) -> dict: return {"messages": [{"role": "user", "content": "我想退款"}]} def assistant_turn(state: ChatState) -> dict: return {"messages": [{"role": "assistant", "content": "好的，我来帮您处理"}]}

这里的关键是Annotated[list, add_messages]。Annotated是 Python 自带的类型标注工具，LangGraph 会读取它的第二个参数作为这个字段的 Reducer。add_messages的作用是：如果新旧值都是消息列表，就把它们合并；如果存在带id的消息对象，还能按id去重更新。

你不需要手动把历史消息拼进返回值，只要返回“这一次新增的消息”，运行时就会自动追加到状态里的messages上。这就是正确用法：节点只返回增量，合并交给 Reducer。

3.3 自定义 Reducer：写一个属于自己的合并规则

add_messages 好用，但它解决不了所有问题。比如你希望某个字段做“取最大值”“数值累加”“字典合并”，那就得自己写 Reducer。Reducer 的签名很固定：接收两个参数，第一个是当前状态里已有的值，第二个是节点返回的更新值，返回合并后的新值。

from typing import TypedDict, Annotated, Annotated def sum_reducer(current: int, update: int) -> int: return current + update class ScoreState(TypedDict): total: Annotated[int, sum_reducer] name: str def node_a(state: ScoreState) -> dict: return {"total": 10} def node_b(state: ScoreState) -> dict: return {"total": 5}

这两个节点无论怎么排序执行，最终total都是 15 而不是“后写的 5”。因为每个节点返回的数值都会通过sum_reducer与当前值合并，顺序是先加 10 再加 5，或者反过来。

这里有一个特别重要的细节：Reducer 是一对一合并的，也就是说它每次只处理“一个更新值”和“当前累积值”。所以在并行分支场景下，你不能在节点里返回一个“已经包含旧值的完整快照”，否则会把已有内容再算一遍，导致重复累加。正确的做法永远是：返回增量，让 Reducer 去做累积。

# 错误示范：节点里自己拼了历史 def wrong_node(state: ScoreState) -> dict: return {"total": state["total"] + 100} # 如果配了 sum_reducer，这里最终会多加一次 # 正确示范：只返回增量 def right_node(state: ScoreState) -> dict: return {"total": 100} # sum_reducer 自己处理累加

我把这个坑单独拿出来讲，是因为它在真实项目里特别常见。凡是配了 Reducer 的字段，节点里的返回值含义就变了：不再是“最终值”，而是“这次贡献的更新量”。这个思维转变是练状态代码最核心的一关。

4. 状态驱动的条件分支与多路汇合

4.1 条件边的路由约定

LangGraph 最吸引人的能力就是条件分支：根据当前状态的内容，决定下一步走哪个节点。实现方式是用add_conditional_edges挂一个路由函数。

路由函数接收当前状态，返回一个节点名字符串，或者返回一个能被映射表翻译的 key。我推荐直接返回节点名字符串，少一层映射，少踩一个坑。

def route(state: ChatState) -> str: if state["intent"] == "refund": return "refund" if state["intent"] == "tech": return "tech" return "human"

然后这样挂边：

builder.add_conditional_edges("classify", route)

运行时会在classify节点执行完毕后，根据返回的字符串决定下一步进哪个节点。返回的名字必须在图里已经注册过，否则会报错说找不到节点。

4.2 一个带分支的完整练习：意图路由客服系统

把前面的内容串起来，我练过一个完整的小例子，模拟一个客服入口按意图分流到不同处理节点：

from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages class SupportState(TypedDict): messages: Annotated[list, add_messages] intent: str handled: bool def classify(state: SupportState) -> dict: text = state["messages"][-1]["content"] if "退" in text: intent = "refund" elif "卡" in text or "慢" in text: intent = "tech" else: intent = "human" return {"intent": intent} def refund_handler(state: SupportState) -> dict: return {"messages": [{"role": "assistant", "content": "退款流程已启动"}], "handled": True} def tech_handler(state: SupportState) -> dict: return {"messages": [{"role": "assistant", "content": "已创建技术支持工单"}], "handled": True} def human_handler(state: SupportState) -> dict: return {"messages": [{"role": "assistant", "content": "转接人工客服"}], "handled": True} def route(state: SupportState) -> str: return state["intent"] # 直接返回节点名 builder = StateGraph(SupportState) builder.add_node("classify", classify) builder.add_node("refund", refund_handler) builder.add_node("tech", tech_handler) builder.add_node("human", human_handler) builder.add_edge(START, "classify") builder.add_conditional_edges("classify", route) builder.add_edge("refund", END) builder.add_edge("tech", END) builder.add_edge("human", END) graph = builder.compile() result = graph.invoke({"messages": [{"role": "user", "content": "我想退款"}], "intent": "", "handled": False}) print(result)

注意这里route直接返回state["intent"]，而intent是classify节点写进状态里的。所以路由的决策完全依赖状态里的字段，这就是“状态驱动”四个字的含义。我建议你把输入改成“网络很卡”和“我要投诉”各跑一遍，观察它走的分支完全不同，但代码里没有写死任何顺序。

4.3 分支汇合时的状态同步细节

很多时候，分支结束后还要汇合到一个公共节点继续处理。比如客服所有分支最后都要进入一个“记录日志”节点。汇合本身不复杂，只要在每条分支末尾都连到同一个节点就行：

builder.add_edge("refund", "record_log") builder.add_edge("tech", "record_log") builder.add_edge("human", "record_log")

但这里有个非常隐蔽的状态问题：如果多个分支是并行执行的（比如用add_edge让两个节点同时指向同一个下游），它们会同时写同一个状态字段。对于配了 Reducer 的字段（比如messages），更新会按顺序合并，不会丢；但对于没配 Reducer 的字段，比如普通int或者str，后写完的覆盖先写完的，最终结果取决于执行顺序，而这个顺序在并行场景里是不确定的。

所以我的经验法则是：并行分支里，尽量让每个分支写自己的独立字段（refund_result、tech_result），不要在多个分支里改同一个无 Reducer 的字段。如果确实要汇总，就定义一个results字段配上 Reducer，让各方把结果塞进去再统一处理。这个设计思路我在真实项目里用得很频繁，能避免掉 80% 的“状态被莫名覆盖”问题。

5. 状态练习中的常见坑与排查实录

5.1 状态没更新？先按这三个方向查

我在练习状态代码时，最常碰到的问题就是“我明明 return 了，为什么状态没变”。排查顺序基本固定：

• 第一，检查返回字典的 key 拼写有没有跟状态定义里完全一致，包括大小写和下划线。TypedDict 在编辑器里会提示，但你如果用动态字符串构造字典，提示就失效了，必须肉眼核对。
• 第二，检查节点函数是否真的执行了 return。很多人写节点时先写了一段打印逻辑，最后忘了 return，函数默认返回 None，LangGraph 会把它当成“没有更新”，状态自然原封不动。
• 第三，检查这个字段是否配了 Reducer。如果配了累加型 Reducer，而你想“设置新值”，返回单个值的结果可能跟你预期不一样。比如total配了sum_reducer，某个节点想重置为 0，简单返回{"total": 0}并不会变成 0，而是继续累加。

排查技巧也很直接：在每个节点开头打印当前状态、return 前打印准备返回的更新值，跑一遍输出就能定位是哪一层的问题。别靠猜，状态图一旦节点多了，靠猜是猜不出来的。

5.2 Reducer 的“双刃剑”效应

给字段加 Reducer 能解决累积问题，但也会带来意想不到的副作用。最典型的是：你在开发初期给messages加上了add_messages，后期某个节点想“清空消息列表重新开始”，结果发现清不掉，因为任何返回的列表都会跟现有列表合并。

解决这类问题，我常用的办法有三个：一是换一个字段名来承载“重置后的新列表”，让旧列表留在状态里但不再被读取；二是在 Reducer 里加入特殊标记值，比如检测到更新值是某个特定字符串时直接替换而不是合并；三是干脆把“重置”也做成一个明确的业务节点，它直接返回一个全新的消息列表，并且你通过设计让它的上游节点不再往同一个字段写增量。

这里面没有银弹，关键是想清楚：一旦某个字段配置了 Reducer，全图所有节点对这个字段的更新语义都已经变了，写节点时必须时刻记住这一点。我在团队里跟人协作时，经常在状态定义旁边写清楚每个字段的合并语义，避免别人把普通字段当成覆盖字段来写。

5.3 TypedDict 的类型边界与运行时校验

TypedDict 虽然在定义阶段给了你类型提示，但它的约束主要停留在编辑器层面。LangGraph 在运行时对状态的校验深度有限，尤其是嵌套结构，比如state["profile"]是一个 dict，里面少了一个 key 时，TypedDict 不一定能拦住你，等到节点里访问时才报 KeyError。

我的建议是：状态结构尽量保持扁平。字段就是一层，每个字段的值要么是标量，要么是列表，要么是明确类型的数据对象。如果确实有复杂的结构化数据，用 dataclass 或 pydantic 模型作为字段类型，在节点边界做显式校验。不要试图在状态里塞三层以上的嵌套字典，那会让整个图的可排查性变得非常差。

这里还有一个相关细节：LangGraph 对“节点返回了状态定义里不存在的 key”这件事，在不同版本里处理方式不完全一致。我建议永远不要依赖旧版本的宽松行为，返回的 key 必须都定义在状态结构里，否则直接视为实现错误，尽早暴露。

5.4 Checkpointer 相关：thread_id 不配就白改

状态练习做到后面，你一定会碰到持久化和多轮对话的需求。LangGraph 的记忆机制依赖 Checkpointer，当你把图编译成带记忆的版本后，调用方式会发生变化：

from langgraph.checkpoint.memory import MemorySaver saver = MemorySaver() graph = builder.compile(checkpointer=saver) config = {"configurable": {"thread_id": "thread-1"}} graph.invoke({"messages": [{"role": "user", "content": "第一轮"}]}, config) graph.invoke({"messages": [{"role": "user", "content": "第二轮"}]}, config)

没有传config的话，每次invoke仍然是从干净的初始状态开始跑，之前写入状态的内容不会被保留。我见过不少人配了 Checkpointer 却忘了传thread_id，然后来问“为什么我的图没有记忆”，问题就出在这。

另外，同一个thread_id的多次调用之间，状态是累积的。你想在调试时重新开始，就换一个新的thread_id，或者用graph.get_state(config)查看当前状态、用graph.update_state(config, {...})手动修改某一步的状态。这些 API 都是练状态代码时迟早要碰的，我建议你在基础练习通过后，专门花半天时间只练 Checkpointer 相关的操作。

6. 我的练习心法与最后几条建议

6.1 按什么顺序练，效率最高

如果让我给一个建议的练习顺序，我会把它分成四步：第一步，只练三个节点的直线图，把invoke的输入输出彻底搞清楚，重点理解“节点返回的是更新补丁”这件事；第二步，练习各种 Reducer，自己写一个累加器、一个add_messages的聊天累积、一个取最大值的合并函数，把Annotated的用法吃透；第三步，练条件分支，从最简单的二分路由开始，慢慢加分支、加汇合；第四步，练带 Checkpointer 的多轮对话，理解thread_id和状态持久化的关系。

每完成一步，就写一个独立的小脚本保存下来，脚本里加几个断言，判断输出的状态是否符合预期。比如跑完CounterState后断言result["count"] == 4。这样你后面改动状态结构时，一跑就能知道有没有破坏旧行为，非常值得。

6.2 最值得带进真实项目的一个习惯

最后分享一个我个人非常受用的习惯：每个状态字段的定义旁边，用注释写清楚它的更新语义。比如：

class OrderState(TypedDict): step: str # 覆盖，后写为准 logs: Annotated[list, operator.add] # 累加，各节点只追加增量 messages: Annotated[list, add_messages] # 聊天消息自动合并

这短短几行注释，能避免掉大量的“跟我预期不一样”的 bug。状态代码本来就是 LangGraph 项目里最容易被忽视、却最影响全局稳定性的部分，前期多花一个下午把状态模型吃透，后面搭真正的多代理应用时，你会感谢自己当初没有跳过这一步。