在智能体开发领域很多开发者都面临一个共同的困境面对复杂的业务场景如何构建一个既灵活又可控的多智能体系统传统的线性流程在处理多轮对话、动态路由和状态管理时往往力不从心。本文将基于价值2万多的AI大模型全套教程完整分享LangChainLangGraphMCP智能体开发实战经验。无论你是刚接触智能体开发的新手还是希望提升系统架构能力的中高级开发者本文都将为你提供从基础概念到项目实战的完整指导。通过本文你将掌握如何构建一个支持多轮对话、动态路由和外部工具调用的智能客服系统。1. 智能体开发技术栈概述1.1 LangChain智能体开发的基础框架LangChain是一个用于开发由语言模型驱动的应用程序的框架它简化了与LLM的交互过程提供了链式调用、记忆管理和工具集成等核心功能。核心组件包括Models支持多种LLM提供商OpenAI、Anthropic、本地模型等Prompts模板化提示词管理Chains将多个组件组合成工作流Agents根据输入动态选择工具和操作Memory维护对话和历史状态1.2 LangGraph复杂工作流的解决方案LangGraph是建立在LangChain之上的库专门用于构建有状态、多智能体的应用程序。与LangChain的线性链不同LangGraph使用图结构来定义复杂的工作流。关键特性基于状态机的图结构支持循环和条件分支多智能体协作可视化调试工具1.3 MCP模型上下文协议工具集成标准MCP是Anthropic提出的标准协议用于标准化AI模型与外部工具和数据的集成方式。它解决了传统API集成的多个痛点。MCP的核心优势动态工具发现运行时查询可用工具结构化调用校验自动验证参数格式统一安全管控集中管理访问权限多轮交互支持自然的对话式工具调用2. 环境准备与工具配置2.1 系统环境要求在开始项目之前确保你的开发环境满足以下要求# 检查Python版本 python3 --version # 需要 3.10 # 检查Node.js版本 node --version # 需要 18 # 检查pip版本 pip3 --version2.2 AWS Bedrock权限配置由于我们的示例使用Amazon Bedrock作为LLM服务需要配置相应的访问权限# 安装AWS CLI curl https://awscli.amazonaws.com/awscli-exe-linux-x86_64.zip -o awscliv2.zip unzip awscliv2.zip sudo ./aws/install # 配置AWS凭证 aws configure # 输入Access Key ID、Secret Access Key、区域如us-west-2 # 验证Bedrock访问权限 aws bedrock list-foundation-models --region us-west-22.3 项目依赖安装创建项目目录并安装必要的Python包# 创建项目目录 mkdir customer_service_mcp cd customer_service_mcp # 创建虚拟环境 python3 -m venv venv source venv/bin/activate # 安装核心依赖 pip install langchain0.1.0 pip install langchain_community pip install langchain_mcp_adapters0.1.0 pip install boto31.34.0 pip install mcp-server0.1.0 pip install aioconsole0.7.0 pip install python-dotenv1.0.0 pip install regex2023.0.03. 智能客服系统架构设计3.1 系统整体架构我们的智能客服系统采用分层架构设计用户界面层 → 路由层 → 专业代理层 → 工具服务层 → 数据存储层各层职责说明用户界面层处理用户输入和系统输出路由层识别用户意图并分发给相应代理专业代理层订单代理、物流代理等专业处理模块工具服务层MCP服务器封装的外部工具调用数据存储层订单数据、对话历史等持久化存储3.2 数据流设计系统采用状态驱动的数据流模式from typing import TypedDict, Optional, Dict, Any class ConversationState(TypedDict): 对话状态数据结构 user_input: str intent: str # ORDER or LOGISTICS order_id: Optional[str] conversation_history: list current_agent: str resolved: bool4. 核心模块实现4.1 基础代理类设计首先创建所有代理的基类实现公共功能# 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 class BaseAgent(ABC): 所有客服代理的基类 def __init__(self, model_id: str anthropic.claude-3-sonnet-20240229-v1:0, region: str us-west-2): 初始化代理使用的LLM模型 self.llm BedrockChat( model_idmodel_id, model_kwargs{temperature: 0.7, max_tokens: 2048}, region_nameregion ) self.conversation_history: Dict[str, List[Dict]] {} def _get_conversation_history(self, conversation_id: str) - List[Dict]: 获取指定对话的历史记录 return self.conversation_history.get(conversation_id, []) def _update_conversation_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): 处理用户输入的核心方法 pass4.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) - tuple[str, str]: 处理用户输入识别意图 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_id4.3 订单问题代理实现订单问题代理专门处理与订单相关的问题# agents/order_issue_agent.py import uuid from typing import Optional, List, Dict from langchain.prompts import ChatPromptTemplate from agents.base_agent import BaseAgent from services.order_service import OrderService from services.sop_service import SOPService class OrderIssueAgent(BaseAgent): 订单问题代理处理订单相关的客户问题 def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.order_service OrderService() self.sop_service SOPService() self.prompt ChatPromptTemplate.from_messages([ (system, 你是处理订单问题的客服代理。 请按照以下决策树处理客户咨询 {decision_tree} 之前的对话 {history} 指导原则 - 请严格遵循决策树不要随意回答 - 如果不确定询问的对象请要求更多细节 - 这是即时消息保持回复简短精炼 - 不要使用此致敬礼等正式结束语 - 不要以客户的身份回复), (human, 订单信息 {order_info} 客户问题{question}) ]) def _format_order_info(self, order_info: Optional[dict]) - str: 格式化订单信息用于提示词 if not order_info: return 没有提供具体的订单信息。 return ( f订单详情\n f- 订单ID{order_info[order_id]}\n f- 客户{order_info[customer_name]}\n f- 商品{, .join(order_info[items])}\n f- 状态{order_info[status]}\n f- 配送地址{order_info[address]} ) def _format_history(self, history: List[Dict[str, str]]) - str: 格式化对话历史 if not history: return 没有之前的对话。 return \n.join([ f{msg[role].capitalize()}: {msg[content]} for msg in history ]) def process(self, user_input: str, conversation_id: Optional[str] None, order_id: Optional[str] None, history: List[Dict[str, str]] None, **kwargs): 处理订单相关的客户咨询 if not conversation_id: conversation_id str(uuid.uuid4()) # 获取订单信息如果提供了订单ID order_info None if order_id: order_info self.order_service.get_order_info(order_id) # 准备处理链 chain self.prompt | self.llm # 获取响应 response chain.invoke({ decision_tree: self.sop_service.order_decision_tree, history: self._format_history(history or []), order_info: self._format_order_info(order_info), question: user_input }) # 更新对话历史 self._update_conversation_history(conversation_id, user_input, response.content) return response.content, conversation_id4.4 物流问题代理实现物流问题代理专门处理配送相关的问题# agents/logistics_issue_agent.py import uuid import re from typing import Optional, List, Dict from langchain.prompts import ChatPromptTemplate from agents.base_agent import BaseAgent from services.order_service import OrderService from services.sop_service import SOPService class LogisticsIssueAgent(BaseAgent): 物流问题代理处理配送相关的客户问题 def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.order_service OrderService() self.sop_service SOPService() self.prompt ChatPromptTemplate.from_messages([ (system, 你是处理物流问题的客服代理。 请按照以下决策树处理客户咨询 {decision_tree} 之前的对话 {history} 指导原则 - 请严格遵循决策树不要随意回答 - 如果不确定询问的对象请要求更多细节 - 这是即时消息保持回复简短精炼 - 不要使用此致敬礼等正式结束语 - 不要以客户的身份回复 - 特别注意配送时间框架和补偿政策), (human, 订单信息 {order_info} 客户问题{question}) ]) def _extract_order_id(self, text: str) - Optional[str]: 从文本中提取订单ID patterns [ r订单\s*(?:ID|id)?\s*[#]?\s*(\d), rorder\s*(?:ID|id)?\s*[#]?\s*(\d), r订单号?\s*[:]\s*(\d) ] for pattern in patterns: match re.search(pattern, text, re.IGNORECASE) if match: return match.group(1) return None def process(self, user_input: str, conversation_id: Optional[str] None, order_id: Optional[str] None, history: List[Dict[str, str]] None, **kwargs): 处理物流相关的客户咨询 if not conversation_id: conversation_id str(uuid.uuid4()) # 尝试从用户输入中提取订单ID extracted_order_id self._extract_order_id(user_input) target_order_id order_id or extracted_order_id # 获取订单信息 order_info None if target_order_id: order_info self.order_service.get_order_info(target_order_id) # 格式化信息 formatted_order_info self._format_order_info(order_info) if hasattr(self, _format_order_info) else str(order_info) formatted_history self._format_history(history or []) if hasattr(self, _format_history) else str(history) # 准备处理链 chain self.prompt | self.llm # 获取响应 response chain.invoke({ decision_tree: self.sop_service.logistics_decision_tree, history: formatted_history, order_info: formatted_order_info, question: user_input }) # 更新对话历史 self._update_conversation_history(conversation_id, user_input, response.content) return response.content, conversation_id5. 服务层实现5.1 订单服务实现订单服务负责管理订单数据的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 _load_order_data(self) - List[Dict]: 从文件加载订单数据 try: with open(self.data_file, r, encodingutf-8) as file: return json.load(file) except (FileNotFoundError, json.JSONDecodeError) as e: print(f加载订单数据错误{e}) return [] def _save_order_data(self, order_data: List[Dict]) - bool: 保存订单数据到文件 try: with open(self.data_file, w, encodingutf-8) as file: json.dump(order_data, file, indent2, ensure_asciiFalse) return True except Exception as e: print(f保存订单数据错误{e}) return False def get_order_info(self, order_id: str) - Optional[Dict]: 根据订单ID获取订单信息 orders self._load_order_data() for order in orders: if order[order_id] order_id: return order return None def update_order_address(self, order_id: str, new_address: str) - bool: 更新订单配送地址 orders self._load_order_data() for order in orders: if order[order_id] order_id: order[address] new_address return self._save_order_data(orders) return False def get_all_orders(self) - List[Dict]: 获取所有订单信息 return self._load_order_data()5.2 SOP服务实现SOP标准操作程序服务提供决策树支持# services/sop_service.py class SOPService: SOP服务管理标准操作程序的决策树 property def order_decision_tree(self) - str: 订单问题决策树 return 订单问题决策树 1. 订单状态查询 1.1. 我的订单在哪里 → 使用订单ID检查订单状态 1.2. 订单进度如何 → 提供当前处理阶段和预计时间 2. 订单修改 2.1. 可以修改/取消订单吗 → 检查订单是否仍在处理中 2.1.1. 如果状态为Processing → 允许修改 2.1.2. 如果状态为Shipped → 无法修改提供物流信息 2.2. 想给订单添加商品 → 检查订单状态 2.2.1. 如果状态为Processing → 建议重新下单或联系客服 2.2.2. 如果状态为Shipped → 无法添加建议单独下单 3. 支付问题 3.1. 支付失败 → 验证支付信息建议重试或更换支付方式 3.2. 退款查询 → 检查退款状态提供预计到账时间 property def logistics_decision_tree(self) - str: 物流问题决策树 return 物流问题决策树 1. 包裹位置查询 1.1. 包裹超过预计送达时间 1.1.1. 检查承运商网站跟踪信息 1.1.1.1. 超过ETA 7天 → 建议再等待2-3天 1.1.1.2. 超过ETA 7天且有跟踪更新 → 建议等待2-3天并联系承运商 1.1.1.2.1. 客户不愿等待 → 提供100积分补偿 1.1.1.2.2. 客户非常不满 → 提供100%店铺信用最终方案100%现金退款 1.1.1.3. 超过ETA 7天且无跟踪更新 → 提供100%店铺信用或重新发货选项 1.2. 跟踪信息4天无更新 1.2.1. 仍在ETA内 → 升级至物流团队调查 1.2.2. 超过ETA → 遵循包裹超过预计送达时间流程 2. 配送地址 2.1. 修改配送地址 → 如果订单未发货则更新地址 2.2. 地址验证 → 确认地址详细信息 3. 包裹标记为已送达但未收到 3.1. 检查是整个包裹未收到还是缺少物品 3.2. 与客户共享并验证地址 3.3. 首次客户 3.3.1. 地址正确 → 提供重新发货或100%现金退款 3.3.2. 地址不正确 → 提供50%店铺信用最终重新发货或100%现金退款 3.4. 回头客 3.4.1. 地址正确且订单$200 → 提供100%店铺信用 3.4.2. 地址不正确且订单$200 → 提供50%店铺信用 3.4.3. 订单$200 → 升级至团队负责人 注意事项 - 如果客户已支付关税不要提供重新发货 - 对于BNPL支付方式Klarna/Afterpay强调店铺信用不是真实货币 - 对于$200订单的特殊情况升级至团队负责人 6. MCP服务器集成6.1 MCP服务器配置创建MCP服务器来暴露我们的工具# server.py from mcp.server.fastmcp import FastMCP import json import uuid from typing import Optional, Dict, Any 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_asciiFalse) 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_asciiFalse) 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_asciiFalse) 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 get_sop_tree(sop_type: str) - str: 获取特定的SOP决策树 try: if sop_type.lower() order: return json.dumps({ decision_tree: system.sop_service.order_decision_tree }, ensure_asciiFalse) elif sop_type.lower() logistics: return json.dumps({ decision_tree: system.sop_service.logistics_decision_tree }, ensure_asciiFalse) return json.dumps({ error: f未知的SOP类型{sop_type}, sop_type: sop_type }) except Exception as e: return json.dumps({ error: f发生错误{str(e)}, sop_type: sop_type }) if __name__ __main__: mcp.run(transportsse)6.2 主系统协调器创建主系统来协调各个代理和服务# main.py import uuid import json import asyncio import aioconsole import re from typing import Optional, Dict, Any from langchain_community.chat_models import BedrockChat from langchain_mcp_adapters.client import MultiServerMCPClient 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_idmodel_id, regionregion) self.order_agent OrderIssueAgent(model_idmodel_id, regionregion) self.logistics_agent LogisticsIssueAgent(model_idmodel_id, regionregion) # 初始化服务 self.order_service OrderService() self.sop_service SOPService() # 存储活跃对话 self.conversations: Dict[str, Dict[str, Any]] {} def _extract_order_id(self, text: str) - Optional[str]: 从文本中提取订单ID patterns [ r订单\s*(?:ID|id)?\s*[#]?\s*(\d), rorder\s*(?:ID|id)?\s*[#]?\s*(\d), r订单号?\s*[:]\s*(\d) ] for pattern in patterns: match re.search(pattern, text, re.IGNORECASE) if match: return match.group(1) return None def process_question(self, user_question: str, conversation_id: Optional[str] None) - tuple[str, str]: 通过多代理系统处理客户问题 # 生成对话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, historyself.conversations[conversation_id][history] ) print(f识别到的意图{intent}) # 从问题中提取订单ID如果存在 extracted_order_id self._extract_order_id(user_question) if extracted_order_id: self.conversations[conversation_id][order_id] extracted_order_id # 第二层基于意图处理 current_order_id self.conversations[conversation_id].get(order_id) if intent ORDER: response, _ self.order_agent.process( user_question, conversation_id, order_idcurrent_order_id, historyself.conversations[conversation_id][history] ) elif intent LOGISTICS: response, _ self.logistics_agent.process( user_question, conversation_id, order_idcurrent_order_id, historyself.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) # 初始化MCP客户端 client MultiServerMCPClient({ customer_service: { url: http://localhost:8000/sse, transport: sse, } }) tools await client.get_tools() process_question_tool next( (tool for tool in tools if tool.name process_question), None ) if not process_question_tool: print(错误未找到process_question工具) return while True: try: user_input await aioconsole.ainput(\n客户) if user_input.lower() exit: print(感谢使用我们的客服系统。再见) break result await process_question_tool.arun({ question: user_input, conversation_id: conversation_id }) response_data json.loads(result) print(f\n客服{response_data[response]}) conversation_id response_data[conversation_id] except Exception as e: print(f\n错误{str(e)}) if __name__ __main__: asyncio.run(interactive_session())7. 项目部署与运行7.1 启动脚本配置创建启动脚本来简化部署过程# start_server.sh #!/bin/bash # 创建虚拟环境如果不存在 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# start_client.sh #!/bin/bash # 激活虚拟环境 source venv/bin/activate # 启动客户端 python3 main.py # 停用虚拟环境 deactivate7.2 依赖文件配置创建requirements.txt文件# requirements.txt langchain0.1.0 langchain_community0.0.1 langchain_mcp_adapters0.1.0 boto31.34.0 python-dotenv1.0.0 regex2023.0.0 mcp-server0.1.0 aioconsole0.7.0 uvicorn0.24.0 fastapi0.104.07.3 运行系统按照以下步骤运行完整的智能客服系统# 1. 给启动脚本添加执行权限 chmod x start_server.sh chmod x start_client.sh # 2. 启动MCP服务器终端1 ./start_server.sh # 3. 启动客户端终端2 ./start_client.sh8. 系统测试与验证8.1 功能测试用例测试系统的各项功能# test_system.py import asyncio from main import CustomerServiceSystem async def test_system(): 测试客服系统功能 system CustomerServiceSystem() test_cases [ # 订单状态查询 订单123的状态是什么, # 地址修改 我想把订单123的地址改成朝阳区XXX路123号, # 物流问题 订单456显示已送达但我没收到, # 混合场景 我的订单789和订单123都出了问题 ] for i, question in enumerate(test_cases, 1): print(f\n测试用例 {i}: {question}) response, conv_id system.process_question(question) print(f响应: {response}) print(f对话ID: {conv_id}) print(- * 50) if __name__ __main__: asyncio.run(test_system())8.2 性能优化建议基于实际测试结果提供性能优化建议内存优化定期清理过期的对话历史使用LRU缓存存储频繁访问的订单数据实现对话历史的压缩存储响应速度优化对LLM调用实现缓存机制使用异步处理提高并发能力对常用查询实现预计算9. 常见问题与解决方案9.1 环境配置问题问题1AWS Bedrock权限错误Error: Unable to locate credentials解决方案# 检查AWS配置 aws configure list # 重新配置AWS凭证 aws configure # 验证Bedrock访问 aws bedrock list-foundation-models --region us-west-2问题2Python包版本冲突ImportError: cannot import name xxx from langchain解决方案# 清理并重新安装 pip uninstall -y langchain langchain_community pip install langchain0.1.0 langchain_community0.0.19.2 运行时问题问题3MCP服务器启动失败Address already in use解决方案# 查找占用端口的进程 lsof -i :8000 # 终止占用进程 kill -9 PID # 或者更改服务器端口 python server.py --port 8080问题4订单数据文件权限错误Permission denied: order_data.txt解决方案# 更改文件权限 chmod 644 order_data.txt # 或者指定可写目录 export DATA_DIR/path/to/writable/directory10. 扩展与进阶功能10.1 集成LangGraph实现复杂工作流将现有系统升级到LangGraph架构# langgraph_system.py from typing import Typed
