企业官网建设流程全解析

这次我们来看一个关于 LangGraph 智能体开发的实战教程。LangGraph 作为 LangChain 生态中用于构建复杂、有状态多智能体应用的核心框架，正受到越来越多开发者的关注。它解决了传统链式调用在处理循环、分支和持久化状态时的局限性，让构建能“思考”和“协作”的 AI 智能体变得更加直观。本文不是空谈概念，而是聚焦于实战：从核心架构拆解到代码逐行实现，带你快速掌握如何用 LangGraph 搭建一个具备记忆、工具调用和决策能力的智能体，并探讨其与 API、MCP 协议等生态的集成。

如果你关心如何将大模型 API 转化为可执行、可协作的智能工作流，或者正在寻找比简单提示工程更强大的智能体构建方案，这篇文章值得你仔细阅读。我们将重点关注 LangGraph 的核心设计思想、关键组件的实际用法，以及如何将其部署为一个可复用的服务。整个过程不依赖特定付费平台，完全基于开源框架和标准 API，你可以用自己的开发环境复现。

1. 核心能力速览

2. 适用场景与使用边界

LangGraph 适合需要超越简单问答和文本生成的场景。当你的应用逻辑涉及“先思考，再行动，根据结果决定下一步”的循环，或者需要多个 AI “角色”分工协作时，它就是理想选择。

典型适用场景包括：

1. 复杂任务分解与执行：例如，用户提出“帮我分析某公司的市场竞争力”，智能体可以自动分解为“搜索最新财报”、“分析行业趋势”、“总结竞争优势与风险”等子任务，并依次执行。
2. 持续对话与状态维护：在客服或游戏场景中，需要记住之前的对话历史和用户偏好，并基于此做出连贯的后续响应。
3. 多智能体协作：模拟一个团队，例如一个“研究员”智能体负责搜集资料，一个“分析师”智能体负责总结，一个“评审员”智能体负责检查质量，它们通过共享的状态进行交互。
4. 工具增强的AI应用：智能体可以根据需要调用搜索引擎、数据库、代码执行环境、绘图工具等外部 API，实现功能扩展。

使用边界与注意事项：

• 并非万能：对于简单的、无状态的文本生成任务，使用基础的 Chain 或直接调用模型 API 可能更轻量、高效。
• 复杂性管理：图（Graph）的设计可能变得复杂，需要良好的架构设计来保证可维护性。
• 成本与延迟：每个节点的执行都可能涉及一次大模型 API 调用，在复杂图中可能导致较高的 token 消耗和累积延迟。
• 合规与安全：当智能体能够自动调用外部工具（如网络搜索、数据库写入）时，必须实施严格的权限控制和输入验证，防止越权操作或注入攻击。所有生成内容需符合法律法规。

3. 环境准备与前置条件

在开始编写 LangGraph 智能体之前，你需要准备好基础的 Python 开发环境，并确保能够访问所需的大模型服务。

基础环境清单：

• 操作系统：Windows 10/11, macOS, 或 Linux (推荐 Ubuntu 20.04+)。
• Python 版本：3.8 或更高版本。建议使用 3.10 以获得最佳兼容性。
• 包管理工具：pip或conda。
• 网络：能够访问你选择的大模型 API 服务（如 OpenAI, Anthropic, DeepSeek 等）。如果使用本地模型（如通过 Ollama），则需确保本地服务已启动。

核心依赖包：LangGraph 是 LangChain 生态系统的一部分，通常需要一并安装。

# 创建并激活虚拟环境（推荐） python -m venv langgraph-env source langgraph-env/bin/activate # Linux/macOS # 或 langgraph-env\Scripts\activate # Windows # 安装核心包 pip install langgraph langchain langchain-core

大模型 API 配置：你需要一个可用的 API Key。本文以广泛使用的 OpenAI 兼容 API 为例（也可以是 DeepSeek、通义千问等提供的兼容接口）。

# 安装 OpenAI SDK (如果使用 OpenAI 或兼容接口) pip install openai

然后，在代码中或通过环境变量设置 API Key：

# 在终端中设置环境变量（临时） export OPENAI_API_KEY="your-api-key-here" # Linux/macOS # set OPENAI_API_KEY=your-api-key-here # Windows CMD # $env:OPENAI_API_KEY="your-api-key-here" # Windows PowerShell

4. LangGraph 核心概念与架构拆解

理解 LangGraph 的关键在于掌握其几个核心抽象，它们共同构成了智能体的“骨架”。

4.1 状态（State）

状态是智能体的“记忆单元”，是一个字典（Dict），在图的整个执行生命周期中传递和更新。你需要在定义图时声明状态的模式（Schema）。

from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages import operator class State(TypedDict): # 消息历史，`add_messages` 操作符能智能地追加消息 messages: Annotated[List[str], add_messages] # 其他自定义状态，如当前任务步骤、工具调用结果等 current_step: str research_findings: List[str]

Annotated和add_messages是 LangGraph 提供的语法糖，用于简化消息列表的追加操作。

4.2 节点（Node）

节点是图中的一个执行单元，通常是一个函数。它接收当前状态，执行一些操作（如调用 LLM、运行工具），并返回一个更新后的状态字典。

def llm_node(state: State) -> dict: """一个简单的LLM响应节点""" from langchain_openai import ChatOpenAI model = ChatOpenAI(model="gpt-4o-mini", temperature=0) # 从状态中获取对话历史 conversation_history = state['messages'] # 构造提示词 prompt = f"基于以下对话历史，请给出专业回复：\n{conversation_history[-1]}" # 调用模型 response = model.invoke(prompt) # 更新状态：将模型回复添加到消息历史 new_messages = state['messages'] + [response.content] return {"messages": new_messages, "current_step": "responded"}

4.3 边（Edge）

边决定了执行流程。分为两种：

• 条件边（Conditional Edge）：根据某个条件函数的结果，决定下一步走向哪个节点。
• 普通边：无条件地从一个节点指向下一个节点。

4.4 图（Graph）

图是节点和边的容器。你通过StateGraph类来构建图，添加节点和边，并最终编译成一个可执行的CompiledGraph。

from langgraph.graph import StateGraph, END # 1. 创建图，并指定状态结构 workflow = StateGraph(State) # 2. 添加节点 workflow.add_node("llm_agent", llm_node) workflow.add_node("research_tool", research_node) # 假设有另一个研究工具节点 # 3. 设置入口点 workflow.set_entry_point("llm_agent") # 4. 添加边 workflow.add_edge("llm_agent", "research_tool") # llm_agent 执行完后总是去 research_tool workflow.add_edge("research_tool", END) # research_tool 执行完后结束 # 5. 编译图 app = workflow.compile()

编译后的app就是一个可执行的智能体。你通过传入初始状态来运行它：final_state = app.invoke({"messages": ["用户问题"]})。

5. 实战：构建一个具备研究与总结能力的智能体

让我们构建一个相对完整的智能体，它能够根据用户问题决定是否需要联网研究，然后进行总结。

5.1 定义状态与工具

首先，定义更丰富的状态，并模拟一个“网络搜索”工具。

from typing import TypedDict, Annotated, List, Literal from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): messages: Annotated[List[str], add_messages] needs_research: bool # 是否需要研究 research_data: List[str] # 研究得到的数据 final_answer: str # 最终答案 # 模拟一个网络搜索工具函数 def web_search_tool(query: str) -> str: """模拟工具：根据查询返回模拟的网络搜索结果。""" # 真实场景中，这里会调用 SerperAPI、Google Search API 等 print(f"[工具调用] 正在搜索: {query}") # 返回模拟数据 simulated_results = [ f"关于'{query}'的摘要信息点 A。", f"关于'{query}'的最新报告指出观点 B。", f"专家对'{query}'的看法 C。" ] return "\n".join(simulated_results)

5.2 创建决策节点（路由）

这个节点负责分析用户问题，决定是否需要调用研究工具。

def decision_node(state: AgentState) -> dict: """决策节点：判断是否需要深入研究。""" from langchain_openai import ChatOpenAI model = ChatOpenAI(model="gpt-4o-mini", temperature=0) last_message = state['messages'][-1] if state['messages'] else "" prompt = f""" 用户的问题是：{last_message} 请判断回答这个问题是否需要实时或最新的网络信息进行研究。 如果你的判断是**需要**，请回复单词 `research`。 如果你的判断是**不需要**，仅凭你的知识足以回答，请回复单词 `answer`。 只输出 `research` 或 `answer`，不要有其他任何内容。 """ decision = model.invoke(prompt).content.strip().lower() return {"needs_research": decision == "research"}

5.3 创建研究节点

如果决策节点判定需要研究，则执行此节点。

def research_node(state: AgentState) -> dict: """研究节点：调用搜索工具获取信息。""" query = state['messages'][-1] search_results = web_search_tool(query) return {"research_data": [search_results]} # 将结果存入状态

5.4 创建回答节点

这个节点综合对话历史和研究数据（如果有），生成最终答案。

def answer_node(state: AgentState) -> dict: """回答节点：生成最终答案。""" from langchain_openai import ChatOpenAI model = ChatOpenAI(model="gpt-4o-mini", temperature=0) last_message = state['messages'][-1] research_info = "\n".join(state.get('research_data', [])) prompt = f""" 请回答用户的问题。 用户问题：{last_message} {f'以下是相关的网络研究信息：\n{research_info}' if research_info else '请基于你的知识进行回答。'} 请提供清晰、有条理、准确的回答。 """ response = model.invoke(prompt) return {"final_answer": response.content}

5.5 组装智能体图

现在，我们将节点和条件边组合起来，形成一个完整的智能体工作流。

from langgraph.graph import StateGraph, END # 创建图 workflow = StateGraph(AgentState) # 添加所有节点 workflow.add_node("decision", decision_node) workflow.add_node("research", research_node) workflow.add_node("answer", answer_node) # 设置入口点 workflow.set_entry_point("decision") # 添加条件边：决策后路由 def decide_next_step(state: AgentState) -> Literal["research", "answer"]: """根据决策节点的结果，决定下一步是研究还是直接回答。""" return "research" if state.get("needs_research") else "answer" workflow.add_conditional_edges( "decision", decide_next_step, { "research": "research", "answer": "answer", } ) # 添加普通边 workflow.add_edge("research", "answer") # 研究完成后，进入回答节点 workflow.add_edge("answer", END) # 回答完成后，结束 # 编译图 research_agent = workflow.compile()

5.6 运行与测试智能体

编译完成后，我们就可以像调用函数一样调用这个智能体。

# 初始化输入 initial_state = { "messages": ["请解释一下什么是 LangGraph，以及它和 LangChain 的区别？"], "needs_research": False, "research_data": [], "final_answer": "" } # 运行智能体 print("开始运行智能体...") try: final_state = research_agent.invoke(initial_state) print("\n=== 智能体运行完成 ===") print(f"最终答案：\n{final_state['final_answer']}") print(f"是否触发了研究：{final_state.get('needs_research')}") except Exception as e: print(f"运行出错：{e}")

运行上述代码，你会看到智能体首先经过decision节点判断（可能会输出[工具调用]信息），然后根据判断结果，要么走research -> answer路径，要么直接走answer路径，最终输出一个结构化的答案。

6. 接口 API 封装与批量任务处理

将编译好的智能体图封装成 API 服务，是投入生产环境的关键一步。这里我们使用 FastAPI 来创建一个简单的 Web 服务。

6.1 创建 FastAPI 应用

首先，确保安装了 FastAPI 和 Uvicorn。

pip install fastapi uvicorn

然后，创建一个app.py文件：

from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List, Optional import asyncio # 导入之前编译好的智能体 `research_agent` # 假设上面的代码已封装在一个函数 `create_agent()` 中 from your_agent_module import create_agent app = FastAPI(title="LangGraph 智能体 API 服务") agent = create_agent() # 获取编译好的图实例 # 定义请求体模型 class AgentRequest(BaseModel): message: str session_id: Optional[str] = None # 用于区分不同会话，实现状态隔离 class BatchAgentRequest(BaseModel): tasks: List[AgentRequest] # 单次查询接口 @app.post("/v1/chat/completions") async def chat_completion(request: AgentRequest): try: # 这里可以基于 session_id 从数据库或缓存中恢复历史状态 # 为简化示例，我们每次都是新的会话 initial_state = { "messages": [request.message], "needs_research": False, "research_data": [], "final_answer": "" } # 注意：invoke 是同步方法，在异步环境中使用 `asyncio.to_thread` 避免阻塞 final_state = await asyncio.to_thread(agent.invoke, initial_state) return { "session_id": request.session_id, "answer": final_state["final_answer"], "needs_research": final_state.get("needs_research", False) } except Exception as e: raise HTTPException(status_code=500, detail=f"智能体执行失败: {str(e)}") # 批量任务接口（简易版，顺序执行） @app.post("/v1/batch/completions") async def batch_chat_completion(batch_request: BatchAgentRequest): results = [] for task in batch_request.tasks: try: initial_state = { "messages": [task.message], "needs_research": False, "research_data": [], "final_answer": "" } final_state = await asyncio.to_thread(agent.invoke, initial_state) results.append({ "session_id": task.session_id, "answer": final_state["final_answer"], "success": True }) except Exception as e: results.append({ "session_id": task.session_id, "answer": None, "error": str(e), "success": False }) return {"results": results} if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)

6.2 启动与调用 API 服务

在终端中启动服务：

python app.py

服务启动后，你可以使用curl或 Pythonrequests库进行调用。

# 单次调用 curl -X POST "http://127.0.0.1:8000/v1/chat/completions" \ -H "Content-Type: application/json" \ -d '{"message": "什么是MCP协议？", "session_id": "test_123"}' # 批量调用 curl -X POST "http://127.0.0.1:8000/v1/batch/completions" \ -H "Content-Type: application/json" \ -d '{ "tasks": [ {"message": "解释一下AI幻觉", "session_id": "task1"}, {"message": "Python的GIL是什么", "session_id": "task2"} ] }'

# Python 调用示例 import requests import json url = "http://127.0.0.1:8000/v1/chat/completions" payload = {"message": "什么是MCP协议？", "session_id": "test_123"} headers = {'Content-Type': 'application/json'} response = requests.post(url, data=json.dumps(payload), headers=headers) print(response.json())

6.3 实现真正的状态持久化（进阶）

上面的示例每次都是全新的状态。在生产环境中，你需要根据session_id将会话状态（即AgentState）持久化到数据库（如 Redis、PostgreSQL）或文件系统中。在每次invoke时，先加载历史状态，执行后再保存新状态。这可以实现真正的多轮对话记忆。

7. 与 MCP（Model Context Protocol）协议集成

MCP 协议是 LangChain 推出的一种标准化协议，用于让 AI 应用（如智能体）安全、可控地访问外部工具和数据源（如数据库、文件系统、第三方 API）。LangGraph 智能体可以很方便地集成 MCP Server 来扩展能力。

核心思路：

1. 部署或连接 MCP Server：例如，一个提供数据库查询工具的 MCP Server。
2. 在 LangGraph/LangChain 中配置 MCP Client：通过langchain-mcp-adapters等包，将 MCP Server 提供的工具注册到 LangChain 的工具列表中。
3. 在智能体节点中调用这些工具：就像调用普通的@tool装饰的函数一样。

简化示例步骤：

# 假设已有一个运行在 localhost:8001 的 MCP Server (例如提供 SQL 查询) # 安装 MCP 适配器 pip install langchain-mcp-adapters

from langchain_mcp_adapters import ClientAdapter import httpx # 1. 创建 MCP 客户端，连接到 MCP Server async with httpx.AsyncClient() as client: mcp_client = ClientAdapter.create( client=client, transport="sse", # 或 stdio，取决于 Server 类型 url="http://localhost:8001/sse" # MCP Server 的 SSE 端点 ) # 2. 获取 MCP Server 提供的所有工具 tools = await mcp_client.get_tools() # 3. 将工具绑定到 LLM from langchain_openai import ChatOpenAI model = ChatOpenAI(model="gpt-4o-mini") model_with_tools = model.bind_tools(tools) # 4. 在 LangGraph 的节点函数中，就可以让模型决定是否调用这些 MCP 工具了 def agent_node_with_mcp(state: State): # ... 构造消息 ... response = model_with_tools.invoke(messages) if response.tool_calls: # 执行工具调用，这里会通过 MCP 协议与 Server 通信 results = execute_tool_calls(response.tool_calls, tools) # ... 处理结果并更新状态 ... return new_state

通过集成 MCP，你的 LangGraph 智能体就能安全地访问企业内部数据库、CRM 系统等，而无需将敏感逻辑直接写在智能体代码中。

8. 资源占用、性能观察与优化

LangGraph 框架本身是轻量级的，性能瓶颈主要在于大模型 API 调用和工具执行。

性能观察点：

1. Token 消耗：每个节点中的 LLM 调用都会消耗 Token。在复杂图中，累计成本可能很高。建议在关键节点打印或记录输入的 prompt 长度和返回的 response 长度。
2. API 延迟：网络延迟和模型推理时间占主导。对于顺序执行的图，总耗时是各节点耗时的总和。考虑对无依赖的节点进行异步并行处理。
3. 状态大小：如果状态中存储了大量历史消息或数据，可能会影响序列化和传输效率。定期清理或摘要历史消息。
4. 工具调用开销：如果工具是网络服务（如搜索、数据库查询），其延迟也需要计入。

优化建议：

• 异步执行：对于可以并行的节点（例如，同时查询多个不相关的数据源），使用asyncio.gather在节点函数中并发执行。
• 缓存：对频繁查询且结果变化不大的工具调用（如某些知识查询）引入缓存机制（如functools.lru_cache或 Redis 缓存）。
• 流式输出：如果最终答案是文本生成，考虑使用模型的流式响应，提升用户体验。
• 超时与重试：为工具调用和模型调用设置合理的超时和重试机制，增强鲁棒性。
• 图结构优化：审视你的图，是否存在不必要的循环或分支？能否合并一些简单的节点？

9. 常见问题与排查方法

10. 最佳实践与使用建议

1. 从简单开始，逐步复杂化：先构建一个只有2-3个节点的最小可行图（MVP），确保它能跑通。再逐步添加新的节点、分支和状态字段。
2. 状态设计要精简：只把真正需要在节点间传递的数据放入状态。避免存储过大的中间结果（如整张图片），可以存储文件路径或引用 ID。
3. 为节点和边起好名字：使用有意义的名称（如"classify_intent","call_search_api"），这能让图的可视化和调试更容易。
4. 实现完善的日志记录：在每个节点的开始和结束记录日志，包括输入状态摘要、工具调用详情、耗时等。这对于调试复杂工作流至关重要。
5. 进行彻底的单元测试：为每个节点函数编写单元测试，模拟输入状态，验证输出状态。然后为整个图编写集成测试。
6. 版本化你的图：当对智能体逻辑进行重大修改时，考虑对图定义进行版本控制，以便回滚和对比。
7. 安全第一：特别是当智能体能够执行写操作（如发送邮件、修改数据库）或访问敏感信息时，必须在工具调用前加入权限验证和用户确认机制（可在状态中设计user_confirmed字段）。
8. 成本监控：在关键节点记录 LLM 调用的 Token 使用量，并设置预算告警，避免意外的高额费用。

LangGraph 提供了一个强大而灵活的范式来构建新一代的 AI 智能体应用。它最大的优势在于将复杂的、有状态的交互流程清晰地建模成一张图，使得开发、调试和维护都变得更加直观。要掌握它，最好的方法就是动手实践：从一个具体的任务出发，画出它的执行流程图，然后用 LangGraph 的节点和边将其实现出来。当你成功地将一个复杂的业务逻辑转化为一个自运行的智能体时，你会感受到这种开发方式的巨大潜力。建议将本文的示例代码作为起点，尝试改造和扩展，构建属于你自己的智能体。