zhaojie/assist
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 重大功能更新 v1.4.0 - 飞书集成、AI语义相似度、前端优化

Browse Source 
主要更新内容:
- 🚀 飞书多维表格集成，支持工单数据同步
- 🤖 AI建议与人工描述语义相似度计算
- 🎨 前端UI全面优化，现代化设计
- 📊 智能知识库入库策略（AI准确率<90%使用人工描述）
- 🔧 代码重构，模块化架构优化
- 📚 完整文档整合和更新
- 🐛 修复配置导入和数据库字段问题

技术特性:
- 使用sentence-transformers进行语义相似度计算
- 快速模式结合TF-IDF和语义方法
- 响应式设计，支持移动端
- 加载状态和动画效果
- 配置化AI准确率阈值
This commit is contained in:
赵杰 Jie Zhao （雄狮汽车科技）
2025-09-19 19:32:42 +01:00
parent 79cf316c63
commit da4736c323
30 changed files with 4778 additions and 1406 deletions
Download Patch File Download Diff File Expand all files Collapse all files

254
src/agent/agent_assistant_core.py Normal file
View File

@@ -0,0 +1,254 @@
# -*- coding: utf-8 -*-
"""
TSP Agent助手核心模块
包含Agent助手的核心功能和基础类
"""
import logging
import asyncio
from typing import Dict, Any, List, Optional
from datetime import datetime
import json
from src.main import TSPAssistant
from src.agent import AgentCore, AgentState
from src.agent.auto_monitor import AutoMonitorService
from src.agent.intelligent_agent import IntelligentAgent, AlertContext, KnowledgeContext
from src.agent.llm_client import LLMManager, LLMConfig
from src.agent.action_executor import ActionExecutor
logger = logging.getLogger(__name__)
class TSPAgentAssistantCore(TSPAssistant):
"""TSP Agent助手核心 - 基础功能"""
def __init__(self, llm_config: Optional[LLMConfig] = None):
# 初始化基础TSP助手
super().__init__()
# 初始化Agent核心
self.agent_core = AgentCore()
# 初始化自动监控服务
self.auto_monitor = AutoMonitorService(self)
# 初始化LLM客户端
self._init_llm_manager(llm_config)
# 初始化智能Agent
self.intelligent_agent = IntelligentAgent(
llm_manager=self.llm_manager,
agent_core=self.agent_core
)
# 初始化动作执行器
self.action_executor = ActionExecutor(self)
# Agent状态
self.agent_state = AgentState.IDLE
self.is_agent_mode = True
self.proactive_monitoring_enabled = False
# 执行历史
self.execution_history = []
self.max_history_size = 1000
logger.info("TSP Agent助手核心初始化完成")
def _init_llm_manager(self, llm_config: Optional[LLMConfig] = None):
"""初始化LLM管理器"""
if llm_config:
self.llm_manager = LLMManager(llm_config)
else:
# 使用默认配置 - 千问模型
try:
from config.llm_config import DEFAULT_CONFIG
self.llm_manager = LLMManager(DEFAULT_CONFIG)
except ImportError:
# 如果配置文件不存在，使用内置配置
default_config = LLMConfig(
provider="openai",
api_key="sk-your-qwen-api-key-here",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
model="qwen-turbo",
temperature=0.7,
max_tokens=2000
)
self.llm_manager = LLMManager(default_config)
def get_agent_status(self) -> Dict[str, Any]:
"""获取Agent状态"""
return {
"agent_state": self.agent_state.value,
"is_agent_mode": self.is_agent_mode,
"proactive_monitoring": self.proactive_monitoring_enabled,
"execution_count": len(self.execution_history),
"llm_status": self.llm_manager.get_status(),
"agent_core_status": self.agent_core.get_status(),
"last_activity": self.execution_history[-1]["timestamp"] if self.execution_history else None
}
def toggle_agent_mode(self, enabled: bool) -> bool:
"""切换Agent模式"""
try:
self.is_agent_mode = enabled
if enabled:
self.agent_state = AgentState.IDLE
logger.info("Agent模式已启用")
else:
self.agent_state = AgentState.DISABLED
logger.info("Agent模式已禁用")
return True
except Exception as e:
logger.error(f"切换Agent模式失败: {e}")
return False
def start_proactive_monitoring(self) -> bool:
"""启动主动监控"""
try:
if not self.proactive_monitoring_enabled:
self.proactive_monitoring_enabled = True
self.auto_monitor.start_monitoring()
logger.info("主动监控已启动")
return True
return False
except Exception as e:
logger.error(f"启动主动监控失败: {e}")
return False
def stop_proactive_monitoring(self) -> bool:
"""停止主动监控"""
try:
if self.proactive_monitoring_enabled:
self.proactive_monitoring_enabled = False
self.auto_monitor.stop_monitoring()
logger.info("主动监控已停止")
return True
return False
except Exception as e:
logger.error(f"停止主动监控失败: {e}")
return False
def run_proactive_monitoring(self) -> Dict[str, Any]:
"""运行主动监控检查"""
try:
if not self.proactive_monitoring_enabled:
return {"success": False, "message": "主动监控未启用"}
# 获取系统状态
system_health = self.get_system_health()
# 检查预警
alerts = self.check_alerts()
# 检查工单状态
workorders_status = self._check_workorders_status()
# 运行智能分析
analysis = self.intelligent_agent.analyze_system_state(
system_health=system_health,
alerts=alerts,
workorders=workorders_status
)
# 执行建议的动作
actions_taken = []
if analysis.get("recommended_actions"):
for action in analysis["recommended_actions"]:
result = self.action_executor.execute_action(action)
actions_taken.append(result)
return {
"success": True,
"analysis": analysis,
"actions_taken": actions_taken,
"timestamp": datetime.now().isoformat()
}
except Exception as e:
logger.error(f"主动监控检查失败: {e}")
return {"success": False, "error": str(e)}
def _check_workorders_status(self) -> Dict[str, Any]:
"""检查工单状态"""
try:
from src.core.database import db_manager
from src.core.models import WorkOrder
with db_manager.get_session() as session:
total_workorders = session.query(WorkOrder).count()
open_workorders = session.query(WorkOrder).filter(WorkOrder.status == 'open').count()
resolved_workorders = session.query(WorkOrder).filter(WorkOrder.status == 'resolved').count()
return {
"total": total_workorders,
"open": open_workorders,
"resolved": resolved_workorders,
"resolution_rate": resolved_workorders / total_workorders if total_workorders > 0 else 0
}
except Exception as e:
logger.error(f"检查工单状态失败: {e}")
return {"error": str(e)}
def run_intelligent_analysis(self) -> Dict[str, Any]:
"""运行智能分析"""
try:
# 获取系统数据
system_health = self.get_system_health()
alerts = self.check_alerts()
workorders = self._check_workorders_status()
# 创建分析上下文
context = {
"system_health": system_health,
"alerts": alerts,
"workorders": workorders,
"timestamp": datetime.now().isoformat()
}
# 运行智能分析
analysis = self.intelligent_agent.comprehensive_analysis(context)
# 记录分析结果
self._record_execution("intelligent_analysis", analysis)
return analysis
except Exception as e:
logger.error(f"智能分析失败: {e}")
return {"error": str(e)}
def _record_execution(self, action_type: str, result: Any):
"""记录执行历史"""
execution_record = {
"timestamp": datetime.now().isoformat(),
"action_type": action_type,
"result": result,
"agent_state": self.agent_state.value
}
self.execution_history.append(execution_record)
# 保持历史记录大小限制
if len(self.execution_history) > self.max_history_size:
self.execution_history = self.execution_history[-self.max_history_size:]
def get_action_history(self, limit: int = 50) -> List[Dict[str, Any]]:
"""获取动作执行历史"""
return self.execution_history[-limit:] if self.execution_history else []
def clear_execution_history(self) -> Dict[str, Any]:
"""清空执行历史"""
try:
self.execution_history.clear()
logger.info("执行历史已清空")
return {"success": True, "message": "执行历史已清空"}
except Exception as e:
logger.error(f"清空执行历史失败: {e}")
return {"success": False, "error": str(e)}
def get_llm_usage_stats(self) -> Dict[str, Any]:
"""获取LLM使用统计"""
try:
return self.llm_manager.get_usage_stats()
except Exception as e:
logger.error(f"获取LLM使用统计失败: {e}")
return {"error": str(e)}

243
src/agent/agent_message_handler.py Normal file
View File

@@ -0,0 +1,243 @@
# -*- coding: utf-8 -*-
"""
TSP Agent消息处理模块
处理Agent的消息处理和对话功能
"""
import logging
import asyncio
from typing import Dict, Any, List, Optional
from datetime import datetime
from .agent_assistant_core import TSPAgentAssistantCore
from .intelligent_agent import IntelligentAgent
logger = logging.getLogger(__name__)
class AgentMessageHandler:
"""Agent消息处理器"""
def __init__(self, agent_core: TSPAgentAssistantCore):
self.agent_core = agent_core
self.intelligent_agent = agent_core.intelligent_agent
self.action_executor = agent_core.action_executor
async def process_message_agent(self, message: str, user_id: str = "admin",
work_order_id: Optional[int] = None,
enable_proactive: bool = True) -> Dict[str, Any]:
"""使用Agent处理消息"""
try:
# 更新Agent状态
self.agent_core.agent_state = self.agent_core.agent_core.AgentState.PROCESSING
# 创建对话上下文
context = {
"message": message,
"user_id": user_id,
"work_order_id": work_order_id,
"timestamp": datetime.now().isoformat(),
"enable_proactive": enable_proactive
}
# 使用智能Agent处理消息
agent_response = await self.intelligent_agent.process_message(context)
# 执行建议的动作
actions_taken = []
if agent_response.get("recommended_actions"):
for action in agent_response["recommended_actions"]:
action_result = self.action_executor.execute_action(action)
actions_taken.append(action_result)
# 生成响应
response = {
"response": agent_response.get("response", "Agent已处理您的请求"),
"actions": actions_taken,
"status": "completed",
"confidence": agent_response.get("confidence", 0.8),
"context": context
}
# 记录执行历史
self.agent_core._record_execution("message_processing", response)
# 更新Agent状态
self.agent_core.agent_state = self.agent_core.agent_core.AgentState.IDLE
return response
except Exception as e:
logger.error(f"Agent消息处理失败: {e}")
self.agent_core.agent_state = self.agent_core.agent_core.AgentState.ERROR
return {
"response": f"处理消息时发生错误: {str(e)}",
"actions": [],
"status": "error",
"error": str(e)
}
async def process_conversation_agent(self, conversation_data: Dict[str, Any]) -> Dict[str, Any]:
"""使用Agent处理对话"""
try:
# 提取对话信息
user_message = conversation_data.get("message", "")
user_id = conversation_data.get("user_id", "anonymous")
session_id = conversation_data.get("session_id")
# 创建对话上下文
context = {
"message": user_message,
"user_id": user_id,
"session_id": session_id,
"conversation_history": conversation_data.get("history", []),
"timestamp": datetime.now().isoformat()
}
# 使用智能Agent处理对话
agent_response = await self.intelligent_agent.process_conversation(context)
# 执行建议的动作
actions_taken = []
if agent_response.get("recommended_actions"):
for action in agent_response["recommended_actions"]:
action_result = self.action_executor.execute_action(action)
actions_taken.append(action_result)
# 生成响应
response = {
"response": agent_response.get("response", "Agent已处理您的对话"),
"actions": actions_taken,
"status": "completed",
"confidence": agent_response.get("confidence", 0.8),
"context": context,
"session_id": session_id
}
# 记录执行历史
self.agent_core._record_execution("conversation_processing", response)
return response
except Exception as e:
logger.error(f"Agent对话处理失败: {e}")
return {
"response": f"处理对话时发生错误: {str(e)}",
"actions": [],
"status": "error",
"error": str(e)
}
async def process_workorder_agent(self, workorder_data: Dict[str, Any]) -> Dict[str, Any]:
"""使用Agent处理工单"""
try:
# 提取工单信息
workorder_id = workorder_data.get("workorder_id")
action_type = workorder_data.get("action_type", "analyze")
# 创建工单上下文
context = {
"workorder_id": workorder_id,
"action_type": action_type,
"workorder_data": workorder_data,
"timestamp": datetime.now().isoformat()
}
# 使用智能Agent处理工单
agent_response = await self.intelligent_agent.process_workorder(context)
# 执行建议的动作
actions_taken = []
if agent_response.get("recommended_actions"):
for action in agent_response["recommended_actions"]:
action_result = self.action_executor.execute_action(action)
actions_taken.append(action_result)
# 生成响应
response = {
"response": agent_response.get("response", "Agent已处理工单"),
"actions": actions_taken,
"status": "completed",
"confidence": agent_response.get("confidence", 0.8),
"context": context
}
# 记录执行历史
self.agent_core._record_execution("workorder_processing", response)
return response
except Exception as e:
logger.error(f"Agent工单处理失败: {e}")
return {
"response": f"处理工单时发生错误: {str(e)}",
"actions": [],
"status": "error",
"error": str(e)
}
async def process_alert_agent(self, alert_data: Dict[str, Any]) -> Dict[str, Any]:
"""使用Agent处理预警"""
try:
# 创建预警上下文
context = {
"alert_data": alert_data,
"timestamp": datetime.now().isoformat()
}
# 使用智能Agent处理预警
agent_response = await self.intelligent_agent.process_alert(context)
# 执行建议的动作
actions_taken = []
if agent_response.get("recommended_actions"):
for action in agent_response["recommended_actions"]:
action_result = self.action_executor.execute_action(action)
actions_taken.append(action_result)
# 生成响应
response = {
"response": agent_response.get("response", "Agent已处理预警"),
"actions": actions_taken,
"status": "completed",
"confidence": agent_response.get("confidence", 0.8),
"context": context
}
# 记录执行历史
self.agent_core._record_execution("alert_processing", response)
return response
except Exception as e:
logger.error(f"Agent预警处理失败: {e}")
return {
"response": f"处理预警时发生错误: {str(e)}",
"actions": [],
"status": "error",
"error": str(e)
}
def get_conversation_suggestions(self, context: Dict[str, Any]) -> List[str]:
"""获取对话建议"""
try:
return self.intelligent_agent.get_conversation_suggestions(context)
except Exception as e:
logger.error(f"获取对话建议失败: {e}")
return []
def get_workorder_suggestions(self, workorder_data: Dict[str, Any]) -> List[str]:
"""获取工单建议"""
try:
return self.intelligent_agent.get_workorder_suggestions(workorder_data)
except Exception as e:
logger.error(f"获取工单建议失败: {e}")
return []
def get_alert_suggestions(self, alert_data: Dict[str, Any]) -> List[str]:
"""获取预警建议"""
try:
return self.intelligent_agent.get_alert_suggestions(alert_data)
except Exception as e:
logger.error(f"获取预警建议失败: {e}")
return []

405
src/agent/agent_sample_actions.py Normal file
View File

@@ -0,0 +1,405 @@
# -*- coding: utf-8 -*-
"""
TSP Agent示例动作模块
包含Agent的示例动作和测试功能
"""
import logging
import asyncio
from typing import Dict, Any, List
from datetime import datetime, timedelta
from .agent_assistant_core import TSPAgentAssistantCore
logger = logging.getLogger(__name__)
class AgentSampleActions:
"""Agent示例动作处理器"""
def __init__(self, agent_core: TSPAgentAssistantCore):
self.agent_core = agent_core
async def trigger_sample_actions(self) -> Dict[str, Any]:
"""触发示例动作"""
try:
logger.info("开始执行示例动作")
# 执行多个示例动作
actions_results = []
# 1. 系统健康检查
health_result = await self._sample_health_check()
actions_results.append(health_result)
# 2. 预警分析
alert_result = await self._sample_alert_analysis()
actions_results.append(alert_result)
# 3. 工单处理
workorder_result = await self._sample_workorder_processing()
actions_results.append(workorder_result)
# 4. 知识库更新
knowledge_result = await self._sample_knowledge_update()
actions_results.append(knowledge_result)
# 5. 性能优化
optimization_result = await self._sample_performance_optimization()
actions_results.append(optimization_result)
# 记录执行历史
self.agent_core._record_execution("sample_actions", {
"actions_count": len(actions_results),
"results": actions_results
})
return {
"success": True,
"message": f"成功执行 {len(actions_results)} 个示例动作",
"actions_results": actions_results,
"timestamp": datetime.now().isoformat()
}
except Exception as e:
logger.error(f"执行示例动作失败: {e}")
return {
"success": False,
"error": str(e),
"timestamp": datetime.now().isoformat()
}
async def _sample_health_check(self) -> Dict[str, Any]:
"""示例：系统健康检查"""
try:
# 获取系统健康状态
health_data = self.agent_core.get_system_health()
# 模拟健康检查逻辑
health_score = health_data.get("health_score", 0)
if health_score > 80:
status = "excellent"
message = "系统运行状态良好"
elif health_score > 60:
status = "good"
message = "系统运行状态正常"
elif health_score > 40:
status = "fair"
message = "系统运行状态一般，建议关注"
else:
status = "poor"
message = "系统运行状态较差，需要优化"
return {
"action_type": "health_check",
"status": status,
"message": message,
"health_score": health_score,
"timestamp": datetime.now().isoformat()
}
except Exception as e:
logger.error(f"健康检查失败: {e}")
return {
"action_type": "health_check",
"status": "error",
"error": str(e)
}
async def _sample_alert_analysis(self) -> Dict[str, Any]:
"""示例：预警分析"""
try:
# 获取预警数据
alerts = self.agent_core.check_alerts()
# 分析预警
alert_count = len(alerts)
critical_alerts = [a for a in alerts if a.get("level") == "critical"]
warning_alerts = [a for a in alerts if a.get("level") == "warning"]
# 生成分析结果
if alert_count == 0:
status = "no_alerts"
message = "当前无活跃预警"
elif len(critical_alerts) > 0:
status = "critical"
message = f"发现 {len(critical_alerts)} 个严重预警，需要立即处理"
elif len(warning_alerts) > 0:
status = "warning"
message = f"发现 {len(warning_alerts)} 个警告预警，建议关注"
else:
status = "info"
message = f"发现 {alert_count} 个信息预警"
return {
"action_type": "alert_analysis",
"status": status,
"message": message,
"alert_count": alert_count,
"critical_count": len(critical_alerts),
"warning_count": len(warning_alerts),
"timestamp": datetime.now().isoformat()
}
except Exception as e:
logger.error(f"预警分析失败: {e}")
return {
"action_type": "alert_analysis",
"status": "error",
"error": str(e)
}
async def _sample_workorder_processing(self) -> Dict[str, Any]:
"""示例：工单处理"""
try:
# 获取工单状态
workorders_status = self.agent_core._check_workorders_status()
total = workorders_status.get("total", 0)
open_count = workorders_status.get("open", 0)
resolved_count = workorders_status.get("resolved", 0)
resolution_rate = workorders_status.get("resolution_rate", 0)
# 分析工单状态
if total == 0:
status = "no_workorders"
message = "当前无工单"
elif open_count > 10:
status = "high_backlog"
message = f"工单积压严重，有 {open_count} 个待处理工单"
elif resolution_rate > 0.8:
status = "good_resolution"
message = f"工单处理效率良好，解决率 {resolution_rate:.1%}"
else:
status = "normal"
message = f"工单处理状态正常，待处理 {open_count} 个"
return {
"action_type": "workorder_processing",
"status": status,
"message": message,
"total_workorders": total,
"open_workorders": open_count,
"resolved_workorders": resolved_count,
"resolution_rate": resolution_rate,
"timestamp": datetime.now().isoformat()
}
except Exception as e:
logger.error(f"工单处理分析失败: {e}")
return {
"action_type": "workorder_processing",
"status": "error",
"error": str(e)
}
async def _sample_knowledge_update(self) -> Dict[str, Any]:
"""示例：知识库更新"""
try:
from src.core.database import db_manager
from src.core.models import KnowledgeEntry
with db_manager.get_session() as session:
# 获取知识库统计
total_knowledge = session.query(KnowledgeEntry).count()
verified_knowledge = session.query(KnowledgeEntry).filter(
KnowledgeEntry.is_verified == True
).count()
unverified_knowledge = total_knowledge - verified_knowledge
# 分析知识库状态
if total_knowledge == 0:
status = "empty"
message = "知识库为空，建议添加知识条目"
elif unverified_knowledge > 0:
status = "needs_verification"
message = f"有 {unverified_knowledge} 个知识条目需要验证"
else:
status = "up_to_date"
message = "知识库状态良好，所有条目已验证"
return {
"action_type": "knowledge_update",
"status": status,
"message": message,
"total_knowledge": total_knowledge,
"verified_knowledge": verified_knowledge,
"unverified_knowledge": unverified_knowledge,
"timestamp": datetime.now().isoformat()
}
except Exception as e:
logger.error(f"知识库更新分析失败: {e}")
return {
"action_type": "knowledge_update",
"status": "error",
"error": str(e)
}
async def _sample_performance_optimization(self) -> Dict[str, Any]:
"""示例：性能优化"""
try:
# 获取系统性能数据
system_health = self.agent_core.get_system_health()
# 分析性能指标
cpu_usage = system_health.get("cpu_usage", 0)
memory_usage = system_health.get("memory_usage", 0)
disk_usage = system_health.get("disk_usage", 0)
# 生成优化建议
optimization_suggestions = []
if cpu_usage > 80:
optimization_suggestions.append("CPU使用率过高，建议优化计算密集型任务")
if memory_usage > 80:
optimization_suggestions.append("内存使用率过高，建议清理缓存或增加内存")
if disk_usage > 90:
optimization_suggestions.append("磁盘空间不足，建议清理日志文件或扩容")
if not optimization_suggestions:
status = "optimal"
message = "系统性能良好，无需优化"
else:
status = "needs_optimization"
message = f"发现 {len(optimization_suggestions)} 个性能优化点"
return {
"action_type": "performance_optimization",
"status": status,
"message": message,
"cpu_usage": cpu_usage,
"memory_usage": memory_usage,
"disk_usage": disk_usage,
"optimization_suggestions": optimization_suggestions,
"timestamp": datetime.now().isoformat()
}
except Exception as e:
logger.error(f"性能优化分析失败: {e}")
return {
"action_type": "performance_optimization",
"status": "error",
"error": str(e)
}
async def run_performance_test(self) -> Dict[str, Any]:
"""运行性能测试"""
try:
start_time = datetime.now()
# 执行多个测试
test_results = []
# 1. 响应时间测试
response_time = await self._test_response_time()
test_results.append(response_time)
# 2. 并发处理测试
concurrency_test = await self._test_concurrency()
test_results.append(concurrency_test)
# 3. 内存使用测试
memory_test = await self._test_memory_usage()
test_results.append(memory_test)
end_time = datetime.now()
total_time = (end_time - start_time).total_seconds()
return {
"success": True,
"message": "性能测试完成",
"total_time": total_time,
"test_results": test_results,
"timestamp": datetime.now().isoformat()
}
except Exception as e:
logger.error(f"性能测试失败: {e}")
return {
"success": False,
"error": str(e),
"timestamp": datetime.now().isoformat()
}
async def _test_response_time(self) -> Dict[str, Any]:
"""测试响应时间"""
start_time = datetime.now()
# 模拟处理任务
await asyncio.sleep(0.1)
end_time = datetime.now()
response_time = (end_time - start_time).total_seconds()
return {
"test_type": "response_time",
"response_time": response_time,
"status": "good" if response_time < 0.5 else "slow"
}
async def _test_concurrency(self) -> Dict[str, Any]:
"""测试并发处理"""
try:
# 创建多个并发任务
tasks = []
for i in range(5):
task = asyncio.create_task(self._simulate_task(i))
tasks.append(task)
# 等待所有任务完成
results = await asyncio.gather(*tasks)
return {
"test_type": "concurrency",
"concurrent_tasks": len(tasks),
"successful_tasks": len([r for r in results if r.get("success")]),
"status": "good"
}
except Exception as e:
return {
"test_type": "concurrency",
"status": "error",
"error": str(e)
}
async def _simulate_task(self, task_id: int) -> Dict[str, Any]:
"""模拟任务"""
try:
await asyncio.sleep(0.05) # 模拟处理时间
return {
"task_id": task_id,
"success": True,
"result": f"Task {task_id} completed"
}
except Exception as e:
return {
"task_id": task_id,
"success": False,
"error": str(e)
}
async def _test_memory_usage(self) -> Dict[str, Any]:
"""测试内存使用"""
try:
import psutil
# 获取当前内存使用情况
memory_info = psutil.virtual_memory()
return {
"test_type": "memory_usage",
"total_memory": memory_info.total,
"available_memory": memory_info.available,
"used_memory": memory_info.used,
"memory_percentage": memory_info.percent,
"status": "good" if memory_info.percent < 80 else "high"
}
except Exception as e:
return {
"test_type": "memory_usage",
"status": "error",
"error": str(e)
}