企业官网建设流程全解析

1. 为什么我最终选择了LangGraph 1.0来搭智能体工作流

先说说我自己的经历。早先做智能体应用，我试过多种方案：要么是纯靠代码手写状态机，把每一步转移逻辑硬编码在if/else里；要么用一些轻量级编排框架，但一旦流程复杂，状态维护就成了一场灾难。直到LangGraph升级到1.0，我把一个多轮对话、工具调用、人工审核混合的流程迁移过来之后，才真正感受到"用图来编排智能体"这件事的价值。

LangGraph 1.0核心就三样东西：状态（State）、节点（Node）、边（Edge）。思路并不复杂：把一次智能体任务拆成几步，每一步是一个节点，节点之间通过边连接，而整个运行过程中的数据存在一个全局状态里，节点读写这个状态，边决定下一步走向哪个节点。你可以把它理解为一张流程图，但这是一张"活的"流程图——状态里的数据会驱动流程的走向，节点可以在运行过程中动态改变执行逻辑。

这篇文章适合谁看？如果你已经写过一些基于大模型的应用，比如调过API做过简单的工具调用，但对如何搭建"多个节点协作、带条件分支、可人工介入"的完整工作流还比较模糊，那这篇内容正好帮你把骨架搭起来。我会从代码层面讲清楚：状态怎么定义、节点怎么编写、边怎么连接，最后落一个能跑的Demo。

2. 状态、节点、边：LangGraph工作流的三个基本组件

2.1 状态是智能体的"便签本"

在LangGraph里，状态不是某个节点的私有变量，而是整个图上所有节点共享的数据容器。每个节点执行完，都可以往状态里写入内容，其他节点也能读到。这个设计让节点之间的数据交换变得很自然——你不需要显式地把A节点的输出传给B节点，只需要各自读写同一个状态。

我习惯把状态理解成一张"便签本"，每个节点干活之前看一眼便签本，干完活再往上添几笔。LangGraph会对这张便签本做版本化管理，每一步的中间结果都可以被追踪和回放，这给后面调试带来了很大的便利。

定义一个状态很简单，在1.0版本里通常用TypedDict配合注解：

from typing import TypedDict, Annotated from langgraph.graph import StateGraph class AgentState(TypedDict): user_input: str intent: str # 节点写入的中间结果 tool_name: str tool_result: str final_answer: str

需要注意一个细节：LangGraph对状态字段有一套默认的更新规则——后写的覆盖先写的。但在某些场景下，你希望字段是"追加"而非"覆盖"，比如对话历史，就需要定义一个reducer去指定合并逻辑。这个后面单独展开讲。

2.2 节点是一个个"干活的小工"

节点就是执行单位。你可以把节点理解为带输入输出函数的函数。在LangGraph 1.0里，一个节点接收当前状态作为输入，返回一个字典作为对状态的更新。

一个典型的节点长这样：

from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4o-mini") def parse_intent_node(state: AgentState) -> dict: """把用户输入翻译成明确的意图""" prompt = f"请判断以下用户请求的意图，只返回一个词：查询、计算、还是闲聊。\n输入：{state['user_input']}" resp = llm.invoke(prompt) intent = resp.content.strip() return {"intent": intent}

这里有一个关键点：节点返回的字典里，只会更新你写的那几个字段，其他字段保持不变。这个设计让每个节点都尽量小、只干一件事——你不需要考虑"会不会动到别人的数据"，只需声明自己负责的字段即可。

2.3 边描述了节点之间的转移条件

边定义了节点执行的先后顺序。最简单的是普通边（normal edge），表示"执行完A，下一步走B"。还有一种条件边（conditional edge），它像一个路口的分岔标记——根据当前状态里的某个字段，决定下一步到底走哪个节点。

在LangGraph 1.0里，边不仅规定了执行的顺序，还给整张图引入了"结构化分支"的能力。很多智能体应用之所以需要在图里跑而不是纯写代码，就是因为代价分支逻辑往往会随着业务规则变化而变化，散落在代码里的if/else很难维护，而图上几条边就能把规则清晰勾勒出来。

2.4 图的运行时从哪里开始，到哪里结束

一张图至少要有一个入口节点和一个结束点。入口节点就是整个工作流开始时第一个执行的节点；结束时，让某个节点返回一个特殊标记或连接到一个"结束边"。

在LangGraph 1.0中，比较推荐的做法是为工作流定义合理的入口状态和结束条件。如果你不确定某个流程链条该怎么拆，一个简单的方法是：把用户的每一次"触发→处理→响应"拆成一个节点，再把那些需要决策的分岔点做成条件边。

3. 从0到1实现第一个LangGraph工作流

3.1 安装和最小运行环境

我建议直接用虚拟环境安装，避免污染全局环境：

python -m venv .venv source .venv/bin/activate pip install langgraph langchain-openai

现在LangGraph 1.0已经发布，依赖包相对精简。注意langgraph和langchain的版本要匹配，避免遇到API签名不一致的问题。我这里测试时用的版本组合是langgraph>=1.0、langchain-openai>=0.2，跑起来没有冲突。

3.2 一张最小可执行的图

我们来做一个场景：用户提了一个问题，系统先判断是否和"天气查询"相关，是的话走天气查询节点，否则走普通问答节点。

from typing import TypedDict, Literal from langgraph.graph import StateGraph class WeatherState(TypedDict): user_input: str need_weather: bool response: str def classify_node(state: WeatherState) -> dict: # 实践中这里会调LLM，这里用简单关键词示意 text = state["user_input"] need_weather = "天气" in text or "下雨" in text or "温度" in text return {"need_weather": need_weather} def weather_node(state: WeatherState) -> dict: return {"response": f"抱歉，我还无法实时获取天气。你问的是：{state['user_input']}"} def chat_node(state: WeatherState) -> dict: return {"response": f"这是一个普通问答节点，你输入的是：{state['user_input']}"} # 条件路由函数 def route_after_classify(state: WeatherState) -> Literal["weather_node", "chat_node"]: if state["need_weather"]: return "weather_node" return "chat_node" builder = StateGraph(WeatherState) builder.add_node("classify_node", classify_node) builder.add_node("weather_node", weather_node) builder.add_node("chat_node", chat_node) builder.set_entry_point("classify_node") builder.add_conditional_edges("classify_node", route_after_classify) builder.add_edge("weather_node", "__end__") builder.add_edge("chat_node", "__end__") graph = builder.compile()

运行起来非常直接：

result = graph.invoke({"user_input": "明天会下雨吗？", "need_weather": False, "response": ""}) print(result)

你在结果里能看到整个状态的最终快照。这就是最简单的LangGraph工作流：一个判断节点+两条条件分支，加上两个终态节点。

3.3 在LangGraph 1.0里如何"跑"图

在1.0版本里，invoke是同步执行，ainvoke是异步执行。如果你的节点里有异步IO（比如调用大模型API的异步版本），就用ainvoke。

另一个差异是流式输出。对大模型应用来说，用户往往希望"字一个个蹦出来"而不是等很久一次性返回。LangGraph 1.0提供了stream方法，可以逐节点流式输出状态更新，也可以配合大模型的stream模式输出token级的结果。

我实际项目中的经验：优先用astream配合stream_mode="messages"，这样实时性最好。普通非流式场景下，直接用invoke即可。

4. 理解状态属性与Reducer机制：从覆盖到合并

4.1 为什么需要Reducer

默认情况下，一个节点返回的字段值会覆盖状态里的旧值。这在很多场景下没问题，但有些字段天然是"累积型"的。比如多轮对话的消息列表，每轮结束都要往同一个字段追加新消息，而不是把旧的冲掉。

LangGraph 1.0提供了一个Annotated类型机制，配合reducer函数来定义字段的合并方式。下面的代码展示如何把messages字段设为追加模式：

from typing import Annotated, TypedDict from langgraph.graph import MessagesState from langchain_core.messages import AnyMessage def add_messages(left, right): # 实践中直接复用LangChain的实现，这里示意 return left + right class MultiTurnState(TypedDict): messages: Annotated[list[AnyMessage], add_messages]

你可能会问：为什么不直接在节点里自己写state["messages"] = state["messages"] + [new_msg]？因为Reducer的价值在于"多个节点并行或循环执行时仍能合并"，你不需要在每个节点里关心合并细节，声明一次就够了。

4.2 状态字段的默认覆盖与显式追加

来看一个实际对比：

写状态时有一条朴素的原则：作为流程控制依据的字段用覆盖即可，作为上下文积累的字段用Reducer追加。搞混了会出现典型的"历史消息丢失"问题——排查起来费时费力。

4.3 状态初始化和跨图传递

每次调用invoke时，传入的字典就作为初始状态。如果图比较大，建议在入口处预留一个"默认值工厂"函数，把缺省字段补全，而不是在每个节点里去if "xxx" not in state这样判断。

def default_state(user_input: str) -> dict: return { "user_input": user_input, "intent": "", "tool_name": "", "tool_result": "", "final_answer": "", }

这样所有节点拿到的状态结构永远完整，节点内就不需要写多余的防御性判断了。

5. 边与路由控制：智能体的决策核心

5.1 普通边 vs 条件边怎么选

普通边适合顺序执行，条件边适合分岔路由。判断标准很朴素：执行完这个节点，下一步是不是"看情况而定"？如果是，就走条件边。

我在实际项目里把条件边归类成三类：

• 内容路由：比如根据LLM输出的意图标签，分发到对应处理节点。
• 路由到工具：当用户请求需要调用外部API时，先走工具选择节点，再由条件边决定调用哪个工具节点。
• 循环与退回：比如生成的答案质量不过关，条件边可以把流程拉回"重新生成"节点；或者执行完工具后回到主对话节点，让LLM基于工具结果继续推理。

5.2 条件边函数的定义和注册方式

看一个更完整的条件边写法：

def router(state: AgentState) -> str: intent = state["intent"] if intent == "查询": return "query_node" elif intent == "计算": return "calculator_node" return "fallback_node" builder.add_conditional_edges("parse_intent_node", router, { "query_node": "query_node", "calculator_node": "calculator_node", "fallback_node": "fallback_node", })

在LangGraph 1.0里，add_conditional_edges接收三个参数：源节点、路由函数、路由映射字典。路由函数的返回值必须是映射字典里的key之一，这样才能保证图的结构是确定的。

需要特别提醒一个坑：路由函数里尽量不要做大模型调用。如果你让路由逻辑也依赖大模型输出，整体延迟会叠加，而且不容易调试。我通常把大模型调用放在数据解析节点里，让路由只读解析好的字段，这样所有分支判断都变成了纯规则，稳定且快。

5.3 循环边与递归限制

LangGraph天然支持循环，这是它比很多编排框架强的地方。比如一个"反思再回答"的流程：生成回答 → 检查质量 → 不合格则回到生成节点 → 重新生成。

builder.add_node("generate_node", generate_answer) builder.add_node("check_node", check_quality) builder.add_node("regenerate_node", regenerate_answer) builder.add_edge("generate_node", "check_node") builder.add_conditional_edges("check_node", route_after_check) builder.add_edge("regenerate_node", "generate_node")

但注意，无限循环会卡死整个工作流，所以LangGraph在编译时可以设置递归上限：

graph = builder.compile(recursion_limit=10)

一旦执行步数超过限制，就会抛出异常。我在项目里会让check_node在第三或第四次循环时强制放行，保证用户体验。

6. 实战：搭建一个多步骤智能体工作流

6.1 场景设定与流程拆解

下面我们做一个更接近真实需求的例子：一个"商品推荐助手"。用户输入需求，系统先判断是否需要查商品库，如果需要，先调用检索工具拿到商品列表，再做分析推荐；如果不需要，直接回答。

整个流程拆成四个节点：

1. parse_intent_node：解析用户需求，判断意图和关键实体。
2. search_products_node：根据实体去模拟商品库检索。
3. recommend_node：基于检索结果生成推荐语。
4. fallback_node：处理非推荐类闲聊。

6.2 关键代码实现

from typing import TypedDict, Literal from langgraph.graph import StateGraph class RecState(TypedDict): user_input: str need_search: bool keyword: str product_list: list output: str def parse_intent_node(state: RecState) -> dict: # 简单规则代替LLM text = state["user_input"] if ("买" in text or "推荐" in text) and ("手机" in text or "电脑" in text or "耳机" in text): return {"need_search": True, "keyword": text} return {"need_search": False, "keyword": ""} def search_products_node(state: RecState) -> dict: keyword = state["keyword"] # 实际项目里这里会调用向量检索或数据库 mock_products = [ {"name": "智能降噪耳机", "price": 599}, {"name": "轻薄商务本", "price": 5999}, {"name": "旗舰手机", "price": 3999}, ] if "耳机" in keyword: products = [mock_products[0]] elif "电脑" in keyword: products = [mock_products[1]] elif "手机" in keyword: products = [mock_products[2]] else: products = mock_products return {"product_list": products} def recommend_node(state: RecState) -> dict: items = "、".join([p["name"] for p in state["product_list"]]) return {"output": f"根据你的需求，为你推荐：{items}。需要更详细信息可以继续追问。"} def fallback_node(state: RecState) -> dict: return {"output": "我现在主要负责商品推荐，你可以问我手机、电脑、耳机等产品。"} def route_after_parse(state: RecState) -> Literal["search_products_node", "fallback_node"]: return "search_products_node" if state["need_search"] else "fallback_node" builder = StateGraph(RecState) builder.add_node("parse_intent_node", parse_intent_node) builder.add_node("search_products_node", search_products_node) builder.add_node("recommend_node", recommend_node) builder.add_node("fallback_node", fallback_node) builder.set_entry_point("parse_intent_node") builder.add_conditional_edges("parse_intent_node", route_after_parse) builder.add_edge("search_products_node", "recommend_node") builder.add_edge("recommend_node", "__end__") builder.add_edge("fallback_node", "__end__") rec_graph = builder.compile()

调用测试：

result = rec_graph.invoke({ "user_input": "想买一个降噪耳机，有推荐吗？", "need_search": False, "keyword": "", "product_list": [], "output": "", }) print(result["output"])

6.3 这个架构能怎么扩展

这套骨架最大的好处是可以无痛加节点。比如后续想加"比价"功能，只需要：

• 新增一个compare_price_node
• 在search_products_node和recommend_node之间加一条条件边，决定是否先比价再推荐

因为节点之间完全通过状态解耦，新增节点时不需要改动已存在的节点函数，改图结构即可。这就是用图编排相比硬编码状态机最明显的高级感。

7. 调试与可观测性：LangGraph的隐藏优势

7.1 使用内置绘制功能检查图结构

LangGraph 1.0内置了图的可视化能力。编译后的图对象可以直接生成Mermaid或PNG，这个功能在刚开始搭流程时特别好用——你不需要在脑内推演节点之间的连接，直接看图画就一目了然。

一个典型做法是：

# 生成并保存图结构 with open("graph.png", "wb") as f: f.write(rec_graph.get_graph().draw_mermaid_png())

当然具体API可能随版本略有变化，但思路是一致的：编译后的graph对象提供了查看和导出图结构的方法。我每次调新流程，第一件事就是把图画出来看一眼，常常能发现一些"自以为连对了但实际连错"的边。

7.2 运行时状态追踪

调用invoke时，加一个config参数就能开启详细的调试信息：

from langchain_core.runnables import RunnableConfig config = RunnableConfig(recursion_limit=10, tags=["demo"]) result = rec_graph.invoke(input_state, config=config)

结合LangSmith之类的追踪工具（也可以用LangGraph自带的回调机制），每一步节点的输入输出、耗时都能被记录下来。排查问题时，我最常用的是"回溯最近几步的状态快照"——先看看某一步状态里的字段是不是和预期一致，再判断是节点问题还是边路由问题。

7.3 常见的失败模式与排查链路

在实际使用中，我遇到的失败大多集中在以下几类：

一是节点返回了状态中不存在的字段。LangGraph一般不会马上报错，但这个字段会被忽略掉，最终结果里就是你想要的字段没更新。排查方法是把每个节点的返回值打出来，挨个字段对照。

二是条件边路由函数返回了映射里没有的key。这个会直接抛KeyError，报错信息里会把路由函数名和返回的key都显示出来，定位很快。

三是循环递归超限。这种问题发生在节点之间形成了"死循环"，比如某个校验节点永远判定为不合格，流程就会不断循环，直到触发recursion_limit。解决思路是在校验节点里增加"已重试次数"字段，达到一定次数后强制走出口。

8. 从固定流程到自适应智能体：下一步可以这样扩展

8.1 让图自己决定"下一个工具"

现在很多智能体应用不满足于固定的流程顺序，而是希望大模型根据用户输入动态决定要调用哪些工具。LangGraph里可以实现一个"工具选择节点"，让LLM先输出所需工具列表，再通过条件边逐一执行。

def tool_selector_node(state: AgentState) -> dict: prompt = "根据用户输入，选择需要调用的工具，可选：search_web, calculate, get_time。只返回工具名，多个工具用逗号分隔。" resp = llm.invoke(prompt) tools = [t.strip() for t in resp.content.split(",")] return {"selected_tools": tools}

然后后续的路由就读取selected_tools里的工具名，逐一dispatch到对应的工具节点。这个模式在1.0里非常顺滑，因为状态天然支持列表类型的字段，工具执行结果也能累积在状态里。

8.2 人工审批与暂停恢复

很多业务流程需要人工参与，比如"AI先生成内容 → 人工审核 → 通过则发送，不通过则退回重写"。LangGraph 1.0支持中断和恢复机制。

你可以把"需要人工审核"的节点设置为一个中断点：图执行到这里会暂停，等待外部传入新的状态后再继续。这样实际落地时，运营人员可以介入每个关键环节，而不是完全交给模型自动跑。

8.3 多智能体协作的前景

最后提一个方向：多个LangGraph图之间可以通过"作为节点"的方式嵌套组合。也就是说，一个小图可以作为一个大图里的节点存在，从而实现多智能体分层协作。比如一个"客服主管图"里，用户提问先分诊，再分别调用"售后子图"和"技术支持子图"。

我个人的感受是，LangGraph 1.0最大的价值在于把"流程"和"逻辑"解耦了。以前写复杂的智能体，最痛苦的是维护那一堆隐性的流转关系。现在图结构本身就承载了流程，状态管理交给框架，代码里剩下的全是业务逻辑。对一个开发者来说，这种体验确实值得专门写一篇文章来分享。