160_社交媒体分析：舆情监测 - 2025年LLM驱动的实时流情感聚类与多模态舆情洞察技术实现

目录
├── 1. 社交媒体舆情监测的现状与挑战
├── 2. LLM在情感分析与聚类中的应用
├── 3. 实时流处理架构设计
├── 4. 情感聚类算法实现
├── 5. 多模态舆情数据融合
├── 6. 系统优化与性能调优
├── 7. 实际应用案例分析
├── 8. 未来发展与技术展望

class LLMSentimentAnalyzer:
    def __init__(self, llm_client, model_name="gpt-5"):
        """
        初始化LLM情感分析器
        
        参数:
        llm_client: LLM客户端，用于与LLM服务交互
        model_name: 使用的模型名称
        """
        self.llm_client = llm_client
        self.model_name = model_name
        self.prompt_template = ""
        self.prompt_template += "请分析以下社交媒体文本的情感，并提供详细的情感分析结果。\n"
        self.prompt_template += "文本: {text}\n\n"
        self.prompt_template += "请以JSON格式返回以下信息:\n"
        self.prompt_template += "1. 主要情感: positive, negative, neutral中的一个\n"
        self.prompt_template += "2. 情感强度: 0-100的数值，越高表示情感越强烈\n"
        self.prompt_template += "3. 情感关键词: 文本中表达情感的主要关键词\n"
        self.prompt_template += "4. 潜在意图: 作者表达此情感的可能意图\n"
        self.prompt_template += "5. 情感分析依据: 为什么认为这是这种情感\n"

    def analyze_sentiment(self, text, max_tokens=500):
        """
        分析文本情感
        
        参数:
        text: 要分析的文本
        max_tokens: 生成的最大token数
        
        返回:
        情感分析结果字典
        """
        # 构建提示
        prompt = self.prompt_template.format(text=text)
        
        # 调用LLM
        response = self.llm_client.generate(
            model=self.model_name,
            prompt=prompt,
            max_tokens=max_tokens,
            temperature=0.1,  # 低温度以确保一致性
            response_format={"type": "json_object"}
        )
        
        # 解析结果
        try:
            sentiment_result = json.loads(response["content"])
            # 添加置信度和处理时间等元数据
            sentiment_result["confidence"] = response.get("confidence", 0.9)
            sentiment_result["processing_time_ms"] = response.get("processing_time_ms", 0)
            return sentiment_result
        except Exception as e:
            # 处理解析错误
            return {
                "error": str(e),
                "raw_response": response,
                "fallback_sentiment": "neutral"
            }

    def batch_analyze(self, texts, batch_size=10, max_workers=4):
        """
        批量分析文本情感
        
        参数:
        texts: 文本列表
        batch_size: 批处理大小
        max_workers: 最大工作线程数
        
        返回:
        情感分析结果列表
        """
        results = []
        
        # 分批处理
        for i in range(0, len(texts), batch_size):
            batch = texts[i:i+batch_size]
            
            # 并行处理批次中的文本
            with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
                batch_results = list(executor.map(self.analyze_sentiment, batch))
            
            results.extend(batch_results)
        
        return results

情感聚类流程:
文本预处理 → 特征提取 → 向量嵌入 → 相似度计算 → 聚类算法应用 → 主题提取 → 结果可视化

社交媒体平台 → 数据采集层 → 消息队列(Kafka) → 流处理引擎(Flink) → 分析处理层(LLM+聚类) → 存储层 → 服务层 → 终端用户
                                  ↑                                 |
                                  └────────────── 反馈循环 ─────────┘

# Flink高可用配置示例
high-availability: zookeeper
high-availability.storageDir: hdfs:///flink/recovery
high-availability.zookeeper.quorum: zk1:2181,zk2:2181,zk3:2181
high-availability.zookeeper.path.root: /flink
high-availability.cluster-id: /flink/舆情监测集群

# 检查点配置
state.backend: rocksdb
state.checkpoints.dir: hdfs:///flink/checkpoints
state.checkpoints.interval: 60000  # 60秒
state.savepoints.dir: hdfs:///flink/savepoints

# 并行度配置
parallelism.default: 128

# 容错配置
taskmanager.numberOfTaskSlots: 8
jobmanager.execution.failover-strategy: region

class LLMEmbeddingClustering:
    def __init__(self, llm_embedding_client, clustering_algorithm="density_based"):
        """
        初始化LLM嵌入聚类器
        
        参数:
        llm_embedding_client: LLM嵌入客户端
        clustering_algorithm: 聚类算法类型，支持"density_based"、"hierarchical"、"streaming"
        """
        self.llm_embedding_client = llm_embedding_client
        self.clustering_algorithm = clustering_algorithm
        self.clusters = {}
        self.cluster_counter = 0
        self.embeddings_cache = {}
        self.config = self._get_default_config()
    
    def _get_default_config(self):
        """获取默认配置"""
        return {
            "density_based": {
                "eps": 0.5,  # 最大距离参数
                "min_samples": 5,  # 最小样本数
                "metric": "cosine"  # 距离度量
            },
            "hierarchical": {
                "threshold": 0.6,  # 聚类阈值
                "linkage": "ward"  # 链接方法
            },
            "streaming": {
                "window_size": 1000,  # 窗口大小
                "similarity_threshold": 0.7,  # 相似度阈值
                "decay_factor": 0.9  # 衰减因子
            }
        }
    
    def get_embedding(self, text):
        """
        获取文本的嵌入向量，带缓存机制
        
        参数:
        text: 输入文本
        
        返回:
        嵌入向量
        """
        # 检查缓存
        if text in self.embeddings_cache:
            return self.embeddings_cache[text]
        
        # 调用LLM获取嵌入
        embedding = self.llm_embedding_client.generate_embedding(text)
        
        # 存入缓存
        self.embeddings_cache[text] = embedding
        
        # 缓存管理：如果缓存过大，清理部分旧缓存
        if len(self.embeddings_cache) > 10000:
            self._prune_cache(5000)
        
        return embedding
    
    def _prune_cache(self, keep_size):
        """清理缓存"""
        # 简单实现：保留最近添加的部分
        keys = list(self.embeddings_cache.keys())[:keep_size]
        new_cache = {k: self.embeddings_cache[k] for k in keys}
        self.embeddings_cache = new_cache
    
    def cluster(self, texts, batch_size=100):
        """
        对文本进行聚类
        
        参数:
        texts: 文本列表
        batch_size: 批处理大小
        
        返回:
        聚类结果字典
        """
        results = []
        
        # 分批处理
        for i in range(0, len(texts), batch_size):
            batch = texts[i:i+batch_size]
            batch_results = self._process_batch(batch)
            results.extend(batch_results)
        
        return results
    
    def _process_batch(self, texts):
        """处理一批文本"""
        # 获取嵌入向量
        embeddings = [self.get_embedding(text) for text in texts]
        
        # 根据算法类型执行聚类
        if self.clustering_algorithm == "density_based":
            return self._density_based_clustering(texts, embeddings)
        elif self.clustering_algorithm == "hierarchical":
            return self._hierarchical_clustering(texts, embeddings)
        elif self.clustering_algorithm == "streaming":
            return self._streaming_clustering(texts, embeddings)
        else:
            raise ValueError(f"不支持的聚类算法: {self.clustering_algorithm}")
    
    def _density_based_clustering(self, texts, embeddings):
        """基于密度的聚类(DBSCAN变体)"""
        # 使用sklearn的DBSCAN或自定义实现
        from sklearn.cluster import DBSCAN
        
        config = self.config["density_based"]
        dbscan = DBSCAN(
            eps=config["eps"],
            min_samples=config["min_samples"],
            metric=config["metric"]
        )
        
        # 执行聚类
        labels = dbscan.fit_predict(embeddings)
        
        # 处理结果
        results = []
        for i, (text, label) in enumerate(zip(texts, labels)):
            if label == -1:  # 噪声点，分配到新聚类
                self.cluster_counter += 1
                cluster_id = f"cluster_{self.cluster_counter}"
                self.clusters[cluster_id] = {
                    "centroid": embeddings[i],
                    "size": 1,
                    "samples": [text]
                }
                results.append({
                    "text": text,
                    "cluster_id": cluster_id,
                    "is_new_cluster": True
                })
            else:
                # 分配到现有聚类
                cluster_id = f"cluster_{label + 1}"  # 调整标签以匹配我们的ID格式
                results.append({
                    "text": text,
                    "cluster_id": cluster_id,
                    "is_new_cluster": False
                })
        
        return results

class IncrementalClustering:
    def __init__(self, embedding_dim=1024, similarity_threshold=0.7, max_clusters=1000):
        """
        初始化增量聚类器
        
        参数:
        embedding_dim: 嵌入向量维度
        similarity_threshold: 相似度阈值，超过此值的文本将被归为一类
        max_clusters: 最大聚类数量
        """
        self.embedding_dim = embedding_dim
        self.similarity_threshold = similarity_threshold
        self.max_clusters = max_clusters
        self.clusters = {}
        self.cluster_counter = 0
        
        # 初始化向量索引
        try:
            import faiss
            self.index = faiss.IndexHNSWFlat(embedding_dim, 32)
            self.use_faiss = True
        except ImportError:
            # 降级到简单实现
            self.use_faiss = False
    
    def add_document(self, text, embedding, metadata=None):
        """
        添加单个文档到聚类
        
        参数:
        text: 文档文本
        embedding: 嵌入向量
        metadata: 文档元数据
        
        返回:
        聚类结果
        """
        metadata = metadata or {}
        
        # 查找最相似的聚类
        best_cluster_id = None
        best_similarity = 0
        
        if self.clusters and self.use_faiss:
            # 使用FAISS进行快速近邻搜索
            distances, indices = self.index.search(np.array([embedding]), 1)
            if indices[0][0] != -1:
                # FAISS返回的是索引，需要映射回聚类ID
                # 注意：这里简化了实现，实际需要维护索引到聚类ID的映射
                pass
        elif self.clusters:
            # 线性搜索最相似的聚类
            for cluster_id, cluster in self.clusters.items():
                # 计算余弦相似度
                similarity = self._cosine_similarity(embedding, cluster["centroid"])
                if similarity > best_similarity:
                    best_similarity = similarity
                    best_cluster_id = cluster_id
        
        # 判断是否创建新聚类
        if best_similarity < self.similarity_threshold or not self.clusters:
            # 创建新聚类
            self.cluster_counter += 1
            cluster_id = f"cluster_{self.cluster_counter}"
            self.clusters[cluster_id] = {
                "centroid": embedding,
                "size": 1,
                "texts": [text],
                "embeddings": [embedding],
                "metadata": [metadata],
                "created_at": time.time(),
                "updated_at": time.time()
            }
            
            # 更新索引
            if self.use_faiss and len(self.clusters) <= self.max_clusters:
                self.index.add(np.array([embedding]))
            
            return {
                "cluster_id": cluster_id,
                "is_new_cluster": True,
                "similarity": 0
            }
        else:
            # 更新现有聚类
            cluster = self.clusters[best_cluster_id]
            
            # 更新聚类中心（简单平均，可使用更复杂的方法）
            new_size = cluster["size"] + 1
            cluster["centroid"] = self._update_centroid(
                cluster["centroid"], 
                embedding, 
                cluster["size"], 
                new_size
            )
            
            # 更新聚类信息
            cluster["size"] = new_size
            cluster["texts"].append(text)
            cluster["embeddings"].append(embedding)
            cluster["metadata"].append(metadata)
            cluster["updated_at"] = time.time()
            
            # 对于FAISS，需要重建索引或使用支持动态更新的索引结构
            # 这里简化处理，实际应用中需要更复杂的索引更新策略
            
            return {
                "cluster_id": best_cluster_id,
                "is_new_cluster": False,
                "similarity": best_similarity
            }
    
    def _cosine_similarity(self, vec1, vec2):
        """计算余弦相似度"""
        return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))
    
    def _update_centroid(self, old_centroid, new_embedding, old_size, new_size):
        """更新聚类中心"""
        # 简单加权平均
        return ((old_centroid * old_size) + new_embedding) / new_size
    
    def get_cluster_info(self, cluster_id):
        """
        获取聚类信息
        
        参数:
        cluster_id: 聚类ID
        
        返回:
        聚类信息字典
        """
        return self.clusters.get(cluster_id, None)
    
    def get_all_clusters(self):
        """获取所有聚类"""
        return self.clusters
    
    def prune_old_clusters(self, max_age_seconds=3600):
        """
        清理旧聚类
        
        参数:
        max_age_seconds: 聚类最大年龄（秒）
        """
        current_time = time.time()
        old_clusters = []
        
        for cluster_id, cluster in self.clusters.items():
            if current_time - cluster["updated_at"] > max_age_seconds:
                old_clusters.append(cluster_id)
        
        for cluster_id in old_clusters:
            del self.clusters[cluster_id]
        
        # 重建索引
        if self.use_faiss and old_clusters:
            self.index = faiss.IndexHNSWFlat(self.embedding_dim, 32)
            centroids = [cluster["centroid"] for cluster in self.clusters.values()]
            if centroids:
                self.index.add(np.array(centroids))
        
        return len(old_clusters)

class ClusterEvaluation:
    def __init__(self):
        """初始化聚类评估器"""
        self.metrics = {
            "silhouette": self._silhouette_score,
            "davies_bouldin": self._davies_bouldin_score,
            "calinski_harabasz": self._calinski_harabasz_score
        }
    
    def evaluate_clustering(self, embeddings, labels, metrics=None):
        """
        评估聚类质量
        
        参数:
        embeddings: 嵌入向量列表
        labels: 聚类标签列表
        metrics: 要使用的评估指标列表
        
        返回:
        评估结果字典
        """
        if metrics is None:
            metrics = self.metrics.keys()
        
        results = {}
        
        for metric in metrics:
            if metric in self.metrics:
                try:
                    results[metric] = self.metrics[metric](embeddings, labels)
                except Exception as e:
                    results[metric] = {"error": str(e)}
        
        return results
    
    def _silhouette_score(self, embeddings, labels):
        """计算轮廓系数"""
        from sklearn.metrics import silhouette_score
        try:
            return silhouette_score(embeddings, labels)
        except ValueError:
            # 如果聚类数小于2，无法计算
            return -1
    
    def _davies_bouldin_score(self, embeddings, labels):
        """计算Davies-Bouldin指数"""
        from sklearn.metrics import davies_bouldin_score
        try:
            return davies_bouldin_score(embeddings, labels)
        except ValueError:
            # 如果聚类数小于2，无法计算
            return float('inf')
    
    def _calinski_harabasz_score(self, embeddings, labels):
        """计算Calinski-Harabasz指数"""
        from sklearn.metrics import calinski_harabasz_score
        try:
            return calinski_harabasz_score(embeddings, labels)
        except ValueError:
            # 如果聚类数小于2，无法计算
            return 0
    
    def optimize_clustering_params(self, embeddings, param_grid, algorithm="dbscan"):
        """
        优化聚类参数
        
        参数:
        embeddings: 嵌入向量
        param_grid: 参数网格
        algorithm: 聚类算法
        
        返回:
        最佳参数和评估得分
        """
        if algorithm == "dbscan":
            return self._optimize_dbscan(embeddings, param_grid)
        elif algorithm == "kmeans":
            return self._optimize_kmeans(embeddings, param_grid)
        else:
            raise ValueError(f"不支持的算法: {algorithm}")
    
    def _optimize_dbscan(self, embeddings, param_grid):
        """优化DBSCAN参数"""
        from sklearn.cluster import DBSCAN
        best_score = -1
        best_params = None
        
        for eps in param_grid.get("eps", [0.5]):
            for min_samples in param_grid.get("min_samples", [5]):
                # 执行聚类
                dbscan = DBSCAN(eps=eps, min_samples=min_samples)
                labels = dbscan.fit_predict(embeddings)
                
                # 计算评估指标
                n_clusters = len(set(labels)) - (1 if -1 in labels else 0)
                if n_clusters > 1:  # 确保至少有2个有效聚类
                    silhouette = self._silhouette_score(embeddings, labels)
                    
                    # 更新最佳参数
                    if silhouette > best_score:
                        best_score = silhouette
                        best_params = {
                            "eps": eps,
                            "min_samples": min_samples,
                            "n_clusters": n_clusters
                        }
        
        return {
            "best_params": best_params,
            "best_score": best_score,
            "metric": "silhouette"
        }

class MultimodalDataProcessor:
    def __init__(self, text_processor, image_processor, audio_processor=None):
        """
        初始化多模态数据处理器
        
        参数:
        text_processor: 文本处理器
        image_processor: 图像处理器
        audio_processor: 音频处理器（可选）
        """
        self.text_processor = text_processor
        self.image_processor = image_processor
        self.audio_processor = audio_processor
        self.fusion_strategy = "early_fusion"
    
    def process_item(self, item):
        """
        处理单个多模态项目
        
        参数:
        item: 包含多模态数据的字典
        
        返回:
        处理结果
        """
        results = {}
        
        # 处理文本
        if "text" in item:
            results["text"] = self.text_processor.process(item["text"])
        
        # 处理图像
        if "image_url" in item or "image_data" in item:
            image_input = item.get("image_data", None) or item["image_url"]
            results["image"] = self.image_processor.process(image_input)
        
        # 处理音频
        if self.audio_processor and ("audio_url" in item or "audio_data" in item):
            audio_input = item.get("audio_data", None) or item["audio_url"]
            results["audio"] = self.audio_processor.process(audio_input)
        
        # 执行多模态融合
        if len(results) > 1:
            results["fused"] = self._fuse_multimodal(results)
        
        return results
    
    def _fuse_multimodal(self, modality_results):
        """
        融合多模态结果
        
        参数:
        modality_results: 各模态处理结果的字典
        
        返回:
        融合结果
        """
        if self.fusion_strategy == "early_fusion":
            return self._early_fusion(modality_results)
        elif self.fusion_strategy == "late_fusion":
            return self._late_fusion(modality_results)
        elif self.fusion_strategy == "hybrid_fusion":
            return self._hybrid_fusion(modality_results)
        else:
            raise ValueError(f"不支持的融合策略: {self.fusion_strategy}")
    
    def _early_fusion(self, modality_results):
        """
        早期融合：在特征层面融合
        """
        embeddings = []
        
        # 收集各模态的嵌入向量
        if "text" in modality_results and "embedding" in modality_results["text"]:
            embeddings.append(modality_results["text"]["embedding"])
        
        if "image" in modality_results and "embedding" in modality_results["image"]:
            embeddings.append(modality_results["image"]["embedding"])
        
        if "audio" in modality_results and "embedding" in modality_results["audio"]:
            embeddings.append(modality_results["audio"]["embedding"])
        
        if not embeddings:
            return {"error": "没有可融合的嵌入向量"}
        
        # 简单连接融合
        # 实际应用中可能需要更复杂的融合方法
        fused_embedding = np.concatenate(embeddings)
        
        return {
            "embedding": fused_embedding,
            "modality_weights": {}
        }
    
    def _late_fusion(self, modality_results):
        """
        晚期融合：在决策层面融合
        """
        # 收集各模态的决策结果
        sentiment_scores = {}
        confidence_scores = {}
        
        if "text" in modality_results and "sentiment" in modality_results["text"]:
            sentiment_scores["text"] = modality_results["text"]["sentiment"]
            confidence_scores["text"] = modality_results["text"].get("confidence", 1.0)
        
        if "image" in modality_results and "sentiment" in modality_results["image"]:
            sentiment_scores["image"] = modality_results["image"]["sentiment"]
            confidence_scores["image"] = modality_results["image"].get("confidence", 1.0)
        
        if "audio" in modality_results and "sentiment" in modality_results["audio"]:
            sentiment_scores["audio"] = modality_results["audio"]["sentiment"]
            confidence_scores["audio"] = modality_results["audio"].get("confidence", 1.0)
        
        if not sentiment_scores:
            return {"error": "没有可融合的决策结果"}
        
        # 计算加权平均情感
        # 这里简化实现，实际应用中可能需要更复杂的融合方法
        weighted_sentiment = {}
        total_confidence = sum(confidence_scores.values())
        
        for modality, sentiment in sentiment_scores.items():
            weight = confidence_scores[modality] / total_confidence
            for sentiment_type, score in sentiment.items():
                if sentiment_type not in weighted_sentiment:
                    weighted_sentiment[sentiment_type] = 0
                weighted_sentiment[sentiment_type] += score * weight
        
        return {
            "sentiment": weighted_sentiment,
            "modality_weights": confidence_scores
        }
    
    def set_fusion_strategy(self, strategy):
        """
        设置融合策略
        
        参数:
        strategy: 融合策略，支持"early_fusion", "late_fusion", "hybrid_fusion"
        """
        valid_strategies = ["early_fusion", "late_fusion", "hybrid_fusion"]
        if strategy not in valid_strategies:
            raise ValueError(f"无效的融合策略，支持的策略: {valid_strategies}")
        self.fusion_strategy = strategy

class CrossModalSentimentAnalyzer:
    def __init__(self, multimodal_processor):
        """
        初始化跨模态情感分析器
        
        参数:
        multimodal_processor: 多模态处理器
        """
        self.multimodal_processor = multimodal_processor
    
    def analyze_consistency(self, multimodal_item):
        """
        分析多模态情感一致性
        
        参数:
        multimodal_item: 多模态项目
        
        返回:
        一致性分析结果
        """
        # 处理多模态数据
        processed_results = self.multimodal_processor.process_item(multimodal_item)
        
        # 提取各模态情感
        modalities = ["text", "image", "audio"]
        sentiment_results = {}
        
        for modality in modalities:
            if modality in processed_results and "sentiment" in processed_results[modality]:
                sentiment_results[modality] = processed_results[modality]["sentiment"]
        
        # 计算一致性
        if len(sentiment_results) < 2:
            # 单模态或无模态数据，返回基本结果
            if sentiment_results:
                first_modality = list(sentiment_results.keys())[0]
                return {
                    "sentiment": sentiment_results[first_modality],
                    "consistency": 1.0,  # 单模态一致性为1
                    "conflict_detection": "single_modality",
                    "dominant_modality": first_modality
                }
            else:
                return {
                    "error": "没有有效的情感分析结果",
                    "consistency": 0.0
                }
        
        # 计算模态间一致性
        pairwise_consistencies = {}
        sentiment_directions = {}
        
        # 确定各模态的情感方向（积极/消极/中性）
        for modality, sentiment in sentiment_results.items():
            sentiment_directions[modality] = self._get_sentiment_direction(sentiment)
        
        # 计算两两模态间的一致性
        modalities_list = list(sentiment_results.keys())
        for i in range(len(modalities_list)):
            for j in range(i+1, len(modalities_list)):
                mod1 = modalities_list[i]
                mod2 = modalities_list[j]
                consistency = self._calculate_pairwise_consistency(
                    sentiment_results[mod1], 
                    sentiment_results[mod2]
                )
                pairwise_consistencies[f"{mod1}_vs_{mod2}"] = consistency
        
        # 检测情感冲突
        conflict_detection = self._detect_sentiment_conflict(sentiment_directions, pairwise_consistencies)
        
        # 确定主导情感和模态
        dominant = self._determine_dominant_sentiment(sentiment_results, pairwise_consistencies)
        
        # 计算整体一致性得分
        overall_consistency = sum(pairwise_consistencies.values()) / len(pairwise_consistencies)
        
        return {
            "sentiment": dominant["sentiment"],
            "consistency": overall_consistency,
            "conflict_detection": conflict_detection,
            "dominant_modality": dominant["modality"],
            "pairwise_consistencies": pairwise_consistencies,
            "modality_sentiments": sentiment_results
        }
    
    def _get_sentiment_direction(self, sentiment):
        """
        获取情感方向
        
        参数:
        sentiment: 情感字典
        
        返回:
        情感方向: "positive", "negative", "neutral"
        """
        if "positive" in sentiment and "negative" in sentiment:
            if sentiment["positive"] > sentiment["negative"]:
                return "positive"
            elif sentiment["negative"] > sentiment["positive"]:
                return "negative"
            else:
                return "neutral"
        elif "positive" in sentiment:
            return "positive"
        elif "negative" in sentiment:
            return "negative"
        else:
            return "neutral"
    
    def _calculate_pairwise_consistency(self, sentiment1, sentiment2):
        """
        计算两个模态情感的一致性
        
        参数:
        sentiment1: 模态1的情感字典
        sentiment2: 模态2的情感字典
        
        返回:
        一致性得分(0-1)
        """
        # 简化实现：计算情感分布的余弦相似度
        # 确保情感类别一致
        all_categories = set(sentiment1.keys()).union(set(sentiment2.keys()))
        vec1 = [sentiment1.get(cat, 0) for cat in all_categories]
        vec2 = [sentiment2.get(cat, 0) for cat in all_categories]
        
        # 计算余弦相似度
        dot_product = np.dot(vec1, vec2)
        norm1 = np.linalg.norm(vec1)
        norm2 = np.linalg.norm(vec2)
        
        if norm1 == 0 or norm2 == 0:
            return 0.0
        
        return dot_product / (norm1 * norm2)
    
    def _detect_sentiment_conflict(self, sentiment_directions, pairwise_consistencies):
        """
        检测情感冲突
        
        参数:
        sentiment_directions: 各模态的情感方向
        pairwise_consistencies: 模态间一致性
        
        返回:
        冲突检测结果
        """
        # 检查情感方向是否一致
        directions = list(sentiment_directions.values())
        if all(d == directions[0] for d in directions):
            return "no_conflict"
        
        # 检查低一致性对
        low_consistency_pairs = []
        for pair, score in pairwise_consistencies.items():
            if score < 0.5:  # 一致性阈值
                low_consistency_pairs.append(pair)
        
        if low_consistency_pairs:
            return "conflict_detected", {"low_consistency_pairs": low_consistency_pairs}
        
        return "potential_conflict"
    
    def _determine_dominant_sentiment(self, sentiment_results, pairwise_consistencies):
        """
        确定主导情感和模态
        
        参数:
        sentiment_results: 各模态情感结果
        pairwise_consistencies: 模态间一致性
        
        返回:
        主导情感和模态
        """
        # 简化实现：基于情感强度和一致性确定
        # 在实际应用中，可能需要更复杂的逻辑
        
        # 计算各模态的情感强度
        modality_strengths = {}
        for modality, sentiment in sentiment_results.items():
            # 假设情感字典包含强度值
            if "positive" in sentiment or "negative" in sentiment:
                pos_score = sentiment.get("positive", 0)
                neg_score = sentiment.get("negative", 0)
                strength = max(pos_score, neg_score)
                modality_strengths[modality] = strength
            else:
                modality_strengths[modality] = 0
        
        # 找出强度最高的模态
        dominant_modality = max(modality_strengths, key=modality_strengths.get)
        
        return {
            "modality": dominant_modality,
            "sentiment": sentiment_results[dominant_modality]
        }

class VisionSentimentAnalyzer:
    def __init__(self, vision_llm_client, model_name="gpt-5-vision"):
        """
        初始化视觉情感分析器
        
        参数:
        vision_llm_client: 视觉LLM客户端
        model_name: 使用的模型名称
        """
        self.vision_llm_client = vision_llm_client
        self.model_name = model_name
        self.prompt_template = ""
        self.prompt_template += "请分析以下图像的情感内容，并提供详细的情感分析结果。\n"
        self.prompt_template += "请考虑图像中的以下因素：\n"
        self.prompt_template += "1. 人物表情和姿态\n"
        self.prompt_template += "2. 色彩和构图\n"
        self.prompt_template += "3. 场景和环境\n"
        self.prompt_template += "4. 整体氛围\n"
        self.prompt_template += "5. 可能传达的情绪\n"
        self.prompt_template += "\n请以JSON格式返回以下信息:\n"
        self.prompt_template += "1. 主要情感: positive, negative, neutral中的一个\n"
        self.prompt_template += "2. 情感强度: 0-100的数值，越高表示情感越强烈\n"
        self.prompt_template += "3. 情感细节: 包含多种情感类别的详细分布\n"
        self.prompt_template += "4. 视觉元素: 影响情感判断的主要视觉元素\n"
        self.prompt_template += "5. 分析依据: 为什么认为图像表达这种情感\n"
    
    def analyze_image(self, image_url, text_context=None, max_tokens=1000):
        """
        分析图像情感
        
        参数:
        image_url: 图像URL或本地路径
        text_context: 可选的文本上下文
        max_tokens: 最大生成token数
        
        返回:
        情感分析结果
        """
        # 构建提示
        prompt = self.prompt_template
        if text_context:
            prompt += f"\n图像相关文本上下文: {text_context}\n"
        
        # 调用视觉LLM
        response = self.vision_llm_client.analyze_image(
            model=self.model_name,
            image=image_url,
            prompt=prompt,
            max_tokens=max_tokens,
            temperature=0.1,
            response_format={"type": "json_object"}
        )
        
        # 解析结果
        try:
            sentiment_result = json.loads(response["content"])
            # 添加置信度和处理时间等元数据
            sentiment_result["confidence"] = response.get("confidence", 0.9)
            sentiment_result["processing_time_ms"] = response.get("processing_time_ms", 0)
            return sentiment_result
        except Exception as e:
            # 处理解析错误
            return {
                "error": str(e),
                "raw_response": response,
                "fallback_sentiment": "neutral"
            }
    
    def analyze_video(self, video_url, num_frames=10, text_context=None):
        """
        分析视频情感
        
        参数:
        video_url: 视频URL或本地路径
        num_frames: 采样帧数
        text_context: 可选的文本上下文
        
        返回:
        视频情感分析结果
        """
        # 提取视频关键帧
        frames = self._extract_key_frames(video_url, num_frames)
        
        # 分析每一帧
        frame_results = []
        for frame_path in frames:
            try:
                frame_result = self.analyze_image(frame_path, text_context)
                frame_results.append(frame_result)
            except Exception as e:
                frame_results.append({"error": str(e), "frame_path": frame_path})
        
        # 融合帧分析结果
        video_sentiment = self._aggregate_frame_sentiments(frame_results)
        
        return {
            "overall_sentiment": video_sentiment,
            "frame_sentiments": frame_results,
            "num_processed_frames": len(frame_results)
        }
    
    def _extract_key_frames(self, video_url, num_frames):
        """
        提取视频关键帧
        
        参数:
        video_url: 视频URL或路径
        num_frames: 要提取的帧数
        
        返回:
        帧路径列表
        """
        # 实际实现需要使用视频处理库如OpenCV
        # 这里简化为模拟实现
        frames = []
        
        try:
            import cv2
            video = cv2.VideoCapture(video_url)
            total_frames = int(video.get(cv2.CAP_PROP_FRAME_COUNT))
            frame_interval = max(1, total_frames // num_frames)
            
            current_frame = 0
            while current_frame < total_frames and len(frames) < num_frames:
                video.set(cv2.CAP_PROP_POS_FRAMES, current_frame)
                ret, frame = video.read()
                if ret:
                    # 保存帧
                    frame_path = f"temp_frame_{len(frames)}.jpg"
                    cv2.imwrite(frame_path, frame)
                    frames.append(frame_path)
                current_frame += frame_interval
            
            video.release()
        except Exception as e:
            print(f"视频帧提取错误: {e}")
        
        return frames
    
    def _aggregate_frame_sentiments(self, frame_results):
        """
        聚合帧情感结果
        
        参数:
        frame_results: 帧情感分析结果列表
        
        返回:
        聚合的视频情感
        """
        if not frame_results:
            return {"sentiment": "neutral", "confidence": 0.5}
        
        # 提取有效结果
        valid_results = [r for r in frame_results if "error" not in r]
        if not valid_results:
            return {"sentiment": "neutral", "confidence": 0.3}
        
        # 简单平均聚合
        sentiment_counts = {"positive": 0, "negative": 0, "neutral": 0}
        sentiment_intensities = {"positive": 0, "negative": 0, "neutral": 0}
        
        for result in valid_results:
            sentiment = result.get("主要情感", "neutral")
            intensity = result.get("情感强度", 50)
            sentiment_counts[sentiment] += 1
            sentiment_intensities[sentiment] += intensity
        
        # 计算平均强度
        for sentiment in sentiment_intensities:
            if sentiment_counts[sentiment] > 0:
                sentiment_intensities[sentiment] /= sentiment_counts[sentiment]
        
        # 确定主导情感
        dominant_sentiment = max(sentiment_counts, key=sentiment_counts.get)
        
        return {
            "sentiment": dominant_sentiment,
            "confidence": sentiment_counts[dominant_sentiment] / len(valid_results),
            "sentiment_distribution": {
                "positive": sentiment_counts["positive"] / len(valid_results),
                "negative": sentiment_counts["negative"] / len(valid_results),
                "neutral": sentiment_counts["neutral"] / len(valid_results)
            },
            "average_intensities": sentiment_intensities
        }

class MultiLevelCache:
    def __init__(self, embedding_dim=1024, lru_cache_size=10000, ttl=3600):
        """
        初始化多级缓存
        
        参数:
        embedding_dim: 嵌入向量维度
        lru_cache_size: LRU缓存大小
        ttl: 缓存过期时间(秒)
        """
        # 内存LRU缓存
        self.memory_cache = lru.LRUCache(max_size=lru_cache_size)
        
        # Redis分布式缓存
        try:
            import redis
            self.redis_client = redis.Redis(host='localhost', port=6379, db=0)
            self.redis_enabled = True
        except ImportError:
            self.redis_enabled = False
        
        self.embedding_dim = embedding_dim
        self.ttl = ttl
        self.metrics = {
            "hits": 0,
            "misses": 0,
            "total": 0
        }
    
    def get(self, key):
        """
        从缓存获取数据
        
        参数:
        key: 缓存键
        
        返回:
        缓存值或None
        """
        self.metrics["total"] += 1
        
        # 1. 检查内存缓存
        value = self.memory_cache.get(key)
        if value is not None:
            self.metrics["hits"] += 1
            return value
        
        # 2. 检查Redis缓存
        if self.redis_enabled:
            try:
                redis_value = self.redis_client.get(key)
                if redis_value:
                    # 反序列化
                    value = json.loads(redis_value)
                    # 更新到内存缓存
                    self.memory_cache.set(key, value)
                    self.metrics["hits"] += 1
                    return value
            except Exception as e:
                print(f"Redis读取错误: {e}")
        
        # 缓存未命中
        self.metrics["misses"] += 1
        return None
    
    def set(self, key, value):
        """
        设置缓存
        
        参数:
        key: 缓存键
        value: 缓存值
        """
        # 更新内存缓存
        self.memory_cache.set(key, value)
        
        # 更新Redis缓存
        if self.redis_enabled:
            try:
                # 序列化
                redis_value = json.dumps(value)
                self.redis_client.setex(key, self.ttl, redis_value)
            except Exception as e:
                print(f"Redis写入错误: {e}")
    
    def get_hit_ratio(self):
        """
        获取缓存命中率
        
        返回:
        命中率
        """
        if self.metrics["total"] == 0:
            return 0
        return self.metrics["hits"] / self.metrics["total"]
    
    def clear(self):
        """
        清除缓存
        """
        self.memory_cache.clear()
        if self.redis_enabled:
            try:
                self.redis_client.flushdb()
            except Exception as e:
                print(f"Redis清除错误: {e}")

class BatchProcessor:
    def __init__(self, batch_size=32, max_wait_time=0.1):
        """
        初始化批处理器
        
        参数:
        batch_size: 批处理大小
        max_wait_time: 最大等待时间(秒)
        """
        self.batch_size = batch_size
        self.max_wait_time = max_wait_time
        self.batch_queue = []
        self.callback_map = {}
        self.lock = threading.Lock()
        self.processing = False
        self.last_process_time = time.time()
        
        # 启动批处理线程
        self.thread = threading.Thread(target=self._process_loop, daemon=True)
        self.thread.start()
    
    def add_item(self, item, callback=None):
        """
        添加项目到批处理队列
        
        参数:
        item: 要处理的项目
        callback: 处理完成后的回调函数
        
        返回:
        项目ID
        """
        item_id = str(uuid.uuid4())
        
        with self.lock:
            self.batch_queue.append((item_id, item))
            if callback:
                self.callback_map[item_id] = callback
            
            # 检查是否需要立即处理
            current_time = time.time()
            if (len(self.batch_queue) >= self.batch_size or 
                current_time - self.last_process_time >= self.max_wait_time):
                self._trigger_processing()
        
        return item_id
    
    def _trigger_processing(self):
        """
        触发批处理
        """
        if not self.processing:
            self.processing = True
            # 通知处理线程
            # 实际实现可能需要条件变量或事件
    
    def _process_loop(self):
        """
        批处理循环
        """
        while True:
            items_to_process = []
            
            with self.lock:
                if self.processing and self.batch_queue:
                    # 取出当前批次
                    items_to_process = self.batch_queue[:self.batch_size]
                    self.batch_queue = self.batch_queue[self.batch_size:]
                    self.processing = len(self.batch_queue) >= self.batch_size
                    self.last_process_time = time.time()
            
            if items_to_process:
                try:
                    # 处理批次
                    results = self._process_batch([item[1] for item in items_to_process])
                    
                    # 调用回调
                    for (item_id, _), result in zip(items_to_process, results):
                        if item_id in self.callback_map:
                            try:
                                self.callback_map[item_id](result)
                            except Exception as e:
                                print(f"回调错误: {e}")
                            del self.callback_map[item_id]
                except Exception as e:
                    print(f"批处理错误: {e}")
            
            # 短暂休眠
            time.sleep(0.01)
    
    def _process_batch(self, items):
        """
        处理批次项目
        子类需要实现此方法
        
        参数:
        items: 项目列表
        
        返回:
        处理结果列表
        """
        # 示例实现，实际应用中需要根据具体任务重写
        return [None] * len(items)

# 示例：LLM嵌入批处理器
class EmbeddingBatchProcessor(BatchProcessor):
    def __init__(self, llm_client, batch_size=32, max_wait_time=0.1):
        super().__init__(batch_size, max_wait_time)
        self.llm_client = llm_client
    
    def _process_batch(self, texts):
        """
        批量获取文本嵌入
        
        参数:
        texts: 文本列表
        
        返回:
        嵌入向量列表
        """
        try:
            # 调用LLM的批量嵌入API
            responses = self.llm_client.generate_embeddings_batch(texts)
            return responses
        except Exception as e:
            print(f"嵌入批处理错误: {e}")
            # 返回空嵌入向量作为降级
            embedding_dim = 1024  # 假设嵌入维度
            return [np.zeros(embedding_dim) for _ in texts]

# Kubernetes部署配置示例
apiVersion: apps/v1
kind: Deployment
metadata:
  name: sentiment-analysis-deployment
spec:
  replicas: 10  # 初始副本数
  selector:
    matchLabels:
      app: sentiment-analysis
  template:
    metadata:
      labels:
        app: sentiment-analysis
    spec:
      containers:
      - name: sentiment-analysis
        image: sentiment-analysis:latest
        resources:
          requests:
            memory: "2Gi"
            cpu: "1"
          limits:
            memory: "4Gi"
            cpu: "2"
        env:
        - name: BATCH_SIZE
          value: "64"
        - name: CACHE_SIZE
          value: "100000"
        - name: REDIS_URL
          value: "redis://redis-master:6379/0"
        ports:
        - containerPort: 8080
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: sentiment-analysis-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: sentiment-analysis-deployment
  minReplicas: 5
  maxReplicas: 100
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80