企业官网建设流程全解析

这次我们直接看 LangGraph。它不是一个概念 Demo，而是 LangChain 团队开源的智能体编排框架，用来把 Agent 从“模型调用一次就结束”的线性流程，改造成“可以循环调用工具、可以分支、可以暂停等人工确认、还可以断点续跑”的图结构工作流。很多人学 Agent 开发卡在两点：一是组件太多不知道先看哪个，二是照抄代码后一跑就报错、不知道去哪查。这篇文章把智能体架构、核心组件、代码实战、接口批量任务一次性拆开讲，所有示例都按最小可运行的标准给出来，看完可以直接在本地跑通。

先列几个最值得关注的点：LangGraph 支持有状态多轮会话、支持条件分支和循环、支持人工审批打断点、支持执行状态持久化、支持流式输出，并且它有很强的服务化能力，可以包装成本地接口或接入批量任务队列。从算力角度看，LangGraph 本身只是编排层，跑纯逻辑演示 CPU 就够了；真正吃显存的是底层 LLM 和向量检索部分，这一点我们在后面资源占用章节单独说。

这篇文章的实操路线是：先讲架构和核心组件，再给环境准备清单，然后依次跑通“最小 StateGraph”“带工具调用的 ReAct Agent”“人工审批与流式输出”“API 与批量任务”。适合正在学 Agent 开发、想把 LLM 接进业务流程、或者被“结果不可控、多轮对话断上下文、没法人工干预”这类问题卡住的人。下面直接进入正文。

1. LangGraph 智能体核心能力速览

补充一点：LangGraph 不是替代 LangChain 的存在，而是继承了 LangChain 的组件生态，比如 Tool、Message、模型接口等，同时把“执行流程”的控制权从框架手里拿回用户手里。你完全可以只学 LangGraph 不碰 LangChain 的高阶封装，但通常还是建议把基础概念过一遍。

2. 智能体架构拆解：从 ReAct 到图编排

早期用 LangChain 写 Agent，很多方案落在 AgentExecutor 上。它的逻辑可以理解为“模型决定下一步 → 执行工具 → 把结果放回消息列表 → 再次交给模型”，这个循环在内部执行，用户能干预的点非常有限。遇到复杂业务，比如步骤多、需要人工审批、需要不同分支策略，AgentExecutor 的线性封装就开始别扭。

LangGraph 的思路是把 Agent 画成一张图。每个“操作”是一个节点，比如“模型推理”“调用工具”“输出最终回答”；节点之间用边连接，普通边表示必然跳转，条件边表示根据状态决定下一步走向哪里。这样设计之后，三个最大优势是：

LangGraph 智能体的核心组件可以浓缩为一张表：

这套组件说起来抽象，但实际写代码只有几个动作：定义状态、写节点函数、画边、编译图、调用。接下来我们把环境准备好，然后开始跑最小示例。

3. 环境准备与依赖安装

LangGraph 是一个 Python 库，安装不复杂，但建议不要直接装在全局 Python 环境里，避免和项目依赖冲突。推荐用 venv 或 conda 隔离环境。

先从基础环境开始：

• Python 版本：建议 Python 3.10 以上，具体以你使用的 LangGraph 版本说明为准。
• 操作系统：Windows / macOS / Linux 都可以，Linux 部署最省心。
• 基础依赖：langgraph、langchain-core、langchain-community（可选）、transformers 或 ollama 等模型接入方式。

安装命令示例：

python -m venv .venv source .venv/bin/activate # Windows 下执行 .venv\Scripts\activate pip install -U langgraph langchain-core

如果后续要用到工具调用和人机协作，再装可选组件：

pip install langchain-community langchain-openai # 如果希望打印图结构用于调试，可额外安装 pip install grandalf

模型接入这一块需要单独说明：LangGraph 本身不带模型。你可以接 OpenAI、Ollama、HuggingFace Transformers、vLLM 等多种后端。在纯本地演示阶段，最省钱的做法是用一个固定逻辑的“假 LLM”代替真实模型，先把图和节点跑通。等确认流程没有问题了，再接真实模型。

这样安排的好处是：不用先付 API 费用，也不用担心本地显存不够。后面实战示例我会用注释里的方式说明如何替换成真实模型。

4. 最小可运行示例：先跑通一个 StateGraph

先不搞复杂 Agent，我们从最小状态图开始。目的是理解三个概念：State 怎么定义、Node 怎么执行、Edge 怎么跳转。

from typing import Annotated, TypedDict from langgraph.graph import StateGraph, END def merge_logs(left: list, right: list) -> list: """将多条日志合并追加，避免新返回覆盖旧状态。""" return left + right class State(TypedDict): counter: int log: Annotated[list, merge_logs] def node_a(state: State) -> dict: current = state.get("counter", 0) + 1 return {"counter": current, "log": [f"node_a 执行，当前计数: {current}"]} def node_b(state: State) -> dict: current = state.get("counter", 0) + 10 return {"counter": current, "log": [f"node_b 执行，当前计数: {current}"]} graph = StateGraph(State) graph.add_node("a", node_a) graph.add_node("b", node_b) graph.set_entry_point("a") graph.add_edge("a", "b") graph.add_edge("b", END) app = graph.compile() result = app.invoke({"counter": 0, "log": []}) print(result)

这段代码做了什么：

1. 定义了一个State，它有两个字段：counter和log。
2. log字段用Annotated指定了 reducer 为merge_logs，这样每一次节点返回的日志不会覆盖之前的值，而是追加。
3. 图从a节点开始，执行完后跳到b，再从b结束。
4. compile()将图编译成可调用对象，invoke()传入初始状态执行。

执行后预期输出类似：

{'counter': 11, 'log': ['node_a 执行，当前计数: 1', 'node_b 执行，当前计数: 11']}

这里最需要理解的是 reducer 的作用。如果不加Annotated[list, merge_logs]，节点b返回的log会直接覆盖node_a写入的日志，最后你只能看到node_b的那一条。这个规则对 Agent 的消息列表尤其重要，因为每一轮模型输出和工具结果都要追加到历史里，而不是互相覆盖。

跑通最小示例后，可以试一个简单改动：把graph.add_edge("a", "b")换成条件边，让node_a自己决定下一步是去b还是直接结束。

def choose_next(state: State) -> str: if state["counter"] >= 5: return "b" return END graph.add_conditional_edges("a", choose_next, {"b": "b", END: END})

条件边是 LangGraph 智能体最重要的能力之一，因为“是否继续调用工具”本质上就是一个条件判断。下一节我们进入实战，把条件边和工具调用结合起来。

5. 实战：构建带工具调用的 ReAct Agent

这一节的示例会模拟一个真正 Agent 的思考循环：输入用户问题 → 判断是否需要调用工具 → 调用工具 → 把工具结果放回消息列表 → 再次判断是否结束。

我们用一个简单的天气查询工具做演示。为了让你跑起来不依赖任何外部 API，示例里先用固定逻辑代替模型决策，代码注释里也给了替换成真实模型的方式。

from typing import Annotated, TypedDict from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langchain_core.messages import HumanMessage, AIMessage from langchain_core.tools import tool from langgraph.prebuilt import ToolNode @tool def get_weather(city: str) -> str: """查询指定城市当前的天气情况。""" return f"{city} 当前 24℃，多云，有微风。" class AgentState(TypedDict): messages: Annotated[list, add_messages] def agent_node(state: AgentState) -> dict: # 演示环境不接外部模型，用固定规则模拟模型决策。 # 真实开发中替换为类似： # model = chat_ollama(...) 或 model = ChatOpenAI(...) # message = model.bind_tools(tools).invoke(state["messages"]) # return {"messages": [message]} last_content = state["messages"][-1].content if "天气" in last_content: # 模拟模型返回一个 tool_call，要求执行 get_weather 工具 return { "messages": [ AIMessage( content="", tool_calls=[ { "id": "call_demo_001", "name": "get_weather", "args": {"city": "上海"}, } ], ) ] } return {"messages": [AIMessage(content="我没有理解你的意图，当前演示只支持天气查询。")]} def should_continue(state: AgentState) -> str: last_message = state["messages"][-1] if getattr(last_message, "tool_calls", None): return "tools" return END # 初始化图 tools = [get_weather] tool_node = ToolNode(tools) graph = StateGraph(AgentState) graph.add_node("agent", agent_node) graph.add_node("tools", tool_node) graph.set_entry_point("agent") graph.add_edge("tools", "agent") graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END}) app = graph.compile() result = app.invoke( {"messages": [HumanMessage(content="上海天气怎么样？")]}, ) for message in result["messages"]: print(message)

运行后你会看到完整的事件链：用户消息 → 带tool_calls的 AIMessage → 工具执行结果 ToolMessage → 再次回到 agent 节点。这里的关键点：

• add_messages是 LangGraph 内置的消息合并 reducer，专门用于给消息列表追加新消息。
• ToolNode会自动读取 AIMessage 中的tool_calls字段，执行对应工具，并把结果包装成 ToolMessage 写回状态。
• should_continue通过判断最后一条消息是否带tool_calls来决定是否继续调用工具。如果模型还在要工具，图就继续循环，直到模型返回纯文本答案。

如果把示例中的问题改成“你好”，流程会走另一个分支：should_continue发现没有工具调用，图直接结束。这就是条件边的作用：一个图里可以同时存在“继续循环”和“结束”两条路径。

替换成真实模型时，只需要把agent_node中间的模拟逻辑替换为：

from langchain_ollama import ChatOllama # 以本地 Ollama 为例 model = ChatOllama(model="qwen3:8b", temperature=0) model_with_tools = model.bind_tools(tools) def agent_node(state: AgentState) -> dict: response = model_with_tools.invoke(state["messages"]) return {"messages": [response]}

这样你就得到了一个能真正调用本地模型的 LangGraph Agent。Ollama 等本地模型服务只要按各自方式启动即可，调用接口不占用 LangGraph 本身的资源。

6. 人工审批、流式输出与持久化

跑通基础 Agent 后，下一步解决企业场景里最常见的三个问题：怎么在敏感操作前插入人审、怎么实现打字机式输出、怎么保存多轮会话状态。

6.1 人工审批节点

早期 Agent 最大的风险是“不可控”：模型可能直接执行危险工具。LangGraph 的interrupt机制可以把流程停在某个节点，等服务端拿到人工输入后才继续。

from langgraph.types import interrupt from langgraph.checkpoint.memory import MemorySaver def approval_node(state: AgentState) -> dict: user_decision = interrupt( { "type": "approval", "content": "是否批准执行工具调用？请回复 yes 或 no。", } ) if user_decision.get("approve") is True: return {"messages": [HumanMessage(content="审批通过，继续执行。")]} return {"messages": [HumanMessage(content="审批拒绝，流程终止。")]} # 在图中加入审批节点，并把它放在工具调用之前 graph.add_node("approval", approval_node) graph.add_edge("approval", "tools") graph.add_edge("agent", "approval") # 实际按你的流程调整边的连接

注意interrupt的导入路径在不同版本中可能有差异，旧版本可能从langgraph.constants导入。如果你照代码写报错 ModuleNotFoundError，先检查当前安装的 langgraph 版本，再按对应的包路径调整导入。

6.2 检查点持久化

多轮对话要求每次请求都能读写同一个会话上下文，LangGraph 的compile(checkpointer=...)就是干这个的。最简单的内存版检查点，适合熟悉机制：

checkpointer = MemorySaver() app = graph.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "user-001"}} # 第一轮对话 app.invoke({"messages": [HumanMessage(content="上海天气怎么样？")]}, config) # 第二轮对话，同一 thread_id 下状态延续 app.invoke({"messages": [HumanMessage(content="那明天呢？")]}, config)

同一个thread_id下的消息历史会一直保留，不同thread_id的会话彼此隔离。这也是 LangGraph 实现批量任务的基础。生产环境建议把检查点换成数据库或文件存储方案，避免进程重启后数据丢失；MemorySaver 适合验证，不适合正式用。

6.3 流式输出

Llama、ChatGPT 式的打字机输出在 LangGraph 里也支持。最常用的方式是按节点更新来流式读取：

inputs = {"messages": [HumanMessage(content="上海天气怎么样？")]} for event in app.stream(inputs, config, stream_mode="updates"): for node_name, node_output in event.items(): print(node_name, node_output)

stream_mode="updates"会在每个节点执行完后返回该节点的输出。如果你接的是真实 LLM 接口，还有更细粒度的 token 流式模式，可以在推理阶段拿到 token 逐步输出。流式输出的价值不仅仅是体验好，更重要的是长任务里用户可以尽早看到进度，而不是干等着接口超时。

7. LangGraph 接口 API 与批量任务

LangGraph 跑通之后，要接到真实业务里，大多数情况下需要包一层接口服务。 LangGraph 本身没有规定你是用 FastAPI、Flask 还是官方 Server，社区常见的两种思路都可用。下面给一个用 FastAPI 包装的最小示例，最简单也最容易控制权限和扩展。

from fastapi import FastAPI from pydantic import BaseModel from langchain_core.messages import HumanMessage server = FastAPI() class ChatRequest(BaseModel): thread_id: str message: str class ChatResponse(BaseModel): answer: str @server.post("/chat", response_model=ChatResponse) def chat(req: ChatRequest): config = {"configurable": {"thread_id": req.thread_id}} result = app.invoke({"messages": [HumanMessage(content=req.message)]}, config) return ChatResponse(answer=result["messages"][-1].content)

启动接口服务：

uvicorn main:server --host 0.0.0.0 --port 8000

启动后访问http://127.0.0.1:8000/docs可以直接调试接口。用 curl 测试请求：

curl -X POST http://127.0.0.1:8000/chat \ -H "Content-Type: application/json" \ -d '{"thread_id": "user-001", "message": "上海天气怎么样？"}'

返回结果类似：

{"answer":"上海 当前 24℃，多云，有微风。"}

接口跑通之后，批量任务的思路就简单了：每个任务分配一个独立thread_id，循环调用同一个接口，收集返回结果。生产级别要注意三件事：并发限流、失败重试、日志记录。

import time questions = [ "北京天气怎么样？", "广州天气怎么样？", "深圳天气怎么样？", ] def batch_run(questions: list[str]) -> list[dict]: results = [] for idx, question in enumerate(questions): try: config = {"configurable": {"thread_id": f"batch-{idx}"}} result = app.invoke({"messages": [HumanMessage(content=question)]}, config) results.append({"question": question, "answer": result["messages"][-1].content}) except Exception as exc: # 单条失败不影响整批任务 results.append({"question": question, "error": str(exc)}) time.sleep(0.1) return results

批量任务的工程要点：

• 每个线程一个thread_id，不要复用，否则上下文串话。
• 重试策略放在外层循环，超过 2 次失败的单独记录，避免死循环。
• 用队列服务承接大批量任务比 for 循环更稳，比如 Redis RQ、Celery、Arq 都可以。
• 每次调用要记录输入输出的 token 数和耗时，方便成本核算。

如果你想用官方服务化方式跑，可以按官方文档安装langgraph-cli并把graph.py放在指定目录启动本地开发服务器，它会自动生成 REST API 和可视化面板。不过不同版本配置差异较大，首次上手更推荐上面这段 FastAPI 代码，它完全可控且不依赖云服务。

8. 资源占用、性能观察与问题排查

8.1 资源占用怎么观察

LangGraph 编排层本身占用的内存和 CPU 很少。真正影响资源的是底层模型和检索组件：

本地跑模型时，显存占用不是固定数字，它会随上下文长度增加而上涨。输入越长、并发越多，显存和延迟越高。如果你的业务是几十个会话并行，建议认真做上下文裁剪，不要把所有历史永远堆在 messages 里。

调用云端 API 模型时，客户端基本无显存问题，但每个请求的 token 和成本都要记录。编排层可以做的优化是：控制单轮工具调用次数、裁剪历史消息、避免每次 reconstruct 整个图。

8.2 性能观察方法

给节点计时是一个简单而有效的做法：

import time start = time.perf_counter() result = app.invoke(inputs, config) print("total time:", time.perf_counter() - start)

想要更细，就在每个节点函数里打点。配合app.stream可以观察每一步的间隔，找出卡在模型调用还是卡在工具执行。

8.3 常见问题排查表

这里额外提一个高频问题：用MemorySaver跑多轮测试时第一次正常、第二次重启后上下文没了。这不是 bug，而是内存版检查点重启即清空。正式环境换用持久化后端。

9. 最佳实践与合规边界

9.1 工程实践建议

• 先假后真。先用固定逻辑跑通图结构，再接入真实 LLM，能节约大量调试时间。
• State 字段尽量少。只放节点真正需要的数据，削掉中间缓存字段，否则每个节点都要处理无关状态。
• 工具权限最小化。不要一股脑把“文件删除”“数据库执行”“发邮件”等高风险工具开放给模型，先只挂必要工具。
• 敏感步骤加 interrupt。涉及转账、删除、外发数据、权限变更的节点，必须人工确认。
• 所有工具函数加清晰 docstring，模型通过 docstring 理解工具用途，写清楚参数含义能明显提升工具调用准确率。
• 上下文要裁剪。长期会话会给 prompt 塞入大量历史，要按 token 阈值合并或摘要，避免成本爆炸。
• 每次调用记录 thread_id、模型名、token 数、耗时。出问题的时候才能快速定位。

9.2 合规与安全边界

智能体框架把模型从“回答问题”扩展成“操作业务系统”，这带来明显的能力提升，也带来责任。开发时建议注意以下几点：

• 输入用户数据和 prompt 内容涉及个人信息时，先评估数据出境和存储要求，不能直接把敏感信息抛给第三方模型服务。
• 模型生成结果必须人工复核后再进入正式业务环节，尤其是通知类、审批类、对外发布类场景。
• 工具调用要有白名单、权限校验、操作日志。模型只能调用当前会话授权范围内的工具。
• 涉及人脸、声音、版权素材等场景，必须先确认授权；涉及自动化操作外部系统，要确认使用边界，避免绕过平台限制或产生误操作。
• 开源组件要保留许可证信息；第三方库升级前先看 breaking change，避免生产环境静默升级。

10. 总结与下一步

LangGraph 最值得尝试的点是：它让 Agent 的流程真正变得可见、可控、可恢复。你不再需要在一个黑盒循环里赌模型能不能完成整个任务，而是可以把“意图判断”“工具执行”“人工审批”“最终回复”拆成一条条带名字的路径。状态由你定义，分支由你画，复杂流程也能拆成一个一个小节点来调试。

拿到这篇文章后建议按顺序做三件事：先跑通第四节中最小的 StateGraph，理解状态合并；再把第五节中的 ReAct Agent 粘到本地，把问句从“上海天气”换成你自己的场景，看条件边和工具节点如何联动；最后用第七节的 FastAPI 包装一层接口，接一个批量任务测试稳定性。最容易踩的坑不是语法，而是条件边路由写错、状态字段没有定义、interrupt 导入路径对不上版本，遇到报错先去看路由表和 state schema。想继续深入，下一步可以尝试给 Agent 挂上知识库检索节点、接入长时记忆存储、或者把同一个图部署到多 worker 上做水平扩展。