企业官网建设流程全解析

这一章聊聊 LangGraph。做 Agent 工程化的人，大概率都经历过一段迷茫期：模型能力有了，工具也接了不少，但流程一旦复杂起来，要么状态散落得找不着北，要么多轮会话稍一偏移就乱套。LangGraph 的出现，相当于把“写死流程”变成了“画一张可执行的有向图”，让每一步节点、每一条跳转、每一份状态都清清楚楚落在代码里。本文不会给你背一遍文档，而是把我在实际项目里用 LangGraph 搭 Agent 工作流的完整思考拆开，讲清楚它到底解决了什么问题、核心概念怎么用、以及哪些坑我替你踩过了。适合正在做 Agent 落地的开发者，无论是刚入门还是已经在用 LangChain 生态，都能从里面找到能直接抄作业的东西。

1. 为什么我最终选了 LangGraph 而不是手撸状态机

1.1 Agent 编排的三个真实痛点

先说痛点。你如果写过稍微复杂一点的 Agent，一定遇到过这三件事。

第一，多步任务的状态分散。模型生成结果要暂存，工具调用要传参，中间步骤的中间产物散落在各个模块里，全靠一个庞大的 context 对象传来传去。等步骤多了，你根本分不清某个字段到底是在哪个环节被覆盖的，排查起来像考古。

第二，分支跳转难以维护。你今天觉得流程是 A 到 B 到 C，明天产品说要根据用户意图走分支 D，用 if-else 堆确实能写，但一旦分支嵌套三层以上，代码的可读性就崩了，改一处可能要带回滚三天。

第三，人机协同难以切入。真实业务里，Agent 不可能永远全自动跑，经常会遇到“这个工单需要人工审核”“这个回答需要主管确认”的情况。手写状态机时，想在中间暂停、等人确认、再接着跑，实现起来异常痛苦，你会发现自己要额外维护一套“会话暂停和恢复”的机制。

这三个痛点叠加起来，指向同一个结论：Agent 工作流缺的不是代码能力，而是建模能力。你得先想清楚“这一步之后可能去哪几边走”，再去谈怎么写功能。

1.2 图编排方案对比：LangGraph、AutoGen、CrewAI 与手写状态机

当时我在选型时，把主流方案都过了一遍。

AutoGen 主打多 Agent 对话，它的核心是让多个智能体互相聊，系统自动协调。这个思路在探索型任务里很好用，但问题在于行为不可预测，你很难约束它“必须先查知识库再回答”。它的对话驱动模型也不适合固定业务流。

CrewAI 则强在角色分工和任务委派，如果你需要的是一组 Agent 按角色协作，它上手很快。但它的抽象层级较高，你要精确控制每个节点在什么条件下走向哪条路径时，会感到明显的阻力。

手写状态机当然能实现，我之前项目里也这么干过。但问题在于状态机的逻辑分散在事件处理函数里，当流程变得复杂，光把图结构和业务逻辑分开维护就很费劲。更麻烦的是，图形化想清楚之后，你还要手动实现状态持久化、回滚、断点恢复，这些底层基建非常繁琐。

LangGraph 的定位恰好补上了这块空白。它直接用图来建模工作流，节点是 Python 函数，边是路由逻辑，状态作为唯一的传递中枢贯穿全程，还内置了检查点机制和人工介入能力。这套设计正好命中了我上面说的三个痛点。

1.3 “状态图”的天然优势

LangGraph 的核心抽象叫做 StateGraph，翻译过来就是状态图。它跟传统状态机的最大区别在于：状态机的“状态”是离散的、有限的，适合描述设备开关这种明确场景；而 LangGraph 里的“状态”是一个持久的数据结构，允许你在任意节点读取、修改、部分更新。它更像一个贯穿整个执行过程的“黑板书”，所有节点都可以往上写内容，然后下一个节点接着读。

这个设计带来两个直接好处。第一，数据流天然清晰。你只要看状态定义，就知道整个工作流要处理哪些数据，节点的输入输出都围绕状态展开，不会出现散落的全局变量。第二，流程图和代码一一对应。你在设计文档里画出的方框箭头，在代码里就是 add_node 和 add_edge，产品同学和开发同学沟通时，拿图就能对齐，不用再对着代码脑补流程。

我用下来最明显的感觉是：以前做完一个 Agent，我心里没底，不知道它在极端情况下会走到哪条分支；用 LangGraph 之后，所有可能路径都摆在图上，分析问题时可以直接从“走到哪个节点了”开始排查，定位效率提高了很多。

2. 核心概念拆解：把工作流拆成一张可执行的有向图

2.1 State：贯穿全程的中枢缓存

State 在 LangGraph 里是一切的基础。它本质上是一个你定义的 TypedDict 或 Pydantic 模型，里面包含需要在整个工作流中传递的字段。

from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class WorkflowState(TypedDict): user_input: str history: Annotated[list, add_messages] # 自动追加消息 search_results: list final_answer: str

这里有三个细节值得注意。

一是字段之间相互独立，节点根据业务需求选择更新哪些字段，未更新的字段保持不变。这非常关键，你不需要在每次节点返回时把所有字段都带上。

二是 Annotated 和 reducer 的搭配。如果你希望某个字段是累加式的，比如多轮对话记录，需要给这个字段绑定一个 reducer（比如 add_messages），LangGraph 在收到节点返回值后会用 reducer 把新值和旧值合并，而不是直接覆盖。

三是“约定优于配置”之外，其实也暗含了约束。建议尽量把 State 定义得精简，别把临时计算结果都塞进去。我在最初设计时习惯把所有中间结果全部放进 State，后来发现状态越来越大，排查问题时看整个状态都要翻好几页。最佳实践是只保留节点之间需要传递的数据，临时变量放在节点内部处理完就销毁。

2.2 Node：一个节点就是一段可复用的逻辑

Node 是图里的执行单元，它就是一个普通的 Python 函数。输入是当前 State，返回值是一份部分更新的状态字典。

def search_node(state: WorkflowState): query = state["user_input"] results = do_search(query) # 你封装的搜索函数 return {"search_results": results}

节点函数不需要感知图的全局结构，它只负责“拿到需要的状态，处理，返回更新”。这种低耦合设计让节点可以单独测试、单独复用。我在实际项目里，采购了专门的测试脚本对单个节点做单元测试，确保搜索节点、重写节点各自独立正确，再组合到图里跑，排查问题的范围大幅缩小。

一个容易忽略的细节是：节点里尽量避免做一个异步长任务阻塞整条链路。如果某个节点要调用外部 API 且耗时较长，建议在节点内部做好超时控制、重试机制和降级策略。毕竟 LangGraph 本身不负责网络层面的可靠性，它只负责调度。

2.3 边与条件分支：图的“路标”

有节点还不够，你得告诉 LangGraph 执行完 A 之后去哪个节点。这就是边的作用。

普通边表示无条件跳转，比如执行完分类节点之后固定进入查询节点；条件边则根据当前的 State 动态决定走哪条路径。

from typing import Literal def route_after_classify(state: WorkflowState) -> Literal["rag", "human"]: if state.get("need_human"): return "human" return "rag"

然后把它接到图里：

builder.add_conditional_edges( "classify_node", route_after_classify, {"rag": "rag_node", "human": "human_node"} )

条件边是 LangGraph 最核心的进阶用法。它让你把“业务的判断逻辑”单独抽成一个函数，这个函数本身也是可以单测的。我经常先定义一个恒假的条件边跑通主流程，再逐步把真实路由逻辑加进去，这样每一步出问题都能定位到是路由问题还是节点问题。

2.4 Checkpointer：断点保存与记忆恢复

Checkpointer 是 LangGraph 区别于普通编排框架的重型武器。

它能在每次节点执行之后保存图的完整状态，并关联到某个会话标识上。这样即使进程重启，只要拿到相同的会话 ID，就能从保存的位置恢复执行。它带来的能力非常实用：

• 中断恢复：流程中途发生异常，恢复后从上次保存点继续，而不是从头开始。
• 多轮会话：一个 Agent 服务多个用户时，用 thread_id 区分不同会话，互相不干扰。
• 时间旅行：通过读取历史状态快照，可以查看某一轮执行结束时的完整上下文。

from langgraph.checkpoint.memory import MemorySaver memory = MemorySaver() app = builder.compile(checkpointer=memory)

调用时只要带上 config：

config = {"configurable": {"thread_id": "user_001"}} result = app.invoke({"user_input": "你好"}, config)

我用 MemorySaver 做过原型，线上则把它替换成持久化存储方案。需要提醒的是，MemorySaver 只是存在内存里，进程退出就没了，生产环境一定要接数据库或者对象存储。

3. 实操：构建一个可控的多工具客服工单 Agent

3.1 场景定义与状态设计

理论讲完，我们来做个实际的东西。假设要为一家电商平台做一个客服工单 Agent，目标是根据用户留言自动分类，并决定走知识库回复还是转人工，最终给出一个可落地的回复方案。

这个场景足够典型，它既有分类判断，又有分支路由，还带有“转人工”这种典型的人机协同需求。

先定义状态：

from typing import TypedDict, Annotated, Literal, List from langgraph.graph.message import add_messages class TicketState(TypedDict): ticket_id: str # 工单号 user_message: str # 用户原始留言 messages: Annotated[list, add_messages] # 对话历史 category: str # 分类结果 need_human: bool # 是否需要人工 kb_hits: List[str] # 知识库命中结果 final_reply: str # 最终回复内容

这里我把user_message单列，同时保留messages作为对话历史。原因很简单：用户留言需要被多个节点读取，直接state["user_message"]取，比从 messages 里翻最后一条要直观得多。messages字段我留给后续多轮交互时累加用。

3.2 节点实现与条件路由

图里需要四个节点：入口分类节点 classify_node、知识库查询节点 rag_node、人工处理节点 human_node、回复生成节点 reply_node。

分类节点用大模型判断这条留言属于哪个类别以及是否需要人工介入：

def classify_node(state: TicketState): prompt = f"对以下用户留言分类，并判断是否需要人工处理。\n留言：{state['user_message']}" resp = llm.call(prompt) category, need_human = parse_classify_result(resp) return {"category": category, "need_human": need_human}

rag 节点做知识库检索：

def rag_node(state: TicketState): hits = knowledge_base.search(state["category"], state["user_message"], top_k=3) return {"kb_hits": hits}

human 节点的核心逻辑不是做自动化回复，而是把工单置为待人工状态并通知值班人员：

def human_node(state: TicketState): notify_staff(state["ticket_id"], state["user_message"], state["category"]) return {"final_reply": "已转人工处理，请稍候，客服人员将尽快联系您。"}

路由函数：

def route_after_classify(state: TicketState) -> Literal["rag", "human"]: return "human" if state["need_human"] else "rag"

核心判断逻辑被抽成了一个纯函数，哪怕你不想接 LangGraph，直接单元测这个函数也行。

3.3 编译与调用：invoke、stream、断点续跑

节点和边都定义好后，组装图：

from langgraph.graph import StateGraph, START, END builder = StateGraph(TicketState) builder.add_node("classify", classify_node) builder.add_node("rag", rag_node) builder.add_node("human", human_node) builder.add_node("reply", reply_node) builder.add_edge(START, "classify") builder.add_conditional_edges("classify", route_after_classify) builder.add_edge("rag", "reply") builder.add_edge("human", END) builder.add_edge("reply", END) app = builder.compile(checkpointer=MemorySaver())

执行时，invoke一跑到底，适合离线任务；stream则按节点产出中间结果，适合需要实时展示给用户的场景。

config = {"configurable": {"thread_id": "ticket_10086"}} for chunk in app.stream( {"user_message": "我在订单里看到一个很大的破损痕跡，想申请退款，但找不到售后入口。"}, config, stream_mode="updates" ): print(chunk)

stream_mode="updates"返回的是每个节点产生的状态更新片段，你可以直接把它推到前端做流式展示。注意，这些更新是“片段”而不是完整的最终状态，如果前端需要完整上下文，建议再单独 get_state 拉取。

如果想让流程在某个节点前暂停，比如在human_node之前强制插入审批：

app = builder.compile( checkpointer=MemorySaver(), interrupt_before=["human"] )

第一次 invoke 会停在 human 节点前，确认之后再下发继续执行的命令。这个机制是 LangGraph 做 Human-in-the-loop 的标准姿势，也是它能胜任业务系统集成的关键能力。

3.4 断点续跑和时间旅行

断点续跑在客服场景里有非常实际的价值。比如某天凌晨系统执行到一半，数据库连接断了，进程直接崩溃。如果用惯了普通的函数调用链，你的第一反应一定是“完蛋了，这个工单处理到哪一步了”。但在 LangGraph 里，只要 checkpointer 有持久化存储，进程恢复后你只要去查历史状态，就能知道这个工单卡在哪个节点、当时的状态是什么、然后决定是继续跑，还是修正状态后重新执行。

state = app.get_state(config) print(state.next) # 查看下一步要执行什么

如果觉得当时的某步结果不对，还可以用update_state修正某个字段之后重新恢复：

app.update_state(config, {"kb_hits": correct_hits}) app.invoke(None, config)

这比删掉整个会话重跑要优雅得多，在真实生产环境里这能救你很多次。

4. 常见问题与排查心得

4.1 状态无故丢失或互相覆盖

这是刚开始用 LangGraph 最容易踩的坑。症状是：某节点明明返回了字段 A，但到下一个节点读出来却是空。排查时先怀疑两件事。

第一，State 定义上字段类型是否正确。Python 是动态类型，TypedDict 只是在类型层面做提示，真正的运行时行为取决于节点返回的字典到底带什么 key。如果某个节点返回了state，但你的状态定义里没有这个 key，刷新状态时这个值会被静默丢弃。

第二，更新语义是覆盖还是合并。字段如果没绑 reducer，新值会整体覆盖旧值；字段绑定了 reducer，LangGraph 会调用 reducer 把新旧值合并。很多问题都源于误以为“节点返回值会自动追加”，结果旧值被覆盖了。

我在项目里定了一条规矩：State 里每个字段都必须明确标注更新语义，是“覆盖”还是“累加”，写进代码注释里。这样团队成员维护起来也不会纠结。

4.2 条件边返回的路径不在映射表里

你定义一个条件边函数，返回一个字符串，但忘了在映射表里注册这个字符串对应的目标节点。运行时会直接报错，但报错信息通常不会特别明显，只提示找不到目标。

我的排查经验是：先看路由函数的返回值类型，用Literal约束能避免大半问题；再看add_conditional_edges第三个参数传的映射字典是否覆盖了所有返回值。另外一个隐蔽的问题是节点 id 拼写，add_node 的字符串标识和 add_edge 里引用的标识必须完全一致，稍微多个空格都会出问题。

4.3 长流程超时与资源泄漏

LangGraph 本身不限制单次运行的时长。如果你的某个节点需要调一个大模型三次，单次模型调用30秒，那这个节点就可能卡一两分钟。在同步 HTTP 场景下，客户端早就超时断了，但后端图还在继续跑。

这里有两个层面的解法。请求层，把 invoke 放到异步任务队列，客户端通过轮询或 webhook 获取结果。框架层，在每个节点内部增加超时控制和失败降级。比如模型调用失败就返回兜底文案并标记错误，而不是让异常往上抛导致整张图中断。

我还习惯给复杂工作流加 max_steps 之类的总超时，避免出现死循环走到天荒地老。

4.4 恢复会话后上下文错乱

这是多轮会话场景的常见问题。现象是：用户会话恢复后，模型对上下文的记忆是断层的，可能只拿到了最后一轮，也可能拿到了好几轮之前的状态。

排查时盯住两个变量：一是 thread_id，恢复会话时 thread_id 必须和保存时完全一致，否则 Checkpointer 会在新的会话分支上保存，读不到旧状态。二是你定义 State 时给 messages 字段绑的 reducer，必须是add_messages而不是add之类的自定义函数。

历史上我见过一个案例，某开发者在 messages 字段上绑了一个简单 list.extend 函数，结果多轮对话里模型无法区分哪条是用户哪条是助手，上下文混乱。换回 add_messages 之后立刻正常，因为这个 reducer 本身就把消息角色和顺序一起管理了。

4.5 常见问题速查表

5. 一点个人心得

我在实际项目中彻底切换到 LangGraph 之后，最深的感受是：编排框架带来的不只是一堆 API，而是一种思考方式的改变。以前写 Agent，我脑子里的图是模糊的，边走边看。现在写 Agent，我第一件事就是打开画图工具，把节点、边、状态一条条画出来，画清楚了再开始写代码。

这个习惯帮我提前发现了至少三分之一的设计问题。比如你画图的时候就会发现：分类节点之后其实有三条路，而不是两条，你还要考虑“无法分类”这种情况怎么办。如果直接写代码，这种分支很容易在事后被硬塞进去，然后破坏原有的状态结构。

如果你现在正在做 Agent 落地，我对你的建议是：先从一个小而完整的场景开始，比如一个 3 节点的工作流，把 State 设计、条件边、Checkpointer 全走一遍，再去处理复杂业务。不要上来就铺开十几个节点，图越大，调试越痛苦。等你对节点的边界、状态的定义、路由函数的职责都心里有数了，LangGraph 的威力才会真正释放出来。