企业官网建设流程全解析

在实际企业级 AI 应用开发中，一个常见的困境是：传统的线性处理流程难以应对复杂的、需要记忆、决策和工具调用的多轮交互场景。对于有后端开发经验的工程师而言，转向 AI 应用层开发，最大的挑战往往不是模型本身，而是如何将 AI 能力（如大语言模型）工程化地嵌入到具备状态、逻辑和外部工具调用的业务系统中。这正是 LangGraph 要解决的核心问题。它不是一个简单的 SDK，而是一个用于构建有状态、多步骤工作流（特别是 Agent 和多智能体系统）的框架，其设计哲学与后端开发中的状态机、工作流引擎有异曲同工之妙。

本文旨在为具备后端开发背景的工程师提供一条平滑切入 AI Agent 开发的实践路径。我们将绕过繁杂的理论，直接聚焦于如何利用 LangGraph 构建一个具备完整状态管理、工具调用和人机交互能力的智能体系统。你将理解 LangGraph 的核心抽象，掌握其与 LangChain 的协同关系，并通过一个从零开始的教程，最终完成一个可运行、可扩展的企业级多智能体原型。学完后，你将能够将后端工程中的模块化、状态持久化和 API 设计思想，应用于构建更智能、更自主的 AI 应用。

1. 理解 LangGraph：为什么它是后端工程师的“状态机”

在深入代码之前，必须厘清 LangGraph 的定位及其与 LangChain 的关系，这决定了你如何组织你的 AI 项目结构。

1.1 LangGraph 与 LangChain：分工与协同

很多初学者会混淆 LangChain 和 LangGraph。你可以这样理解：LangChain 是“乐高积木”，而 LangGraph 是“拼装说明书和动力系统”。

• LangChain提供了构建 AI 应用所需的基础组件（Components）。这包括：

  • 模型抽象：统一调用 OpenAI、Anthropic、本地模型等。
  • 提示词模板：管理和大语言模型对话的文本。
  • 文档加载器与向量存储：处理知识库和检索。
  • 工具（Tools）：封装外部能力，如搜索、计算、数据库查询、API 调用。
  • 链（Chains）：将多个组件按固定顺序组合成线性流程。

  LangChain 解决了“用什么”和“怎么连”的基础问题，但其核心链（LLMChain, SequentialChain）本质上是线性的、无状态的。对于需要根据中间结果动态决定下一步行动（如思考、调用工具、等待用户输入）的复杂场景，它显得力不从心。

• LangGraph则专注于解决“有状态的、循环的、可分支的”工作流问题。它的核心抽象是图（Graph）。在图里：

  • 节点（Nodes）代表一个执行单元（如调用 LLM、执行工具、处理逻辑）。
  • 边（Edges）定义了节点之间的流转条件。
  • 状态（State）是一个贯穿整个图执行过程的共享数据结构，所有节点都可以读取和修改它。

  这完美对应了后端系统中的状态机或工作流引擎。LangGraph 让你能够清晰地定义智能体的“思考-行动-观察”循环，处理多轮对话的上下文，并协调多个智能体之间的协作。

结论：在现代 LangChain 生态中，最佳实践是“用 LangChain 提供的基础组件，在 LangGraph 的框架内组装成智能体”。你的项目依赖通常会同时包含langchain-core、langchain-community（或其他集成包）和langgraph。

1.2 核心概念：State, Node, Edge

理解这三个概念是使用 LangGraph 的基石。

1. 状态（State）：

   • 是什么：一个类似字典（Dict）或 Pydantic 模型的结构，用于存储工作流执行过程中的所有信息。它是工作流的“记忆体”。
   • 为什么重要：在传统的无状态 HTTP 请求中，会话状态通常保存在外部（如数据库、Redis）。在 LangGraph 的工作流中，状态是显式传递的，使得推理过程可追踪、可调试。
   • 如何设计：这是架构的关键。你需要预先定义状态包含哪些字段。例如，一个聊天智能体的状态可能包括：messages（对话历史），next（指示下一步该执行哪个节点），intermediate_steps（工具调用和结果），sender（消息发送者）等。

2. 节点（Node）：

   • 是什么：一个 Python 函数，它接收当前State作为输入，执行一些操作（如调用 LLM、运行工具），并返回一个包含对State修改内容的字典。
   • 职责：节点应该职责单一。例如，call_model节点只负责调用大语言模型并生成回复；execute_tool节点只负责运行指定的工具。

3. 边（Edge）：

   • 是什么：决定在某个节点执行完毕后，接下来应该执行哪个节点的规则。边可以是固定的（always_go_to），也可以是基于状态内容的条件判断（conditional_edge）。
   • 关键边类型：
     • 入口点（Entry Point）：工作流的起点。
     • 普通边：无条件指向下一个节点。
     • 条件边：根据状态中的某个值（如 LLM 返回的tool_calls列表是否为空）来决定下一步是继续调用工具还是结束。

1.3 LangGraph 的执行模型：编译与运行

LangGraph 的工作流是“编译时定义，运行时执行”的。

1. 定义图：你通过代码创建节点、定义边，构建出图的结构。这个过程是静态的。
2. 编译图：调用graph.compile()将图结构编译成一个可执行对象（CompiledGraph）。
3. 运行图：向编译后的图传入初始状态，图引擎会按照你定义的逻辑，依次执行节点，并根据边的条件流转，直到到达终点。

这种模式与后端定义 API 路由（定义）和处理 HTTP 请求（运行）非常相似。

2. 环境准备与项目初始化

我们将构建一个“研究助手”智能体，它能根据用户的问题，自动决定是否需要联网搜索，并整合信息给出回答。这个例子涵盖了状态管理、工具调用和条件逻辑。

2.1 环境与依赖配置

首先，确保你的 Python 环境（建议 3.10+）并安装必要依赖。我们将使用 OpenAI 的模型和 Tavily 搜索工具作为示例。

# 创建项目目录并进入 mkdir langgraph-research-agent && cd langgraph-research-agent # 创建虚拟环境（可选但推荐） python -m venv venv # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate # 安装核心依赖 pip install langgraph langchain-openai langchain-community # 安装用于搜索的工具包（Tavily 是一个不错的搜索 API） pip install tavily-python # 安装用于结构化状态定义的 Pydantic（推荐） pip install pydantic

2.2 项目结构与关键文件

一个清晰的项目结构有助于管理复杂度。建议如下：

langgraph-research-agent/ ├── agents/ # 智能体定义 │ ├── __init__.py │ └── research_agent.py # 研究助手智能体图定义 ├── tools/ # 工具定义 │ ├── __init__.py │ └── search_tool.py # 搜索工具封装 ├── schemas/ # 数据模型（状态定义） │ ├── __init__.py │ └── state.py # 智能体状态模型 ├── config.py # 配置文件（API Keys， 模型设置） ├── main.py # 应用入口，运行示例 └── requirements.txt

2.3 配置 API 密钥

在config.py中管理你的密钥，切勿将密钥硬编码在代码中或提交到版本控制系统。

# config.py import os from dotenv import load_dotenv load_dotenv() # 从 .env 文件加载环境变量 class Config: # OpenAI OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") OPENAI_MODEL = "gpt-4o-mini" # 或 gpt-4-turbo, 根据实际情况选择 # Tavily Search TAVILY_API_KEY = os.getenv("TAVILY_API_KEY") # 创建配置实例 config = Config()

在项目根目录创建.env文件：

# .env OPENAI_API_KEY=sk-your-openai-key-here TAVILY_API_KEY=tvly-your-tavily-key-here

3. 构建研究助手智能体：从状态定义到图编译

现在，我们从最核心的状态定义开始，逐步构建完整的智能体图。

3.1 定义智能体状态（State Schema）

状态是工作流的基石。我们使用 Pydantic 的BaseModel来定义，这能提供类型提示和验证。在schemas/state.py中：

# schemas/state.py from typing import List, Optional, Any, Dict from pydantic import BaseModel, Field from langchain_core.messages import BaseMessage class AgentState(BaseModel): """研究助手智能体的状态模型""" # 核心：对话消息历史 messages: List[BaseMessage] = Field(default_factory=list, description="完整的对话历史") # 下一步要执行的节点名称（由条件边或节点设置） next: Optional[str] = Field(default=None, description="下一步执行的节点名") # 中间步骤：记录工具调用及其结果，用于给模型提供上下文 intermediate_steps: List[Dict[str, Any]] = Field(default_factory=list, description="工具调用和结果的列表") # 发送者（可用于多智能体场景区分） sender: Optional[str] = Field(default="user", description="当前消息的发送者") class Config: # 允许任意类型，因为 intermediate_steps 里可能存各种工具结果 arbitrary_types_allowed = True

关键解释：

• messages：使用 LangChain 标准的BaseMessage列表（如HumanMessage,AIMessage,ToolMessage）。这是与 LLM 交互的载体。
• next：这是 LangGraph 控制流的关键。节点可以通过修改state.next来指示下一步去哪里。
• intermediate_steps：这是一个通用列表，用于存储形如(tool_call, tool_result)的元组。这是实现 ReAct（Reasoning and Acting）模式的关键，让模型知道它之前调用工具得到了什么结果。
• 使用Field和description能让代码更清晰，对后续的调试和文档化也有帮助。

3.2 封装工具（Tools）

工具是智能体与外界交互的桥梁。在tools/search_tool.py中，我们封装一个 Tavily 搜索工具。

# tools/search_tool.py from langchain.tools import tool from langchain_community.tools.tavily_search import TavilySearchResults from config import config # 初始化 Tavily 搜索工具 # TavilySearchResults 本身已经是一个 LangChain Tool 对象 tavily_tool = TavilySearchResults( api_key=config.TAVILY_API_KEY, max_results=3, # 每次搜索返回的结果数 search_depth="advanced" # 搜索深度 ) # 为了更好的控制，我们可以用 @tool 装饰器再包装一层，添加描述 @tool def web_search_tool(query: str) -> str: """ 使用 Tavily 搜索引擎在互联网上搜索最新信息。 当用户的问题涉及实时信息、新闻、未知事件或需要最新数据时，使用此工具。 Args: query: 搜索查询字符串，应具体、明确。 Returns: 一个格式化的字符串，包含搜索结果的摘要。 """ try: results = tavily_tool.invoke({"query": query}) # Tavily 返回的结果是一个列表，每个元素是字典 formatted_results = [] for i, res in enumerate(results, 1): formatted_results.append( f"[{i}] {res.get('title', 'No Title')}\n" f" URL: {res.get('url', 'No URL')}\n" f" Content: {res.get('content', 'No Content')[:200]}..." # 截取部分内容 ) return "\n\n".join(formatted_results) if formatted_results else "未找到相关结果。" except Exception as e: return f"搜索工具执行出错: {str(e)}"

工具设计要点：

1. 清晰的描述：@tool装饰器会自动使用函数的 docstring 作为工具描述。这个描述对于 LLM 决定何时调用此工具至关重要。务必写清楚工具的用途和适用场景。
2. 健壮的错误处理：工具内部应捕获异常并返回友好的错误信息，避免因工具失败导致整个智能体崩溃。
3. 格式化输出：将原始 API 响应格式化为 LLM 易于理解和处理的文本。

3.3 定义图节点（Nodes）

节点是执行具体工作的函数。我们在agents/research_agent.py中定义两个核心节点：call_model（调用 LLM）和execute_tools（执行工具）。

首先，初始化 LLM 和工具集：

# agents/research_agent.py from typing import Dict, Any from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, END from schemas.state import AgentState from tools.search_tool import web_search_tool from config import config # 1. 初始化 LLM llm = ChatOpenAI( api_key=config.OPENAI_API_KEY, model=config.OPENAI_MODEL, temperature=0.1, # 低温度，让输出更确定 ) # 2. 将工具绑定到 LLM（让 LLM 知道它可以调用哪些工具） # 首先，将工具包装成 LangChain 可识别的列表 tools = [web_search_tool] llm_with_tools = llm.bind_tools(tools) # 3. 创建图构建器，并指定状态模式 graph_builder = StateGraph(AgentState)

接下来，定义call_model节点。这个节点的职责是：基于当前对话历史和之前的工具调用结果，让 LLM 决定下一步是直接回答还是调用工具。

# agents/research_agent.py (续) def call_model(state: AgentState) -> Dict[str, Any]: """ 调用大语言模型节点。 输入：当前状态（包含消息历史和中间步骤）。 输出：更新后的状态，其中 messages 包含了 AI 的新回复。 """ print(f"\n[节点 call_model] 被调用。历史消息数：{len(state.messages)}") # 准备给模型的消息：历史消息 + 工具执行结果（如果有） messages = state.messages # 如果有中间步骤，需要将工具执行结果转换为 ToolMessage 并添加到消息列表 # 这是 ReAct 模式的关键：让模型看到它上次行动的结果 if state.intermediate_steps: # 将中间步骤转换为 LangChain 能识别的 AIMessage (包含 tool_calls) 和 ToolMessage # 注意：这里简化处理，实际应根据 intermediate_steps 的结构来构建 # 假设 intermediate_steps 的每个元素是 (tool_call_dict, tool_result) for step in state.intermediate_steps[-1:]: # 通常只处理最近一次的结果 # 这里需要根据实际存储的结构调整。一个更标准的做法是： # 1. 在上一个 `execute_tools` 节点，将 tool_call 和 result 分别存储为特定格式。 # 2. 在这里，构造对应的 AIMessage 和 ToolMessage。 # 为了示例清晰，我们采用一种更直接的方式，在 `execute_tools` 节点直接更新 messages。 pass # 在我们的设计中，工具结果会由 `execute_tools` 节点直接以 ToolMessage 形式添加到 state.messages 中。 # 因此，这里直接使用最新的 messages 列表即可。 # 调用绑定了工具的 LLM response = llm_with_tools.invoke(messages) # 将 AI 的响应消息添加到状态中 new_messages = state.messages + [response] # 注意：我们不在这个节点修改 `next`，由条件边根据 response 中是否有 tool_calls 来决定 return {"messages": new_messages}

然后，定义execute_tools节点。这个节点的职责是：执行 LLM 要求调用的工具，并将结果记录到状态中。

# agents/research_agent.py (续) def execute_tools(state: AgentState) -> Dict[str, Any]: """ 执行工具节点。 输入：状态，其中最后一条消息是 AI 的响应，且该响应包含 tool_calls。 输出：更新后的状态，其中 messages 添加了工具执行结果，intermediate_steps 被更新。 """ print(f"\n[节点 execute_tools] 被调用。") messages = state.messages last_message = messages[-1] # 初始化返回的更新字典 updates = {"intermediate_steps": []} # 检查上一条 AI 消息是否包含工具调用 if not hasattr(last_message, 'tool_calls') or not last_message.tool_calls: print(" 警告：最后一条消息没有 tool_calls，无需执行工具。") # 如果没有工具调用，直接返回，并指示下一步去 END updates["next"] = END return updates # 遍历并执行每一个工具调用 for tool_call in last_message.tool_calls: tool_name = tool_call['name'] tool_args = tool_call['args'] print(f" 执行工具: {tool_name}，参数: {tool_args}") # 根据工具名找到对应的工具函数 tool_to_call = None for tool in tools: if tool.name == tool_name: tool_to_call = tool break if tool_to_call is None: result = f"错误：未知工具 '{tool_name}'" else: try: # 执行工具 result = tool_to_call.invoke(tool_args) except Exception as e: result = f"工具 '{tool_name}' 执行出错: {str(e)}" print(f" 工具结果摘要: {str(result)[:100]}...") # 记录中间步骤（可选，用于更复杂的流程控制） updates["intermediate_steps"].append({ "tool_call": tool_call, "result": result }) # 将工具执行结果作为 ToolMessage 添加到消息历史中 # 这是关键：让下一次模型调用能看到工具的输出 from langchain_core.messages import ToolMessage tool_message = ToolMessage( content=str(result), tool_call_id=tool_call['id'], # 必须与 AI 请求中的 tool_call_id 对应 ) messages.append(tool_message) # 更新消息列表和中间步骤 updates["messages"] = messages # 执行完工具后，默认应该让模型再次思考，所以设置 next 为 “call_model” updates["next"] = "call_model" return updates

3.4 构建图结构：添加节点与边

定义了节点函数后，需要将它们添加到图中，并定义节点之间的流转逻辑。

# agents/research_agent.py (续) # 4. 将节点添加到图中 graph_builder.add_node("call_model", call_model) graph_builder.add_node("execute_tools", execute_tools) # 5. 设置入口点：工作流从 `call_model` 开始（模型先思考用户问题） graph_builder.set_entry_point("call_model") # 6. 定义条件边：在 `call_model` 之后，根据 AI 响应决定下一步 def should_continue(state: AgentState) -> str: """ 条件路由函数。 检查最后一条 AI 消息是否包含工具调用。 如果有，去 `execute_tools`；如果没有，工作流结束（去 END）。 """ messages = state.messages last_message = messages[-1] if hasattr(last_message, 'tool_calls') and last_message.tool_calls: print(f"[条件路由] 检测到工具调用，前往 execute_tools") return "execute_tools" else: print(f"[条件路由] 未检测到工具调用，工作流结束") return END # 添加从 `call_model` 出发的条件边 graph_builder.add_conditional_edges( "call_model", should_continue, # 路由函数 { "execute_tools": "execute_tools", # 如果返回 “execute_tools”，则去往该节点 END: END # 如果返回 END，则结束 } ) # 7. 添加从 `execute_tools` 出发的普通边 # 在 `execute_tools` 节点中，我们已经通过修改 `state.next` 来指示下一步（通常是回到 call_model）。 # 这里添加一条边，其目标由 `state.next` 决定。 graph_builder.add_edge("execute_tools", "call_model") # 注意：更精细的控制可以在 `execute_tools` 节点中设置 `state.next` 为 END 来提前结束。 # 8. 编译图，得到可执行对象 research_agent_graph = graph_builder.compile()

图结构解读：

1. 入口是call_model。
2. call_model执行后，运行should_continue函数判断。
   • 如果 AI 响应中包含tool_calls，则前往execute_tools。
   • 否则，前往END，工作流结束。
3. execute_tools执行后，无条件返回call_model（因为我们希望在得到工具结果后，让模型再次思考如何回应）。execute_tools节点内部也可以根据情况将state.next设为END来提前终止循环。

这就构成了一个经典的“思考-行动”循环：call_model-> （可能）execute_tools->call_model-> ... ->END。

4. 运行与验证：与智能体对话

现在，我们创建一个主程序来运行这个智能体。在main.py中：

# main.py from langchain_core.messages import HumanMessage from agents.research_agent import research_agent_graph from schemas.state import AgentState def run_agent(query: str): """ 运行研究助手智能体。 """ print(f"\n{'='*50}") print(f"用户提问: {query}") print(f"{'='*50}") # 1. 构建初始状态 initial_state = AgentState( messages=[HumanMessage(content=query)], sender="user" ) # 2. 运行编译好的图，传入初始状态 # config 参数可以控制执行细节，如递归深度限制，防止无限循环 final_state = research_agent_graph.invoke( initial_state, config={"recursion_limit": 10} # 限制循环次数，避免意外 ) # 3. 从最终状态中提取 AI 的最终回复 final_messages = final_state['messages'] # 最后一条消息应该是 AI 的最终回答（不包含工具调用） final_ai_response = None for msg in reversed(final_messages): if msg.type == "ai": final_ai_response = msg.content break print(f"\n{'='*50}") print("智能体最终回答:") print(f"{'='*50}") print(final_ai_response) print(f"{'='*50}") # 可选：打印完整的交互历史，用于调试 print("\n完整交互历史:") for i, msg in enumerate(final_messages): print(f"[{i}] {msg.type}: {msg.content[:150]}...") if __name__ == "__main__": # 测试不同的问题 test_queries = [ "你好，请介绍一下你自己。", # 无需搜索 "截至2024年7月，OpenAI 最新的多模态模型叫什么？有什么特点？", # 需要搜索 "帮我计算一下 125 的平方根。", # 我们没提供计算器工具，模型应直接回答或说明无法计算 ] for q in test_queries: run_agent(q) input("\n按 Enter 键继续下一个问题...")

运行python main.py，观察控制台输出。你应该能看到类似以下的流程：

================================================== 用户提问: 截至2024年7月，OpenAI 最新的多模态模型叫什么？有什么特点？ ================================================== [节点 call_model] 被调用。历史消息数：1 [条件路由] 检测到工具调用，前往 execute_tools [节点 execute_tools] 被调用。 执行工具: web_search_tool，参数: {'query': 'OpenAI 2024年7月 最新 多模态 模型 名称 特点'} 工具结果摘要: [1] OpenAI 发布新模型 GPT-4o... (实际搜索结果) [节点 call_model] 被调用。历史消息数：3 (用户问题 + AI工具调用请求 + 工具结果) [条件路由] 未检测到工具调用，工作流结束 ================================================== 智能体最终回答: ================================================== 截至2024年7月，OpenAI 最新的多模态模型是 GPT-4o（“o”代表“omni”）... (整合了搜索结果的回答) ==================================================

对于第一个问题“介绍一下你自己”，由于不涉及实时信息，LLM 不会调用工具，工作流会直接从call_model到END。

5. 常见问题排查与调试技巧

在开发 LangGraph 智能体时，你可能会遇到以下典型问题。

5.1 状态更新不生效

现象：在节点函数中修改了state的某个字段（如state.messages.append(...)），但后续节点读取时发现没有变化。原因与解决：

1. 根本原因：LangGraph 的状态更新机制要求节点函数返回一个字典，字典的键对应状态字段名，值是要替换的（对于列表/字典）或设置的新值。在节点函数内部直接修改传入的state对象是无效的。
2. 正确做法：在节点函数末尾，构造并返回更新字典。

   # 错误做法（无效）： def node_func(state): state.messages.append(new_message) # 无效！ return {} # 正确做法： def node_func(state): new_messages = state.messages + [new_message] # 创建新列表 return {"messages": new_messages} # 返回更新字典

3. 检查：在节点函数中多使用print或日志输出state的关键字段，确认你的更新字典是否正确返回。

5.2 工具不被调用或错误调用

现象：LLM 应该调用工具时没有调用，或者调用了错误的工具/参数。原因与解决：

1. 工具描述不清：检查@tool装饰器下函数的 docstring。确保描述清晰说明了工具的用途、适用场景和输入参数格式。LLM 依赖这个描述做决策。
2. LLM 绑定问题：确认llm.bind_tools(tools)调用成功，且tools列表包含了所有需要的工具对象。
3. 提示词影响：如果你在messages中提供了系统提示（SystemMessage），确保它没有过度限制模型的行为（例如，禁止模型使用工具）。
4. 参数格式错误：检查工具函数的参数是否与 LLM 生成的tool_call['args']匹配。LLM 有时会生成 JSON 字符串，而工具期望的是字典。bind_tools机制通常会处理好这个转换。

5.3 图陷入无限循环

现象：智能体在call_model和execute_tools之间反复循环，无法停止。原因与解决：

1. 缺少终止条件：在call_model后的条件边中，你的should_continue函数逻辑可能有问题，导致即使 AI 给出了最终答案，仍然被路由到execute_tools。仔细检查last_message.tool_calls的判断逻辑。
2. 工具结果导致再次调用工具：AI 在收到工具结果后，可能仍然认为需要进一步搜索。这可能是工具结果不完整或问题本身需要多轮搜索。可以：
   • 在execute_tools节点中，加入逻辑判断（如工具执行次数超过阈值），然后设置state.next = END强制结束。
   • 在编译图时，通过config={"recursion_limit": N}设置全局递归深度限制（如上文示例）。
3. 使用intermediate_steps：在状态中维护一个intermediate_steps列表，并在should_continue函数中检查其长度，超过一定次数则强制结束。

5.4 多轮对话上下文丢失

现象：在连续的对话中，智能体忘记了之前的对话历史。原因与解决：

1. 状态未持久化：LangGraph 的State是单次invoke调用范围内的。每次新的用户提问，你都需要将之前的历史消息作为初始状态的一部分传入。
2. 解决方案：你需要在外层（如 Web 服务器、对话管理服务）维护每个会话（Session）的完整messages历史。当用户发起新消息时，构建初始状态AgentState(messages=all_history_messages)，然后调用图。图执行完成后，将新的state.messages保存回会话存储。

5.5 调试与可视化

LangGraph 提供了强大的调试和可视化支持。

• 打印状态：在每个节点开始和结束时，打印state的关键字段，这是最直接的调试方式。
• 使用LangGraph Studio：这是一个本地开发工具，可以可视化你的图结构，并逐步执行、检查状态。通过pip install langgraph-cli安装，然后使用langgraph studio命令启动。
• 查看编译后的图：compiled_graph.get_graph().draw_mermaid()可以生成 Mermaid 图代码，帮助你理解结构。

6. 进阶：构建企业级多智能体系统

单一智能体能力有限。企业级应用往往需要多个智能体协作。LangGraph 通过“多智能体”和“监督器（Supervisor）”模式来支持这一点。

6.1 多智能体协作模式

假设我们扩展研究助手，引入一个“写作专家”智能体。流程变为：

1. 研究助手：负责分析问题，决定是否需要搜索，并调用搜索工具。
2. 写作专家：负责将研究助手收集的信息，整理成结构清晰、文笔优美的报告。

这可以通过两个子图（research_agent_graph,writer_agent_graph）和一个监督器节点来实现。监督器根据当前任务和状态，决定将工作分配给哪个子图（智能体）。

6.2 关键设计：共享状态与消息路由

在多智能体系统中，状态设计更为关键。通常需要一个全局状态，包含：

• messages: 所有智能体都能看到的对话历史。
• current_agent: 当前应该活跃的智能体名称。
• task_description: 原始任务描述。
• research_findings: 研究助手收集的发现。
• draft_report: 写作专家生成的草稿。

监督器节点的逻辑类似于之前的should_continue，但它不是检查tool_calls，而是检查current_agent或任务完成状态，从而路由到不同的智能体节点。

6.3 实现简述

1. 定义多智能体状态：扩展AgentState，添加上述字段。
2. 创建子图：将之前的研究助手编译成一个子图research_agent_graph。同样，创建一个写作专家子图writer_agent_graph（它可能调用不同的工具或提示词）。
3. 创建监督器节点：

   def supervisor_node(state: AgentState): # 根据 state 的内容，决定下一步调用哪个智能体 if state.current_agent == "researcher": # 调用研究助手子图 new_state = research_agent_graph.invoke(state) # 根据研究助手的结果，可能更新 state.current_agent 为 “writer” if research_is_done(new_state): return {"current_agent": "writer", **new_state} else: return new_state elif state.current_agent == "writer": # 调用写作专家子图 return writer_agent_graph.invoke(state) else: # 初始路由逻辑 if problem_needs_research(state): return {"current_agent": "researcher"} else: return {"current_agent": "writer"}

4. 构建主图：以supervisor_node为核心节点，构建主图。监督器节点调用子图，子图执行完毕后返回主图，由监督器决定下一步。

这种模式极大地增强了系统的模块化和复杂性处理能力。每个智能体可以独立开发、测试，然后通过监督器协调。

7. 生产环境最佳实践

将 LangGraph 智能体部署到生产环境，需要考虑以下方面：

具体建议：

1. 将图包装为服务：使用 FastAPI 或 Flask 将智能体暴露为 HTTP 端点。请求体包含会话 ID 和用户输入，服务层负责加载/保存会话状态，调用图，并返回响应。
2. 实现超时控制：为graph.invoke()设置超时，防止单个请求长时间占用资源。
3. 缓存策略：对于频繁且结果不变的查询（如“介绍你自己”），可以在服务层或图入口节点添加缓存，直接返回历史结果，避免不必要的 LLM 调用。
4. 可观察性：在状态中增加metadata字段，记录请求 ID、时间戳、用户 ID 等，便于全链路追踪。

对于后端工程师而言，LangGraph 带来的最大价值是将 AI 的“非确定性”推理过程，纳入了熟悉的“确定性”工程管控体系。通过状态、节点、边这些抽象，你可以像设计微服务工作流一样设计复杂的 AI 智能体，并运用已有的后端知识来保障其可靠性、可观测性和可维护性。从构建一个单一的研究助手开始，逐步扩展到多智能体协作系统，是掌握这一强大框架的合理路径。下一步，你可以尝试为智能体集成更多工具（如数据库查询、代码执行、内部 API），或者探索 LangGraph 的持久化检查点（Checkpoint）功能，以实现更长的、可中断恢复的工作流。