企业官网建设流程全解析

这次我们来看一个在 AI Agent 开发里讨论度非常高的开源框架：LangGraph。它是 LangChain 团队推出的智能体编排框架，解决的问题很具体——大模型单次问答做不了复杂任务，但多步调用、工具选择、状态流转、人工审核、断点续跑这些事，手写代码又容易写成一团乱麻。LangGraph 用图结构把整个流程组织起来，节点干什么、边按什么条件走、中间状态存在哪，全部显式定义，代码可读性比传统 ReAct 手写循环高一个量级。

这个项目的核心特点可以概括为五条：第一，图状态编排，智能体的每一步用节点和边来表达；第二，内置状态管理与 checksum 持久化，支持断点续跑；第三，支持人工介入，流程可以在指定节点暂停，确认后再继续；第四，支持流式输出与并发分支，适合复杂工作流；第五，框架本身不依赖显卡，CPU 也能跑，显存占用完全取决于你接在哪一个大模型服务上。本文会从环境准备、安装部署、核心组件、两套代码实战、服务化封装到性能排查，完整走一遍 Agent 开发流程。

先说明门槛。LangGraph 本身是 Python 框架，不直接推理大模型，所以不需要高配显卡；如果你接 OpenAI 或国内大模型 API，只需要网络和 Key；想完全本地化，也可以接 Ollama 这类本地模型服务。本文演示用的设备是普通开发机，重点观察的是代码能不能跑通、流程可读性如何、接口好不好接，而不是显卡参数。所以写这篇教程时不会出现“4060 占用 7G”这类结论，实际资源占用以你选择的大模型服务为准。

适合的读者很明确：已经会调 Prompt，但想从“单轮对话”升级到“多步骤智能体”的开发者；RAG 项目里流程已经复杂到不知道如何继续维护的人；需要在团队里快速搭建一套可解释、可审计、可人工介入的 Agent 工作流的人。下面直接进入正题。

1. LangGraph 核心能力速览

先把最关键的参数和规格放在前面，方便你快速判断这个框架适不适合你的项目。

从材料看，LangGraph 最大的竞争力不是某一个组件，而是“把流程画成图，让每一条路径都可解释”。构建出的图可以直接用invoke同步执行，也可以astream异步流式输出，这给了工程化集成的空间。

2. 智能体架构与核心组件

LangGraph 的架构本质上是用图论组织智能体的执行流程。一个智能体不再是一个黑盒 Prompt，而是多个节点组成的拓扑结构。节点负责计算，边负责跳转，状态对象在节点之间传递。

2.1 State：整个流程的数据模型

State 定义了这个工作流“每一刻都拿着什么数据”。你可以把它理解成一张放在所有节点中间的消息总线。LangGraph 允许多个节点往 State 里追加字段，也支持用 reducer 控制合并逻辑。

最常用的就是对话消息列表。LangGraph 提供的add_messagesreducer 会自动把新消息追加到已有消息列表中，而不是覆盖掉，这对 Agent 的上下文累积至关重要。State 设计直接影响后续所有节点的输入输出，是整个图的“地基”。

2.2 Node：节点就是 Python 函数

Node 是图的最小执行单元，本质就是一个普通函数：接收 State，返回 State 的一部分更新。你不需要继承任何基类，不需要写装饰器标记，这降低了入门门槛。

通常一个节点只做一件事，比如“调用大模型生成回复”“解析工具结果”“检查文本质量”。做多件事的节点会让图不可读，也不利于复用。这就是为什么 LangGraph 智能体需要拆节点，而不是在函数里堆逻辑。

2.3 Edge 与条件边：决定下一步去哪

边分为普通边和条件边。普通边表示“执行完 A 之后一定去 B”，条件边表示“根据当前 State 的计算结果动态决定下一步”。条件边是 Agent 和固定流水线最大的区别：模型决定是否需要调用工具，工具调用结果决定是继续调用还是输出最终回答。

条件边的返回结果会映射到不同节点名称。例如tools_condition返回"tools"，图就进入工具执行节点；返回END，图就结束。

2.4 Checkpointer 与持久化

Checkpointer 是 LangGraph 支持断点续跑的关键组件。它把每一步的 State 快照保存下来，每个会话用thread_id标识。当流程被人工暂停或中断时，同一thread_id可以通过传入相同配置继续执行。

默认常用MemorySaver，把快照放在内存里，适合本地开发和测试；生产环境建议换成数据库存储的 checkpointer。这个机制让“人审 + Agent 自动执行”的组合落地非常自然。

核心组件对照表如下：

3. 适用场景与使用边界

LangGraph 适合的典型场景包括：多步骤任务拆解、RAG 问答后需要自动执行操作的流程、根据用户输入动态选择工具的 Agent、需要加入人工确认环节的内容生产流水线、跨轮次任务状态保持的客服系统。

使用边界要拎清楚。它不适合单轮 Prompt 直出任务，这类任务直接用大模型 API 更轻；它也不适合承载重型训练或数据处理，图编排不负责计算加速；它更不适合替代业务系统，Agent 只是其中一层决策和执行框架。

需要特别强调的是，接入在线大模型 API 时，会把文本内容发送到模型服务商，涉及敏感数据要确定脱敏方案；调用工具时，要考虑该工具是否允许高频访问。本文所有演示均在本地测试环境完成，输出结果用于功能验证，不做公网生产部署。

4. 环境准备与前置条件

动手之前先检查环境，这套清单适用于大部分 LangGraph 项目。

• Python 3.9+，建议 3.10 或 3.11，避免部分依赖在未来版本出现兼容问题。
• 建议使用虚拟环境，例如venv或conda，不要让依赖污染系统 Python。
• 需要一个可访问的大模型服务地址和 API Key。
• 如果接本地模型，需要额外配置 Ollama 或 vLLM 等推理服务。
• Windows 用户建议确认终端使用的是 PowerShell 或 Git Bash，个别依赖编译报错需要先装对应版本的 Visual C++ 运行库。
• 防火墙或代理策略需要允许访问你选择的模型服务域名。

虚拟环境创建示例：

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

安装依赖这一步在下一章统一说明。第一次接触 LangGraph 的同学不要跳步，虚拟环境隔离能省掉大量“为什么别人能跑我不能跑”的问题。

5. 安装部署与项目初始化

LangGraph 的安装非常简单。以下是基础依赖，实际项目按需增减。这里以常见组合为例：

pip install langgraph langchain-openai langchain-core

如果你接的是其他厂商的模型，也可以安装对应的 LangChain 适配包，比如langchain-anthropic、langchain-community等。部分国内模型服务支持 OpenAI 兼容接口，可以直接用langchain-openai指定base_url。

项目目录建议这样组织：

langgraph-demo/ ├── .venv/ # 虚拟环境 ├── config.yaml # 模型配置 ├── tools/ # 自定义工具 │ └── device_tools.py ├── graphs/ # 图定义 │ ├── basic_agent.py │ └── workflow.py └── server.py # API 服务

比较关键的是模型配置。不要把 API Key 直接写死在代码里，用环境变量或者.env文件管理更稳妥：

OPENAI_API_BASE=https://your-endpoint.example.com/v1 OPENAI_API_KEY=your-api-key MODEL_NAME=gpt-4o-mini

代码里读取环境变量：

import os from langchain_openai import ChatOpenAI model = ChatOpenAI( base_url=os.getenv("OPENAI_API_BASE"), api_key=os.getenv("OPENAI_API_KEY"), model=os.getenv("MODEL_NAME", "gpt-4o-mini"), )

到这里，LangGraph 项目已经初始化完成，可以开始写第一张图。

6. 代码实战一：基础 Agent 与工具调用

这一节直接演示一个带工具调用的 ReAct 风格智能体。场景是“设备状态查询 Agent”：用户提供设备编号，Agent 判断需要调工具，工具返回状态，Agent 汇总回答。

6.1 定义工具

from langchain_core.tools import tool @tool def get_device_status(device_id: str) -> str: """查询设备在线状态，输入设备编号，返回正常或离线。""" status_map = { "A003": "正常", "B204": "离线", "C999": "未知", } return status_map.get(device_id, "没有该设备的记录")

工具函数最上面必须写清楚描述，因为大模型靠这个描述决定要不要调用它。返回字符串是 LangChain 工具的标准做法。

6.2 定义 State 与节点

from typing import Annotated from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode, tools_condition class State(TypedDict): messages: Annotated[list, add_messages] model_with_tools = model.bind_tools([get_device_status]) def agent_node(state: State): response = model_with_tools.invoke(state["messages"]) return {"messages": [response]} def build_basic_agent(model): builder = StateGraph(State) builder.add_node("agent", agent_node) builder.add_node("tools", ToolNode([get_device_status])) builder.add_edge(START, "agent") builder.add_conditional_edges("agent", tools_condition, {"tools": "tools", END: END}) builder.add_edge("tools", "agent") return builder.compile()

这段代码一共做了四件事：定义全局消息状态；把大模型绑定工具，让模型具备工具调用能力；注册 agent 节点和 tools 节点；用条件边让模型自行决定是调工具还是直接结束。tools_condition是 LangGraph 预置逻辑，检查模型返回的内容中是否包含工具调用请求，无需自己判断 message 结构。

6.3 运行测试

graph = build_basic_agent(model) result = graph.invoke({ "messages": [("user", "帮我查一下设备 A003 的状态")] }) print(result["messages"][-1].content)

预期输出大致是这样一句话：设备 A003 的状态是正常。如果模型判断需要工具，但工具结果为空，则回答可能包含查询失败提示。

判断标准：进程正常结束；控制台能看到回答；工具相关日志显示节点从 agent 走到 tools 再回到 agent。

第一次跑如果报错，先确认三件事：模型地址是否正确、API Key 是否有权限、模型名称是否与服务商提供的一致。容易忽略的是，部分模型服务商不兼容工具调用格式，导致tools_condition永远返回 END，这类问题要查模型文档确认是否支持 OpenAI 风格的 function calling。

7. 代码实战二：工作流、分支循环与人工确认

基础 Agent 演示了动态循环，这一节看两个在工作中更常用的能力：条件分支和人工确认。场景是内容生产流水线：生成草稿 -> 质量检查 -> 人工审核 -> 发布。没有通过质量检查的稿子直接重新生成。

7.1 条件分支与循环

import random def generate_draft_node(state: State): # 实际项目中这里替换为具体的生成逻辑 content = model_with_tools.invoke([("system", "写一段产品介绍，200字以内"), ("user", state["topic"])]) return {"messages": [content], "draft": content.content} def quality_check_node(state: State): # 简化判断：随机返回 pass 或 retry score = random.randint(0, 10) return {"quality_score": score} def should_publish(state: State): if state.get("quality_score", 0) >= 7: return "publish" return "retry"

from langgraph.checkpoint.memory import MemorySaver builder = StateGraph(State) builder.add_node("generate", generate_draft_node) builder.add_node("check", quality_check_node) builder.add_node("publish", lambda state: {"final_output": "已发布：" + state["draft"]}) builder.add_edge(START, "generate") builder.add_edge("generate", "check") builder.add_conditional_edges("check", should_publish, {"publish": "publish", "retry": "generate"}) builder.add_edge("publish", END) saver = MemorySaver() compiled_workflow = builder.compile(checkpointer=saver)

这里 retry 边指回 generate 节点，就构成了循环。你要限制重试次数的话，可以在 State 里加一个retry_count字段，在条件函数里判断次数，超过上限就走到人工节点。

7.2 人工确认节点

LangGraph 的人工介入不是写一个特殊节点，而是用interrupt_before参数让图跑到指定节点前停下：

config = {"configurable": {"thread_id": "daily-task-20260101"}} graph_paused = builder.compile( checkpointer=saver, interrupt_before=["publish"], ) partial_result = graph_paused.invoke( {"topic": "智能硬件设备的产品介绍"}, config=config, ) print("流程已在 publish 前暂停，等待人工确认") print(partial_result["draft"])

此时进程控制权交还给你。确认没问题后，用同一个thread_id再次调用，图会从暂停点继续：

final_result = graph_paused.invoke(None, config=config) print(final_result["final_output"])

注意第二次invoke的第一个参数传None，意思是“不新增输入，继续执行剩余节点”。这是 LangGraph 中断续跑的标准用法。人工确认内容正确后，信息流会继续进入 publish 节点，否则你可以在代码里选择重新生成或者丢弃本次任务。

7.3 判断效果

判断工作流改造是否成功，主要看三点：图能够停在 publish 前；第二次调用可以继续执行而不是从头开始；连贯执行时，条件分支在多次运行中表现出差异化路径。运行时可以在控制台打印节点名称，确认每一步。使用 MemorySaver 时，重启进程后快照会丢失，这只是测试用法，生产环境要把 checkpointer 持久化到数据库。

8. 接口 API、批量任务与服务化扩展

图编译好之后，下一步就是把它暴露成服务给上游系统调用。最直接的方式是包一层 FastAPI。

8.1 用 FastAPI 包装 Agent

from fastapi import FastAPI from pydantic import BaseModel app = FastAPI(title="LangGraph Agent Demo") class AgentRequest(BaseModel): prompt: str thread_id: str = "default-thread" class AgentResponse(BaseModel): status: int message: str thread_id: str @app.post("/agent/run", response_model=AgentResponse) async def run_agent(req: AgentRequest): try: config = {"configurable": {"thread_id": req.thread_id}} result = graph.invoke( {"messages": [("user", req.prompt)]}, config=config, ) return AgentResponse( status=0, message=result["messages"][-1].content, thread_id=req.thread_id, ) except Exception as exc: return AgentResponse(status=1, message=str(exc), thread_id=req.thread_id)

启动服务：

uvicorn server:app --host 127.0.0.1 --port 8000

这里要多说一句：同一个thread_id会对应同一个 checkpointer 快照状态，所以多轮会话只要复用thread_id，Agent 就能自动记住上下文。这是接口设计里最值得利用的一个点。

8.2 curl 调用

curl -X POST http://127.0.0.1:8000/agent/run \ -H "Content-Type: application/json" \ -d '{"prompt": "查一下设备 A003 的状态", "thread_id": "demo-001"}'

后端会返回 JSON，结构包含status、message和thread_id。接口能跑通，就可以接入其他工具链了。

8.3 批量任务设计

import csv import time tasks = [ {"id": "t001", "prompt": "查设备 A003"}, {"id": "t002", "prompt": "查设备 B204"}, {"id": "t003", "prompt": "查设备 C999"}, ] results = [] for task in tasks: try: config = {"configurable": {"thread_id": f"batch-{task['id']}"}} result = graph.invoke({"messages": [("user", task["prompt"])]}, config=config) results.append({ "task_id": task["id"], "answer": result["messages"][-1].content, }) print(f"任务 {task['id']} 完成") except Exception as exc: results.append({"task_id": task["id"], "error": str(exc)}) print(f"任务 {task['id']} 失败: {exc}") with open("batch_results.csv", "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=["task_id", "answer", "error"]) writer.writeheader() writer.writerows(results)

批量任务的要点是每条任务用独立thread_id，互不污染；任务处理要记录日志，失败任务单独标记；并发量如果大，可以用concurrent.futures.ThreadPoolExecutor控制并发数，但要注意模型服务的限流，这里不展开讲具体限流数值，以实际服务商为准。

9. 性能观察与常见问题排查

LangGraph 本身是编排层，节点之间的跳转开销很小，瓶颈几乎都被大模型推理占走。评判性能时不要看框架微秒级调度，要看以下几个维度：单次模型调用的延迟、工具调用次数、图在数据库中持久化的读写耗时、以及用户在感知上的等待时间。

流式输出是改善观感的一个重要手段。LangGraph 提供了astream方法，可以让前端在模型生成过程中逐 token 收到内容，而不是等整段完成后一次返回：

async for chunk in graph.astream( {"messages": [("user", "查一下 A003 状态")]}, config={"configurable": {"thread_id": "stream-001"}}, ): print(chunk)

为了让排查效率更高，把常见问题整理成了这张表：

排查时建议先打印节点执行顺序。在编译后的图对象上，可以通过graph.get_graph()打印出图结构的文本表示，确认拓扑是否符合预期。

10. 最佳实践与后续扩展

最后这部分给出几条工程化建议，都是从实际项目里沉淀出来的经验。

第一，先小后大。任何新项目第一次运行都先用最小图验证：只有 start、一个写死输出的节点、end。模型都先不接，确认图能跑通，再逐步加节点和工具。这一步能省掉大量反复排查的精力。

第二，State 设计要克制。不要把状态对象膨胀成“什么都有”的数据库表，节点之间传参与收参遵循最小化原则。需要持久化的业务字段单独设计，不要全部塞进 messages 里。

第三，人工确认只放在关键节点。用户不可能每次都审核整条链路，把interrupt_before用在“发布前”“确认执行外部动作前”这类真正需要人把关的位置即可。自动化能覆盖的检查，优先交给条件节点判断。

第四，日志和可观测性提前做。每个节点函数入口打印出当前执行节点名和时间，失败时记录完整 State 快照。生产环境里的 Agent 调试问题，九成发生在日志缺失的角落。

第五，接口服务要控制访问范围。FastAPI 示例里用的 127.0.0.1，只允许本机调用；跨机器调用需要配合认证鉴权。涉及人脸、声音、版权素材识别与生成时，必须在确认授权后才纳入工作流；调用第三方工具接口时，要遵守对方平台的使用条款，工具执行的每一步都应该在日志里留痕。

后续扩展方向可以考虑三个方面。一是多 Agent 协作，在同一个图里注册多个 agent 节点，让不同节点负责不同领域；二是并行分支，在节点返回send指令实现 fan-out 并行处理，适合 RAG 多路召回；三是生产部署，把 MemorySaver 换成数据库 checkpointer，并考虑 LangGraph Platform 或自建服务。本文没有编造任何具体部署平台参数，生产环境的架构选择需要你按实际项目规模和成本来评估。

最后说一个比较容易踩的坑：版本。LangGraph 迭代很快，网上很多教程的 API 写法在你安装的版本里可能已经变了。遇到报错时不必纠结旧代码，优先打开本地安装的langgraph包看实际方法名和参数。本文所有代码以常见 0.7.x 版本 API 风格为基准，运行时要结合你本机安装版本微调。

最有价值的第一个验证目标，是把基础工具调用 Agent 跑通。这个目标实现之后，你对 State、Node、条件边和工具绑定的理解就串起来了，后面再学工作流、持久化和服务化都会快很多。先把最小图跑起来，比背十张架构图都有用。