企业官网建设流程全解析

在实际企业级 AI 应用开发中，构建一个能处理复杂、多步骤任务的智能体（Agent）往往比训练一个单一的大语言模型更具挑战性。开发者常常面临状态管理混乱、工具调用顺序失控、多角色协作困难等问题。LangGraph 作为 LangChain 生态中用于构建有状态、多智能体工作流的核心框架，通过将智能体行为建模为图（Graph），清晰地定义了状态流转和节点执行逻辑，为复杂 Agent 架构提供了工程化的解决方案。本文将从 LangGraph 的核心概念入手，逐步构建一个包含监督者（Supervisor）和长期记忆（State）的多智能体系统，并最终部署一个可运行的本地 AI 智能体。无论你是希望理解 Agent 架构设计，还是需要将 LangGraph 应用于实际项目，本文都将提供一条从入门到实战的清晰路径。

1. 理解 LangGraph：为什么图是构建智能体的最佳抽象

在深入代码之前，必须理解 LangGraph 解决的核心问题。传统的链式调用（Chain）在处理线性任务时表现良好，但面对需要循环、分支、回溯或并行执行的复杂场景时，其表达能力就显得捉襟见肘。例如，一个客服机器人可能需要根据用户意图决定调用知识库查询、订单状态检查或人工坐席转接等多个工具，并且这些调用可能不是一次性的，而是根据中间结果动态调整的。

1.1 图（Graph）与状态（State）模型

LangGraph 将整个智能体的工作流抽象为一个有向图。图中的节点（Node）代表一个可执行单元，例如调用一个大语言模型、执行一个工具（Tool），或者进行逻辑判断。边（Edge）则定义了节点之间的流转条件，决定了执行完一个节点后，下一步应该走向哪里。这种抽象天然适合描述包含判断、循环和并发的业务流程。

与图紧密相关的是状态（State）模型。在 LangGraph 中，一个类型化的State对象贯穿整个图的执行过程。每个节点都可以读取和修改这个共享状态。这解决了传统链式调用中状态传递隐式、易丢失的问题。常见的状态类型是MessageState，它专门用于管理对话历史，但你可以定义任何符合业务需求的TypedDict作为状态。

1.2 LangGraph 与 LangChain 的关系与区别

这是一个常见的困惑点。LangChain 是一个更广泛的框架，提供了与各种大语言模型、向量数据库、工具等集成的组件。你可以把 LangChain 看作是一个“工具箱”和“连接器”。而 LangGraph 是 LangChain 生态系统中的一个专门用于构建有状态、多步骤工作流的库。它依赖于 LangChain 的核心组件（如 LLM、 Tools），但提供了更强大的流程控制能力。

简单来说：

• LangChain：擅长“做什么”——集成模型、调用工具、检索文档。
• LangGraph：擅长“按什么顺序、在什么条件下做”——编排复杂的、有状态的执行流程。

当你需要构建一个能进行多轮交互、根据历史决策的智能体时，LangGraph 是你的首选。

1.3 核心概念：节点（Node）、边（Edge）与检查点（Checkpoint）

• 节点（Node）：一个接收状态、执行操作、返回新状态的函数。它可以是调用 LLM、运行工具，或者一个简单的逻辑函数。
• 边（Edge）：连接节点的路径。分为条件边（Conditional Edge）和普通边。条件边允许根据当前状态的值动态决定下一个节点，这是实现分支和循环的关键。
• 检查点（Checkpoint）：LangGraph 支持在执行的特定点保存状态快照。这使得工作流可以暂停、恢复，甚至实现类似“长期记忆”的机制，对于构建复杂的、可中断的对话系统至关重要。

理解了这些，我们就知道 LangGraph 不是替代 LLM，而是为 LLM 驱动的智能体提供了一个可靠、可调试的执行引擎。

2. 环境准备与核心依赖配置

在开始构建智能体之前，需要搭建一个稳定的 Python 开发环境。本文将使用开源模型（通过 Ollama 运行）和 LangChain/LangGraph 的最新稳定版本来演示，确保所有步骤都可以在本地复现。

2.1 创建虚拟环境与安装依赖

首先，创建一个独立的 Python 环境以避免包冲突。

# 创建并激活虚拟环境（以 conda 为例，也可使用 venv） conda create -n langgraph-demo python=3.10 -y conda activate langgraph-demo # 安装核心框架 pip install langgraph langchain langchain-community # 安装用于本地模型交互的库 pip install ollama # 可选：用于可视化图的库 pip install pygraphviz

注意：pygraphviz的安装可能需要系统级的 Graphviz 开发库。在 Ubuntu 上可以运行sudo apt-get install graphviz graphviz-dev，在 macOS 上可以运行brew install graphviz。

2.2 启动本地模型服务（Ollama）

我们将使用 Ollama 在本地运行开源大语言模型，这比调用远程 API 更快速、私密且无成本。

1. 前往 Ollama 官网 下载并安装对应操作系统的客户端。
2. 安装完成后，在终端拉取一个轻量级模型，例如llama3.2或qwen2.5。

# 拉取模型 ollama pull llama3.2:3b # 或 ollama pull qwen2.5:3b

3. 确保 Ollama 服务在后台运行。安装后通常会自动启动服务。

2.3 验证环境

创建一个简单的 Python 脚本，测试 LangChain 能否成功调用本地 Ollama 模型。

# test_env.py from langchain_community.llms import Ollama # 初始化本地 LLM llm = Ollama(model="llama3.2:3b") # 进行一次简单调用 response = llm.invoke("请用中文回答：什么是人工智能？") print(response)

运行此脚本python test_env.py，如果能看到模型返回的连贯中文回答，说明环境配置成功。

3. 构建你的第一个 LangGraph 智能体：单节点工作流

我们从最简单的图开始：一个只包含一个节点（调用 LLM）的工作流。目标是理解如何定义状态、创建图和运行它。

3.1 定义状态（State）

状态是一个TypedDict，它规定了图中可以流转哪些信息。我们从最简单的对话状态开始。

# simple_agent.py from typing import TypedDict, List from langgraph.graph import StateGraph, END # 1. 定义状态：包含消息列表 class AgentState(TypedDict): messages: List[str] # 存储对话历史 # 2. 定义节点函数 def call_model(state: AgentState) -> AgentState: """节点：调用LLM生成回复""" from langchain_community.llms import Ollama llm = Ollama(model="llama3.2:3b") # 获取最新的用户消息（这里简单处理，取最后一条） user_input = state["messages"][-1] if state["messages"] else "你好" prompt = f"用户说：{user_input}\n请给出友好、简洁的回复：" # 调用模型 response = llm.invoke(prompt) # 更新状态，将AI回复加入消息列表 new_messages = state["messages"] + [f"AI: {response}"] return {"messages": new_messages} # 3. 构建图 graph_builder = StateGraph(AgentState) graph_builder.add_node("assistant", call_model) # 添加名为“assistant”的节点 graph_builder.set_entry_point("assistant") # 设置入口节点 graph_builder.add_edge("assistant", END) # 设置出口，执行完即结束 # 编译图 graph = graph_builder.compile() # 4. 运行图 initial_state = {"messages": ["用户: 今天的天气怎么样？"]} result = graph.invoke(initial_state) print("最终状态中的消息：", result["messages"])

这个例子中，我们定义了一个状态AgentState，它只有一个字段messages。图只有一个节点assistant，该节点读取状态中的最后一条消息，调用 LLM 生成回复，并将回复追加到messages中，然后流程结束。

3.2 理解图的编译与执行

graph.compile()是关键一步，它将我们定义的节点和边编译成一个可执行的对象。graph.invoke(initial_state)则是以初始状态启动图的执行。图会从入口节点开始，按照边的定义，依次执行节点，直到到达END。

执行上述代码，你会看到类似['用户: 今天的天气怎么样？', 'AI: 今天天气晴朗，适合外出。']的输出。虽然简单，但这已经是一个完整的有状态工作流。

4. 实现多智能体与监督者（Supervisor）架构

单智能体能力有限。现实任务往往需要多个专家（智能体）协作，并由一个监督者（Supervisor）来协调。例如，一个任务可能先由“分类器”判断类型，再路由给“翻译器”或“总结器”处理。

4.1 设计多智能体系统

我们将构建一个包含三个智能体的系统：

1. 翻译智能体（Translator）：负责将输入翻译成英文。
2. 总结智能体（Summarizer）：负责总结文本内容。
3. 问答智能体（QA）：负责回答基于文本的问题。
4. 监督者（Supervisor）：根据用户输入的意图，决定将任务派发给哪个智能体。

4.2 定义包含意图识别的状态

我们需要扩展状态，以包含路由决策。

# multi_agent.py from typing import TypedDict, List, Literal, Optional from langgraph.graph import StateGraph, END from langchain_community.llms import Ollama # 定义更丰富的状态 class MultiAgentState(TypedDict): messages: List[str] current_agent: Optional[str] # 当前执行的智能体名称 next_agent: Optional[str] # 监督者决定的下一个智能体 final_output: Optional[str] # 最终输出 # 初始化一个共享的LLM，避免重复创建 llm = Ollama(model="qwen2.5:3b") def supervisor_node(state: MultiAgentState) -> MultiAgentState: """监督者节点：分析用户意图，路由到对应智能体""" user_input = state["messages"][-1] prompt = f""" 请分析用户意图，并只返回以下三个选项之一： - translator: 如果用户要求翻译或内容涉及多语言。 - summarizer: 如果用户要求总结、概括或提炼要点。 - qa: 如果用户提出了一个具体问题需要回答。 用户输入：{user_input} 意图： """ intent = llm.invoke(prompt).strip().lower() # 简单清理LLM输出，确保是三个选项之一 if "translator" in intent: next_agent = "translator" elif "summarizer" in intent: next_agent = "summarizer" elif "qa" in intent: next_agent = "qa" else: next_agent = "translator" # 默认路由 return {"next_agent": next_agent} def translator_node(state: MultiAgentState) -> MultiAgentState: """翻译智能体节点""" user_input = state["messages"][-1] prompt = f"将以下中文文本翻译成英文：{user_input}" translation = llm.invoke(prompt) return {"final_output": f"翻译结果：{translation}", "current_agent": "translator"} def summarizer_node(state: MultiAgentState) -> MultiAgentState: """总结智能体节点""" user_input = state["messages"][-1] prompt = f"用中文总结以下文本的核心内容：{user_input}" summary = llm.invoke(prompt) return {"final_output": f"总结结果：{summary}", "current_agent": "summarizer"} def qa_node(state: MultiAgentState) -> MultiAgentState: """问答智能体节点""" user_input = state["messages"][-1] prompt = f"请基于你的知识回答以下问题：{user_input}" answer = llm.invoke(prompt) return {"final_output": f"答案：{answer}", "current_agent": "qa"}

4.3 构建带有条件路由的图

关键步骤在于，监督者节点执行后，需要根据其输出的next_agent值，动态选择下一个节点。这需要使用add_conditional_edges。

# 继续 multi_agent.py # 构建图 builder = StateGraph(MultiAgentState) # 添加节点 builder.add_node("supervisor", supervisor_node) builder.add_node("translator", translator_node) builder.add_node("summarizer", summarizer_node) builder.add_node("qa", qa_node) # 设置入口点为监督者 builder.set_entry_point("supervisor") # 定义条件路由函数 def route_after_supervisor(state: MultiAgentState) -> str: """根据监督者决定的 next_agent 返回下一个节点名""" return state.get("next_agent", "translator") # 添加条件边：从supervisor出发，根据条件路由到三个工作节点之一 builder.add_conditional_edges( "supervisor", route_after_supervisor, { "translator": "translator", "summarizer": "summarizer", "qa": "qa", } ) # 为每个工作节点添加指向 END 的边（执行完即结束） builder.add_edge("translator", END) builder.add_edge("summarizer", END) builder.add_edge("qa", END) # 编译图 graph = builder.compile() # 运行测试 print("=== 测试多智能体系统 ===") test_inputs = [ "将‘你好，世界’翻译成英文", "概括一下《红楼梦》的主要情节", "珠穆朗玛峰的高度是多少？" ] for inp in test_inputs: print(f"\n用户输入：{inp}") initial_state = {"messages": [inp], "current_agent": None, "next_agent": None, "final_output": None} result = graph.invoke(initial_state) print(f"执行代理：{result['current_agent']}") print(f"最终输出：{result['final_output']}")

运行此脚本，你会看到对于不同的输入，监督者成功地将任务路由到了不同的智能体，并得到了相应的输出。这便是一个最基本的多智能体协作系统。

5. 集成长期记忆与复杂状态管理

上述示例中，状态在单次执行后即丢弃。为了实现多轮对话和上下文感知，我们需要引入长期记忆机制。LangGraph 的Checkpointer和更复杂的状态设计是实现这一目标的关键。

5.1 使用 MessageState 管理对话历史

LangGraph 预定义了MessageState，它专门用于处理对话场景，内部使用list[BaseMessage]来存储消息。我们改造之前的例子，使用MessageState并加入记忆。

# memory_agent.py from typing import Literal from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver # 内存检查点，用于演示 from langchain_core.messages import HumanMessage, AIMessage from langchain_community.llms import Ollama # 1. 导入预定义的MessageState from langgraph.graph import MessageState llm = Ollama(model="llama3.2:3b") # 2. 定义节点函数，现在接收和返回的是MessageState def call_llm(state: MessageState): """节点：调用LLM，基于完整对话历史生成回复""" # state[“messages”] 是一个 BaseMessage 列表 messages = state["messages"] # 调用模型，传入整个历史 response = llm.invoke(messages) # 将AI回复作为AIMessage加入状态 return {"messages": [AIMessage(content=response)]} # 3. 构建图，并配置检查点（Checkpointer） memory = MemorySaver() # 使用内存存储检查点，生产环境可换为数据库 builder = StateGraph(state_schema=MessageState) builder.add_node("assistant", call_llm) builder.set_entry_point("assistant") builder.add_edge("assistant", END) # 编译图时传入检查点管理器 graph = builder.compile(checkpointer=memory) # 4. 运行图，并保存线程（Thread）ID以实现多轮对话 config = {"configurable": {"thread_id": "user_123"}} # 唯一线程ID # 第一轮对话 initial_state = {"messages": [HumanMessage(content="我叫小明。")]} result1 = graph.invoke(initial_state, config=config) print("第一轮回复：", result1["messages"][-1].content) # 第二轮对话：图会从检查点恢复状态，包含历史消息 result2 = graph.invoke({"messages": [HumanMessage(content="我刚才说我叫什么名字？")]}, config=config) print("第二轮回复（有记忆）：", result2["messages"][-1].content)

通过使用MemorySaver和唯一的thread_id，我们为对话创建了持久化的线程。每次调用invoke时，如果传入相同的thread_id，LangGraph 会从检查点加载之前的状态，从而实现跨轮次的记忆。MessageState自动处理了消息的累加。

5.2 设计自定义的长期记忆模块

对于更复杂的场景，如需要从大量历史中筛选相关记忆，可以结合向量数据库。思路是：将对话摘要或关键信息存入向量库，在需要时进行检索。

# 伪代码示例：结合向量数据库的记忆模块 class LongTermMemory: def __init__(self, vector_store): self.store = vector_store def remember(self, query: str, k: int=3): """检索相关记忆""" return self.store.similarity_search(query, k=k) def memorize(self, text: str, metadata: dict): """存储新的记忆""" self.store.add_texts([text], metadatas=[metadata]) # 在智能体节点中，可以先检索记忆，再将记忆作为上下文注入给LLM

这超出了本文基础范围，但指出了扩展方向：将 LangGraph 的状态管理与外部的知识库相结合，可以构建出能力强大的、有长期记忆的智能体。

6. 运行、调试与可视化

构建复杂的图之后，如何调试和验证其执行流程至关重要。

6.1 使用 LangGraph Studio 进行可视化（可选）

LangGraph 提供了 Studio 工具，可以可视化图结构并逐步调试。安装后，可以通过编写一个简单的描述文件来启动。

# 安装 langgraph-cli pip install langgraph-cli # 在项目目录下创建 langgraph.json 描述文件 # 然后运行 langgraph dev

这会在本地启动一个 Web 服务，允许你上传图定义并可视化执行步骤。对于复杂工作流，这是一个强大的调试助手。

6.2 在代码中跟踪执行状态

更直接的方式是在invoke时启用流式输出，观察执行路径。

# 使用 stream 模式运行图，观察节点执行顺序 for event in graph.stream(initial_state, config=config, stream_mode="values"): node_name = list(event.keys())[0] print(f"执行节点: {node_name}") # 可以进一步打印 state 的变化 # print(f"状态: {event[node_name]}")

此外，在每个节点函数内部添加详细的日志打印，是定位问题最有效的方法。

7. 常见问题排查与优化实践

在实际开发中，你会遇到各种问题。下表列出了一些典型问题及其排查思路：

7.1 最佳实践建议

1. 状态设计最小化：只将需要在节点间传递的数据放入 State。避免将整个应用上下文塞进去。
2. 节点职责单一：每个节点应只完成一件明确的事情。这有利于测试、复用和调试。
3. 善用检查点：对于耗时长的流程或需要暂停/恢复的场景，检查点是必备功能。
4. 错误处理：在关键节点（尤其是调用外部 API 或工具时）添加try...except，并考虑将错误信息写入状态，由专门的错误处理节点来响应。
5. 测试驱动：为每个节点函数编写单元测试，为整个图编写集成测试，模拟各种输入和状态。
6. 生产环境部署：考虑使用更持久化的检查点存储（如 PostgreSQL），并为图执行添加超时、重试和监控机制。

8. 从演示到生产：架构扩展思考

本文的示例均在单机、单进程内运行。要将其发展为可服务大量用户的生产级系统，需要考虑以下扩展：

• 分布式执行：对于计算密集或 IO 密集的节点，可以将其部署为独立的微服务，LangGraph 通过远程调用（RPC）来协调。Ray 是一个值得考虑的分布式执行框架，可以与 LangGraph 结合。
• 高可用检查点：将MemorySaver替换为基于数据库（如 Redis, PostgreSQL）的检查点存储实现，确保状态持久化和多实例共享。
• 可观测性：在图执行过程中注入追踪（Tracing）信息，收集每个节点的耗时、输入输出和错误，集成到如 LangSmith 或 OpenTelemetry 等可观测性平台。
• 版本化管理：图的定义（节点和边）会随着业务迭代而变化。需要建立图的版本管理机制，实现灰度发布和回滚。

LangGraph 提供的是一种强大的编排范式，它将智能体的“决策逻辑”（图定义）与“执行环境”（节点实现）解耦。掌握这种范式后，你可以根据业务复杂度，灵活地选择从简单的内存图到复杂的分布式工作流引擎，构建出真正可靠、可维护的企业级 Agent 架构。