import torch
import torch.nn as nn
from transformers import BertModel, BertTokenizer
class IntentClassifier(nn.Module):
    """基于BERT的意图分类器"""
    
    def __init__(self, bert_model_name, num_intents, dropout_rate=0.1):
        super(IntentClassifier, self).__init__()
        self.bert = BertModel.from_pretrained(bert_model_name)
        self.dropout = nn.Dropout(dropout_rate)
        self.classifier = nn.Linear(self.bert.config.hidden_size, num_intents)
        
        # 意图层级定义
        self.intent_hierarchy = {
            "consultation": {  # 咨询类
                "product_info": ["产品功能", "价格查询", "规格参数"],
                "service_policy": ["退换货", "保修政策", "配送方式"],
                "account_issue": ["账号登录", "密码重置", "信息修改"]
            },
            "complaint": {  # 投诉类
                "product_quality": ["质量问题", "功能异常", "外观缺陷"],
                "service_attitude": ["服务态度", "响应时间", "专业程度"],
                "logistics_issue": ["配送延迟", "包装破损", "地址错误"]
            },
            "transaction": {  # 交易类
                "order_inquiry": ["订单状态", "物流跟踪", "配送信息"],
                "payment_issue": ["支付失败", "退款查询", "发票申请"],
                "after_sales": ["退货申请", "换货流程", "维修预约"]
            }
        }
    
    def forward(self, input_ids, attention_mask):
        """前向传播"""
        outputs = self.bert(input_ids=input_ids, attention_mask=attention_mask)
        pooled_output = outputs.pooler_output
        pooled_output = self.dropout(pooled_output)
        logits = self.classifier(pooled_output)
        return logits
    
    def predict_intent_with_confidence(self, text, tokenizer, device):
        """预测意图并返回置信度"""
        self.eval()
        with torch.no_grad():
            encoding = tokenizer(
                text, 
                truncation=True, 
                padding=True, 
                max_length=512, 
                return_tensors='pt'
            )
            
            input_ids = encoding['input_ids'].to(device)
            attention_mask = encoding['attention_mask'].to(device)
            
            logits = self.forward(input_ids, attention_mask)
            probabilities = torch.softmax(logits, dim=-1)
            confidence, predicted_class = torch.max(probabilities, dim=-1)
            
            return predicted_class.item(), confidence.item()

import json
from typing import Dict, List, Any
from collections import defaultdict
class DialogueStateTracker:
    """对话状态跟踪器"""
    
    def __init__(self):
        self.dialogue_state = {
            "user_info": {},           # 用户信息
            "current_intent": None,    # 当前意图
            "slots": {},              # 槽位信息
            "context_history": [],    # 上下文历史
            "turn_count": 0,          # 对话轮次
            "last_action": None,      # 上次动作
            "confidence_scores": []   # 置信度历史
        }
        
        # 定义槽位依赖关系
        self.slot_dependencies = {
            "order_inquiry": ["order_number", "phone_number"],
            "product_consultation": ["product_category", "specific_model"],
            "complaint_handling": ["complaint_type", "order_number", "description"]
        }
    
    def update_state(self, user_input: str, intent: str, entities: Dict, 
                    confidence: float, system_action: str):
        """更新对话状态"""
        
        # 更新基本信息
        self.dialogue_state["current_intent"] = intent
        self.dialogue_state["turn_count"] += 1
        self.dialogue_state["last_action"] = system_action
        self.dialogue_state["confidence_scores"].append(confidence)
        
        # 更新槽位信息
        for entity_type, entity_value in entities.items():
            self.dialogue_state["slots"][entity_type] = entity_value
        
        # 维护上下文历史
        turn_info = {
            "turn": self.dialogue_state["turn_count"],
            "user_input": user_input,
            "intent": intent,
            "entities": entities,
            "confidence": confidence,
            "system_action": system_action,
            "timestamp": self._get_timestamp()
        }
        self.dialogue_state["context_history"].append(turn_info)
        
        # 清理过期上下文(保留最近10轮)
        if len(self.dialogue_state["context_history"]) > 10:
            self.dialogue_state["context_history"] = \
                self.dialogue_state["context_history"][-10:]
    
    def check_slot_completeness(self, intent: str) -> Dict[str, bool]:
        """检查当前意图所需槽位是否完整"""
        required_slots = self.slot_dependencies.get(intent, [])
        slot_status = {}
        
        for slot in required_slots:
            slot_status[slot] = slot in self.dialogue_state["slots"]
        
        return slot_status
    
    def get_missing_slots(self, intent: str) -> List[str]:
        """获取缺失的槽位"""
        slot_status = self.check_slot_completeness(intent)
        return [slot for slot, filled in slot_status.items() if not filled]
    
    def resolve_coreference(self, current_input: str) -> str:
        """指代消解处理"""
        # 简化的指代消解逻辑
        pronouns_map = {
            "它": self._get_last_product_mention(),
            "这个": self._get_last_entity_mention(),
            "那个": self._get_previous_entity_mention(),
            "我的订单": self._get_user_order()
        }
        
        resolved_input = current_input
        for pronoun, reference in pronouns_map.items():
            if pronoun in current_input and reference:
                resolved_input = resolved_input.replace(pronoun, reference)
        
        return resolved_input
    
    def _get_last_product_mention(self):
        """获取最近提到的产品"""
        for turn in reversed(self.dialogue_state["context_history"]):
            if "product_name" in turn["entities"]:
                return turn["entities"]["product_name"]
        return None
    
    def _get_timestamp(self):
        """获取当前时间戳"""
        import time
        return int(time.time())
    
    def export_state(self) -> str:
        """导出对话状态用于调试"""
        return json.dumps(self.dialogue_state, ensure_ascii=False, indent=2)

class ContextAwareResponseGenerator:
    """上下文感知的回复生成器"""
    
    def __init__(self, dst: DialogueStateTracker):
        self.dst = dst
        self.response_templates = self._load_response_templates()
    
    def generate_response(self, intent: str, entities: Dict, 
                         knowledge_result: Dict = None) -> Dict:
        """生成上下文感知的回复"""
        
        # 检查槽位完整性
        missing_slots = self.dst.get_missing_slots(intent)
        
        if missing_slots:
            # 生成槽位询问回复
            return self._generate_slot_inquiry(intent, missing_slots)
        else:
            # 生成业务回复
            return self._generate_business_response(intent, entities, knowledge_result)
    
    def _generate_slot_inquiry(self, intent: str, missing_slots: List[str]) -> Dict:
        """生成槽位询问回复"""
        slot_questions = {
            "order_number": "请提供您的订单号，我来帮您查询订单状态。",
            "phone_number": "请提供您注册时使用的手机号码进行身份验证。",
            "product_category": "请告诉我您想了解哪类产品？",
            "complaint_type": "请描述一下您遇到的具体问题。"
        }
        
        # 优先询问最重要的槽位
        primary_slot = missing_slots[0]
        question = slot_questions.get(primary_slot, f"请提供{primary_slot}信息。")
        
        return {
            "response_type": "slot_inquiry",
            "text": question,
            "missing_slots": missing_slots,
            "next_action": "collect_slot_info"
        }
    
    def _generate_business_response(self, intent: str, entities: Dict, 
                                  knowledge_result: Dict) -> Dict:
        """生成业务回复"""
        
        if knowledge_result and knowledge_result.get("success"):
            # 基于知识库结果生成回复
            template = self.response_templates.get(intent, {}).get("success")
            response_text = template.format(**entities, **knowledge_result["data"])
            
            return {
                "response_type": "business_success",
                "text": response_text,
                "knowledge_source": knowledge_result.get("source"),
                "next_action": "satisfaction_survey"
            }
        else:
            # 处理失败情况
            fallback_response = self.response_templates.get(intent, {}).get("fallback")
            
            return {
                "response_type": "business_fallback", 
                "text": fallback_response,
                "next_action": "transfer_to_human"
            }
    
    def _load_response_templates(self) -> Dict:
        """加载回复模板"""
        return {
            "order_inquiry": {
                "success": "您的订单{order_number}当前状态为：{status}，预计{delivery_date}送达。",
                "fallback": "抱歉，暂时无法查询到您的订单信息，正在为您转接人工客服。"
            },
            "product_consultation": {
                "success": "{product_name}的主要特点：{features}，当前价格：{price}元。",
                "fallback": "关于这个产品的详细信息，我为您转接专业顾问。"
            }
        }

import numpy as np
from sentence_transformers import SentenceTransformer
import faiss
from typing import List, Dict, Tuple
import json
class KnowledgeRetrievalSystem:
    """知识检索系统"""
    
    def __init__(self, model_name: str = "distilbert-base-chinese"):
        self.encoder = SentenceTransformer(model_name)
        self.knowledge_base = []
        self.knowledge_vectors = None
        self.faiss_index = None
        
        # 知识库结构
        self.kb_structure = {
            "product_info": {},      # 产品信息库
            "policy_rules": {},      # 政策规则库
            "faq_pairs": {},        # 常见问题库
            "procedure_steps": {}    # 操作流程库
        }
    
    def build_knowledge_base(self, knowledge_data: List[Dict]):
        """构建知识库"""
        self.knowledge_base = knowledge_data
        
        # 提取文本进行向量化
        texts = [item["content"] for item in knowledge_data]
        self.knowledge_vectors = self.encoder.encode(texts)
        
        # 构建FAISS索引
        dimension = self.knowledge_vectors.shape[1]
        self.faiss_index = faiss.IndexFlatIP(dimension)  # 内积检索
        
        # 归一化向量
        faiss.normalize_L2(self.knowledge_vectors)
        self.faiss_index.add(self.knowledge_vectors)
        
        print(f"知识库构建完成，包含{len(knowledge_data)}条知识")
    
    def retrieve_knowledge(self, query: str, top_k: int = 5, 
                          threshold: float = 0.7) -> List[Dict]:
        """检索相关知识"""
        
        # 查询向量化
        query_vector = self.encoder.encode([query])
        faiss.normalize_L2(query_vector)
        
        # FAISS检索
        similarities, indices = self.faiss_index.search(query_vector, top_k)
        
        results = []
        for i, (similarity, idx) in enumerate(zip(similarities[0], indices[0])):
            if similarity >= threshold:
                knowledge_item = self.knowledge_base[idx].copy()
                knowledge_item["similarity"] = float(similarity)
                knowledge_item["rank"] = i + 1
                results.append(knowledge_item)
        
        # 二次排序：结合业务规则
        results = self._rerank_results(query, results)
        
        return results
    
    def _rerank_results(self, query: str, initial_results: List[Dict]) -> List[Dict]:
        """基于业务规则的二次排序"""
        
        # 定义业务权重
        category_weights = {
            "urgent": 1.3,      # 紧急问题
            "common": 1.1,      # 常见问题
            "product": 1.0,     # 产品相关
            "policy": 0.9       # 政策相关
        }
        
        for result in initial_results:
            base_score = result["similarity"]
            category = result.get("category", "common")
            weight = category_weights.get(category, 1.0)
            
            # 考虑知识的时效性
            freshness_factor = self._calculate_freshness_factor(result)
            
            # 综合评分
            result["final_score"] = base_score * weight * freshness_factor
        
        # 按综合评分排序
        return sorted(initial_results, key=lambda x: x["final_score"], reverse=True)
    
    def _calculate_freshness_factor(self, knowledge_item: Dict) -> float:
        """计算知识时效性因子"""
        import time
        
        current_time = time.time()
        knowledge_time = knowledge_item.get("update_time", current_time)
        
        # 30天内的知识权重为1.0，超过30天开始衰减
        days_old = (current_time - knowledge_time) / (24 * 3600)
        if days_old <= 30:
            return 1.0
        else:
            return max(0.8, 1.0 - (days_old - 30) * 0.01)
    
    def update_knowledge_item(self, knowledge_id: str, new_content: Dict):
        """动态更新知识条目"""
        
        # 找到对应的知识条目
        for i, item in enumerate(self.knowledge_base):
            if item.get("id") == knowledge_id:
                # 更新内容
                self.knowledge_base[i].update(new_content)
                
                # 重新计算向量
                new_vector = self.encoder.encode([new_content["content"]])
                faiss.normalize_L2(new_vector)
                
                # 更新FAISS索引中的向量
                self.knowledge_vectors[i] = new_vector[0]
                self.faiss_index = self._rebuild_faiss_index()
                
                return True
        
        return False
    
    def _rebuild_faiss_index(self):
        """重建FAISS索引"""
        dimension = self.knowledge_vectors.shape[1]
        new_index = faiss.IndexFlatIP(dimension)
        new_index.add(self.knowledge_vectors)
        return new_index

import numpy as np
from typing import Dict, List, Tuple
from enum import Enum
class HandoffReason(Enum):
    """转人工原因枚举"""
    LOW_CONFIDENCE = "置信度过低"
    COMPLEX_QUERY = "查询过于复杂"
    EMOTIONAL_ESCALATION = "情绪升级"
    REPEATED_FAILURE = "重复失败"
    EXPLICIT_REQUEST = "明确要求"
    BUSINESS_CRITICAL = "业务关键"
class HumanHandoffDecisionModel:
    """人机切换决策模型"""
    
    def __init__(self):
        # 决策权重配置
        self.decision_weights = {
            "confidence_score": 0.25,      # 置信度权重
            "emotion_score": 0.20,         # 情绪权重
            "complexity_score": 0.20,      # 复杂度权重
            "failure_count": 0.15,         # 失败次数权重
            "user_satisfaction": 0.10,     # 用户满意度权重
            "business_priority": 0.10      # 业务优先级权重
        }
        
        # 阈值设置
        self.handoff_threshold = 0.6
        
        # 情绪词典
        self.emotion_keywords = {
            "negative": ["生气", "愤怒", "不满", "投诉", "差评", "垃圾"],
            "urgent": ["紧急", "急", "马上", "立即", "尽快"],
            "confused": ["不懂", "不明白", "搞不清", "糊涂"]
        }
    
    def should_handoff_to_human(self, dialogue_context: Dict) -> Tuple[bool, str, float]:
        """判断是否需要转人工"""
        
        # 计算各项评分
        confidence_score = self._calculate_confidence_score(dialogue_context)
        emotion_score = self._calculate_emotion_score(dialogue_context)
        complexity_score = self._calculate_complexity_score(dialogue_context)
        failure_score = self._calculate_failure_score(dialogue_context)
        satisfaction_score = self._calculate_satisfaction_score(dialogue_context)
        business_score = self._calculate_business_priority_score(dialogue_context)
        
        # 加权计算总分
        total_score = (
            confidence_score * self.decision_weights["confidence_score"] +
            emotion_score * self.decision_weights["emotion_score"] +
            complexity_score * self.decision_weights["complexity_score"] +
            failure_score * self.decision_weights["failure_count"] +
            satisfaction_score * self.decision_weights["user_satisfaction"] +
            business_score * self.decision_weights["business_priority"]
        )
        
        # 判断主要原因
        main_reason = self._identify_main_reason({
            "confidence": confidence_score,
            "emotion": emotion_score,
            "complexity": complexity_score,
            "failure": failure_score,
            "satisfaction": satisfaction_score,
            "business": business_score
        })
        
        should_handoff = total_score >= self.handoff_threshold
        
        return should_handoff, main_reason, total_score
    
    def _calculate_confidence_score(self, context: Dict) -> float:
        """计算置信度评分"""
        recent_confidences = context.get("confidence_scores", [])[-3:]  # 最近3轮
        
        if not recent_confidences:
            return 0.5
        
        avg_confidence = np.mean(recent_confidences)
        
        # 置信度越低，转人工评分越高
        if avg_confidence < 0.3:
            return 1.0
        elif avg_confidence < 0.5:
            return 0.8
        elif avg_confidence < 0.7:
            return 0.4
        else:
            return 0.1
    
    def _calculate_emotion_score(self, context: Dict) -> float:
        """计算情绪评分"""
        recent_inputs = [turn["user_input"] for turn in context.get("context_history", [])[-3:]]
        
        emotion_score = 0.0
        for text in recent_inputs:
            for emotion_type, keywords in self.emotion_keywords.items():
                for keyword in keywords:
                    if keyword in text:
                        if emotion_type == "negative":
                            emotion_score += 0.4
                        elif emotion_type == "urgent":
                            emotion_score += 0.3
                        elif emotion_type == "confused":
                            emotion_score += 0.2
        
        return min(emotion_score, 1.0)
    
    def _calculate_complexity_score(self, context: Dict) -> float:
        """计算查询复杂度评分"""
        current_intent = context.get("current_intent")
        turn_count = context.get("turn_count", 0)
        
        # 复杂意图映射
        complex_intents = {
            "complaint_handling": 0.8,
            "refund_request": 0.7,
            "technical_support": 0.6,
            "product_consultation": 0.3
        }
        
        intent_complexity = complex_intents.get(current_intent, 0.2)
        
        # 多轮对话增加复杂度
        turn_complexity = min(turn_count * 0.1, 0.5)
        
        return min(intent_complexity + turn_complexity, 1.0)
    
    def _calculate_failure_score(self, context: Dict) -> float:
        """计算失败次数评分"""
        context_history = context.get("context_history", [])
        
        # 统计最近的失败次数
        failure_count = 0
        for turn in context_history[-5:]:  # 最近5轮
            if turn.get("system_action") in ["fallback", "unclear", "no_result"]:
                failure_count += 1
        
        return min(failure_count * 0.3, 1.0)
    
    def _calculate_satisfaction_score(self, context: Dict) -> float:
        """计算用户满意度评分"""
        # 这里可以集成实时满意度检测模型
        # 简化实现：基于用户反馈关键词
        recent_inputs = [turn["user_input"] for turn in context.get("context_history", [])[-2:]]
        
        dissatisfaction_keywords = ["不行", "没用", "解决不了", "要人工", "转人工"]
        
        for text in recent_inputs:
            for keyword in dissatisfaction_keywords:
                if keyword in text:
                    return 1.0
        
        return 0.2
    
    def _calculate_business_priority_score(self, context: Dict) -> float:
        """计算业务优先级评分"""
        current_intent = context.get("current_intent")
        
        # 高优先级业务
        high_priority_intents = {
            "payment_issue": 0.9,
            "security_concern": 1.0,
            "complaint_handling": 0.8,
            "refund_request": 0.7
        }
        
        return high_priority_intents.get(current_intent, 0.2)
    
    def _identify_main_reason(self, scores: Dict) -> str:
        """识别主要转人工原因"""
        max_score_key = max(scores.keys(), key=lambda k: scores[k])
        
        reason_mapping = {
            "confidence": HandoffReason.LOW_CONFIDENCE.value,
            "emotion": HandoffReason.EMOTIONAL_ESCALATION.value,
            "complexity": HandoffReason.COMPLEX_QUERY.value,
            "failure": HandoffReason.REPEATED_FAILURE.value,
            "satisfaction": HandoffReason.EXPLICIT_REQUEST.value,
            "business": HandoffReason.BUSINESS_CRITICAL.value
        }
        
        return reason_mapping.get(max_score_key, "未知原因")

import time
import json
from typing import Dict, List
from collections import defaultdict, deque
from dataclasses import dataclass
from datetime import datetime, timedelta
@dataclass
class PerformanceMetrics:
    """性能指标数据类"""
    timestamp: float
    response_time: float
    intent_confidence: float
    user_satisfaction: float
    is_resolved: bool
    turn_count: int
    handoff_occurred: bool
class PerformanceMonitor:
    """性能监控系统"""
    
    def __init__(self, window_size: int = 1000):
        self.window_size = window_size
        self.metrics_buffer = deque(maxlen=window_size)
        self.real_time_stats = defaultdict(list)
        
        # 阈值配置
        self.thresholds = {
            "response_time": 1000,      # 1秒
            "intent_confidence": 0.8,   # 80%
            "satisfaction": 4.0,        # 4分(满分5分)
            "resolution_rate": 0.75     # 75%
        }
    
    def record_interaction(self, metrics: PerformanceMetrics):
        """记录交互指标"""
        self.metrics_buffer.append(metrics)
        
        # 实时异常检测
        self._detect_anomalies(metrics)
        
        # 更新实时统计
        self._update_real_time_stats(metrics)
    
    def get_performance_report(self, time_range: str = "1h") -> Dict:
        """生成性能报告"""
        
        # 筛选时间范围内的数据
        cutoff_time = self._get_cutoff_time(time_range)
        filtered_metrics = [
            m for m in self.metrics_buffer 
            if m.timestamp >= cutoff_time
        ]
        
        if not filtered_metrics:
            return {"error": "No data available for the specified time range"}
        
        # 计算各项指标
        report = {
            "time_range": time_range,
            "total_interactions": len(filtered_metrics),
            "performance_metrics": {
                "avg_response_time": self._calculate_avg_response_time(filtered_metrics),
                "intent_accuracy": self._calculate_intent_accuracy(filtered_metrics),
                "resolution_rate": self._calculate_resolution_rate(filtered_metrics),
                "avg_satisfaction": self._calculate_avg_satisfaction(filtered_metrics),
                "handoff_rate": self._calculate_handoff_rate(filtered_metrics),
                "avg_turn_count": self._calculate_avg_turn_count(filtered_metrics)
            },
            "quality_distribution": self._get_quality_distribution(filtered_metrics),
            "anomalies": self._get_recent_anomalies(cutoff_time)
        }
        
        return report
    
    def _detect_anomalies(self, metrics: PerformanceMetrics):
        """实时异常检测"""
        anomalies = []
        
        # 响应时间异常
        if metrics.response_time > self.thresholds["response_time"]:
            anomalies.append({
                "type": "high_response_time",
                "value": metrics.response_time,
                "threshold": self.thresholds["response_time"],
                "timestamp": metrics.timestamp
            })
        
        # 置信度异常
        if metrics.intent_confidence < self.thresholds["intent_confidence"]:
            anomalies.append({
                "type": "low_confidence",
                "value": metrics.intent_confidence,
                "threshold": self.thresholds["intent_confidence"],
                "timestamp": metrics.timestamp
            })
        
        # 满意度异常
        if metrics.user_satisfaction < self.thresholds["satisfaction"]:
            anomalies.append({
                "type": "low_satisfaction",
                "value": metrics.user_satisfaction,
                "threshold": self.thresholds["satisfaction"],
                "timestamp": metrics.timestamp
            })
        
        # 记录异常
        if anomalies:
            self.real_time_stats["anomalies"].extend(anomalies)
    
    def _calculate_avg_response_time(self, metrics: List[PerformanceMetrics]) -> float:
        """计算平均响应时间"""
        total_time = sum(m.response_time for m in metrics)
        return round(total_time / len(metrics), 2)
    
    def _calculate_intent_accuracy(self, metrics: List[PerformanceMetrics]) -> float:
        """计算意图识别准确率"""
        high_confidence_count = sum(
            1 for m in metrics 
            if m.intent_confidence >= self.thresholds["intent_confidence"]
        )
        return round(high_confidence_count / len(metrics), 3)
    
    def _calculate_resolution_rate(self, metrics: List[PerformanceMetrics]) -> float:
        """计算问题解决率"""
        resolved_count = sum(1 for m in metrics if m.is_resolved)
        return round(resolved_count / len(metrics), 3)
    
    def _get_cutoff_time(self, time_range: str) -> float:
        """获取时间范围的截止时间"""
        now = time.time()
        if time_range == "1h":
            return now - 3600
        elif time_range == "24h":
            return now - 86400
        elif time_range == "7d":
            return now - 604800
        else:
            return now - 3600  # 默认1小时

# Docker部署配置示例
version: '3.8'
services:
  # NLU服务
  nlu-service:
    image: intelligent-agent/nlu:latest
    replicas: 3
    resources:
      limits:
        memory: 2G
        cpus: "1.0"
    environment:
      - MODEL_PATH=/models/bert-chinese
      - MAX_SEQUENCE_LENGTH=512
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8080/health"]
      interval: 30s
      timeout: 10s
      retries: 3
  # 对话管理服务
  dialogue-manager:
    image: intelligent-agent/dm:latest
    replicas: 2
    depends_on:
      - redis
      - postgresql
    environment:
      - REDIS_URL=redis://redis:6379
      - DB_URL=postgresql://user:pass@postgresql:5432/dialogue_db
  # 知识检索服务
  knowledge-retrieval:
    image: intelligent-agent/knowledge:latest
    replicas: 2
    volumes:
      - ./knowledge_base:/app/knowledge_base
    environment:
      - FAISS_INDEX_PATH=/app/knowledge_base/faiss_index
      - SENTENCE_TRANSFORMER_MODEL=distilbert-base-chinese