LangGraph与MCP实战：构建智能电商客服系统的完整指南

如果你正在学习AI大模型开发，特别是智能体（Agent）相关的技术，可能会遇到这样的困惑：为什么看了很多教程，还是无法构建真正实用的智能系统？为什么简单的问答模型容易实现，但一到复杂的多轮对话、状态管理和工具调用就束手无策？

这正是LangChain、LangGraph和MCP这三个技术栈要解决的核心问题。它们不是孤立的技术点，而是一个完整的智能体开发体系。本文将基于一个完整的电商客服实战项目，带你从零掌握这套技术栈的真正精髓。

1. 这篇文章真正要解决的问题

很多开发者在学习AI大模型应用开发时，容易陷入"只会调用API"的困境。当需求从简单的问答升级到复杂的业务流程时，传统的线性处理方式就显得力不从心。比如：

• 客服系统中，用户可能先问订单状态，然后要求修改地址，最后又询问物流信息
• 每个请求都需要访问不同的后端系统（订单数据库、物流API等）
• 需要维护对话状态和历史上下文
• 要根据业务规则进行智能路由和决策

这就是为什么需要LangGraph的状态机模型和MCP的工具集成能力。本文将通过一个完整的快时尚电商客服案例，展示如何构建一个真正可用的多智能体系统。

2. 基础概念与核心原理

2.1 LangChain vs LangGraph：不是替代而是互补

LangChain主要解决的是"链式"任务，适合线性流程。比如：用户输入→意图识别→调用工具→返回结果。这种模式在简单场景下很有效，但当业务流程涉及循环、分支、状态维护时就会遇到瓶颈。

LangGraph的核心创新在于"状态驱动的图结构"，它把业务流程建模为一个有向图，其中：

• 节点（Nodes）：执行具体任务的函数单元
• 边（Edges）：控制流程走向，支持条件分支
• 状态（State）：在整个图中传递和共享的数据容器

# LangGraph 状态定义示例 from typing import TypedDict, Optional class CustomerServiceState(TypedDict): intent: str order_details: Optional[dict] compensation_level: int escalation_needed: bool final_resolution: Optional[str]

2.2 MCP（模型上下文协议）：工具集成的标准化方案

MCP解决了AI应用开发中的一个关键痛点：如何让大模型安全、规范地使用外部工具和数据。与传统API集成相比，MCP的优势在于：

• 动态发现：客户端可以在运行时查询可用的工具，无需预先硬编码
• 结构化调用：服务器负责参数校验和错误处理，模型只需关注业务逻辑
• 统一安全：集中的权限管理和审计日志

3. 环境准备与前置条件

在开始实战之前，需要确保开发环境准备就绪：

3.1 系统要求

• 操作系统：Linux/macOS/Windows（推荐Linux环境）
• Python版本：3.10或更高
• Node.js：18或更高（用于MCP相关工具）

3.2 核心依赖安装

创建项目目录并初始化虚拟环境：

# 创建项目目录 mkdir customer-service-mcp cd customer-service-mcp # 创建虚拟环境 python -m venv venv # 激活虚拟环境 # Linux/macOS source venv/bin/activate # Windows venv\Scripts\activate # 安装核心依赖 pip install langchain>=0.1.0 langchain-community langchain-mcp-adapters>=0.1.0 pip install boto3>=1.34.0 python-dotenv>=1.0.0 regex>=2023.0.0 pip install mcp-server>=0.1.0 aioconsole>=0.7.0

3.3 AWS Bedrock配置（可选）

如果使用Amazon Bedrock作为LLM后端，需要配置AWS凭证：

# 安装AWS CLI pip install awscli # 配置AWS凭证 aws configure # 输入AWS Access Key ID、Secret Access Key、默认区域等

4. 项目架构设计

4.1 整体架构概述

我们的电商客服系统采用分层架构：

用户输入 → 意图识别Agent → 路由分发 → 专业Agent处理 → MCP工具调用 → 返回结果

4.2 项目结构规划

customer_service_mcp/ ├── __init__.py ├── agents/ # 智能体模块 │ ├── base_agent.py # 基础智能体类 │ ├── intent_recognition_agent.py # 意图识别 │ ├── order_issue_agent.py # 订单问题处理 │ └── logistics_issue_agent.py # 物流问题处理 ├── services/ # 业务服务层 │ ├── order_service.py # 订单数据服务 │ └── sop_service.py # 标准操作流程 ├── config/ │ └── mcp_config.py # MCP服务器配置 ├── server.py # MCP服务器主程序 ├── main.py # 客户端主程序 ├── requirements.txt # 依赖列表 ├── start_server.sh # 服务器启动脚本 └── start_client.sh # 客户端启动脚本

5. 核心代码实现

5.1 基础智能体类实现

首先创建基础智能体类，封装通用的LLM调用和对话历史管理：

# agents/base_agent.py from abc import ABC, abstractmethod from typing import Dict, Optional, List from langchain_community.chat_models import BedrockChat from langchain.prompts import ChatPromptTemplate from langchain.schema import BaseMessage class BaseAgent(ABC): """所有客服智能体的基类""" def __init__(self, model_id: str = "anthropic.claude-3-sonnet-20240229-v1:0", region: str = "us-west-2"): """使用Bedrock模型初始化智能体""" self.llm = BedrockChat( model_id=model_id, model_kwargs={"temperature": 0.7, "max_tokens": 2048}, region_name=region ) self.conversation_history: Dict[str, List[Dict[str, str]]] = {} def _get_history(self, conversation_id: str) -> List[Dict[str, str]]: """获取特定对话的历史记录""" return self.conversation_history.get(conversation_id, []) def _update_history(self, conversation_id: str, user_message: str, assistant_message: str): """更新对话历史""" if conversation_id not in self.conversation_history: self.conversation_history[conversation_id] = [] self.conversation_history[conversation_id].extend([ {"role": "user", "content": user_message}, {"role": "assistant", "content": assistant_message} ]) @abstractmethod def process(self, user_input: str, conversation_id: Optional[str] = None, **kwargs): """处理用户输入的核心方法""" pass

5.2 意图识别智能体

这是系统的"路由器"，负责分析用户问题并分发给合适的专业智能体：

# agents/intent_recognition_agent.py import uuid from typing import Optional, List, Dict from langchain.prompts import ChatPromptTemplate from agents.base_agent import BaseAgent class IntentRecognitionAgent(BaseAgent): """识别用户意图的智能体""" def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.prompt = ChatPromptTemplate.from_messages([ ("system", """你是时尚电商客服的意图识别系统。 你的任务是分析客户问题并判断属于以下哪类： 1. ORDER ISSUES（订单状态、修改、问题或支付） 2. LOGISTICS ISSUES（配送地址、运输方式、配送问题） 请参考对话历史来理解当前问题的上下文。 只返回意图关键词：ORDER 或 LOGISTICS 不要返回完整句子。"""), ("human", "对话历史：\n{history}\n\n当前问题：{question}") ]) def process(self, user_input: str, conversation_id: Optional[str] = None, history: List[Dict[str, str]] = None, **kwargs): """识别用户意图""" if not conversation_id: conversation_id = str(uuid.uuid4()) # 格式化对话历史 formatted_history = "\n".join([ f"{msg['role'].capitalize()}: {msg['content']}" for msg in (history or []) ]) # 调用LLM进行意图识别 chain = self.prompt | self.llm response = chain.invoke({ "history": formatted_history, "question": user_input }) intent = response.content.strip().upper() # 验证和标准化意图 if "ORDER" in intent: intent = "ORDER" elif "LOGISTICS" in intent: intent = "LOGISTICS" else: intent = "UNKNOWN" return intent, conversation_id

5.3 订单服务实现

订单服务负责管理订单数据，提供基本的CRUD操作：

# services/order_service.py import json import os from typing import List, Dict, Optional class OrderService: """订单数据服务类""" def __init__(self, data_file: str = "order_data.txt"): self.data_file = data_file self._initialize_data() def _initialize_data(self): """初始化订单数据文件""" if not os.path.exists(self.data_file): initial_data = [ { "order_id": "123", "customer_name": "Alice Chen", "items": ["T-shirt", "Jeans"], "address": "Xicheng District, Beijing", "status": "Processing" }, { "order_id": "456", "customer_name": "Bob Wang", "items": ["Dress", "Shoes"], "address": "Haidian District, Beijing", "status": "Shipped" }, { "order_id": "789", "customer_name": "Charlie Liu", "items": ["Jacket", "Hat"], "address": "Dongcheng District, Beijing", "status": "Delivered" } ] self.save_order_data(initial_data) def get_order_data(self) -> List[Dict]: """从文件读取订单数据""" try: with open(self.data_file, 'r', encoding='utf-8') as file: return json.load(file) except Exception as e: print(f"读取订单数据错误: {str(e)}") return [] def save_order_data(self, order_data: List[Dict]) -> bool: """保存订单数据到文件""" try: with open(self.data_file, 'w', encoding='utf-8') as file: json.dump(order_data, file, indent=2, ensure_ascii=False) return True except Exception as e: print(f"保存订单数据错误: {str(e)}") return False def get_order_info(self, order_id: str) -> Optional[Dict]: """获取特定订单信息""" order_data = self.get_order_data() for order in order_data: if order["order_id"] == order_id: return order return None def update_address(self, order_id: str, new_address: str) -> bool: """更新订单配送地址""" order_data = self.get_order_data() for order in order_data: if order["order_id"] == order_id: order["address"] = new_address return self.save_order_data(order_data) return False

5.4 MCP服务器实现

MCP服务器是连接智能体和外部工具的关键桥梁：

# server.py from mcp.server.fastmcp import FastMCP import json import uuid from typing import Optional from main import CustomerServiceSystem # 初始化FastMCP服务器 mcp = FastMCP("CustomerService") system = CustomerServiceSystem() @mcp.tool() async def process_question(question: str, conversation_id: Optional[str] = None) -> str: """处理客服问题并返回响应""" try: response, new_conversation_id = system.process_question(question, conversation_id) result = { "response": response, "conversation_id": new_conversation_id } return json.dumps(result, ensure_ascii=False) except Exception as e: return json.dumps({ "error": f"处理问题时发生错误: {str(e)}", "question": question }) @mcp.tool() async def get_order_info(order_id: str) -> str: """获取订单信息""" try: order_info = system.order_service.get_order_info(order_id) if order_info: return json.dumps({"order": order_info}, ensure_ascii=False) return json.dumps({ "error": f"订单 {order_id} 未找到", "order_id": order_id }) except Exception as e: return json.dumps({ "error": f"获取订单信息时发生错误: {str(e)}", "order_id": order_id }) @mcp.tool() async def update_order_address(order_id: str, new_address: str) -> str: """更新订单配送地址""" try: success = system.order_service.update_address(order_id, new_address) if success: updated_order = system.order_service.get_order_info(order_id) return json.dumps({ "message": "地址更新成功", "order": updated_order }, ensure_ascii=False) return json.dumps({ "error": f"更新订单 {order_id} 地址失败", "order_id": order_id }) except Exception as e: return json.dumps({ "error": f"更新地址时发生错误: {str(e)}", "order_id": order_id }) if __name__ == "__main__": mcp.run(transport="sse")

6. 系统集成与运行

6.1 主系统协调器

这是整个系统的"大脑"，负责协调各个智能体和工作流程：

# main.py import uuid import json import asyncio import aioconsole import re from typing import Optional, Dict, Any from agents.intent_recognition_agent import IntentRecognitionAgent from agents.order_issue_agent import OrderIssueAgent from agents.logistics_issue_agent import LogisticsIssueAgent from services.order_service import OrderService from services.sop_service import SOPService class CustomerServiceSystem: """主客服系统，协调各个智能体和服务""" def __init__(self, model_id: str = "anthropic.claude-3-sonnet-20240229-v1:0", region: str = "us-west-2"): # 初始化各个智能体 self.intent_agent = IntentRecognitionAgent(model_id=model_id, region=region) self.order_agent = OrderIssueAgent(model_id=model_id, region=region) self.logistics_agent = LogisticsIssueAgent(model_id=model_id, region=region) # 初始化服务 self.order_service = OrderService() self.sop_service = SOPService() # 存储活跃对话 self.conversations: Dict[str, Dict[str, Any]] = {} def process_question(self, user_question: str, conversation_id: Optional[str] = None): """处理用户问题的完整流程""" # 生成或使用现有对话ID if not conversation_id: conversation_id = str(uuid.uuid4()) self.conversations[conversation_id] = { "order_id": None, "history": [] } # 记录用户问题 self.conversations[conversation_id]["history"].append({ "role": "user", "content": user_question }) # 第一步：意图识别 intent, _ = self.intent_agent.process( user_question, conversation_id, history=self.conversations[conversation_id]["history"] ) print(f"识别到的意图: {intent}") # 从问题中提取订单ID order_id_match = re.search( r'order\s+(?:id\s+)?(?:number\s+)?(?:#\s*)?(\d+)', user_question, re.IGNORECASE ) if order_id_match: self.conversations[conversation_id]["order_id"] = order_id_match.group(1) # 第二步：根据意图路由到专业智能体 if intent == "ORDER": response, _ = self.order_agent.process( user_question, conversation_id, order_id=self.conversations[conversation_id].get("order_id"), history=self.conversations[conversation_id]["history"] ) elif intent == "LOGISTICS": response, _ = self.logistics_agent.process( user_question, conversation_id, order_id=self.conversations[conversation_id].get("order_id"), history=self.conversations[conversation_id]["history"] ) else: response = "我不确定您的问题是关于订单还是物流问题。请提供更多详细信息。" # 记录智能体响应 self.conversations[conversation_id]["history"].append({ "role": "assistant", "content": response }) return response, conversation_id async def interactive_session(): """运行交互式会话""" system = CustomerServiceSystem() conversation_id = None print("欢迎使用时尚电商客服系统！") print("您可以咨询订单或物流相关问题。") print("输入 'exit' 结束对话。") print("\n可用测试订单: 123, 456, 789") print("-" * 50) while True: try: user_input = await aioconsole.ainput("\n客户: ") if user_input.lower() == 'exit': print("感谢使用我们的客服系统。再见！") break response, conversation_id = system.process_question(user_input, conversation_id) print(f"\n客服: {response}") except Exception as e: print(f"\n错误: {str(e)}") if __name__ == "__main__": asyncio.run(interactive_session())

6.2 启动脚本配置

创建便捷的启动脚本，简化运行流程：

#!/bin/bash # start_server.sh # 创建虚拟环境（如果不存在） if [ ! -d "venv" ]; then echo "创建虚拟环境..." python3 -m venv venv fi # 激活虚拟环境 source venv/bin/activate # 安装依赖（如果尚未安装） if [ ! -f "venv/installed" ]; then echo "安装依赖..." pip install -r requirements.txt touch venv/installed fi # 启动MCP服务器 echo "启动MCP服务器..." python server.py

#!/bin/bash # start_client.sh source venv/bin/activate python3 main.py deactivate

6.3 依赖管理

创建requirements.txt文件管理项目依赖：

# requirements.txt langchain>=0.1.0 langchain-community>=0.0.1 langchain-mcp-adapters>=0.1.0 boto3>=1.34.0 python-dotenv>=1.0.0 regex>=2023.0.0 mcp-server>=0.1.0 aioconsole>=0.7.0 uvicorn>=0.24.0 fastapi>=0.104.0

7. 系统运行与测试

7.1 启动系统

首先启动MCP服务器：

# 给脚本添加执行权限 chmod +x start_server.sh chmod +x start_client.sh # 启动服务器 ./start_server.sh

服务器启动后，在另一个终端启动客户端：

./start_client.sh

7.2 测试对话示例

系统启动后，可以进行完整的对话测试：

欢迎使用时尚电商客服系统！ 您可以咨询订单或物流相关问题。 输入 'exit' 结束对话。 可用测试订单: 123, 456, 789 -------------------------------------------------- 客户: 订单123的状态是什么？ 识别到的意图: ORDER 客服: 让我检查订单#123的状态... Alice Chen的订单#123当前状态为"处理中"。订单包含T恤和牛仔裤，预计配送到北京市西城区。 如需此订单的其他详细信息，请告诉我。 客户: 我能把订单123的地址改为北京市朝阳区Road X 123号吗？ 识别到的意图: ORDER 客服: 由于您的订单#123尚未发货，我可以将配送地址更新为： Road X, 123号 朝阳区 北京 请确认这个新地址是否正确。确认后，我将在系统中进行更改。 客户: 我确认 识别到的意图: ORDER 客服: 好的，我已将订单#123的配送地址更新为： Road X, 123号 朝阳区 北京 新地址已保存在我们的系统中。在订单发货前如需其他更改，请告诉我。

7.3 MCP工具调用验证

通过MCP Inspector可以验证工具的正常工作：

# 安装MCP Inspector npm install -g @modelcontextprotocol/inspector # 启动Inspector连接MCP服务器 mcp-inspector http://localhost:8000

在Inspector界面中可以测试各个工具的功能，确保MCP服务器正常工作。

8. 常见问题与排查思路

8.1 环境配置问题

8.2 运行时问题

8.3 性能优化建议

1. 对话历史管理：对于长时间对话，考虑实现历史摘要机制，避免上下文过长
2. 智能体缓存：对频繁查询的订单信息实现缓存机制
3. 异步处理：对耗时的外部API调用使用异步模式
4. 错误重试：对暂时性故障实现指数退避重试机制

9. 从LangChain到LangGraph的演进

9.1 当前架构的局限性

虽然我们当前的基于LangChain的多智能体系统已经能够处理复杂的客服场景，但随着业务需求的扩展，可能会遇到以下挑战：

• 状态管理复杂：手动维护对话状态和订单信息容易出错
• 流程控制困难：复杂的条件分支和循环逻辑难以清晰表达
• 可观测性不足：难以直观理解系统的决策流程

9.2 LangGraph改造方案

下面是使用LangGraph重构核心流程的示例：

from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END class CustomerServiceState(TypedDict): """定义客服状态结构""" user_input: str intent: str order_id: Optional[str] order_info: Optional[dict] conversation_history: List[dict] current_response: str def intent_recognition_node(state: CustomerServiceState): """意图识别节点""" # 使用现有的意图识别逻辑 intent_agent = IntentRecognitionAgent() intent, _ = intent_agent.process(state["user_input"]) return {"intent": intent} def order_processing_node(state: CustomerServiceState): """订单处理节点""" order_agent = OrderIssueAgent() response, _ = order_agent.process( state["user_input"], order_id=state.get("order_id") ) return {"current_response": response} def logistics_processing_node(state: CustomerServiceState): """物流处理节点""" logistics_agent = LogisticsIssueAgent() response, _ = logistics_agent.process( state["user_input"], order_id=state.get("order_id") ) return {"current_response": response} def route_by_intent(state: CustomerServiceState): """根据意图路由到不同节点""" if state["intent"] == "ORDER": return "order_processing" elif state["intent"] == "LOGISTICS": return "logistics_processing" else: return "fallback_processing" # 构建图结构 graph_builder = StateGraph(CustomerServiceState) graph_builder.add_node("intent_recognition", intent_recognition_node) graph_builder.add_node("order_processing", order_processing_node) graph_builder.add_node("logistics_processing", logistics_processing_node) graph_builder.set_entry_point("intent_recognition") graph_builder.add_conditional_edges( "intent_recognition", route_by_intent, { "order_processing": "order_processing", "logistics_processing": "logistics_processing", "fallback_processing": END } ) graph_builder.add_edge("order_processing", END) graph_builder.add_edge("logistics_processing", END) customer_service_graph = graph_builder.compile()

9.3 LangGraph的优势体现

通过LangGraph重构后，系统获得以下改进：

1. 显式状态管理：所有状态变更都在类型安全的State结构中明确体现
2. 可视化流程：图结构可以直观展示和调试
3. 灵活扩展：新增节点或修改路由逻辑不影响现有架构
4. 错误隔离：单个节点故障不会导致整个系统崩溃

10. 生产环境最佳实践

10.1 安全考虑

1. 输入验证：对所有用户输入进行严格的验证和清理
2. 权限控制：MCP工具调用需要实现细粒度的权限管理
3. 数据脱敏：日志和错误信息中避免泄露敏感数据
4. 速率限制：防止API滥用和DDoS攻击

10.2 监控与日志

实现全面的监控体系：

import logging from datetime import datetime def setup_logging(): """配置结构化日志""" logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('customer_service.log'), logging.StreamHandler() ] ) def log_agent_interaction(conversation_id: str, user_input: str, agent_response: str, intent: str): """记录智能体交互日志""" logging.info( f"Conversation: {conversation_id}, " f"Intent: {intent}, " f"User: {user_input[:100]}, " f"Agent: {agent_response[:100]}" )

10.3 性能优化

1. 连接池：数据库和外部API连接使用连接池
2. 缓存策略：对静态数据实现多级缓存
3. 异步处理：I/O密集型操作使用异步模式
4. 资源限制：设置合理的超时和并发限制

11. 扩展与定制

11.1 添加新的业务领域

要扩展系统支持新的业务领域（如退款、售后等），只需：

1. 创建新的专业智能体类
2. 在意图识别中增加新的分类
3. 在路由逻辑中添加对应的处理分支
4. 实现相应的MCP工具

11.2 集成外部系统

通过MCP协议可以轻松集成各种外部系统：

• 支付系统：查询支付状态、处理退款
• 库存系统：检查商品库存、预约库存
• CRM系统：获取客户信息、更新客户档案
• 物流平台：实时追踪包裹、计算运费

这个基于LangChain+LangGraph+MCP的智能体开发框架，为构建复杂的企业级AI应用提供了完整的解决方案。从简单的客服对话到复杂的业务流程，都能通过模块化的方式优雅地实现。

建议在实际项目中先从核心功能开始，逐步扩展复杂度，同时注重监控和测试，确保系统的稳定性和可靠性。