817 lines
31 KiB
Python
817 lines
31 KiB
Python
#!/usr/bin/env python3
|
||
# -*- 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
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
class TSPAgentAssistant(TSPAssistant):
|
||
"""TSP Agent助手 - 增强版TSP助手，具备完整Agent功能"""
|
||
|
||
def __init__(self):
|
||
# 初始化基础TSP助手
|
||
super().__init__()
|
||
|
||
# 初始化Agent核心
|
||
self.agent_core = AgentCore()
|
||
|
||
# Agent特有功能
|
||
self.is_agent_mode = True
|
||
self.proactive_tasks = []
|
||
self.agent_memory = {}
|
||
|
||
logger.info("TSP Agent助手初始化完成")
|
||
|
||
async def process_message_agent(
|
||
self,
|
||
message: str,
|
||
user_id: str = None,
|
||
work_order_id: int = None,
|
||
enable_proactive: bool = True
|
||
) -> Dict[str, Any]:
|
||
"""Agent模式处理用户消息"""
|
||
try:
|
||
# 构建请求
|
||
request = {
|
||
"message": message,
|
||
"user_id": user_id,
|
||
"work_order_id": work_order_id,
|
||
"context": {
|
||
"session_id": f"session_{user_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}",
|
||
"timestamp": datetime.now().isoformat()
|
||
}
|
||
}
|
||
|
||
# 使用Agent核心处理请求
|
||
agent_result = await self.agent_core.process_request(request)
|
||
|
||
# 如果启用主动模式，检查是否需要主动行动
|
||
if enable_proactive:
|
||
proactive_result = await self.agent_core.proactive_action()
|
||
if proactive_result:
|
||
agent_result["proactive_action"] = proactive_result
|
||
|
||
# 记录Agent执行信息
|
||
agent_result["agent_mode"] = True
|
||
agent_result["agent_status"] = self.agent_core.get_status()
|
||
|
||
return agent_result
|
||
|
||
except Exception as e:
|
||
logger.error(f"Agent模式处理消息失败: {e}")
|
||
# 回退到传统模式
|
||
return await self._fallback_to_traditional_mode(message, user_id, work_order_id)
|
||
|
||
async def _fallback_to_traditional_mode(
|
||
self,
|
||
message: str,
|
||
user_id: str = None,
|
||
work_order_id: int = None
|
||
) -> Dict[str, Any]:
|
||
"""回退到传统模式"""
|
||
logger.info("回退到传统TSP助手模式")
|
||
|
||
# 使用原有的处理方式
|
||
result = self.process_message(message, user_id, work_order_id)
|
||
|
||
# 添加Agent标识
|
||
result["agent_mode"] = False
|
||
result["fallback_reason"] = "Agent处理失败，使用传统模式"
|
||
|
||
return result
|
||
|
||
async def create_intelligent_work_order(
|
||
self,
|
||
user_message: str,
|
||
user_id: str = None,
|
||
auto_categorize: bool = True,
|
||
auto_priority: bool = True
|
||
) -> Dict[str, Any]:
|
||
"""智能创建工单 - 使用Agent能力"""
|
||
try:
|
||
# 使用Agent分析用户消息
|
||
request = {
|
||
"message": user_message,
|
||
"user_id": user_id,
|
||
"context": {"action": "create_work_order"}
|
||
}
|
||
|
||
agent_result = await self.agent_core.process_request(request)
|
||
|
||
if "error" in agent_result:
|
||
return agent_result
|
||
|
||
# 从Agent结果中提取工单信息
|
||
work_order_info = self._extract_work_order_info(agent_result, user_message)
|
||
|
||
# 创建工单
|
||
work_order = self.create_work_order(
|
||
title=work_order_info["title"],
|
||
description=work_order_info["description"],
|
||
category=work_order_info["category"],
|
||
priority=work_order_info["priority"]
|
||
)
|
||
|
||
# 添加Agent分析结果
|
||
work_order["agent_analysis"] = agent_result
|
||
work_order["intelligent_features"] = {
|
||
"auto_categorized": auto_categorize,
|
||
"auto_prioritized": auto_priority,
|
||
"confidence_score": work_order_info.get("confidence", 0.8)
|
||
}
|
||
|
||
return work_order
|
||
|
||
except Exception as e:
|
||
logger.error(f"智能创建工单失败: {e}")
|
||
return {"error": f"智能创建失败: {str(e)}"}
|
||
|
||
def _extract_work_order_info(self, agent_result: Dict[str, Any], user_message: str) -> Dict[str, Any]:
|
||
"""从Agent结果中提取工单信息"""
|
||
# 这里可以根据Agent的分析结果智能提取工单信息
|
||
# 暂时使用简单的提取逻辑
|
||
|
||
# 使用LLM提取关键信息
|
||
from src.core.llm_client import QwenClient
|
||
llm_client = QwenClient()
|
||
|
||
prompt = f"""
|
||
请从以下用户消息中提取工单信息：
|
||
|
||
用户消息: {user_message}
|
||
|
||
请提取：
|
||
1. 工单标题（简洁明了）
|
||
2. 问题描述（详细描述）
|
||
3. 问题类别（技术问题、账户问题、服务问题等）
|
||
4. 优先级（high、medium、low）
|
||
5. 置信度（0-1）
|
||
|
||
请以JSON格式返回。
|
||
"""
|
||
|
||
messages = [
|
||
{"role": "system", "content": "你是一个工单信息提取专家，擅长从用户消息中提取关键信息。"},
|
||
{"role": "user", "content": prompt}
|
||
]
|
||
|
||
result = llm_client.chat_completion(messages, temperature=0.3)
|
||
|
||
if "error" in result:
|
||
# 使用默认值
|
||
return {
|
||
"title": "用户问题",
|
||
"description": user_message,
|
||
"category": "技术问题",
|
||
"priority": "medium",
|
||
"confidence": 0.5
|
||
}
|
||
|
||
try:
|
||
response_content = result["choices"][0]["message"]["content"]
|
||
import re
|
||
json_match = re.search(r'\{.*\}', response_content, re.DOTALL)
|
||
if json_match:
|
||
extracted_info = json.loads(json_match.group())
|
||
return {
|
||
"title": extracted_info.get("title", "用户问题"),
|
||
"description": extracted_info.get("description", user_message),
|
||
"category": extracted_info.get("category", "技术问题"),
|
||
"priority": extracted_info.get("priority", "medium"),
|
||
"confidence": extracted_info.get("confidence", 0.7)
|
||
}
|
||
else:
|
||
return {
|
||
"title": "用户问题",
|
||
"description": user_message,
|
||
"category": "技术问题",
|
||
"priority": "medium",
|
||
"confidence": 0.5
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"提取工单信息失败: {e}")
|
||
return {
|
||
"title": "用户问题",
|
||
"description": user_message,
|
||
"category": "技术问题",
|
||
"priority": "medium",
|
||
"confidence": 0.5
|
||
}
|
||
|
||
async def intelligent_knowledge_search(
|
||
self,
|
||
query: str,
|
||
context: Dict[str, Any] = None,
|
||
use_reasoning: bool = True
|
||
) -> Dict[str, Any]:
|
||
"""智能知识库搜索 - 使用推理能力"""
|
||
try:
|
||
# 基础搜索
|
||
basic_results = self.search_knowledge(query)
|
||
|
||
if not use_reasoning:
|
||
return basic_results
|
||
|
||
# 使用推理引擎增强搜索
|
||
reasoning_result = await self.agent_core.reasoning_engine.reason_about_problem(
|
||
problem=f"搜索知识: {query}",
|
||
available_information={
|
||
"search_results": basic_results.get("results", []),
|
||
"context": context or {}
|
||
},
|
||
reasoning_type="inductive"
|
||
)
|
||
|
||
# 结合搜索结果和推理结果
|
||
enhanced_results = {
|
||
"basic_search": basic_results,
|
||
"reasoning_analysis": reasoning_result,
|
||
"enhanced_results": self._enhance_search_results(
|
||
basic_results.get("results", []),
|
||
reasoning_result
|
||
),
|
||
"search_strategy": "intelligent_with_reasoning"
|
||
}
|
||
|
||
return enhanced_results
|
||
|
||
except Exception as e:
|
||
logger.error(f"智能知识搜索失败: {e}")
|
||
return self.search_knowledge(query)
|
||
|
||
def _enhance_search_results(
|
||
self,
|
||
basic_results: List[Dict[str, Any]],
|
||
reasoning_result: Dict[str, Any]
|
||
) -> List[Dict[str, Any]]:
|
||
"""增强搜索结果"""
|
||
enhanced_results = []
|
||
|
||
for result in basic_results:
|
||
enhanced_result = result.copy()
|
||
|
||
# 添加推理增强信息
|
||
if "analysis" in reasoning_result:
|
||
enhanced_result["reasoning_insights"] = reasoning_result["analysis"]
|
||
|
||
# 计算增强置信度
|
||
original_confidence = result.get("confidence_score", 0.5)
|
||
reasoning_confidence = reasoning_result.get("confidence", 0.5)
|
||
enhanced_result["enhanced_confidence"] = (original_confidence + reasoning_confidence) / 2
|
||
|
||
enhanced_results.append(enhanced_result)
|
||
|
||
return enhanced_results
|
||
|
||
async def proactive_monitoring(self) -> Dict[str, Any]:
|
||
"""主动监控 - Agent主动检查系统状态"""
|
||
try:
|
||
proactive_actions = []
|
||
|
||
# 检查预警
|
||
alerts = self.get_alerts()
|
||
if alerts.get("count", 0) > 0:
|
||
proactive_actions.append({
|
||
"type": "alert_response",
|
||
"description": f"发现 {alerts['count']} 个活跃预警",
|
||
"priority": "high",
|
||
"action": "需要立即处理预警"
|
||
})
|
||
|
||
# 检查系统健康
|
||
system_status = self.get_system_status()
|
||
if system_status.get("health_score", 1.0) < 0.8:
|
||
proactive_actions.append({
|
||
"type": "system_maintenance",
|
||
"description": "系统健康状态不佳",
|
||
"priority": "medium",
|
||
"action": "建议进行系统维护"
|
||
})
|
||
|
||
# 检查知识库质量
|
||
knowledge_stats = self.knowledge_manager.get_knowledge_stats()
|
||
if knowledge_stats.get("average_confidence", 0.8) < 0.6:
|
||
proactive_actions.append({
|
||
"type": "knowledge_improvement",
|
||
"description": "知识库质量需要提升",
|
||
"priority": "low",
|
||
"action": "建议更新知识库"
|
||
})
|
||
|
||
return {
|
||
"proactive_actions": proactive_actions,
|
||
"timestamp": datetime.now().isoformat(),
|
||
"agent_status": self.agent_core.get_status()
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"主动监控失败: {e}")
|
||
return {"error": str(e)}
|
||
|
||
async def intelligent_analytics(
|
||
self,
|
||
analysis_type: str = "comprehensive",
|
||
date_range: str = "last_7_days"
|
||
) -> Dict[str, Any]:
|
||
"""智能分析 - 使用Agent推理能力"""
|
||
try:
|
||
# 基础分析
|
||
basic_analytics = self.generate_analytics(date_range)
|
||
|
||
if analysis_type == "basic":
|
||
return basic_analytics
|
||
|
||
# 使用Agent进行深度分析
|
||
analysis_request = {
|
||
"message": f"分析{date_range}的数据",
|
||
"context": {
|
||
"analysis_type": analysis_type,
|
||
"basic_data": basic_analytics
|
||
}
|
||
}
|
||
|
||
agent_analysis = await self.agent_core.process_request(analysis_request)
|
||
|
||
# 结合基础分析和Agent分析
|
||
intelligent_analytics = {
|
||
"basic_analytics": basic_analytics,
|
||
"agent_insights": agent_analysis,
|
||
"intelligent_recommendations": self._generate_recommendations(
|
||
basic_analytics,
|
||
agent_analysis
|
||
),
|
||
"analysis_confidence": self._calculate_analysis_confidence(agent_analysis)
|
||
}
|
||
|
||
return intelligent_analytics
|
||
|
||
except Exception as e:
|
||
logger.error(f"智能分析失败: {e}")
|
||
return self.generate_analytics(date_range)
|
||
|
||
def _generate_recommendations(
|
||
self,
|
||
basic_analytics: Dict[str, Any],
|
||
agent_analysis: Dict[str, Any]
|
||
) -> List[Dict[str, Any]]:
|
||
"""生成智能推荐"""
|
||
recommendations = []
|
||
|
||
# 基于基础分析生成推荐
|
||
if basic_analytics.get("summary", {}).get("avg_satisfaction", 0) < 0.7:
|
||
recommendations.append({
|
||
"type": "improvement",
|
||
"title": "提升客户满意度",
|
||
"description": "客户满意度较低，建议优化服务质量",
|
||
"priority": "high",
|
||
"action_items": [
|
||
"分析低满意度工单",
|
||
"改进响应时间",
|
||
"提升解决方案质量"
|
||
]
|
||
})
|
||
|
||
if basic_analytics.get("summary", {}).get("avg_resolution_time_hours", 0) > 24:
|
||
recommendations.append({
|
||
"type": "efficiency",
|
||
"title": "缩短解决时间",
|
||
"description": "平均解决时间过长，建议提升处理效率",
|
||
"priority": "medium",
|
||
"action_items": [
|
||
"优化工作流程",
|
||
"增加自动化处理",
|
||
"提升知识库质量"
|
||
]
|
||
})
|
||
|
||
return recommendations
|
||
|
||
def _calculate_analysis_confidence(self, agent_analysis: Dict[str, Any]) -> float:
|
||
"""计算分析置信度"""
|
||
# 基于Agent分析结果计算置信度
|
||
if "error" in agent_analysis:
|
||
return 0.3
|
||
|
||
# 这里可以实现更复杂的置信度计算逻辑
|
||
return 0.8
|
||
|
||
def get_agent_status(self) -> Dict[str, Any]:
|
||
"""获取Agent状态"""
|
||
return {
|
||
"success": True,
|
||
"status": "active" if self.is_agent_mode else "inactive",
|
||
"active_goals": len(self.agent_core.goal_manager.get_active_goals()),
|
||
"available_tools": len(self.agent_core.tool_manager.get_available_tools()),
|
||
"tools": [
|
||
{
|
||
"name": tool.name,
|
||
"usage_count": getattr(tool, 'usage_count', 0),
|
||
"success_rate": getattr(tool, 'success_rate', 0.8)
|
||
}
|
||
for tool in self.agent_core.tool_manager.get_available_tools()
|
||
],
|
||
"execution_history": []
|
||
}
|
||
|
||
def toggle_agent_mode(self, enabled: bool) -> bool:
|
||
"""切换Agent模式"""
|
||
try:
|
||
if enabled:
|
||
# 同步方式切换
|
||
self.is_agent_mode = True
|
||
logger.info("已切换到Agent模式")
|
||
return True
|
||
else:
|
||
return self.switch_to_traditional_mode()
|
||
except Exception as e:
|
||
logger.error(f"切换Agent模式失败: {e}")
|
||
return False
|
||
|
||
def start_proactive_monitoring(self) -> bool:
|
||
"""启动主动监控"""
|
||
try:
|
||
return self.start_agent_monitoring()
|
||
except Exception as e:
|
||
logger.error(f"启动主动监控失败: {e}")
|
||
return False
|
||
|
||
def stop_proactive_monitoring(self) -> bool:
|
||
"""停止主动监控"""
|
||
try:
|
||
return self.stop_agent_monitoring()
|
||
except Exception as e:
|
||
logger.error(f"停止主动监控失败: {e}")
|
||
return False
|
||
|
||
def run_proactive_monitoring(self) -> Dict[str, Any]:
|
||
"""运行主动监控"""
|
||
try:
|
||
# 模拟主动监控结果
|
||
proactive_actions = [
|
||
{"type": "alert_response", "description": "发现系统性能预警"},
|
||
{"type": "knowledge_update", "description": "建议更新知识库"},
|
||
{"type": "user_assistance", "description": "检测到用户可能需要帮助"}
|
||
]
|
||
return {
|
||
"success": True,
|
||
"proactive_actions": proactive_actions
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"运行主动监控失败: {e}")
|
||
return {"success": False, "error": str(e)}
|
||
|
||
def run_intelligent_analysis(self) -> Dict[str, Any]:
|
||
"""运行智能分析"""
|
||
try:
|
||
# 模拟智能分析结果
|
||
analysis = {
|
||
"trends": {
|
||
"dates": ["2024-01-01", "2024-01-02", "2024-01-03"],
|
||
"satisfaction": [0.8, 0.85, 0.82],
|
||
"resolution_time": [2.5, 2.3, 2.1]
|
||
},
|
||
"recommendations": [
|
||
{"type": "improvement", "title": "提升客户满意度", "description": "建议优化响应时间"},
|
||
{"type": "optimization", "title": "知识库优化", "description": "建议增加更多技术问题解答"}
|
||
]
|
||
}
|
||
return analysis
|
||
except Exception as e:
|
||
logger.error(f"运行智能分析失败: {e}")
|
||
return {"error": str(e)}
|
||
|
||
def process_file_to_knowledge(self, file_path: str, filename: str) -> Dict[str, Any]:
|
||
"""处理文件并生成知识库"""
|
||
try:
|
||
import os
|
||
import mimetypes
|
||
|
||
# 检查文件类型
|
||
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:
|
||
return {"success": False, "error": "无法读取文件内容"}
|
||
|
||
# 使用LLM处理内容
|
||
knowledge_entries = self._extract_knowledge_from_content(content, filename)
|
||
|
||
# 保存到知识库
|
||
saved_count = 0
|
||
for i, entry in enumerate(knowledge_entries):
|
||
try:
|
||
logger.info(f"保存知识条目 {i+1}: {entry.get('question', '')[:50]}...")
|
||
success = self.knowledge_manager.add_knowledge_entry(
|
||
question=entry["question"],
|
||
answer=entry["answer"],
|
||
category=entry.get("category", "其他"),
|
||
confidence_score=entry.get("confidence_score", 0.7),
|
||
is_verified=False # 新添加的知识库条目默认为未验证
|
||
)
|
||
if success:
|
||
saved_count += 1
|
||
logger.info(f"知识条目 {i+1} 保存成功")
|
||
else:
|
||
logger.error(f"知识条目 {i+1} 保存失败")
|
||
except Exception as save_error:
|
||
logger.error(f"保存知识条目 {i+1} 时出错: {save_error}")
|
||
logger.error(f"条目内容: {entry}")
|
||
|
||
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':
|
||
# 需要安装 PyPDF2 或 pdfplumber
|
||
try:
|
||
import PyPDF2
|
||
with open(file_path, 'rb') as f:
|
||
reader = PyPDF2.PdfReader(f)
|
||
text = ""
|
||
for page in reader.pages:
|
||
text += page.extract_text() + "\n"
|
||
return text
|
||
except ImportError:
|
||
return "PDF文件需要安装PyPDF2库"
|
||
elif file_ext in ['.doc', '.docx']:
|
||
# 需要安装 python-docx
|
||
try:
|
||
from docx import Document
|
||
doc = Document(file_path)
|
||
text = ""
|
||
for paragraph in doc.paragraphs:
|
||
text += paragraph.text + "\n"
|
||
return text
|
||
except ImportError:
|
||
return "Word文件需要安装python-docx库"
|
||
else:
|
||
return "不支持的文件格式"
|
||
except Exception as e:
|
||
logger.error(f"读取文件失败: {e}")
|
||
return ""
|
||
|
||
def _extract_knowledge_from_content(self, content: str, filename: str) -> List[Dict[str, Any]]:
|
||
"""从内容中提取知识"""
|
||
try:
|
||
# 构建提示词
|
||
prompt = f"""
|
||
请从以下文档内容中提取问答对，用于构建知识库：
|
||
|
||
文档名称：{filename}
|
||
文档内容：
|
||
{content[:2000]}...
|
||
|
||
请按照以下格式提取问答对：
|
||
1. 问题：具体的问题描述
|
||
2. 答案：详细的答案内容
|
||
3. 分类：问题所属类别（技术问题、APP功能、远程控制、车辆绑定、其他）
|
||
4. 置信度：0-1之间的数值
|
||
|
||
请提取3-5个最有价值的问答对，每个问答对都要完整且实用。
|
||
返回格式为JSON数组，例如：
|
||
[
|
||
{{
|
||
"question": "如何远程启动车辆？",
|
||
"answer": "远程启动车辆需要满足以下条件：1. 车辆处于P档 2. 手刹拉起 3. 车门已锁 4. 电池电量充足",
|
||
"category": "远程控制",
|
||
"confidence_score": 0.9
|
||
}}
|
||
]
|
||
"""
|
||
|
||
# 调用LLM
|
||
response = self.llm_client.chat_completion(
|
||
messages=[{"role": "user", "content": prompt}],
|
||
temperature=0.3,
|
||
max_tokens=2000
|
||
)
|
||
|
||
if response and 'choices' in response:
|
||
content_text = response['choices'][0]['message']['content']
|
||
logger.info(f"LLM响应内容: {content_text[:500]}...")
|
||
|
||
# 尝试解析JSON
|
||
try:
|
||
import json
|
||
# 提取JSON部分
|
||
start_idx = content_text.find('[')
|
||
end_idx = content_text.rfind(']') + 1
|
||
if start_idx != -1 and end_idx != 0:
|
||
json_str = content_text[start_idx:end_idx]
|
||
knowledge_entries = json.loads(json_str)
|
||
logger.info(f"成功解析JSON，提取到 {len(knowledge_entries)} 条知识")
|
||
|
||
# 验证每个条目的字段
|
||
for i, entry in enumerate(knowledge_entries):
|
||
if not isinstance(entry, dict):
|
||
logger.error(f"条目 {i} 不是字典格式: {entry}")
|
||
continue
|
||
if 'question' not in entry:
|
||
logger.error(f"条目 {i} 缺少question字段: {entry}")
|
||
continue
|
||
if 'answer' not in entry:
|
||
logger.error(f"条目 {i} 缺少answer字段: {entry}")
|
||
continue
|
||
logger.info(f"条目 {i} 验证通过: {entry.get('question', '')[:50]}...")
|
||
|
||
return knowledge_entries
|
||
except Exception as json_error:
|
||
logger.warning(f"JSON解析失败: {json_error}")
|
||
logger.warning(f"原始内容: {content_text}")
|
||
|
||
# 如果JSON解析失败，尝试手动解析
|
||
manual_entries = self._parse_knowledge_manually(content_text)
|
||
logger.info(f"手动解析提取到 {len(manual_entries)} 条知识")
|
||
return manual_entries
|
||
else:
|
||
logger.error("LLM响应格式错误")
|
||
logger.error(f"响应内容: {response}")
|
||
return []
|
||
|
||
except Exception as e:
|
||
logger.error(f"提取知识失败: {e}")
|
||
return []
|
||
|
||
def _parse_knowledge_manually(self, content: str) -> List[Dict[str, Any]]:
|
||
"""手动解析知识内容"""
|
||
try:
|
||
entries = []
|
||
lines = content.split('\n')
|
||
|
||
current_entry = {}
|
||
for line in lines:
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
|
||
# 检查问题
|
||
if '问题' in line and ('：' in line or ':' in line):
|
||
if current_entry and 'question' in current_entry:
|
||
entries.append(current_entry)
|
||
current_entry = {}
|
||
# 提取问题内容
|
||
if '：' in line:
|
||
question = line.split('：', 1)[1].strip()
|
||
else:
|
||
question = line.split(':', 1)[1].strip()
|
||
current_entry["question"] = question
|
||
|
||
# 检查答案
|
||
elif '答案' in line and ('：' in line or ':' in line):
|
||
if '：' in line:
|
||
answer = line.split('：', 1)[1].strip()
|
||
else:
|
||
answer = line.split(':', 1)[1].strip()
|
||
current_entry["answer"] = answer
|
||
|
||
# 检查分类
|
||
elif '分类' in line and ('：' in line or ':' in line):
|
||
if '：' in line:
|
||
category = line.split('：', 1)[1].strip()
|
||
else:
|
||
category = line.split(':', 1)[1].strip()
|
||
current_entry["category"] = category
|
||
|
||
# 检查置信度
|
||
elif '置信度' in line and ('：' in line or ':' in line):
|
||
try:
|
||
if '：' in line:
|
||
confidence_str = line.split('：', 1)[1].strip()
|
||
else:
|
||
confidence_str = line.split(':', 1)[1].strip()
|
||
current_entry["confidence_score"] = float(confidence_str)
|
||
except:
|
||
current_entry["confidence_score"] = 0.7
|
||
|
||
# 添加最后一个条目
|
||
if current_entry and 'question' in current_entry and 'answer' in current_entry:
|
||
entries.append(current_entry)
|
||
|
||
# 确保每个条目都有必要的字段
|
||
for entry in entries:
|
||
if 'category' not in entry:
|
||
entry['category'] = '其他'
|
||
if 'confidence_score' not in entry:
|
||
entry['confidence_score'] = 0.7
|
||
|
||
logger.info(f"手动解析完成，提取到 {len(entries)} 条知识")
|
||
return entries
|
||
|
||
except Exception as e:
|
||
logger.error(f"手动解析知识失败: {e}")
|
||
return []
|
||
|
||
async def switch_to_agent_mode(self) -> bool:
|
||
"""切换到Agent模式"""
|
||
try:
|
||
self.is_agent_mode = True
|
||
logger.info("已切换到Agent模式")
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"切换到Agent模式失败: {e}")
|
||
return False
|
||
|
||
def switch_to_traditional_mode(self) -> bool:
|
||
"""切换到传统模式"""
|
||
try:
|
||
self.is_agent_mode = False
|
||
logger.info("已切换到传统模式")
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"切换到传统模式失败: {e}")
|
||
return False
|
||
|
||
async def start_agent_monitoring(self) -> bool:
|
||
"""启动Agent监控"""
|
||
try:
|
||
# 启动基础监控
|
||
self.start_monitoring()
|
||
|
||
# 启动Agent主动监控
|
||
asyncio.create_task(self._agent_monitoring_loop())
|
||
|
||
logger.info("Agent监控已启动")
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"启动Agent监控失败: {e}")
|
||
return False
|
||
|
||
async def _agent_monitoring_loop(self):
|
||
"""Agent监控循环"""
|
||
while True:
|
||
try:
|
||
# 每5分钟执行一次主动监控
|
||
await asyncio.sleep(300)
|
||
|
||
proactive_result = await self.proactive_monitoring()
|
||
|
||
if proactive_result.get("proactive_actions"):
|
||
logger.info(f"发现 {len(proactive_result['proactive_actions'])} 个主动行动机会")
|
||
|
||
# 这里可以实现自动处理逻辑
|
||
# 例如：自动发送通知、自动创建工单等
|
||
|
||
except Exception as e:
|
||
logger.error(f"Agent监控循环错误: {e}")
|
||
await asyncio.sleep(60) # 出错后等待1分钟再继续
|
||
|
||
# 使用示例
|
||
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)
|
||
|
||
# 测试智能工单创建
|
||
work_order = await agent_assistant.create_intelligent_work_order(
|
||
user_message="系统经常出现错误，影响正常使用",
|
||
user_id="user456"
|
||
)
|
||
print("智能工单创建:", work_order)
|
||
|
||
# 测试主动监控
|
||
monitoring_result = await agent_assistant.proactive_monitoring()
|
||
print("主动监控结果:", monitoring_result)
|
||
|
||
# 获取Agent状态
|
||
agent_status = agent_assistant.get_agent_status()
|
||
print("Agent状态:", agent_status)
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|