在企业级AI应用开发中面对复杂业务流程和多轮交互场景传统的线性处理链往往显得力不从心。特别是当需要处理状态管理、条件分支、循环执行等复杂逻辑时开发者常常陷入架构设计的困境。本文将基于最新的LangChain、LangGraph和MCP技术栈为企业级AI智能体开发提供完整的实战解决方案。1. 技术栈核心概念解析1.1 LangChainAI应用开发的基础框架LangChain是一个用于构建大语言模型应用的框架它提供了标准化的接口、组件和工具链。在企业级应用中LangChain主要解决以下核心问题组件化设计将复杂的AI应用拆分为可复用的组件如提示模板、记忆模块、输出解析器等链式编排通过链Chain的概念将多个组件串联起来形成完整的工作流程工具集成标准化外部工具和API的集成方式使AI能够调用外部资源# 基础LangChain使用示例 from langchain.chains import LLMChain from langchain.prompts import PromptTemplate from langchain.llms import OpenAI # 创建提示模板 prompt PromptTemplate( input_variables[product], template为{product}写一个营销文案突出其核心优势。 ) # 创建LLM链 llm OpenAI(temperature0.7) chain LLMChain(llmllm, promptprompt) # 执行链 result chain.run(智能客服系统) print(result)1.2 LangGraph复杂工作流的终极解决方案LangGraph是LangChain的扩展专门用于处理复杂的有状态工作流。与传统的线性链不同LangGraph基于图结构支持循环、条件分支和状态持久化。核心概念解析状态State贯穿整个工作流的共享数据容器使用TypedDict或Pydantic模型定义节点Nodes工作流中的执行单元每个节点都是一个函数接收状态并返回更新后的状态边Edges定义节点之间的流转逻辑包括条件边和普通边检查点Checkpoints支持工作流的状态持久化和恢复from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END # 定义状态结构 class CustomerServiceState(TypedDict): customer_query: str intent: str order_details: dict resolution: str conversation_history: Annotated[list, append] # 创建图结构 graph_builder StateGraph(CustomerServiceState) # 定义节点函数 def intent_recognition_node(state: CustomerServiceState): # 意图识别逻辑 if 订单 in state[customer_query]: state[intent] ORDER_ISSUE elif 物流 in state[customer_query]: state[intent] LOGISTICS_ISSUE else: state[intent] GENERAL_QUERY return state def order_processing_node(state: CustomerServiceState): # 订单处理逻辑 if state[intent] ORDER_ISSUE: # 提取订单ID并查询详情 state[order_details] fetch_order_details(state[customer_query]) return state # 添加节点到图中 graph_builder.add_node(intent_recognition, intent_recognition_node) graph_builder.add_node(order_processing, order_processing_node) # 设置入口点 graph_builder.set_entry_point(intent_recognition) # 定义边逻辑 def route_after_intent(state: CustomerServiceState): if state[intent] ORDER_ISSUE: return order_processing else: return END graph_builder.add_conditional_edges( intent_recognition, route_after_intent, { order_processing: order_processing, END: END } ) graph_builder.add_edge(order_processing, END) # 编译图 graph graph_builder.compile()1.3 MCP模型上下文协议标准化工具集成MCP是Anthropic提出的标准协议用于规范AI模型与外部工具的交互方式。相比传统的API集成MCP提供了动态工具发现、结构化调用和安全访问等优势。MCP的核心优势动态工具发现客户端可以在运行时查询服务器可用的工具列表结构化调用验证服务器负责验证请求参数减少调用错误统一安全管控集中的权限管理和访问控制多轮交互支持支持复杂的多步骤工具调用场景2. 环境准备与项目搭建2.1 系统环境要求确保开发环境满足以下要求# 检查Python版本 python3 --version # 需要 3.10 # 检查Node.js版本如需使用MCP Inspector node --version # 需要 18 # 安装必要的系统依赖 sudo apt update sudo apt install python3-pip python3-venv curl -y2.2 项目初始化与依赖管理创建项目目录结构并配置虚拟环境# 创建项目目录 mkdir enterprise-ai-agent cd enterprise-ai-agent # 创建虚拟环境 python3 -m venv venv source venv/bin/activate # 安装核心依赖 pip install langchain langgraph langchain-community pip install mcp-server fastmcp langchain-mcp-adapters pip install boto3 python-dotenv regex aioconsole创建项目配置文件# config/settings.py import os from dotenv import load_dotenv load_dotenv() class Settings: # LLM配置 MODEL_ID os.getenv(MODEL_ID, anthropic.claude-3-sonnet-20240229-v1:0) AWS_REGION os.getenv(AWS_REGION, us-west-2) # MCP服务器配置 MCP_SERVER_HOST os.getenv(MCP_SERVER_HOST, localhost) MCP_SERVER_PORT int(os.getenv(MCP_SERVER_PORT, 8000)) # 应用配置 MAX_CONVERSATION_HISTORY 10 SESSION_TIMEOUT 3600 # 1小时 settings Settings()2.3 项目结构设计采用模块化设计确保代码的可维护性和扩展性enterprise-ai-agent/ ├── src/ │ ├── agents/ # 智能体模块 │ │ ├── base_agent.py │ │ ├── intent_agent.py │ │ ├── order_agent.py │ │ └── logistics_agent.py │ ├── services/ # 业务服务 │ │ ├── order_service.py │ │ ├── customer_service.py │ │ └── sop_service.py │ ├── graphs/ # LangGraph工作流 │ │ ├── customer_service_graph.py │ │ └── order_processing_graph.py │ ├── mcp/ # MCP集成 │ │ ├── servers/ │ │ └── clients/ │ └── utils/ # 工具函数 ├── tests/ # 测试用例 ├── docs/ # 项目文档 ├── requirements.txt # 依赖列表 └── main.py # 应用入口3. 核心架构设计与实现3.1 多智能体系统架构基于LangGraph构建企业级多智能体系统实现智能路由和协作# src/graphs/customer_service_graph.py from typing import TypedDict, Annotated, Literal from langgraph.graph import StateGraph, END from src.agents.intent_agent import IntentRecognitionAgent from src.agents.order_agent import OrderProcessingAgent from src.agents.logistics_agent import LogisticsAgent class CustomerServiceState(TypedDict): 客户服务状态定义 customer_input: str conversation_id: str intent: Literal[ORDER, LOGISTICS, GENERAL, UNKNOWN] extracted_order_id: str current_agent: str agent_response: str conversation_history: Annotated[list, append] requires_human_intervention: bool class CustomerServiceGraph: def __init__(self): self.intent_agent IntentRecognitionAgent() self.order_agent OrderProcessingAgent() self.logistics_agent LogisticsAgent() self.graph self._build_graph() def _build_graph(self): 构建客户服务图 graph_builder StateGraph(CustomerServiceState) # 添加节点 graph_builder.add_node(intent_detection, self._intent_detection_node) graph_builder.add_node(order_processing, self._order_processing_node) graph_builder.add_node(logistics_processing, self._logistics_processing_node) graph_builder.add_node(general_query, self._general_query_node) graph_builder.add_node(human_escalation, self._human_escalation_node) # 设置入口点 graph_builder.set_entry_point(intent_detection) # 定义路由逻辑 graph_builder.add_conditional_edges( intent_detection, self._route_based_on_intent, { order_processing: order_processing, logistics_processing: logistics_processing, general_query: general_query, human_escalation: human_escalation } ) # 添加普通边 graph_builder.add_edge(order_processing, END) graph_builder.add_edge(logistics_processing, END) graph_builder.add_edge(general_query, END) graph_builder.add_edge(human_escalation, END) return graph_builder.compile() def _intent_detection_node(self, state: CustomerServiceState): 意图识别节点 intent, order_id self.intent_agent.detect_intent( state[customer_input], state.get(conversation_history, []) ) state[intent] intent state[extracted_order_id] order_id state[current_agent] intent_detection return state def _route_based_on_intent(self, state: CustomerServiceState): 基于意图的路由逻辑 if state[requires_human_intervention]: return human_escalation intent state[intent] if intent ORDER: return order_processing elif intent LOGISTICS: return logistics_processing else: return general_query def _order_processing_node(self, state: CustomerServiceState): 订单处理节点 response self.order_agent.process_query( state[customer_input], state[extracted_order_id], state.get(conversation_history, []) ) state[agent_response] response state[current_agent] order_processing # 检查是否需要人工介入 if escalate in response.lower(): state[requires_human_intervention] True return state def process_request(self, customer_input: str, conversation_id: str None): 处理客户请求 initial_state { customer_input: customer_input, conversation_id: conversation_id or self._generate_conversation_id(), intent: UNKNOWN, extracted_order_id: , current_agent: , agent_response: , conversation_history: [], requires_human_intervention: False } result self.graph.invoke(initial_state) return result3.2 MCP服务器实现实现标准化的MCP服务器提供工具发现和结构化调用# src/mcp/servers/customer_service_server.py from mcp.server.fastmcp import FastMCP from mcp.server.models import InitializationOptions import json from src.graphs.customer_service_graph import CustomerServiceGraph # 创建FastMCP实例 mcp FastMCP(CustomerService) class CustomerServiceMCPServer: def __init__(self): self.service_graph CustomerServiceGraph() self.active_sessions {} mcp.tool() async def process_customer_query(self, question: str, session_id: str None) - str: 处理客户查询并返回响应 try: result self.service_graph.process_request(question, session_id) # 更新会话状态 if session_id: self.active_sessions[session_id] { last_activity: datetime.now(), conversation_history: result[conversation_history] } response_data { response: result[agent_response], session_id: result[conversation_id], intent: result[intent], requires_human: result[requires_human_intervention] } return json.dumps(response_data, ensure_asciiFalse) except Exception as e: error_response { error: f处理查询时发生错误: {str(e)}, session_id: session_id } return json.dumps(error_response) mcp.tool() async def get_order_information(self, order_id: str) - str: 获取订单详细信息 try: order_service OrderService() order_info order_service.get_order_details(order_id) if order_info: return json.dumps({ order_id: order_id, status: order_info.status, customer_name: order_info.customer_name, items: order_info.items, delivery_address: order_info.delivery_address, last_updated: order_info.last_updated }, ensure_asciiFalse) else: 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(self, order_id: str, new_address: str) - str: 更新订单配送地址 try: order_service OrderService() success order_service.update_delivery_address(order_id, new_address) if success: updated_order order_service.get_order_details(order_id) return json.dumps({ message: 地址更新成功, order_id: order_id, new_address: new_address, updated_order: updated_order.to_dict() }, ensure_asciiFalse) else: 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 }) # 服务器启动配置 def start_mcp_server(host: str localhost, port: int 8000): 启动MCP服务器 server CustomerServiceMCPServer() # 注册工具 mcp.add_tool(server.process_customer_query) mcp.add_tool(server.get_order_information) mcp.add_tool(server.update_order_address) # 启动服务器 mcp.run(hosthost, portport, transportsse) if __name__ __main__: start_mcp_server()4. 企业级功能实现4.1 智能路由与负载均衡实现基于业务复杂度的智能路由机制# src/agents/base_agent.py from abc import ABC, abstractmethod from typing import Dict, List, Optional from langchain_community.chat_models import BedrockChat from langchain.prompts import ChatPromptTemplate import time import logging class BaseAgent(ABC): 智能体基类提供通用功能 def __init__(self, agent_id: str, model_config: Dict): self.agent_id agent_id self.llm BedrockChat( model_idmodel_config.get(model_id), model_kwargsmodel_config.get(model_kwargs, {}), region_namemodel_config.get(region, us-west-2) ) self.conversation_histories: Dict[str, List] {} self.metrics { requests_processed: 0, average_response_time: 0, error_count: 0 } self.logger logging.getLogger(fagent.{agent_id}) def _update_metrics(self, processing_time: float, success: bool True): 更新性能指标 self.metrics[requests_processed] 1 # 计算平均响应时间移动平均 current_avg self.metrics[average_response_time] n self.metrics[requests_processed] self.metrics[average_response_time] ( current_avg * (n-1) processing_time ) / n if n 1 else processing_time if not success: self.metrics[error_count] 1 def _get_conversation_history(self, session_id: str) - List: 获取会话历史 return self.conversation_histories.get(session_id, []) def _update_conversation_history(self, session_id: str, user_message: str, agent_response: str): 更新会话历史 if session_id not in self.conversation_histories: self.conversation_histories[session_id] [] # 限制历史记录长度 history self.conversation_histories[session_id] history.extend([ {role: user, content: user_message}, {role: assistant, content: agent_response} ]) # 保持最近10轮对话 if len(history) 20: # 10轮对话每轮2条消息 self.conversation_histories[session_id] history[-20:] abstractmethod def process_query(self, query: str, context: Dict) - Dict: 处理查询的抽象方法 pass def get_agent_metrics(self) - Dict: 获取智能体性能指标 return self.metrics.copy() class IntentRecognitionAgent(BaseAgent): 意图识别智能体 def __init__(self, model_config: Dict): super().__init__(intent_recognition, model_config) self.prompt_template ChatPromptTemplate.from_messages([ (system, 你是一个专业的意图识别系统专门分析电商客户服务查询。 请分析用户问题并识别其意图类型 1. ORDER - 订单相关问题状态查询、修改、支付问题 2. LOGISTICS - 物流相关问题配送地址、运输方式、配送问题 3. GENERAL - 一般咨询问题产品信息、促销活动等 4. COMPLAINT - 投诉和建议 同时从问题中提取订单ID如果存在。 请严格按照以下JSON格式响应 { intent: ORDER|LOGISTICS|GENERAL|COMPLAINT, order_id: 提取到的订单ID或空字符串, confidence: 0.0到1.0的置信度 }), (human, 用户问题{question}\n对话历史{history}) ]) def process_query(self, query: str, context: Dict) - Dict: 处理意图识别查询 start_time time.time() try: session_id context.get(session_id, default) history self._get_conversation_history(session_id) # 格式化历史记录 formatted_history \n.join([ f{msg[role]}: {msg[content]} for msg in history[-6:] # 最近3轮对话 ]) # 调用LLM chain self.prompt_template | self.llm response chain.invoke({ question: query, history: formatted_history }) # 解析响应 import json result json.loads(response.content) processing_time time.time() - start_time self._update_metrics(processing_time, successTrue) self.logger.info(f意图识别完成: {result}) return result except Exception as e: processing_time time.time() - start_time self._update_metrics(processing_time, successFalse) self.logger.error(f意图识别失败: {str(e)}) return { intent: UNKNOWN, order_id: , confidence: 0.0, error: str(e) }4.2 持久化状态管理与检查点实现企业级的状态持久化和恢复机制# src/utils/checkpoint_manager.py import json import pickle import redis from datetime import datetime, timedelta from typing import Dict, Any, Optional import hashlib class CheckpointManager: 检查点管理器支持状态持久化和恢复 def __init__(self, redis_url: str redis://localhost:6379, ttl: int 86400): self.redis_client redis.from_url(redis_url) self.ttl ttl # 生存时间秒 def _generate_checkpoint_key(self, session_id: str, checkpoint_id: str None) - str: 生成检查点键 if checkpoint_id: return fcheckpoint:{session_id}:{checkpoint_id} else: return fcheckpoint:{session_id}:latest def save_checkpoint(self, session_id: str, state: Dict[str, Any], checkpoint_id: str None) - bool: 保存状态检查点 try: checkpoint_data { state: state, timestamp: datetime.now().isoformat(), session_id: session_id, checkpoint_id: checkpoint_id or self._generate_checkpoint_id() } # 序列化数据 serialized_data pickle.dumps(checkpoint_data) # 存储到Redis key self._generate_checkpoint_key(session_id, checkpoint_id) self.redis_client.setex(key, self.ttl, serialized_data) # 更新最新检查点引用 latest_key self._generate_checkpoint_key(session_id) self.redis_client.setex(latest_key, self.ttl, serialized_data) return True except Exception as e: print(f保存检查点失败: {str(e)}) return False def load_checkpoint(self, session_id: str, checkpoint_id: str None) - Optional[Dict]: 加载状态检查点 try: key self._generate_checkpoint_key(session_id, checkpoint_id) serialized_data self.redis_client.get(key) if serialized_data: checkpoint_data pickle.loads(serialized_data) return checkpoint_data else: return None except Exception as e: print(f加载检查点失败: {str(e)}) return None def list_checkpoints(self, session_id: str) - List[Dict]: 列出会话的所有检查点 try: pattern fcheckpoint:{session_id}:* keys self.redis_client.keys(pattern) checkpoints [] for key in keys: serialized_data self.redis_client.get(key) if serialized_data: checkpoint_data pickle.loads(serialized_data) checkpoints.append({ checkpoint_id: checkpoint_data[checkpoint_id], timestamp: checkpoint_data[timestamp], key: key.decode(utf-8) }) # 按时间排序 checkpoints.sort(keylambda x: x[timestamp], reverseTrue) return checkpoints except Exception as e: print(f列出检查点失败: {str(e)}) return [] def _generate_checkpoint_id(self) - str: 生成检查点ID timestamp datetime.now().isoformat() return hashlib.md5(timestamp.encode()).hexdigest()[:8] # 集成检查点管理的LangGraph示例 from langgraph.checkpoint.memory import MemorySaver from langgraph.checkpoint.redis import RedisSaver class EnterpriseGraphWithCheckpoints: 支持企业级检查点管理的图 def __init__(self, redis_url: str None): # 根据配置选择检查点存储方式 if redis_url: self.checkpointer RedisSaver.from_redis_url(redis_url) else: self.checkpointer MemorySaver() self.graph self._build_graph() def _build_graph(self): 构建带检查点的图 graph_builder StateGraph(CustomerServiceState) # 添加节点... graph_builder.add_node(intent_detection, self._intent_detection_node) graph_builder.add_node(order_processing, self._order_processing_node) # 配置检查点 graph graph_builder.compile(checkpointerself.checkpointer) return graph def process_with_checkpoint(self, input_data: Dict, config: Dict None): 带检查点处理 thread_id config.get(thread_id) if config else None if thread_id: # 从检查点恢复状态 result self.graph.invoke( input_data, config{configurable: {thread_id: thread_id}} ) else: # 新会话 result self.graph.invoke(input_data) return result5. 实战案例电商客服智能体系统5.1 完整系统集成将各个组件集成为完整的电商客服系统# src/main.py import asyncio import uvicorn from fastapi import FastAPI, HTTPException from pydantic import BaseModel from src.graphs.customer_service_graph import CustomerServiceGraph from src.mcp.servers.customer_service_server import CustomerServiceMCPServer app FastAPI(title企业级AI客服系统, version1.0.0) # 全局实例 service_graph CustomerServiceGraph() mcp_server CustomerServiceMCPServer() class CustomerQuery(BaseModel): question: str session_id: str None user_id: str None class QueryResponse(BaseModel): response: str session_id: str intent: str requires_human: bool processing_time: float app.post(/api/query, response_modelQueryResponse) async def process_customer_query(query: CustomerQuery): 处理客户查询API端点 try: start_time asyncio.get_event_loop().time() result service_graph.process_request( query.question, query.session_id ) processing_time asyncio.get_event_loop().time() - start_time return QueryResponse( responseresult[agent_response], session_idresult[conversation_id], intentresult[intent], requires_humanresult[requires_human_intervention], processing_timeprocessing_time ) except Exception as e: raise HTTPException(status_code500, detailf处理查询时发生错误: {str(e)}) app.get(/api/health) async def health_check(): 健康检查端点 return { status: healthy, timestamp: asyncio.get_event_loop().time(), active_sessions: len(service_graph.active_sessions) } app.get(/api/metrics) async def get_system_metrics(): 获取系统指标 metrics { total_requests_processed: service_graph.metrics[requests_processed], average_response_time: service_graph.metrics[average_response_time], error_rate: service_graph.metrics[error_count] / max(service_graph.metrics[requests_processed], 1), active_sessions: len(service_graph.active_sessions), agents_metrics: {} } # 收集各个智能体的指标 for agent_name, agent in service_graph.agents.items(): metrics[agents_metrics][agent_name] agent.get_agent_metrics() return metrics # MCP服务器集成 app.post(/mcp/tools/call) async def mcp_tool_call(tool_request: dict): MCP工具调用端点 try: tool_name tool_request.get(name) arguments tool_request.get(arguments, {}) if tool_name process_customer_query: result await mcp_server.process_customer_query( arguments.get(question), arguments.get(session_id) ) elif tool_name get_order_information: result await mcp_server.get_order_information( arguments.get(order_id) ) else: raise HTTPException(status_code404, detailf工具 {tool_name} 不存在) return {result: result} except Exception as e: raise HTTPException(status_code500, detailstr(e)) if __name__ __main__: # 启动HTTP服务器 uvicorn.run( app, host0.0.0.0, port8000, log_levelinfo )5.2 系统部署与运维提供完整的部署脚本和运维指南#!/bin/bash # deploy.sh - 企业级AI客服系统部署脚本 set -e echo 开始部署企业级AI客服系统... # 检查环境 check_environment() { echo 检查部署环境... # 检查Python if ! command -v python3 /dev/null; then echo 错误: 未找到Python3 exit 1 fi # 检查Redis if ! command -v redis-cli /dev/null; then echo 警告: 未找到Redis将使用内存模式 fi echo 环境检查完成 } # 安装依赖 install_dependencies() { echo 安装Python依赖... # 创建虚拟环境 python3 -m venv venv source venv/bin/activate # 安装依赖包 pip install -r requirements.txt # 安装开发依赖可选 if [ $1 dev ]; then pip install -r requirements-dev.txt fi echo 依赖安装完成 } # 配置系统 setup_configuration() { echo 配置系统... # 创建配置文件 cat .env EOF # 应用配置 MODEL_IDanthropic.claude-3-sonnet-20240229-v1:0 AWS_REGIONus-west-2 MCP_SERVER_HOSTlocalhost MCP_SERVER_PORT8000 # Redis配置 REDIS_URLredis://localhost:6379 # 日志配置 LOG_LEVELINFO LOG_FILElogs/app.log # 安全配置 API_KEY${API_KEY:-$(openssl rand -hex 32)} SESSION_TIMEOUT3600 EOF # 创建目录结构 mkdir -p logs data/checkpoints echo 配置完成 } # 启动服务 start_services() { echo 启动服务... # 启动Redis如果已安装 if command -v redis-server /dev/null; then redis-server --daemonize yes echo Redis服务已启动 fi # 启动应用 source venv/bin/activate nohup python src/main.py logs/app.log 21 echo $! app.pid echo 应用服务已启动PID: $(cat app.pid) } # 健康检查 health_check() { echo 执行健康检查... sleep 5 # 等待服务启动 if curl -f http://localhost:8000/api/health /dev/null 21; then echo ✅ 系统健康检查通过 else echo ❌ 系统健康检查失败 exit 1 fi } # 主部署流程 main() { check_environment install_dependencies $1 setup_configuration start_services health_check echo 企业级AI客服系统部署完成! echo 管理界面: http://localhost:8000/docs echo 日志文件: logs/app.log } # 执行部署 main $6. 性能优化与最佳实践6.1 缓存策略实现# src/utils/cache_manager.py import redis import json import hashlib from functools import wraps from datetime import timedelta class CacheManager: 智能缓存管理器 def __init__(self, redis_url: str redis://localhost:6379): self.redis_client redis.from_url(redis_url) def _generate_cache_key(self, func_name: str, *args, **kwargs) - str: 生成缓存键 key_data f{func_name}:{str(args)}:{str(kwargs)} return hashlib.md5(key_data.encode()).hexdigest() def cache_result(self, ttl: int 300): # 默认5分钟 缓存装饰器 def decorator(func): wraps(func) def wrapper(*args, **kwargs): # 生成缓存键 cache_key self._generate_cache_key(func.__name__, *args, **kwargs) # 尝试从缓存获取 cached_result self.redis_client.get(cache_key) if cached_result: return json.loads(cached_result) # 执行函数 result func(*args, **kwargs) # 缓存结果 self.redis_client.setex( cache_key, ttl, json.dumps(result, ensure_asciiFalse) ) return result return wrapper return decorator # 使用缓存的智能体示例 class CachedOrderAgent(OrderProcessingAgent): 带缓存的订单处理智能体 def __init__(self, model_config: Dict, cache_manager: CacheManager): super().__init__(model_config) self.cache_manager cache_manager cache_manager.cache_result(ttl60) # 缓存1分钟 def get_order_details(self, order_id: str) - Dict: 获取订单详情带缓存 return super().get_order_details(order_id)6.2 监控与告警系统# src/monitoring/system_monitor.py import time import logging from dataclasses import dataclass from typing import Dict, List import psutil import requests dataclass class SystemMetrics: cpu_percent: float memory_percent: float disk_