企业官网建设流程全解析

LangGraph 不是又一个“套着 LangChain 壳”的编排工具，而是真正把 Agent 流程当成一张图来跑的执行引擎。它适合处理条件路由、循环调用、并行分支、子图嵌套、长期记忆这类复杂状态流。本文不是概念搬运，直接按“核心能力 -> 环境搭建 -> StateGraph 最小实例 -> conditional_edge 深度解析 -> 子图 -> 状态修改 -> 持久化 -> API 封装 -> 性能观察 -> 排错”的顺序展开，你可以照着跑通一整套 LangGraph 实战链路。

1. LangGraph 核心能力速览

在动手之前，先把关键信息放在前面，方便判断这个框架适不适合你的项目。

一句话总结：LangGraph 的入门门槛不在硬件，而在理解“状态图”的执行思维。先跑通最小实例，再逐步加入条件路由、子图和持久化。

2. LangGraph 与 LangChain 的区别

LangChain 和 LangGraph 经常被一起讨论，但两者解决的问题层次不同。

2.1 设计哲学差异

LangChain 的核心抽象是 Chain，也就是一个线性调用链：输入 -> Prompt -> 模型 -> 输出。虽然 LCEL（LangChain Expression Language）可以让多个组件组合成管道，但本质上仍是 DAG 的线性或者简单分支结构，面对“模型反复调用工具直到完成任务”这类循环场景会比较别扭。

LangGraph 的核心抽象是 StateGraph。它把应用执行过程建模成一张有向图：

• State 是贯穿整张图的数据对象；
• Node 是实际执行逻辑的函数；
• Edge 决定数据从哪个节点流向哪个节点；
• conditional_edge 根据运行时状态决定下一步走向，因此天然支持循环和分支。

LangGraph 还提供了 Checkpointer 机制，可以把每一步的 state 持久化保存。这意味着一个 Agent 应用可以在多次调用之间保持上下文，而不是每次调用都从零开始。

2.2 选型建议：什么时候用什么

从团队成本考虑，如果项目当前只有线性链路，不要为了“新”而引入 LangGraph；一旦出现循环、人工介入、多 Agent 协作，LangGraph 的优势就会体现出来。

3. 适用场景与使用边界

LangGraph 适合这些项目：

• 客服 Agent：多个工具节点、条件转人工、对话状态持久化；
• 自动化数据分析：先判断用户意图，再决定调用哪个分析工具，结果不满足要求时循环重试；
• 报告生成流水线：检索资料 -> 生成大纲 -> 并行生成多个段落 -> 最终合并审校；
• 多 Agent 协作：主 Agent 根据任务动态调度子 Agent；
• 业务流程自动化：涉及审批、人工审核、异常回滚。

使用边界也要明确：

• LangGraph 不负责模型推理，你需要自己接 OpenAI、Anthropic、Ollama 或任意云模型 API；
• 它不解决模型输出质量问题，图结构只能帮助你控制流程，不能提升单次生成质量；
• 对强数学调度、分布式计算场景，Graph 框架本身不是最优解；
• 在涉及用户隐私、人脸、声音、版权素材的流程中，必须确认数据来源和模型调用符合授权要求，不要将未脱敏数据直接塞进外部模型。

4. 环境准备与安装部署

4.1 环境要求

LangGraph 是纯 Python 库，部署要求非常低。

如果你的环境已经有 Python 和 pip，直接进入安装。

4.2 安装 LangGraph

新建虚拟环境，避免和系统环境冲突：

python -m venv .venv # Windows .venv\Scripts\activate # macOS / Linux source .venv/bin/activate

安装核心依赖：

pip install -U langgraph langchain-openai

如果后续需要持久化到 SQLite 或 PostgreSQL：

pip install langgraph-checkpoint-sqlite # 或者 pip install langgraph-checkpoint-postgres

验证安装是否成功：

python -c "import langgraph; print(langgraph.__version__)"

说明：以上命令中的langgraph为 PyPI 包名，不同版本 API 有细微差异。执行成功后，就可以开始构建第一个图。

4.3 推荐项目目录结构

langgraph-demo/ ├── .venv/ ├── graphs/ │ ├── __init__.py │ ├── state.py │ ├── nodes.py │ ├── router.py │ └── build_graph.py ├── api/ │ └── main.py ├── tests/ │ └── test_graph.py ├── .env └── requirements.txt

把 state、node、graph 构建逻辑分开，后续维护会轻松很多。

5. 第一个 LangGraph 应用：从零构建状态图

直接写代码。下面是一个最小 StateGraph，包含两个节点：先给消息列表追加一段文本，再追加另一段文本。

from typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): messages: list def node_a(state: State) -> dict: return {"messages": state["messages"] + ["A"]} def node_b(state: State) -> dict: return {"messages": state["messages"] + ["B"]} # 1. 创建状态图 graph = StateGraph(State) # 2. 注册节点 graph.add_node("a", node_a) graph.add_node("b", node_b) # 3. 连接边 graph.add_edge(START, "a") graph.add_edge("a", "b") graph.add_edge("b", END) # 4. 编译 app = graph.compile() # 5. 运行 result = app.invoke({"messages": []}) print(result)

运行结果：

{'messages': ['A', 'B']}

这里有一个关键点：节点函数返回的是一个字典，LangGraph 会把返回的 key 与当前的 State 合并。如果节点函数返回None，表示不修改任何状态。

6. 条件路由与分支控制：conditional_edge 深度解析

这是 LangGraph 最核心的进阶能力。条件路由的作用是根据当前 state 动态决定下一步执行哪个节点。

6.1 基础条件路由

假设我们需要判断输入文本属于“紧急”还是“常规”，然后走不同节点。

from typing import TypedDict from langgraph.graph import StateGraph, START, END class QueryState(TypedDict): user_input: str level: str result: str def classify(state: QueryState) -> dict: if "紧急" in state["user_input"] or "加急" in state["user_input"]: return {"level": "urgent"} return {"level": "normal"} def route_by_level(state: QueryState) -> str: if state["level"] == "urgent": return "fast_handler" return "normal_handler" def fast_handler(state: QueryState) -> dict: return {"result": "已进入快速处理通道"} def normal_handler(state: QueryState) -> dict: return {"result": "已进入常规处理队列"} graph = StateGraph(QueryState) graph.add_node("classify", classify) graph.add_node("fast_handler", fast_handler) graph.add_node("normal_handler", normal_handler) graph.add_edge(START, "classify") # 条件边：classify 节点执行完，根据 route_by_level 结果走向不同目标 graph.add_conditional_edges( "classify", route_by_level, { "fast_handler": "fast_handler", "normal_handler": "normal_handler", }, ) graph.add_edge("fast_handler", END) graph.add_edge("normal_handler", END) app = graph.compile() print(app.invoke({"user_input": "这个问题很紧急，请加急处理"}))

运行结果：

{'user_input': '这个问题很紧急，请加急处理', 'level': 'urgent', 'result': '已进入快速处理通道'}

add_conditional_edges的参数有三个：

• 第一个是源节点名；
• 第二个是路由函数，接收当前 state，返回一个字符串；
• 第三个是映射字典，key 是路由函数返回值，value 是目标节点名。

也可以省略第三个参数，让路由函数直接返回目标节点名：

graph.add_conditional_edges("classify", route_by_level)

6.2 循环检测：Agent 工具调用循环

条件路由天然可以实现循环。下图逻辑是：Agent 节点调用模型，如果模型返回工具调用请求，就进入 tools 节点；否则直接结束。由于agent和tools之间存在环，LangGraph 需要限制最大执行步数，避免无限循环。

import os from typing import Annotated, TypedDict from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] llm = ChatOpenAI( model="gpt-4o-mini", api_key=os.getenv("OPENAI_API_KEY"), ) def agent(state: AgentState) -> dict: response = llm.invoke(state["messages"]) return {"messages": [response]} def tools(state: AgentState) -> dict: return {"messages": [{"role": "tool", "content": "模拟工具返回结果"}]} def should_continue(state: AgentState) -> str: last_message = state["messages"][-1] if getattr(last_message, "tool_calls", None): return "tools" return END graph = StateGraph(AgentState) graph.add_node("agent", agent) graph.add_node("tools", tools) graph.add_edge(START, "agent") # 关键：条件边可能回到 agent，形成环 graph.add_conditional_edges( "agent", should_continue, { "tools": "tools", END: END, }, ) graph.add_edge("tools", "agent") app = graph.compile() result = app.invoke( {"messages": [{"role": "user", "content": "调用一次工具并结束"}]}, config={"recursion_limit": 50}, )

这里的recursion_limit就是循环运行的最大步数上限。若没有合理设置，或者模型反复请求调用工具，LangGraph 会抛出GraphRecursionError。生产环境建议把日志加上，观察每个循环轮次的 tool_calls。

6.3 并行分支与 Send API

当多个子任务互相独立时，可以用并行分支加快执行。LangGraph 的经典的 Map-Reduce 模式通过 Send API 实现动态扇出。

from typing import Annotated, TypedDict from langgraph.constants import Send from langgraph.graph import StateGraph, START, END import operator class RepoState(TypedDict): topics: list sections: Annotated[list, operator.add] def plan_sections(state: RepoState) -> dict: return {"topics": state["topics"]} def continue_to_write(state: RepoState) -> list: # 为每个 topic 创建一个并行任务 return [Send("write_section", {"topic": topic}) for topic in state["topics"]] def write_section(state: dict) -> dict: # 单篇生成，放入 sections return {"sections": [f"### {state['topic']}\n内容草稿"]} graph = StateGraph(RepoState) graph.add_node("plan_sections", plan_sections) graph.add_node("write_section", write_section) graph.add_edge(START, "plan_sections") # 从 plan_sections 扇出到多个 write_section graph.add_conditional_edges("plan_sections", continue_to_write, ["write_section"]) graph.add_edge("write_section", END) app = graph.compile() result = app.invoke({"topics": ["LangGraph", "LangChain", "RAG"]}) print(result)

Send的作用是动态生成多个“虚拟边”和“子任务启动”，LangGraph 会尽量并行执行。如果你的流程是“先把一批文件切片，再逐个 embedding”，这种模式非常好用。

7. 子图 Subgraph：复杂流程模块化

子图就是“图里的图”。当主流程包含多个可复用的业务子流程时，可以把子流程封装成独立的 StateGraph，在主节点中调用。

from typing import TypedDict from langgraph.graph import StateGraph, START, END class SubState(TypedDict): sub_input: str sub_output: str def sub_node_1(state: SubState) -> dict: return {"sub_output": state["sub_input"] + " processed"} sub_graph = StateGraph(SubState) sub_graph.add_node("sub_node_1", sub_node_1) sub_graph.add_edge(START, "sub_node_1") sub_graph.add_edge("sub_node_1", END) sub_app = sub_graph.compile() class ParentState(TypedDict): text: str final_output: str def parent_node(state: ParentState) -> dict: # 直接调用子图 sub_result = sub_app.invoke({"sub_input": state["text"]}) return {"final_output": sub_result["sub_output"]} parent_graph = StateGraph(ParentState) parent_graph.add_node("parent_node", parent_node) parent_graph.add_edge(START, "parent_node") parent_graph.add_edge("parent_node", END) parent_app = parent_graph.compile() print(parent_app.invoke({"text": "hello"}))

运行结果：

{'text': 'hello', 'final_output': 'hello processed'}

子图的好处是：

• 让业务逻辑边界清晰；
• 子图可以独立测试；
• 子图可以有自己的 Checkpointer 和配置；
• 父节点只需要关心子图的输入输出结构，不关心内部细节。

需要注意：子图的输入通常需要从父节点 state 中取字段，并把子图输出映射回父节点返回的字典。不要让子图直接修改父节点的 state，而是通过返回值向上合并。

8. 状态管理与长期记忆

8.1 如何在节点函数改变 state 状态值

很多初学者在“如何修改 state”上踩坑。LangGraph 的规则很简单：

• 每个节点函数接收当前 state；
• 函数的返回值变成一个 dict，键值与 state 合并；
• 不返回的字段保持原样；
• 返回None表示不修改状态；
• 带 reducer 注解的字段使用 reducer 规则合并，未注解字段直接覆盖。

看一个字段覆盖与 reducer 的对比：

from typing import Annotated, TypedDict import operator class CounterState(TypedDict): count: int history: Annotated[list, operator.add] messages: Annotated[list, add_messages] def increment(state: CounterState) -> dict: # count 没有 reducer，返回的新值直接覆盖 return {"count": state["count"] + 1} def add_history(state: CounterState) -> dict: # history 有 operator.add reducer，返回的列表会追加而不是覆盖 return {"history": ["step"]}

实际项目中，最常用的 reducer 是add_messages。当节点返回{"messages": [新消息]}时，新消息会被追加进消息列表，而不是覆盖整个列表。这个特性在构建多轮 Agent 时非常关键。

8.2 Checkpointer 持久化

LangGraph 的状态默认只存在于单次invoke调用中。如果想在多次调用之间保留状态，就需要引入 Checkpointer。

from langgraph.checkpoint.memory import InMemorySaver from langgraph.graph import StateGraph, START, END checkpointer = InMemorySaver() # 编译时传入 checkpointer app = graph.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "user-001"}} # 第一次调用 app.invoke({"messages": [{"role": "user", "content": "你好"}]}, config) # 第二次调用，同一 thread_id 下可以读取之前的状态 app.invoke({"messages": [{"role": "user", "content": "你能记住我吗"}]}, config)

thread_id就是“会话标识”。同一个thread_id的多次调用共享一份 Checkpoint 状态；不同thread_id之间相互隔离。

生产环境建议使用持久化后端，例如 SQLite：

import sqlite3 from langgraph.checkpoint.sqlite import SqliteSaver conn = sqlite3.connect("checkpoints.sqlite", check_same_thread=False) checkpointer = SqliteSaver(conn) app = graph.compile(checkpointer=checkpointer)

8.3 长期记忆与 Store

Checkpointer 解决的是“对话轮次之间”的短期记忆。如果要跨用户长期保存用户偏好、永久知识，LangGraph 在较新版本中提供了BaseStore接口。实践中，也可以先使用 Redis、PostgreSQL 保存用户画像，在节点中手动读取写入，这样不依赖于特定框架 API，更稳定。

9. 将 LangGraph 封装为 API 服务

LangGraph 是库，不是服务。项目落地时通常要包一层 HTTP 接口。下面用 FastAPI 做一个最小封装。

9.1 FastAPI 接口服务

import os from typing import TypedDict from fastapi import FastAPI from pydantic import BaseModel from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END app = FastAPI() class QueryState(TypedDict): user_input: str answer: str llm = ChatOpenAI(api_key=os.getenv("OPENAI_API_KEY")) async def reply(state: QueryState) -> dict: # 简化：直接调用模型 resp = await llm.ainvoke(state["user_input"]) return {"answer": resp.content} graph_builder = StateGraph(QueryState) graph_builder.add_node("reply", reply) graph_builder.add_edge(START, "reply") graph_builder.add_edge("reply", END) graph_app = graph_builder.compile() class ChatRequest(BaseModel): message: str @app.post("/chat") def chat(req: ChatRequest): result = graph_app.invoke({"user_input": req.message}) return {"answer": result["answer"]}

启动服务：

uvicorn api:app --host 0.0.0.0 --port 8000

注意：FastAPI 的路由函数名如果叫app，可能会和 LangGraph 实例名冲突，建议区分命名。

9.2 Python 客户端调用示例

import requests url = "http://127.0.0.1:8000/chat" payload = {"message": "帮我写一封周报邮件"} response = requests.post(url, json=payload, timeout=60) print(response.json())

9.3 批量任务与并发

批量任务有两种常见做法：

• 客户端并发请求：用ThreadPoolExecutor并发调用 HTTP 接口；
• 服务端批处理：在节点内部接收一个批量输入列表，循环或并行处理。

用 Send API 做 Map-Reduce 更符合 LangGraph 的并行模式：

def continue_to_process(state: BatchState) -> list: return [Send("process_item", {"item": item}) for item in state["items"]]

生产环境建议给请求加超时和重试机制。比如用tenacity包装节点函数，当模型 API 临时报错时自动重试 3 次。

10. 资源占用与性能观察

10.1 LangGraph 本身不消耗 GPU

很多读者会问“LangGraph 需要多大显存”。答案很明确：LangGraph 这个框架本身就是纯 Python 的图执行引擎，它不运行 LLM，所以几乎不消耗显存。显存占用完全取决于你在节点里接入了什么模型。

如果使用云端 API，本地只需要几百 MB 内存；如果使用本地大模型，显存需求根据模型大小而定，这部分与框架本身无关。

10.2 观察 LLM 调用耗时

性能瓶颈通常在模型调用，而不是图调度。可以在节点函数里加入耗时统计：

import time def llm_node(state: State) -> dict: start = time.perf_counter() response = llm.invoke(state["user_input"]) elapsed = time.perf_counter() - start print(f"[timing] llm_node: {elapsed:.3f}s") return {"answer": response.content}

对于复杂图，可以用 LangGraph 自带的流式接口观察每个中间节点的输出：

for step in app.stream({"user_input": "test"}, config): print(step)

10.3 降低延迟与成本

• 先做小参数测试：减少模型调用次数、缩短输入上下文；
• 合理设置recursion_limit，避免无意义循环；
• 多个模型调用能合并成一次调用的，先合并；
• 可以并行执行的节点，用 Send API 或异步节点；不能并行的步骤，不要强行拆节点；
• 对重复出现的中间结果，考虑缓存或持久化。

11. 常见问题与排查方法

另外，关于“LangGraph 是否有 Rust 版本”：官方目前提供 Python 和 JS/TS 两个实现，没有官方 Rust 版。如果项目需要 Rust 生态接入，通常是内部封装 Python 子进程或者用 HTTP 调用 LangGraph 服务。

关于“ECharts 与 LangGraph 能否实现节点可视化”：LangGraph 官方支持把编译后的图导出为 Mermaid 格式，可以用app.get_graph().draw_mermaid()获取 Mermaid 文本，再放到支持 Mermaid 的文档工具中展示。ECharts 本身不支持 Mermaid 解析，若一定要在 ECharts 中渲染，需要自己把节点和边数据解析为 ECharts graph 类型的 JSON 结构。

12. 最佳实践与工程化建议

12.1 小参数先行，逐步加复杂功能

第一次跑 LangGraph，先构建一个只有两个节点的线性图。确认状态读写和invoke调用没问题后，再加入条件路由、循环和 Checkpointer。复杂图调试起来很痛苦，不要一次性堆所有功能。

12.2 保留一套最小可运行配置

把最小图示例单独存成脚本，例如quickstart.py。后续每次改动只在业务图目录里进行，最小示例始终作为“环境是否正常”的检测基准。

12.3 日志与可观测性

生产环境建议在图中加入关键步骤日志：

• 每个节点进入、离开的时间；
• 每次模型调用的 token 数；
• 条件路由的实际走向；
• 每次invoke的thread_id和状态更新。

这些数据对排查“为什么 Agent 不按预期行动”非常关键。

12.4 模型文件、输入素材、输出结果分目录管理

data/ ├── inputs/ ├── checkpoints/ ├── outputs/ └── logs/

模型权重、缓存文件、临时文件不要混在一起，避免磁盘占满后难以清理。

12.5 接口服务要限制访问范围

如果封装成 HTTP 服务，关注以下几点：

• 服务绑定到127.0.0.1而不是0.0.0.0，除非确实需要外部访问；
• 加接口鉴权，至少是静态 Token；
• 限制单用户请求频率；
• 请求体大小和超时时间需要显式配置。

12.6 合规与授权提醒

LangGraph 本身只是编排框架，但其中的数据流可能涉及敏感内容。使用用户数据、第三方内容、人脸、声音、版权素材时，请确认数据来源合法、模型调用符合服务商条款、输出内容经过人工审核。商用场景尤其要对最终输出做复核，不要盲目全自动化。

13. 总结与学习路线

LangGraph 最值得尝试的点是它的条件路由 + 循环 + Checkpointer 组合。先用最小的 Agent 工具调用循环跑通效果，再叠加并行子任务和长期记忆，最后包成 FastAPI 服务。

这里有一个最容易踩的坑：不要按 LangChain 的“线性 Chain”思维去写 LangGraph，遇到“我想在节点之间跳转”的需求，直接想清楚 State 的状态流转，然后把它拆成节点和条件边描述。

后续可以继续扩展的方向：

• 多 Agent 协作图：一个主图调度多个子图；
• 人工介入节点：遇到审核需求时暂停图执行，等待人工确认后继续；
• 流式输出：把流式 token 传递给前端；
• 可视化监控：把图执行状态导出到外部监控面板。

如果是要从零学习 LangGraph，建议按这条路线走：先读官方文档中的概念部分，再跑一遍本文的环境与最小示例，然后自己改一个带条件路由的小 Demo，最后用 FastAPI 包一个接口。不要把精力花在记 API 名上，多用print(step)和调试小脚本观察状态流转，比死记文档更有效。建议收藏备用，后面做 Agent 项目时可以直接回来对照。