企业官网建设流程全解析

如果你最近在折腾AI Agent，大概率绕不开LangGraph这个名字。我第一次看到它时，心里想的是：又来了一个花架子框架？但把它拆开以后发现，它做的事情其实特别朴实——把你脑子里的“流程状态机”变成可以运行的代码。和普通链式调用不一样，LangGraph把每一步都当成节点，把节点间的跳转规则当成边，状态则交给一个全局对象管理。也就是说，你可以在任意一步停下来查看状态、改变方向、甚至回溯重跑。

这篇文章我打算直接从核心模块入手，用一个完整的小项目把代码讲透。适合已经跑过LangChain、对Agent有点概念、但还没系统看过LangGraph的读者。我先保证你看完能上手写一个带条件分支和状态记忆的图，然后我们再聊持久化、人工介入这些进阶玩法。看完以后，你至少不会再觉得它是个黑盒。

1. LangGraph到底在解决什么问题

1.1 从普通链式调用到有状态编排

先想想以前我们怎么写AI流程。最朴素的做法是链式调用：

llm.invoke(prompt1) -> LLM结果 -> 拼到prompt2 -> llm.invoke(prompt2) -> ...

这种写法的问题在于：流程是固定的、线性的，一旦中间需要根据结果改变路径，代码就变得非常难维护。比如“如果用户情绪是愤怒，就转人工，否则自动回复”，你会发现自己写了一堆if else，还把每一步的临时变量散落在各处。

LangGraph把这个事情抽象成一张图。每个处理步骤是一个节点，节点之间的跳转关系是边。所有的中间结果都写进一个全局状态对象里，谁都能读，谁都可以改。这样一来：

• 流程可以用图的方式直接画出来，和产品/团队交流成本低。
• 分支、循环、回退都是图的拓扑关系，而不是散落的逻辑。
• 运行过程中状态可持久化，相当于给工作流加了“存档”。

我用一个生活化的类比：普通流程就是一条固定传送带，零件从A到B到C，中间没有岔路；LangGraph是一个带分拣口的流水线，传送带可以变轨、可以回流，每个包裹上的标签就是状态，系统靠标签决定下一个该去哪。这个类比我一直跟别人讲，基本都能秒懂。

1.2 核心设计理念：节点、边、状态机

LangGraph的底层核心就三个概念：

• State（状态）：一个数据结构，保存整个流程运行时需要的信息。
• Node（节点）：一个Python函数，接收State，返回对State的增量更新。
• Edge（边）：从哪个节点走到哪个节点的规则。

其中最容易忽略的是State。它不是普通的dict，而是带类型的、可叠加更新的对象。LangGraph允许每个节点只返回需要修改的字段，剩下的字段自动保留。这个设计让代码非常干净，你不需要手动传递一堆上下文。

再往下走，条件边（Conditional Edges）让图有“智能”。一个节点返回结果后，图会根据状态里的某个字段，决定下一步走哪条边。我觉得理解了这三样东西，外加一个编译过程，langgraph的核心就已经拿下一大半了。

2. 核心模块逐个拆解，附代码直击

2.1 StateGraph：整个工作流的骨架

StateGraph是LangGraph的主类，所有模块都围绕它组织。先看一段最基础的代码：

from typing import TypedDict from langgraph.graph import StateGraph class MyState(TypedDict): messages: list user_name: str def node_a(state: MyState) -> dict: # 读取state，返回需要更新的字段 return {"user_name": "张三"} workflow = StateGraph(MyState) workflow.add_node("a", node_a) workflow.set_entry_point("a") workflow.set_finish_point("a") app = workflow.compile()

这里有几个关键点：

• StateGraph的泛型参数是状态结构，不是节点，很多新手看示例会误以为StateGraph(MyState)是传入某个内置类型。
• add_node给节点起名字，名字可以随意，但建议用有业务含义的词。
• set_entry_point和set_finish_point是出入口。不能只有start没有end，否则编译时会直接报错。
• compile()返回一个可调用的对象，它才是真正用来执行工作流的入口。

我实测下来，StateGraph本身没什么魔法，它做的事情就是把节点和边校验一遍，生成一个可执行的编译图。真正决定行为的是State和节点里的逻辑。

2.2 State：不可变状态与Reducer

State定义的是整张图的“公共记忆”。LangGraph推荐用TypedDict定义，好处是IDE能自动补全，字段结构也非常直观。不过麻烦的地方在于更新规则。

默认情况下，如果两个节点都返回同一个字段，后执行的节点会覆盖先执行的值。这通常在链式流程里没问题，但如果你希望“累加”或者“合并”消息列表，就要用到Reducer。

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

Annotated的含义是：这个字段的更新不走默认覆盖，而是走add_messages这个函数。add_messages会把新的消息追加到原有列表末尾，而不是整体替换。

我开发时遇到过一个问题：在一个并行节点里，两个分支都返回messages字段，如果没有Reducer，状态就会变成最后执行分支的消息，前一个分支的结果直接丢掉了。后来我统一给列表字段加上Reducer，才彻底解决。所以建议所有列表类型的字段都显式配置Reducer，这不会带来额外成本，但能少踩很多坑。

2.3 Node与Edge：运算单元和路线规则

节点函数是LangGraph里的最小运算单元。它接收当前State，返回一个字典，字典里的字段会被合并到State中。核心写法如下：

def process(state: MyState) -> dict: new_messages = state["messages"] + [{"role": "assistant", "content": "处理完成"}] return {"messages": new_messages}

注意：节点函数不要返回None，也不要返回一个包含所有字段的大字典。最佳实践是只返回你真正改动过的字段，这样代码更清晰，也能减少潜在的冲突。

边的类型分两种：

• 普通边：从一个节点无条件走到另一个节点。
• 条件边：根据状态或路由函数的输出，动态决定下一步。

普通边的代码：

workflow.add_edge("a", "b") workflow.set_entry_point("a") workflow.set_finish_point("b")

如果流程是线性的，这种写法非常简单。但如果分支多了，我建议不要每个分支都手动连边，而是用条件边集中管理。

2.4 Conditional Edges：让流程自己拿主意

条件边是LangGraph里最体现“智能”的部分。它的路由函数不是直接返回下一个节点的名字，而是返回一个“键值”，再把键值映射到节点。这样可以避免路由函数和具体节点名耦合。

看一个示例：

def route_after_check(state: MyState) -> str: if state["sentiment"] == "negative": return "human" return "auto_reply" workflow.add_conditional_edges( "classify", route_after_check, { "human": "human_handler", "auto_reply": "auto_reply_node", } )

这个写法的好处是，route_after_check只返回业务含义的字符串，比如“human”“auto_reply”；具体哪个节点处理这个字符串，由后面的字典统一管理。以后改节点名字只需要改映射，不需要动路由函数。

我一开始习惯直接在路由函数里返回节点名，比如return "human_handler"。短期没问题，但图上节点一多，路由函数反复修改，很容易不小心把名字写错。后来全部改成语义化返回，维护成本低很多。

条件边完全可以实现循环效果：让一个节点处理完以后，路由回更早的节点。这就解决了“如果LLM输出不合法就重试”这类需求。要注意的是，循环必须设置一个最大次数，否则可能无限跑下去。

3. 一个完整的可复现示例：智能客服工单处理

3.1 需求场景与状态设计

我拿一个常见的场景来说：智能客服工单分类与回复。用户提交一条问题，系统先做意图和情绪识别，如果用户情绪负面或者问题属于售后类，就转人工；否则直接用自动回复模板。

这个场景很适合用LangGraph演示，因为它同时用到了条件分支和状态持久化。我们定义状态如下：

from typing import Annotated, TypedDict from langgraph.graph import add_messages class TicketState(TypedDict): messages: Annotated[list, add_messages] category: str sentiment: str need_human: bool ticket_id: str final_answer: str

• messages：保存对话原文，用add_messages做累积。
• category：问题分类，比如“技术咨询”“售后”“建议”。
• sentiment：情绪标签。
• need_human：是否转人工，由分类节点计算后写入。
• ticket_id：工单号，用来模拟业务标识。
• final_answer：最终回复内容。

状态里既有简单字段又有列表字段，这样能同时展示Reducer和覆盖更新的区别。设计状态的时候，我的经验是先写下流程里所有节点需要读写的字段，再开始写节点函数，不要边写边加。

3.2 节点函数与分支逻辑实现

先做几个工具函数，模拟LLM的输出：

def classify_and_analyze(state: TicketState) -> dict: user_text = state["messages"][-1]["content"] # 实际项目里这里会调用LLM或专门的分类模型 if "退款" in user_text or "投诉" in user_text: category = "售后" sentiment = "negative" elif "怎么用" in user_text or "安装" in user_text: category = "技术咨询" sentiment = "neutral" else: category = "其他" sentiment = "positive" need_human = category == "售后" or sentiment == "negative" return { "category": category, "sentiment": sentiment, "need_human": need_human, }

这个节点读取状态里的最后一条消息，给出分类、情绪和是否转人工的判断。返回的字段会被写入全局状态，供后面的条件边使用。为了让演示简单，我没有真正调大模型，只用了关键词判断；改成LLM调用时，逻辑是一样的，只是返回之前加一个llm.invoke()。

接着写两个下游节点：一个是自动回复，一个是转人工。

def auto_reply(state: TicketState) -> dict: reply = f"【{state['category']}】您好，这是一条自动回复。您的工单号是: {state['ticket_id']}" return {"final_answer": reply, "messages": [{"role": "assistant", "content": reply}]} def human_handle(state: TicketState) -> dict: reply = f"【{state['category']}】已将工单 {state['ticket_id']} 转给人工客服，请稍候。" return {"final_answer": reply, "messages": [{"role": "assistant", "content": reply}]}

自动回复和人工接待的逻辑都很简单，就是读取状态、生成回复、把回复写回状态。注意消息字段用了Reducer，所以回传消息时用列表追加语义；如果没用Reducer，这条回传语句会覆盖之前的messages，那就出问题了。

条件路由函数：

def decide_next(state: TicketState) -> str: if state["need_human"]: return "human" return "auto"

3.3 编译与调用全流程

把节点和图组合起来：

from langgraph.graph import StateGraph workflow = StateGraph(TicketState) workflow.add_node("classify", classify_and_analyze) workflow.add_node("auto_reply", auto_reply) workflow.add_node("human_handle", human_handle) workflow.set_entry_point("classify") workflow.add_conditional_edges( "classify", decide_next, { "human": "human_handle", "auto": "auto_reply", } ) workflow.add_edge("auto_reply", "__end__") workflow.add_edge("human_handle", "__end__") app = workflow.compile()

注意__end__是一个特殊节点名称，表示流程结束。LangGraph要求从set_entry_point进入的每个分支最终都要能到达__end__，否则编译或运行时可能报错。如果你不想显式连终点边，也可以用set_finish_point("auto_reply")指定。

调用它：

initial_state = { "messages": [{"role": "user", "content": "我要退款，你们的服务太差了"}], "ticket_id": "T20250001", } result = app.invoke(initial_state) print(result["final_answer"])

运行后会走到human_handle分支，因为消息里包含“退款”和“投诉”，sentiment被标记为negative，need_human为True。

参数说明：invoke会返回最终完整状态，而不是只返回输出。所以你可以直接result["final_answer"]，也能看到category、sentiment等中间字段。这个特性很实用，调试时可以直接把result打出来看整条链路的状态变化，省去到处打日志的麻烦。

4. 进阶玩法：持久化、人工介入与并行

4.1 Checkpointer：断点续跑和多轮对话

前面那个例子，图运行完状态就丢了。现实场景里我们往往需要保留状态：比如支持多轮对话、允许某次运行中断后再继续。LangGraph用Checkpointer解决这个问题。

from langgraph.checkpoint.memory import MemorySaver checkpointer = MemorySaver() app = workflow.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "thread-1"}} result1 = app.invoke(initial_state, config) result2 = app.invoke({"messages": [{"role": "user", "content": "那具体多久能解决？"}]}, config)

关键就两点：

• 在compile时传入checkpointer。
• 每次invoke都带上同一个thread_id。

带thread_id之后，LangGraph会把每一次运行的状态快照保存在检查点里。第二次invoke时，它能从同一个thread的旧状态继续，而不是从零开始。我在做多轮客服对话时，这个功能几乎是必需的，否则每轮都要把历史消息手动拼进去。

MemorySaver只适合本地调试，进程一重启数据就没了。生产环境建议用数据库类的持久化实现，LangGraph也提供了基于PostgreSQL等适配器的checkpointer，只是我不推荐在初学阶段就把存储搞复杂，先用内存版跑通，再换存储实现。

4.2 人工介入：interrupt与resume

客服系统里有一种经典需求：自动流程走到某一步时停下来，等人工确认或填写信息，然后继续跑。LangGraph的interrupt机制专门做这件事。

from langgraph.types import interrupt def human_approve(state: TicketState) -> dict: approval = interrupt({ "type": "需要人工确认", "content": state["messages"][-1]["content"], }) return {"need_human": approval.get("approved", False)}

当图执行到这个节点时，遇到interrupt会抛出一个中断信号，把运行状态保存下来，然后暂停。之后你通过外部程序把用户的确认结果传回同一个thread，图就会从刚才中断的地方继续执行。

app.invoke(initial_state, config) # 图执行到interrupt时暂停 resume_command = {"approved": True} app.invoke(Command(resume=resume_command), config)

这个概念理解成“工作流自动暂停等人工回话”就好。它的价值在于把人工确认这个动作变成了图里的一个节点，而不是靠外部回调绕过核心流程。我第一次用这个能力时很兴奋，因为它直接把我以前的“sleep+轮询数据库”方案替换掉了。

4.3 并行节点与Map-Reduce模式

处理多个工单或者同时跑多个检索任务时，线性执行效率太低。LangGraph支持通过Send API动态创建并行分支。

from langgraph.constants import Send def continue_to_support_agents(state: TicketState): return [ Send("agent_worker", {"ticket_id": t["id"], "description": t["desc"]}) for t in state["sub_tickets"] ] workflow.add_conditional_edges( "dispatch", continue_to_support_agents, ["agent_worker"] )

Send的含义不是调用一个节点，而是把传入的数据作为子状态，启动一个新的“agent_worker”节点实例。每个子任务可以独立运行，也可以有自己的内部图路径。

我做批量工单分派时，还会配合Map-Reduce模式：先把一批任务分发给多个worker节点，各跑各的，最后用一个“汇总”节点把结果合并回总状态。这里的汇总节点跟普通节点没什么区别，只是需要处理列表类型的返回结果。注意要给列表字段配好Reducer，不然多个并行分支返回时会互相覆盖。

并行不是越多越好，特别是每个worker都要调LLM时，并发太高会导致接口限流。我的习惯是先限制并发数，比如一次最多处理5个子任务，再逐步调大。

5. 踩坑实录与排查技巧

5.1 高频报错与对策

我在本地和项目里跑LangGraph时，遇到过不少问题，挑几个典型的列出来：

5.2 几个建议和最佳实践

先说说状态设计。State字段越少越好，不要把大段临时内容塞进State。有些中间计算结果只在当前节点用，那就不要写到State里，用节点内部局部变量处理。不然调试时会有大量干扰噪声，而且还容易触发Reducer冲突。

再说节点函数风格。我建议所有节点函数都写成“纯函数”式，不对State做隐式修改，只返回更新。这样单元测试也很好写，你用假State调一个节点，断言返回的dict对不对就行。我一直把节点当独立函数测试，最后再组合成图，整体bug率明显下降。

还有条件边路由函数务必保持轻量。它会跟随每次状态更新被调用，如果里面有重IO或慢计算，整个流程会被拖慢。需要复杂判断时，应提前在一个普通节点里计算好结果，把结果放在State里，路由函数只做读操作。

最后，关于调试日志。LangGraph运行时的图信息，可以把这个打开：

from langgraph.pregel import PregelOptions options = PregelOptions(debug=True) result = app.invoke(initial_state, config=options)

实测能看到每个节点的输入状态输出状态，比自己在节点里print干净得多。不过生产环境不建议一直开着，日志量有点大。

我在实际项目中有一个小习惯：每个节点函数内部都会记录一句日志，格式是“当前节点名+输入状态关键字段+输出状态关键字段”。刚开始觉得麻烦，后来排查线上问题全靠这些日志去还原链路。LangGraph的优势就在于状态流清晰，日志能把每一步的状态变化串起来，定位问题非常快。

再提一个容易忽略的细节：LangGraph的State字段命名不要用太通用的单词，比如result、data，因为多个节点都可能返回这些字段，一旦字段冲突，排查成本很高。我吃过一次亏：一个节点的data是列表，另一个节点的data是字符串，运行时报错后查了半天才明白是字段被覆盖了。现在我都用带有业务含义的字段名，比如sub_ticket_list、final_reply_text。

如果你只是想在单机脚本里快速验证逻辑，用MemorySaver就够了。但如果你准备把这个东西接到一个真实服务里，一定要提前想好checkpointer与业务系统的关系。每个thread_id对应一段独立会话，所以thread_id最好用业务侧的会话ID或用户ID，这样即使重启服务，也能通过同样的thread_id恢复上下文。

LangGraph不是一个需要背很多API才能上手的框架，它的核心就是“状态在哪、节点干什么、边怎么走”。把这三点搞清楚，剩下都是技巧问题。希望这篇拆解能让你少走一点弯路，至少遇到报错时，能第一时间去查State的定义和Reducer配置，而不是对着整个框架发呆。