企业官网建设流程全解析

说实话，真正让我决定翻开LangGraph官方文档的，不是因为它火了，而是因为我手头一个AI应用被“链式调用”折磨到想重写整个架构。当时的需求其实很简单：用户提交一个主题，系统先做资料调研，再生成文章大纲，接着写正文，最后还要过一个质量审核关卡，审核不过就得打回重写。用当时熟悉的LangChain表达式链来写，代码里塞满了各种if/else和循环控制，流程稍微调整一下，牵一发动全身。直到我把这个流程改成一张图——节点是步骤，边是跳转规则，所有数据挂在同一个State上——整个逻辑突然就清晰了。这个思路，正是LangGraph这个AI应用开发框架的核心。

如果你也在做Agent应用、多步骤工作流，或者需要“人机协作”的半自动流程，这篇应该对你有用。我会从框架要解决的问题讲起，拆透StateGraph的设计逻辑，再给一个可以直接跑的多Agent协作Demo，最后分享那些只有写代码写多了才会踩到的坑。

1. 从链式调用到图编排：LangGraph想解决的问题

1.1 我对“模型调用”和“应用流程”理解的一次转变

先说说我走过的弯路。我接触大模型应用开发，是从最简单的“输入提示词-输出结果”开始的，那时候根本不需要框架，一个函数搞定。后来需求变复杂，要支持多轮对话、要调用工具、要查知识库，于是开始用LangChain。早期的LangChain给人印象最深的是“链”（Chain），一条线串到底：第一步做什么，第二步做什么，都写死。这适合流程固定不变的情况。

但真实业务里没有那么多“固定不变”。比如我前面说的内容生成流水线：调研完英文资料要清洗，清洗完要总结，总结完还要判断是不是符合用户要求的风格，不符合就得换一种方式重新总结。这种带条件分支、带循环、允许中途改参数的需求，用链式结构写起来非常别扭。我自己写过一版，流程控制散落在各种回调函数里，别人看代码根本不知道整个流程长什么样。后来我才意识到：应用流程本身应该是一等公民，它不应该被写成代码的“副作用”，而应该被显式建模出来。

这正是LangGraph的切入点。它的做法是：把应用流程画成一张有向图。图的每个节点是一个计算步骤（比如调用一次大模型、查一次数据库），每条边代表步骤间的流转规则。数据流（也就是State）在节点之间传递，每个节点都可以读取State、修改State，然后决定下一步走向哪。这种“图编排”的抽象，本质上是一个状态机。

1.2 LangGraph是什么，和LangChain的关系

有人会问，LangGraph和LangChain到底什么关系？简单说，LangGraph是LangChain生态里专门做“有状态、可编排、可控”的Agent应用框架，它不替代LangChain的模型调用、提示词模板、工具封装这些基础能力，而是在更上层替你管理“流程的状态与跳转”。

我个人的感受是：LangChain解决的是“怎么跟模型、工具打交道”的问题，LangGraph解决的是“怎么把多个步骤组织成一个可靠应用”的问题。两者搭在一起用是常态。我在实际项目中，节点内部照样用LangChain的ChatOpenAI、create_react_agent，但节点之间怎么串联、怎么跳转、怎么持久化，交给LangGraph。

这个定位决定了它的适用边界。如果你的应用只是一个“单轮问答”或者“固定三步处理”，那用简单链就够了，上LangGraph反而增加概念负担。但一旦出现下面这些信号，就该考虑图编排了。

1.3 三个最适合LangGraph的场景

我总结了自己做过的项目，有三个场景最值得上LangGraph。

第一类是流程不确定的多步骤AI应用。比如智能客服，用户问“我的订单到哪了”，你得先识别意图，然后查订单系统，拿到物流信息后再组织回答；如果用户追问“为什么还没到”，你可能还得再调一次售后系统。每一步下一步是什么，取决于上一步的结果。这类流程用代码写分支能写，但写多了就是一坨面条代码。

第二类是需要人工介入的半自动流程。典型例子是内容审核、邮件审批、代码审查协助。AI先生成结果，但发布前必须有人点头。LangGraph提供了interrupt机制，可以在某个节点暂停整个图，等人工确认后再恢复运行。这种“人机回环”能力，是普通链式调用很难做到的。

第三类是长时间运行、需要断点续跑的任务。比如一个批量文档处理任务跑了一半系统崩了，重启之后能不能从上次中断的位置继续？LangGraph的检查点（Checkpoint）机制把每一步的状态都存下来，配合线程ID，可以精确恢复到某个历史步骤。这一点我后面专门用一章展开。

2. 把对话系统画成图：StateGraph的核心设计逻辑

2.1 一张图搞定流程：节点、边、状态三者关系

LangGraph最核心的类是StateGraph。它接受一个State类型定义，然后让你往里面添加节点（Node）和边（Edge），最后编译成可执行的图。

我用一个生活化的类比来理解这三者的关系。假设整个应用是一条智能生产线：节点是工位，每个工位负责一道工序（比如“写调研摘要”“写初稿”）；边是传送带，告诉零件加工完当前工序之后，下一道去哪；State是随零件一起流转的工单，工位上可以读取工单信息、写入新的信息，下一个工位就能看到更新后的内容。

节点本身就是一个普通函数。函数接收当前State作为入参，返回一个字典，字典里的键就是要更新到State上的字段。注意它不是要求你返回完整的State，只需要返回“更新部分”。框架会自动把你返回的内容合并到全局State里。这极大降低了每个节点的冗余代码。

边的类型分两种：普通边和条件边。普通边就是无脑跳转，比如节点A执行完必须去节点B。条件边则通过一个路由函数，根据当前State里的数据动态决定去哪个节点。这正是“流程不确定”背后的实现基础。

2.2 State不是“变量集合”，而是“流程契约”

State在本框架里不是一个可有可无的概念，它几乎是整个应用的“数据结构设计文档”。用TypedDict定义时，每个字段都对应业务里的一个数据切片。我用内容流水线举例，State可以长这样：

class ArticleState(TypedDict, total=False): topic: str # 用户输入的写作主题 research: str # 调研结果 outline: str # 文章大纲 draft: str # 文章草稿 passed: bool # 是否通过审核 review_note: str # 审核意见 attempts: int # 重写次数

这里total=False表示不要求所有字段在一次调用中全部提供。设计State时有一个经验：只放节点之间需要共享的数据。千万别什么中间结果都往里塞，否则整个图会变得又慢又难调试。

State还支持“归约器”（Reducer）。默认情况下，节点返回的字段会直接覆盖State里的旧值。但有些字段你想做追加而不是覆盖，比如对话消息列表。这时就可以用Annotated加上特殊的Reducer函数。框架内置了一个处理消息的add_messages，常用来做多轮对话的历史管理。

from typing import Annotated from langgraph.graph.message import add_messages class ChatState(TypedDict): messages: Annotated[list, add_messages]

这个写法看着有点绕，但精神很简单：字段被add_messages装饰之后，更新是“追加”而不是“覆盖”。这在我做多轮Agent对话时几乎是必备的。

2.3 普通边和条件边：让流程学会“自己选择下一步”

先看一个小例子，感受一下图的构建方式。我定义一个计数器状态，节点A给计数加一，然后通过条件边判断要不要继续循环。

from typing import TypedDict from langgraph.graph import StateGraph, START, END class CountState(TypedDict): count: int def increment(state: CountState) -> dict: return {"count": state["count"] + 1} def route(state: CountState) -> str: if state["count"] < 3: return "increment" return "end" builder = StateGraph(CountState) builder.add_node("increment", increment) builder.add_edge(START, "increment") builder.add_conditional_edges("increment", route, { "increment": "increment", "end": END, }) graph = builder.compile() print(graph.invoke({"count": 0}))

add_conditional_edges的第一个参数是“待路由的节点”，第二个参数是路由函数，第三个参数是一个映射表：路由函数返回的字符串，对应图上真实存在的节点名或END。这里我实现了循环：increment节点执行后，只要count小于3，就继续走increment节点自己。这个循环能力，恰恰是传统链式结构极难表达的东西。

条件边是LangGraph里最频繁使用的API之一，我几乎每个真实项目都会用到。判意图、判质量、判权限，都是它。

3. 一个多Agent协作的完整Demo：从需求拆解到最终落地

3.1 Demo定位：做一个“内容调查-撰写-审核”的三段式流程

理论讲了一堆，动手才有感觉。这一节我给一个能直接跑通的最小示例：内容流水线。流程是这样：用户给一个主题，系统先做“调研”，然后基于调研结果生成“大纲”，大纲转成“草稿”，草稿进入“质量审核”。审核不通过，就带着审核意见回到撰写节点重写；通过则结束。

为了让这个Demo不依赖模型API也能跑，我用一个mock_llm函数模拟大模型输出。你在真实项目里把它替换成ChatOpenAI或本地模型调用即可。但我得说清楚：框架的核心逻辑不在模型有多强，而在流程如何被可靠地组织起来。

3.2 完整代码：从State定义到图编译

直接上代码。

from typing import TypedDict from langgraph.graph import StateGraph, START, END class ArticleState(TypedDict, total=False): topic: str research: str outline: str draft: str passed: bool review_note: str attempts: int def call_llm(system_prompt: str, user_prompt: str) -> str: # 真实项目请替换为真实模型调用，例如： # from langchain_openai import ChatOpenAI # llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7) # return llm.invoke([("system", system_prompt), ("user", user_prompt)]).content if "研究" in system_prompt: return f"〔调研摘要〕围绕「{user_prompt}」整理出三条关键发现：1. 市场增长明显；2. 技术门槛集中在推理链路；3. 落地案例偏少。" if "大纲" in system_prompt: return f"〔大纲〕1. {user_prompt}的背景；2. 核心挑战；3. 实践步骤；4. 总结与展望。" return f"〔草稿〕{user_prompt}\n\n正文内容：背景介绍，核心挑战分析，逐步实施说明。" def research_node(state: ArticleState) -> dict: return {"research": call_llm("你是一名行业研究员，请输出研究报告摘要", state["topic"])} def outline_node(state: ArticleState) -> dict: return {"outline": call_llm("你是一名内容策划，请输出文章大纲", state["topic"] + "\n" + state["research"])} def draft_node(state: ArticleState) -> dict: attempts = state.get("attempts", 0) prompt = state["outline"] + "\n草稿要求：内容完整，例证充分。" draft = call_llm("你是一名资深写手，请撰写全文", prompt) if attempts == 0: # 模拟第一次草稿质量不过关：缺少具体案例 draft += "\n\nTODO: 缺少真实案例支撑，需补充。" else: # 重写时补充了案例 draft += "\n\n案例：某团队将类似方案落地后，交付效率提升约40%。" return {"draft": draft, "attempts": attempts + 1} def review_node(state: ArticleState) -> dict: draft = state.get("draft", "") if "TODO" in draft: return {"passed": False, "review_note": "缺少案例支撑，需补充后再审"} return {"passed": True, "review_note": "内容结构完整，可以发布"} def route_after_review(state: ArticleState) -> str: # 通过审核，或者重试次数达到上限，直接结束 if state.get("passed") or state.get("attempts", 0) >= 3: return "pass" return "rework" builder = StateGraph(ArticleState) builder.add_node("research", research_node) builder.add_node("outline", outline_node) builder.add_node("draft", draft_node) builder.add_node("review", review_node) builder.add_edge(START, "research") builder.add_edge("research", "outline") builder.add_edge("outline", "draft") builder.add_edge("draft", "review") builder.add_conditional_edges( "review", route_after_review, {"pass": END, "rework": "draft"}, ) graph = builder.compile() result = graph.invoke({"topic": "AI Agent 在企业落地的实践"}) print("最终结果：", result["passed"]) print("审核意见：", result["review_note"]) print("尝试次数：", result["attempts"]) print("草稿：", result["draft"][:120])

跑一次，你会发现第一次草稿因为带“TODO”被审核打回，然后流程跳回draft节点重写，第二次草稿不再有“TODO”，审核通过，流程走到END。这一圈下来，循环、条件分支、数据流转就全部演示到了。

3.3 关键技术点拆解

这个Demo看着简单，但有几处值得停下来想清楚。

第一处是route_after_review的返回值。它返回的是字符串“pass”或“rework”，对应add_conditional_edges第三个映射表里的键。映射表的值指向真实的节点名或END。如果你返回了一个映射表里没有的键，运行时会直接报错。这个设计很直白，但很容易因为大小写不一致或者拼写错误翻车。

第二处是attempts字段的更新。我在draft_node里读取并递增它。条件边里又用它判断是否强制终止。这样在状态里保存“流程控制所需的元数据”，是LangGraph应用里常见的做法。控制逻辑不是写死在代码流程里，而是体现在State数据和路由函数的配合中。

第三处是节点返回的只是部分更新。例如research_node只返回{"research": ...}，并没有返回topic或outline。框架自动把返回值与旧State合并。这种“部分更新”设计让节点职责单一，也方便单元测试。

第四处是开启调试。我建议在invoke之前先把图对象打印出来看结构，或者用graph.get_graph().print()。这个命令会把节点和边的关系以文本形式打印在控制台，我第一次用的时候对这个可视化能力印象很深，排查流程连接错误能省不少时间。

3.4 跑起来的调试心得

实际调试里，我最常用的不是断点，而是graph.stream。它可以按步骤吐结果，而不是等最后一次性返回：

for event in graph.stream({"topic": "AI Agent 在企业落地的实践"}): for node_name, state_snapshot in event.items(): print(f"节点 {node_name} 执行完成，当前状态摘要：") print({k: state_snapshot[k] for k in ("attempts", "passed") if k in state_snapshot})

这种方式特别适合看“每一步到底发生了什么”。我记得第一次调试自己的Agent循环时，就是靠stream发现某个节点把状态里的关键字段覆盖成了空字符串，一查是节点写错了返回值键名。如果你发现流程图逻辑没问题但结果不对，优先怀疑State字段名是否匹配。

4. 状态的可续性：检查点机制与人机协作的工程价值

4.1 checkpointer解决了什么问题

图本身是瞬态的：调用一次invoke，执行完，状态就没了。但在真实产品里，我们希望流程可以被中断、被恢复、被追溯。比如用户和Agent的对话进行到第三步，服务重启了，能不能继续聊？审核流程里AI生成初稿后要等人工判断，能不能暂停几个小时再继续？

LangGraph的答案是给图挂一个“检查点器”（Checkpointer）。它会在每次节点执行后，把当时的State持久化下来。后续拿到同一个线程ID，就能知道“这台戏演到哪一幕了”。

from langgraph.checkpoint.memory import InMemorySaver checkpointer = InMemorySaver() graph = builder.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "task-001"}} graph.invoke({"topic": "AI Agent 在企业落地的实践"}, config) # 获取当前状态 current = graph.get_state(config) print(current.values.get("attempts")) # 再往下执行 graph.invoke({"topic": "AI Agent 在企业落地的实践"}, config)

InMemorySaver适合开发调试，进程一重启就没了。生产环境建议用持久化的实现，比如SQLite或Postgres对应的Saver。这样即使服务重启，只要thread_id没变，状态就能恢复。

4.2 从“全自动”到“半自动”：用interrupt实现人工审核

检查点机制真正厉害的是它支撑了“人工介入”。框架里用interrupt函数实现这个能力：当节点执行到interrupt时，执行过程会暂停，框架把当前状态保存下来，返回给你一个中断信号。你处理完人工审核之后，用Command(resume=...)让图从中断处继续。

我在内容流水线里加一个人工确认节点：AI写好草稿之后不直接发布，先停在“人工确认”节点，把草稿展示给编辑，编辑选择“通过”或“打回”。

from langgraph.types import interrupt, Command def human_confirm_node(state: ArticleState) -> dict: decision = interrupt({ "question": "是否允许发布这篇草稿？", "draft": state["draft"], }) return {"passed": decision.get("approve", False)} builder.add_node("human_confirm", human_confirm_node) builder.add_edge("draft", "human_confirm") builder.add_edge("human_confirm", "review")

执行方式也有变化：

config = {"configurable": {"thread_id": "task-002"}} graph.invoke({"topic": "AI Agent 在企业落地的实践"}, config) # 此时执行到 human_confirm 节点会被中断，你可以把 draft 展示给人看 # 人工点击“通过”按钮后，恢复执行 graph.invoke(Command(resume={"approve": True}), config)

这套机制让“人机回环”从代码层面变得非常干净。过去实现一个人工审核，你得自建任务表、轮询状态、处理并发恢复，现在框架把这个通用能力给你做好了。我做过一个内容审核后台，服务端的代码量直接少了一半。

4.3 多轮对话和多任务隔离：thread_id的用法

thread_id这个配置值得单独说说。它是LangGraph区分“不同对话/不同任务”的钥匙。同一个图实例可以同时服务成千上万个线程，只要thread_id不同，State互相不干扰。

我做客服机器人时就是这样：每个用户会话一个thread_id，消息列表通过add_messages不断追加，检查点器把每个会话的历史状态存好。用户隔天再回来，机器人能准确接上上下文，靠的就是线程隔离与持久化。

定时清理线程数据也是要提上日程的事。内存型检查点器不清理的话，跑久了内存会一路涨上去。我一般在每次会话结束后调用接口把对应的检查点删掉，或者写个定时任务清理过期线程。

5. 实操中容易踩的坑：状态类型、并行分支与循环终止

5.1 我踩过的坑：把大对象塞进State

第一次我用LangGraph做文档处理，图里要传递一份几十页的PDF全文，我就直接把文本塞进了State。结果每个节点执行完，框架都要把整个State做快照和序列化，流程还没跑完，内存已经快爆了。后来我把PDF拆成本地文件或者数据库记录，State里只存文件路径和分块ID，速度立竿见影。

这个经验后来被我总结成一条规则：State里只放“流程需要的控制信息和轻量数据摘要”，大对象放到外部存储，用引用代替值。大模型产出的长文本可以放，但不要叠加太多中间产物。

5.2 条件边的函数签名和返回值——最容易翻车的地方

条件边路由函数接收的参数是State，返回值必须是你映射表里的键。我翻过的车包括：路由函数返回了None，返回了带空格的大小写错误字符串，还有忘记把END加进映射表。这些错误在执行到那一跳的时候才暴露，排查起来有点烦。

一个减少翻车的技巧：在路由函数里先打印返回值和当前State的关键字段，确认判断逻辑是不是自己想的；或者用类型注解明确返回类型。映射表复杂的时候，我甚至会为它写一个单独的单元测试。

5.3 并行分支与Send API：动态数量的子任务怎么做

很多人一开始不知道LangGraph还能做并行。如果你的前置节点产出了一批数据，要“每个数据单独走一遍同一个子流程”，手动建边是不现实的，因为数量是动态的。这时候要用SendAPI。

from langgraph.types import Send def fan_out_node(state): chunks = state["chunks"] # 假设已经切好文本块 return [Send("analyze", {"chunk": chunk}) for chunk in chunks]

fan_out_node返回一个Send对象列表，每个对象代表一个“动态生成的子任务”。框架会调度这些任务并行执行，所有子任务完成后，再把各自的更新合并回State。这就是典型的map-reduce模式。我第一次在LangGraph里用这个能力做了一个批量文本审核工具，速度比逐个跑快了好几倍。

5.4 其他小坑：递归限制、版本兼容、流式输出

默认每个图执行有步数上限，超出会报“Recursion Limit”错误。如果你的业务里确实需要长循环，可以主动调高限制：

graph.invoke(inputs, {"recursion_limit": 100})

但我会先问自己：流程是不是设计得太绕了？大多数时候，50步以内的图已经能表达非常复杂的业务了。

版本兼容是另一个坑。LangGraph迭代速度不慢，我记得interrupt的导入路径早年跟现在就不一样，早期版本里某些API在langgraph.prebuilt里，现在归类到langgraph.types。升级大版本前，最好去官方迁移指南把变化列表扫一遍。

流式输出上，stream返回的事件结构需要熟悉。我习惯用stream_mode="updates"只拿节点更新，避免把内部消息也一起打出来，控制台清爽很多。

5.5 坑位速查表

6. 最后一个建议：动手前先画图，再写State，再写节点

聊了这么多，最后分享一个我实际养成的习惯。

很多人上手LangGraph的第一反应是开编辑器写节点代码，我的建议是反过来：先在白板上把流程图画出来，哪怕只是画几个圈和箭头。这个动作的价值在于，它会逼你把“流程的判断点”和“数据的传递关系”都显式化。画完之后，再定义State字段，最后动手写每个节点函数。我拿这个顺序做过好几个项目，几乎没有返工过。

如果你手头正好有“流程带分支、有循环、需要人工确认、要断点续跑”的AI场景，LangGraph基本是目前最顺手的骨架。把示例代码跑通了，再改成你自己的业务，会比从零看文档舒服很多。