# -*- coding: utf-8 -*- """ TSP Agent助手 - 简化版本 提供基本的Agent功能和工具管理 """ import logging import asyncio import json from typing import Dict, Any, List, Optional from datetime import datetime from src.config.unified_config import get_config from src.agent.llm_client import LLMManager from src.web.service_manager import service_manager logger = logging.getLogger(__name__) class TSPAgentAssistant: """TSP Agent助手""" def __init__(self): # 初始化基础功能 config = get_config() self.llm_manager = LLMManager(config.llm) self.is_agent_mode = True self.execution_history = [] # 工具注册表 self.tools = {} self.tool_performance = {} # AI监控状态 self.ai_monitoring_active = False self.monitoring_thread = None logger.info("TSP Agent助手初始化完成") def register_tool(self, name: str, func, metadata: Dict[str, Any] = None): """注册工具""" try: self.tools[name] = { "function": func, "metadata": metadata or {}, "usage_count": 0, "success_count": 0, "last_used": None } logger.info(f"工具 {name} 注册成功") return True except Exception as e: logger.error(f"注册工具 {name} 失败: {e}") return False def unregister_tool(self, name: str) -> bool: """注销工具""" try: if name in self.tools: del self.tools[name] logger.info(f"工具 {name} 注销成功") return True return False except Exception as e: logger.error(f"注销工具 {name} 失败: {e}") return False def get_available_tools(self) -> List[Dict[str, Any]]: """获取可用工具列表""" try: tools_list = [] for name, tool_info in self.tools.items(): tools_list.append({ "name": name, "metadata": tool_info["metadata"], "usage_count": tool_info["usage_count"], "success_count": tool_info["success_count"], "last_used": tool_info["last_used"] }) return tools_list except Exception as e: logger.error(f"获取工具列表失败: {e}") return [] async def execute_tool(self, tool_name: str, parameters: Dict[str, Any] = None) -> Dict[str, Any]: """执行工具""" try: if tool_name not in self.tools: return {"error": f"工具 {tool_name} 不存在"} tool_info = self.tools[tool_name] func = tool_info["function"] # 记录使用 tool_info["usage_count"] += 1 tool_info["last_used"] = datetime.now().isoformat() # 执行工具 start_time = datetime.now() try: if asyncio.iscoroutinefunction(func): result = await func(**(parameters or {})) else: result = func(**(parameters or {})) # 记录成功 tool_info["success_count"] += 1 execution_time = (datetime.now() - start_time).total_seconds() # 记录执行历史 self._record_execution(tool_name, parameters, result, True, execution_time) return { "success": True, "result": result, "execution_time": execution_time, "tool_name": tool_name } except Exception as e: execution_time = (datetime.now() - start_time).total_seconds() self._record_execution(tool_name, parameters, str(e), False, execution_time) return { "success": False, "error": str(e), "execution_time": execution_time, "tool_name": tool_name } except Exception as e: logger.error(f"执行工具 {tool_name} 失败: {e}") return {"error": str(e)} def _record_execution(self, tool_name: str, parameters: Dict[str, Any], result: Any, success: bool, execution_time: float): """记录执行历史""" try: execution_record = { "timestamp": datetime.now().isoformat(), "tool_name": tool_name, "parameters": parameters, "result": result, "success": success, "execution_time": execution_time } self.execution_history.append(execution_record) # 保持历史记录在合理范围内 if len(self.execution_history) > 1000: self.execution_history = self.execution_history[-1000:] except Exception as e: logger.error(f"记录执行历史失败: {e}") def get_tool_performance_report(self) -> Dict[str, Any]: """获取工具性能报告""" try: total_tools = len(self.tools) total_executions = sum(tool["usage_count"] for tool in self.tools.values()) total_successes = sum(tool["success_count"] for tool in self.tools.values()) success_rate = (total_successes / total_executions * 100) if total_executions > 0 else 0 return { "total_tools": total_tools, "total_executions": total_executions, "total_successes": total_successes, "success_rate": round(success_rate, 2), "tools": self.get_available_tools() } except Exception as e: logger.error(f"获取工具性能报告失败: {e}") return {} def get_action_history(self, limit: int = 50) -> List[Dict[str, Any]]: """获取动作执行历史""" try: return self.execution_history[-limit:] if limit > 0 else self.execution_history except Exception as e: logger.error(f"获取动作历史失败: {e}") return [] def clear_execution_history(self) -> Dict[str, Any]: """清空执行历史""" try: count = len(self.execution_history) self.execution_history.clear() return { "success": True, "message": f"已清空 {count} 条执行历史" } except Exception as e: logger.error(f"清空执行历史失败: {e}") return {"success": False, "error": str(e)} def get_agent_status(self) -> Dict[str, Any]: """获取Agent状态""" try: return { "success": True, "is_active": self.is_agent_mode, "ai_monitoring_active": self.ai_monitoring_active, "total_tools": len(self.tools), "total_executions": len(self.execution_history), "tools": self.get_available_tools(), "performance": self.get_tool_performance_report() } except Exception as e: logger.error(f"获取Agent状态失败: {e}") return { "success": False, "error": str(e), "is_active": False, "ai_monitoring_active": False } def toggle_agent_mode(self, enabled: bool) -> bool: """切换Agent模式""" try: self.is_agent_mode = enabled logger.info(f"Agent模式: {'启用' if enabled else '禁用'}") return True except Exception as e: logger.error(f"切换Agent模式失败: {e}") return False def start_proactive_monitoring(self) -> bool: """启动主动监控""" try: if not self.ai_monitoring_active: self.ai_monitoring_active = True logger.info("主动监控已启动") return True return True except Exception as e: logger.error(f"启动主动监控失败: {e}") return False def stop_proactive_monitoring(self) -> bool: """停止主动监控""" try: self.ai_monitoring_active = False logger.info("主动监控已停止") return True except Exception as e: logger.error(f"停止主动监控失败: {e}") return False def run_proactive_monitoring(self) -> Dict[str, Any]: """运行主动监控检查""" try: return { "success": True, "message": "主动监控检查完成", "timestamp": datetime.now().isoformat() } except Exception as e: logger.error(f"运行主动监控失败: {e}") return {"success": False, "error": str(e)} def run_intelligent_analysis(self) -> Dict[str, Any]: """运行智能分析""" try: # 分析工具使用情况 tool_performance = self.get_tool_performance_report() # 分析执行历史 recent_executions = self.get_action_history(20) # 生成分析报告 analysis = { "tool_performance": tool_performance, "recent_activity": len(recent_executions), "success_rate": tool_performance.get("success_rate", 0), "recommendations": self._generate_recommendations(tool_performance) } return analysis except Exception as e: logger.error(f"运行智能分析失败: {e}") return {"error": str(e)} def _generate_recommendations(self, tool_performance: Dict[str, Any]) -> List[str]: """生成建议""" recommendations = [] success_rate = tool_performance.get("success_rate", 100) if success_rate < 90: recommendations.append("工具成功率较低，建议检查工具实现") total_executions = tool_performance.get("total_executions", 0) if total_executions < 10: recommendations.append("工具使用频率较低，建议增加工具调用") return recommendations def get_llm_usage_stats(self) -> Dict[str, Any]: """获取LLM使用统计""" try: return { "total_requests": 0, "total_tokens": 0, "cost": 0.0, "last_updated": datetime.now().isoformat() } except Exception as e: logger.error(f"获取LLM使用统计失败: {e}") return {} def process_message_agent_sync(self, message: str, user_id: str = "admin", work_order_id: Optional[int] = None, enable_proactive: bool = True) -> Dict[str, Any]: """处理消息（同步桥接）""" try: loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) return loop.run_until_complete(self.process_message_agent(message, user_id, work_order_id, enable_proactive)) except Exception as e: logger.error(f"同步处理消息失败: {e}") return {"error": str(e)} 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]: """处理消息 (实战化)""" try: logger.info(f"Agent收到消息: {message}") # 1. 识别意图和推荐工具 prompt = f"用户消息: {message}\n请分析用户意图，并从工具列表中选择最合适的工具。工具列表: {json.dumps(self.get_available_tools())}\n请直接返回你的分析和建议响应。" response_text = await self.llm_manager.generate(prompt) # 2. 模拟动作生成 actions = [] if "工单" in message or "查询" in message: actions.append({"type": "tool_call", "tool": "search_work_order", "status": "suggested"}) return { "success": True, "response": response_text, "actions": actions, "user_id": user_id, "work_order_id": work_order_id, "status": "completed", "timestamp": datetime.now().isoformat() } except Exception as e: logger.error(f"处理消息失败: {e}") return {"error": str(e)} def execute_tool_sync(self, tool_name: str, parameters: Dict[str, Any] = None) -> Dict[str, Any]: """执行工具（同步桥接）""" try: loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) return loop.run_until_complete(self.execute_tool(tool_name, parameters)) except Exception as e: return {"error": str(e)} def trigger_sample_actions_sync(self) -> Dict[str, Any]: """触发示例动作（同步桥接）""" try: loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) return loop.run_until_complete(self.trigger_sample_actions()) except Exception as e: return {"success": False, "error": str(e)} async def trigger_sample_actions(self) -> Dict[str, Any]: """触发示例动作""" try: # 执行一个示例工具 result = await self.execute_tool("sample_tool", {"action": "test"}) return { "success": True, "message": "示例动作已执行", "result": result } except Exception as e: logger.error(f"触发示例动作失败: {e}") return {"success": False, "error": str(e)} async def process_file_to_knowledge(self, file_path: str, filename: str) -> Dict[str, Any]: """处理文件并生成知识库""" try: import os import mimetypes logger.info(f"开始处理知识库上传文件: {filename}") # 检查文件类型 mime_type, _ = mimetypes.guess_type(file_path) file_ext = os.path.splitext(filename)[1].lower() # 读取文件内容 content = self._read_file_content(file_path, file_ext) if not content: logger.error(f"文件读取失败或内容为空: {filename}") return {"success": False, "error": "无法读取文件内容"} logger.info(f"文件读取成功: {filename}, 字符数={len(content)}") # 使用LLM进行知识提取 (异步调用) logger.info(f"正在对文件内容进行 AI 知识提取...") knowledge_entries = await self._extract_knowledge_from_content(content, filename) logger.info(f"知识提取完成: 共提取出 {len(knowledge_entries)} 个潜在条目") # 保存到知识库 saved_count = 0 # 获取知识库管理器 try: knowledge_manager = service_manager.get_assistant().knowledge_manager except Exception as e: logger.error(f"无法获取知识库管理器: {e}") knowledge_manager = None for i, entry in enumerate(knowledge_entries): try: logger.info(f"正在保存知识条目 [{i+1}/{len(knowledge_entries)}]: {entry.get('question', '')[:30]}...") if knowledge_manager: # 实际保存到数据库 knowledge_manager.add_knowledge_entry( question=entry.get('question'), answer=entry.get('answer'), category=entry.get('category', '文档导入'), confidence_score=entry.get('confidence_score', 0.8) ) saved_count += 1 else: # 如果无法获取管理器，仅记录日志（降级处理） logger.warning("知识库管理器不可用，跳过保存") except Exception as save_error: logger.error(f"保存知识条目 {i+1} 时出错: {save_error}") logger.info(f"文件处理任务结束: {filename}, 成功入库 {saved_count} 条") return { "success": True, "knowledge_count": saved_count, "total_extracted": len(knowledge_entries), "filename": filename } except Exception as e: logger.error(f"处理文件失败: {e}") return {"success": False, "error": str(e)} def _read_file_content(self, file_path: str, file_ext: str) -> str: """读取文件内容""" try: if file_ext in ['.txt', '.md']: with open(file_path, 'r', encoding='utf-8') as f: return f.read() elif file_ext == '.pdf': return "PDF文件需要安装PyPDF2库" elif file_ext in ['.doc', '.docx']: return "Word文件需要安装python-docx库" else: return "不支持的文件格式" except Exception as e: logger.error(f"读取文件失败: {e}") return "" async def _extract_knowledge_from_content(self, content: str, filename: str) -> List[Dict[str, Any]]: """从内容中提取知识 - 使用LLM""" try: # 限制内容长度，避免超出token限制 # 假设每个汉字2个token，保留前8000个字符作为上下文 truncated_content = content[:8000] if len(content) > 8000: truncated_content += "\n...(后续内容已省略)" prompt = f""" 你是一个专业的知识库构建助手。请分析以下文档内容，提取出关键的"问题"和"答案"对，用于构建知识库。 文档文件名：{filename} 文档内容： {truncated_content} 要求： 1. 提取文档中的核心知识点，转化为"问题(question)"和"答案(answer)"的形式。 2. "问题"应该清晰明确，方便用户搜索。 3. "答案"应该准确、完整，直接回答问题。 4. "分类(category)"请根据内容自动归类（如：故障排查、操作指南、系统配置、业务流程等）。 5. 输出格式必须是合法的 JSON 数组，不要包含Markdown标记。 JSON格式示例： [ {{"question": "如何重置密码？", "answer": "请访问设置页面，点击重置密码按钮...", "category": "操作指南"}}, {{"question": "系统支持哪些浏览器？", "answer": "支持Chrome, Edge, Firefox...", "category": "系统配置"}} ] """ # 调用LLM生成 logger.info("正在调用LLM进行知识提取...") response_text = await self.llm_manager.generate(prompt, temperature=0.3) # 清理响应中的Markdown标记（如果存在） cleaned_text = response_text.strip() if cleaned_text.startswith("```json"): cleaned_text = cleaned_text[7:] if cleaned_text.startswith("```"): cleaned_text = cleaned_text[3:] if cleaned_text.endswith("```"): cleaned_text = cleaned_text[:-3] cleaned_text = cleaned_text.strip() # 解析JSON try: entries = json.loads(cleaned_text) except json.JSONDecodeError: # 尝试修复常见的JSON错误 logger.warning(f"JSON解析失败，尝试简单修复: {cleaned_text[:100]}...") # 这里可以添加更复杂的修复逻辑，或者直接记录错误 return [] # 验证和标准化 valid_entries = [] for entry in entries: if isinstance(entry, dict) and "question" in entry and "answer" in entry: valid_entries.append({ "question": entry["question"], "answer": entry["answer"], "category": entry.get("category", "文档导入"), "confidence_score": 0.9 # LLM生成的置信度较高 }) return valid_entries except Exception as e: logger.error(f"提取知识失败: {e}") return [] # 使用示例 async def main(): """主函数示例""" # 创建Agent助手 agent_assistant = TSPAgentAssistant() # 测试Agent功能 print("=== TSP Agent助手测试 ===") # 测试Agent模式处理消息 response = await agent_assistant.process_message_agent( message="我的账户无法登录，请帮助我解决这个问题", user_id="user123" ) print("Agent模式响应:", response) # 获取Agent状态 agent_status = agent_assistant.get_agent_status() print("Agent状态:", agent_status) if __name__ == "__main__": asyncio.run(main())