企业官网建设流程全解析

这次我们不聊某一个具体模型，来聊一个把大模型应用从“顺序调用链”升级成“可控状态机”的编排框架——LangGraph。过去半年，LangGraph 在 Agent 开发圈子里热度很高，原因是它解决了 LangChain 时代最头疼的问题：多步任务怎么管理状态、分支怎么走、循环怎么跳、对话记忆怎么持久化。如果你写过多轮 Agent 应用，大概率遇到过“流程一复杂，代码就变成一坨 if-else”的情况。LangGraph 的核心价值，就是把这些流程控制从业务代码里抽出来，交给图结构去管理。

这篇文章会把 LangGraph 从安装部署、核心概念、条件路由、子图、并行分支、长期记忆，到接口 API 包装和批量任务完整过一遍。全文不是概念罗列，而是按“先跑通，再深入，最后排错”的思路来写。每一步都有可复制的代码，读完之后你能独立搭出一个带分支、带记忆、可对外提供 API 的 LangGraph 应用。

如果你最近在学 LangGraph，或者正打算把已有的 LangChain 应用改造成图状态管理，这篇可以直接收藏。

1. LangGraph 核心能力速览

先把最关键的信息放在前面，方便你快速判断这个框架适不适合现阶段去学。

需要说明的是，LangGraph 目前官方主推的是 Python 版本，也有 TypeScript 版本，但绝大部分教程、案例和公司落地场景都集中在 Python。热搜里提到的“Rust 版本”目前没有官方实现，不需要等，用 Python 就够了。

2. 适用场景与使用边界

LangGraph 解决的问题很具体：当一个任务需要多个模型调用、多个工具调用、或者需要根据中间结果决定下一步走向时，用图结构来组织远比手写代码维护起来更清晰。

适合用 LangGraph 的场景包括：

• 多步骤 Agent 工作流，比如“先分析用户意图，再决定调哪个工具，最后汇总结果”。
• 需要条件分支的流程，比如“如果文章超过 5000 字就走长文本摘要分支，否则走普通摘要分支”。
• 需要循环和自省能力的 Agent，比如“生成结果后让大模型自己检查一遍，不满意的生成次数上限以内重新生成”。
• 需要对话记忆的应用，LangGraph 的检查点机制可以把每一轮状态保存下来，下一轮直接接着跑。
• 多 Agent 协作，比如一个 Agent 负责生成，另一个 Agent 负责审核，两者通过图节点串起来。

不适合用 LangGraph 的场景也很明显：

• 单次简单的模型调用，比如只做一次文本分类，直接调 API 就行，不需要上框架。
• 不需要状态管理的一次性任务，加图反而增加理解成本。
• 团队里完全没有 Python 工程化经验，纯给大模型写 Prompt 的团队，建议先从简单的 LangChain 链开始。

使用边界方面，需要提醒三点：

• LangGraph 本身不会替你解决模型合规问题，节点里调用什么模型、模型服务是否允许商用，需要你自己看对应 API 的服务条款。
• 不要把敏感数据直接塞进 State 再拿去调第三方大模型接口。本地开发测试可以，生产环境要评估数据合规风险。
• 如果应用中涉及读取用户人脸、声音、身份证、手机号等隐私信息，必须在用户明确授权的前提下进行，并做好脱敏处理。

3. LangGraph 环境准备与安装

LangGraph 的安装非常轻量，本质上就是装一个 Python 库。这里给出一套完整的环境准备流程，建议在虚拟环境里操作，避免污染全局 Python 环境。

3.1 环境检查

先确认本机 Python 版本。在终端执行：

python --version

如果输出是 Python 3.9 及以上，就可以继续。低于 3.9 建议先升级 Python。

3.2 创建虚拟环境

Linux / macOS / Windows PowerShell 通用做法：

python -m venv langgraph_env

激活虚拟环境：

Windows PowerShell：

langgraph_env\Scripts\Activate.ps1

macOS / Linux：

source langgraph_env/bin/activate

3.3 安装 LangGraph

pip install -U langgraph

如果你打算在节点里调用 OpenAI 兼容接口的模型，需要再装 LangChain 的模型集成包：

pip install langchain-openai

如果你想把图应用包装成 API 服务，需要 FastAPI 和 uvicorn：

pip install fastapi uvicorn

为了后面方便调试，建议把常用依赖一次性装好：

pip install -U langgraph langchain-openai fastapi uvicorn

3.4 配置大模型 API Key

LangGraph 只负责编排，真正干活的是节点里的模型。你需要准备一个可用的模型 API Key，比如 OpenAI 兼容接口的 Key，或者本地部署模型的接口地址。

在终端设置环境变量：

export OPENAI_API_KEY="你的-api-key"

Windows PowerShell 下用：

$env:OPENAI_API_KEY="你的-api-key"

如果你用的是本地模型或第三方兼容接口，通过OpenAI客户端的base_url指定接口地址即可，LangGraph 本身不关心你接的是什么模型。

3.5 验证安装

执行以下命令，不报错即安装成功：

from langgraph.graph import StateGraph, START, END print("LangGraph import success")

到这里环境就准备好了。安装过程中最常见的坑是网络超时，如果 pip 下载慢，可以换国内镜像源再装。

4. 从零构建第一个 LangGraph 应用

环境装好之后，先不要急着看复杂概念，直接跑通一个最小图。这个例子包含两个节点，每个节点对 State 做一次数值累加，最终输出结果。

4.1 定义 State

LangGraph 的 State 本质是一个 TypedDict，所有节点读写的共享数据都在这里面：

from typing import TypedDict, Annotated, Literal from langgraph.graph import StateGraph, START, END class State(TypedDict): total: int messages: list[str]

这里定义了total和messages两个字段，分别存计算值和过程日志。

4.2 定义节点函数

每个节点是一个普通函数，接收当前 State，返回一个字典。返回的字典会合并到 State 里：

def add_one(state: State) -> dict: next_total = state["total"] + 1 return { "total": next_total, "messages": state["messages"] + [f"add_one -> {next_total}"], } def add_two(state: State) -> dict: next_total = state["total"] + 2 return { "total": next_total, "messages": state["messages"] + [f"add_two -> {next_total}"], }

4.3 构建图并编译

graph = StateGraph(State) graph.add_node("add_one", add_one) graph.add_node("add_two", add_two) graph.add_edge(START, "add_one") graph.add_edge("add_one", "add_two") graph.add_edge("add_two", END) app = graph.compile()

4.4 运行图

result = app.invoke({"total": 0, "messages": []}) print(result)

预期输出：

{'total': 3, 'messages': ['add_one -> 1', 'add_two -> 3']}

流程非常直观：0 + 1 = 1，然后1 + 2 = 3。两个节点按图结构依次执行，状态被自动传递。

这一步跑通之后，你就已经理解了 LangGraph 最核心的四个要素：State 存数据、Node 处理数据、Edge 决定流向、compile 之后可以 invoke 执行。

5. LangGraph 核心机制深入：节点、状态与条件路由

上一步只是串行执行，没有体现 LangGraph 真正的价值。这一节重点讲状态更新规则和条件路由，这是 LangGraph 和普通函数调用链最大的区别。

5.1 节点返回值如何合并到 State

每个节点函数返回一个 dict，LangGraph 会把它更新到 State 中。默认行为是“覆盖”：如果返回的 key 在 State 中已存在，就替换掉旧值。

上面例子里messages用了state["messages"] + [msg]，这是手动做追加。真实项目中，消息列表通常需要自动追加而不是覆盖。LangGraph 提供了Annotated和 reducer 机制来解决这个问题。

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

add_messages是 LangGraph 内置的 reducer，节点返回新消息时会自动追加到messages列表，而不是覆盖旧消息。这在对话类 Agent 中非常常用。

你可以同时使用两种字段：需要覆盖的用普通类型，需要追加的用Annotated加 reducer。

5.2 条件路由与分支控制

条件路由解决的是“根据当前状态决定下一步走哪个节点”的问题。这是 LangGraph 实战中最常写的代码。

先看一个综合示例：假设有一个工作流，第一节点根据用户输入生成一个数字，然后判断这个数字是否大于等于 10，决定走“直接结束”还是“再加一次”。

class State(TypedDict): user_input: int total: int messages: Annotated[list, add_messages] def generate_number(state: State) -> dict: return {"total": state["user_input"], "messages": ["generate_number 完成"]} def add_one(state: State) -> dict: return {"total": state["total"] + 1, "messages": ["add_one 完成"]} def too_small(state: State) -> dict: return {"messages": [f"当前值 {state['total']} 仍小于 10，流程结束"]} def route_by_total(state: State) -> Literal["add_one", "too_small", END]: if state["total"] < 10: return "add_one" if state["total"] == 10: return END return "too_small"

构建图：

graph = StateGraph(State) graph.add_node("generate_number", generate_number) graph.add_node("add_one", add_one) graph.add_node("too_small", too_small) graph.add_edge(START, "generate_number") graph.add_conditional_edges( "generate_number", route_by_total, { "add_one": "add_one", "too_small": "too_small", END: END, }, ) graph.add_edge("add_one", "generate_number") app = graph.compile()

测试：

print(app.invoke({"user_input": 8, "total": 0, "messages": []})) # 8 < 10，走 add_one 变为 9，重新进入 generate_number，此时 9 < 10，继续 add_one 变为 10，然后流程重新到 generate_number，10 不小于 10 且等于 10，走 END

这个例子展示了条件路由最典型的三种走向：

• 分支到一个普通节点。
• 分支回自身之前的节点形成循环。
• 分支直接到 END 结束流程。

热搜里提到的“conditional_edge 深度解析、循环检测”指的就是这里。LangGraph 允许图中有循环，因为generate_number -> add_one -> generate_number的路径是合法的。但要注意，循环必须有退出条件，否则会无限执行。LangGraph 默认每个节点最多执行一定次数，超出后会抛出GraphRecursionError，需要在compile(recursion_limit=...)或invoke(config={"recursion_limit": 50})中调整上限。

6. 高级特性：子图、并行分支与长期记忆

基础的路由和状态管理掌握之后，LangGraph 的高阶能力可以大幅简化复杂系统的开发。

6.1 子图嵌套

子图就是把一个编译好的图当作另一个图的节点来使用。适用于“先跑一个独立子流程，再继续主流程”的场景。

# 子图定义 sub_graph = StateGraph(State) sub_graph.add_node("sub_add", add_one) sub_graph.add_edge(START, "sub_add") sub_graph.add_edge("sub_add", END) sub_app = sub_graph.compile() # 主图使用子图 main_graph = StateGraph(State) main_graph.add_node("main_gen", generate_number) main_graph.add_node("sub", sub_app) # 直接把编译后的子图作为节点 main_graph.add_edge(START, "main_gen") main_graph.add_edge("main_gen", "sub") main_graph.add_edge("sub", END) main_app = main_graph.compile()

子图的好处是每个子图可以独立测试、独立复用，主图的逻辑不膨胀。多 Agent 协作时，可以把每个 Agent 封装成子图，在主图里统一调度。

6.2 并行分支

LangGraph 支持从一个节点出发，同时进入多个节点，最后在某个节点汇合。适合做“同时调用多个工具，最后汇总结果”的场景。

def tool_a(state: State) -> dict: return {"messages": ["工具 A 完成"]} def tool_b(state: State) -> dict: return {"messages": ["工具 B 完成"]} def aggregate(state: State) -> dict: return {"messages": ["汇总完成"]} graph = StateGraph(State) graph.add_node("tools_a", tool_a) graph.add_node("tools_b", tool_b) graph.add_node("aggregate", aggregate) graph.add_edge(START, "tools_a") graph.add_edge(START, "tools_b") graph.add_edge("tools_a", "aggregate") graph.add_edge("tools_b", "aggregate") graph.add_edge("aggregate", END) app = graph.compile()

这里需要理解一点，LangGraph 的并行是异步并发执行的，但最终结果仍然会汇总到同一个 State。汇合节点aggregate会在所有上游节点都完成后才执行，这种机制叫 Fan-out / Fan-in。

6.3 长期记忆与检查点

LangGraph 的“长期记忆”能力来自 Checkpointer。每次invoke结束后，可以把这一轮的状态保存到内存或数据库里，下一轮invoke时把历史状态加载出来继续。这就是热搜里“langgraph 长期记忆”的核心机制。

先用最简单的内存检查点：

from langgraph.checkpoint.memory import MemorySaver memory = MemorySaver() app = graph.compile(checkpointer=memory) # 第一轮调用，thread_id 用于区分会话 config = {"configurable": {"thread_id": "user-001"}} result1 = app.invoke({"user_input": 10, "total": 0, "messages": []}, config=config) print(result1) # 第二轮调用，带上同一个 thread_id，可以接着上一轮的状态继续 result2 = app.invoke({"user_input": 0, "total": 0, "messages": []}, config=config) print(result2)

生产环境中不要用内存检查点，服务一重启就没了。可以使用 SQLite 或 Postgres 作为检查点存储后端，LangGraph 官方提供对应实现，具体安装方式以官方文档为准。

检查点机制意味着每个用户会话可以拥有独立的记忆空间。只需要在请求时传入不同的thread_id，LangGraph 就会自动隔离不同会话的状态。这对做聊天机器人、客服系统非常关键。

7. 与 LangChain 集成：模型调用和工具编排

这一节解决一个高频疑问：LangGraph 和 LangChain 到底是什么关系，实际项目中怎么配合使用？

LangChain 是一套模型调用和工具集成的抽象层。你可以用langchain-openai的ChatOpenAI在节点里调用大模型，用langchain_core.tools定义工具，让模型决定是否调用工具。

LangGraph 则是状态和工作流编排层。它不关心模型是哪家的，也不关心工具的具体实现，只管理节点的执行顺序、状态传递和分支逻辑。

实际代码中可以这样配合：

from langchain_openai import ChatOpenAI from langchain_core.tools import tool @tool def get_weather(city: str) -> str: """查询城市天气""" # 这里做真实的天气查询 return f"{city} 天气晴" def call_model(state: State) -> dict: llm = ChatOpenAI(model="gpt-4o-mini") llm_with_tools = llm.bind_tools([get_weather]) result = llm_with_tools.invoke(state["messages"]) return {"messages": [result]} def call_tool(state: State) -> dict: # 解析模型返回的 tool_call，执行工具 return {"messages": [get_weather.invoke("北京")]} def route_by_tool(state: State) -> Literal["call_tool", END]: last_message = state["messages"][-1] if getattr(last_message, "tool_calls", None): return "call_tool" return END

这个结构就是 ReAct Agent 的原型：模型决定要不要调工具，调完工具把结果写回消息队列，再次交给模型判断。这种“思考-行动-观察”的循环是 LangGraph 最擅长处理的场景。

给一个选型建议：如果你的应用只是单个模型调用，不需要 LangGraph；如果你已经用了 LangChain 写工具调用，但分支和状态控制靠手写代码，强烈建议迁移到 LangGraph。它不是取代 LangChain，而是补上 LangChain 最不擅长的流程控制能力。

8. 可观测性与调试：如何看清每一步状态

图节点一多，最怕的是不知道流程走到哪里、状态变成什么样。LangGraph 提供了两种调试手段。

8.1 打印中间状态

最简单的方法是每个节点里打印关键信息：

def add_one(state: State) -> dict: next_total = state["total"] + 1 print(f"[add_one] total from {state['total']} to {next_total}") return {"total": next_total}

这种方式的缺点是日志散落在各节点里，图大了以后不好统一管理。建议在节点函数里用logging模块统一输出。

8.2 可视化图结构

LangGraph 可以把编译后的图导出为 Mermaid 格式，方便你在查看图结构时使用。在 Jupyter 或 Python 脚本里执行：

# 导出 Mermaid 文本，可以复制到支持 Mermaid 的工具中查看 print(app.get_graph().draw_mermaid())

这里说明一下：draw_mermaid()返回的是图结构的 Mermaid 文本，你可以复制到支持 Mermaid 渲染的工具（比如 draw.io、Typora、语雀等）中查看节点流向。这比用 echarts 自己实现节点可视化要省事得多。

如果你想做更精细的 Web 端可视化调试，可以考虑 LangSmith。它是 LangChain 团队提供的全链路追踪平台，能看到每一步节点的输入输出、耗时、token 消耗。使用方式是在环境变量里配置 LangSmith 的 API Key：

export LANGCHAIN_TRACING_V2=true export LANGCHAIN_API_KEY="你的-langsmith-api-key"

配置之后，每次invoke都会自动上报整条链路的执行记录。不过需要注意，使用 LangSmith 会上传少量追踪数据到官方平台，如果对数据敏感，关闭 tracing 只用本地日志即可。

9. 使用 FastAPI 包装接口服务与批量任务

LangGraph 本身不是一个服务，但编译后的app可以非常方便地嵌入到 FastAPI 服务中，对外提供 HTTP 接口。

9.1 创建 FastAPI 服务

示例代码：

from fastapi import FastAPI from pydantic import BaseModel from langgraph.graph import StateGraph, START, END from typing import TypedDict class State(TypedDict): total: int messages: list[str] def add_one(state: State) -> dict: next_total = state["total"] + 1 return {"total": next_total, "messages": state["messages"] + [f"add_one -> {next_total}"]} graph = StateGraph(State) graph.add_node("add_one", add_one) graph.add_edge(START, "add_one") graph.add_edge("add_one", END) app = graph.compile(checkpointer=MemorySaver()) fastapi_app = FastAPI() class RunRequest(BaseModel): thread_id: str total: int = 0 messages: list[str] = [] @fastapi_app.post("/run") def run(req: RunRequest): config = {"configurable": {"thread_id": req.thread_id}} result = app.invoke( {"total": req.total, "messages": req.messages}, config=config, ) return result @fastapi_app.get("/health") def health(): return {"status": "ok"}

启动服务：

uvicorn main:fastapi_app --host 127.0.0.1 --port 8000

9.2 curl 调用接口

curl -X POST http://127.0.0.1:8000/run \ -H "Content-Type: application/json" \ -d '{"thread_id": "u-001", "total": 0, "messages": []}'

返回结果自带上一次thread_id的历史状态，实现多轮会话记忆。

9.3 批量任务处理

LangGraph 的invoke是同步阻塞的，直接在循环里跑即可，但建议加上错误隔离和重试：

import time import logging def run_batch(inputs: list[dict], max_retries: int = 3) -> list: results = [] for item in inputs: for attempt in range(max_retries): try: result = app.invoke( {"total": item["total"], "messages": []}, config={"configurable": {"thread_id": item["thread_id"]}}, ) results.append(result) break except Exception as exc: logging.error(f"task {item['thread_id']} failed attempt {attempt + 1}: {exc}") if attempt == max_retries - 1: results.append({"error": str(exc)}) time.sleep(2 ** attempt) # 指数退避重试 return results

批量任务的关键点有三个：失败重试、任务日志、结果隔离。不要让一个失败任务拖垮整批任务。

10. 资源占用与性能观察

LangGraph 本身是纯 Python 编排层，CPU 和内存开销非常低。真正的资源瓶颈在节点内调用的模型。

分两种情况看：

• 调用云端 API 模型：本机只承担 HTTP 请求和状态管理，显存基本不占用。CPU 完全够用，瓶颈在 API 响应延迟和 Rate Limit。
• 调用本地模型：显存占用完全取决于你加载的模型大小。比如 7B 模型通常需要 6G 到 8G 显存，13B 模型需要 12G 到 16G 显存。具体数字以本地部署工具的实测为准。

性能观察建议：

• 用time.perf_counter()统计每个节点的耗时，定位哪一步最慢。
• 如果节点内部有多个模型调用，优先做并行化，配合 LangGraph 的 Fan-out 机制。
• 给每个节点加超时控制。避免某次 API 调用卡住整个流程。
• 如果相同请求频繁执行，考虑加缓存。LangGraph 允许在节点内部做结果缓存，但缓存 key 需要自己设计。

需要强调一点，不要把 LangGraph 的编排延迟和模型推理延迟混在一起。大多数情况下，一次图编排的纯框架开销在毫秒级，真正的耗时是大模型生成 token 的时间。

11. 常见问题与排查方法

下面是 LangGraph 开发中比较高频的问题和排查思路。

如果你的问题不在表里，优先看三样东西：报错堆栈、节点返回值的类型、State 里字段的当前值。绝大多数 LangGraph 问题都能通过这三样定位。

12. 学习路线与实战建议

如果你计划在七天左右从入门到能写项目，可以参考下面这个节奏，但核心是“先跑通再深入”：

• 第 1 天：安装环境，跑通最小图，理解 State、Node、Edge 三个概念。
• 第 2 天：写条件路由，掌握add_conditional_edges的用法。
• 第 3 天：写带循环的 Agent，理解recursion_limit。
• 第 4 天：给图加 Checkpointer，实现多轮对话记忆。
• 第 5 天：把 LangChain 的模型和工具集成到节点里。
• 第 6 天：用 FastAPI 包装接口，跑通批量任务。
• 第 7 天：选一个真实小项目，比如“PDF 问答 Agent”，把前面所有能力串起来。

工程实践上有几条建议：

• 第一版图不要设计得太复杂，先用最少的节点跑通主流程，再逐步加分支。
• 所有节点函数保持单一职责，一个节点只做一件事。
• 模型相关参数（模型名、温度、超时时间）放到配置里，不要写死在节点代码中。
• 批量任务必须加日志和错误隔离，否则上游一个 API 抖动，整批任务都会失败。
• 接口服务部署到公网前，务必加鉴权。不要在公网裸奔一个能让你花钱调模型的接口。
• 涉及用户隐私数据、版权素材时，先确认授权，再上线。

13. 总结

LangGraph 最值得尝试的点，是它把复杂 Agent 应用的流程控制从手写 if-else 中解放出来，让状态、分支、循环、记忆都变成图结构的一部分。而且它和 LangChain 天然互补，团队如果已经在用 LangChain，迁移成本并不高。

拿到一个 LangGraph 项目，最先应该验证三件事：最小图能不能跑通、条件路由是否正确走到目标分支、跨轮记忆是否按 thread_id 隔离。这三件事验证通过，框架的骨架就算掌握了。

最容易踩的坑也有三个：第一个是节点返回值覆盖了 State 里的旧字段，导致信息丢失；第二个是条件路由分支名写错，GraphRecursionError反复出现；第三个是生产环境用了内存检查点，服务重启后对话记忆全部丢失。

后续可以继续扩展的方向是：把本地模型接入 LangGraph，做多 Agent 协作系统，以及结合 LangSmith 做更精细的链路分析。LangGraph 的生态还在快速迭代，社区里已经有很多基于它搭建的 Agent 产品，值得跟进。

建议先跑通第 4 节的最小图，再花半天时间把条件路由和记忆两个机制玩熟。这套东西学会之后，写复杂 Agent 的维护成本会明显下降。