""" Memory Base Service 提供记忆服务的基础功能和共享辅助方法。 """ from datetime import datetime from typing import Optional from app.core.logging_config import get_logger from app.repositories.neo4j.neo4j_connector import Neo4jConnector from app.services.emotion_analytics_service import EmotionAnalyticsService logger = get_logger(__name__) class MemoryBaseService: """记忆服务基类,提供共享的辅助方法""" def __init__(self): self.neo4j_connector = Neo4jConnector() @staticmethod def parse_timestamp(timestamp_value) -> Optional[int]: """ 将时间戳转换为毫秒级时间戳 支持多种输入格式: - Neo4j DateTime 对象 - ISO格式的时间戳字符串 - Python datetime 对象 Args: timestamp_value: 时间戳值(可以是多种类型) Returns: 毫秒级时间戳,如果解析失败则返回None """ if not timestamp_value: return None try: # 处理 Neo4j DateTime 对象 if hasattr(timestamp_value, 'to_native'): dt_object = timestamp_value.to_native() return int(dt_object.timestamp() * 1000) # 处理 Python datetime 对象 if isinstance(timestamp_value, datetime): return int(timestamp_value.timestamp() * 1000) # 处理字符串格式 if isinstance(timestamp_value, str): dt_object = datetime.fromisoformat(timestamp_value.replace("Z", "+00:00")) return int(dt_object.timestamp() * 1000) # 其他情况尝试转换为字符串再解析 dt_object = datetime.fromisoformat(str(timestamp_value).replace("Z", "+00:00")) return int(dt_object.timestamp() * 1000) except (ValueError, TypeError, AttributeError) as e: logger.warning(f"无法解析时间戳: {timestamp_value}, error={str(e)}") return None async def extract_episodic_emotion( self, summary_id: str, end_user_id: str ) -> Optional[str]: """ 提取情景记忆的主要情绪 查询MemorySummary节点关联的Statement节点, 返回emotion_intensity最大的emotion_type。 Args: summary_id: Summary节点的ID end_user_id: 终端用户ID (group_id) Returns: 最大emotion_intensity对应的emotion_type,如果没有则返回None """ try: query = """ MATCH (s:MemorySummary) WHERE elementId(s) = $summary_id AND s.group_id = $group_id MATCH (s)-[:DERIVED_FROM_STATEMENT]->(stmt:Statement) WHERE stmt.emotion_type IS NOT NULL AND stmt.emotion_intensity IS NOT NULL RETURN stmt.emotion_type AS emotion_type, stmt.emotion_intensity AS emotion_intensity ORDER BY emotion_intensity DESC LIMIT 1 """ result = await self.neo4j_connector.execute_query( query, summary_id=summary_id, group_id=end_user_id ) if result and len(result) > 0: emotion_type = result[0].get("emotion_type") logger.info(f"成功提取 summary_id={summary_id} 的情绪: {emotion_type}") return emotion_type else: logger.info(f"summary_id={summary_id} 没有情绪信息") return None except Exception as e: logger.error(f"提取情景记忆情绪时出错: {str(e)}", exc_info=True) return None async def get_episodic_memory_count( self, end_user_id: Optional[str] = None ) -> int: """ 获取情景记忆数量 查询 MemorySummary 节点的数量。 Args: end_user_id: 可选的终端用户ID,用于过滤特定用户的节点 Returns: 情景记忆的数量 """ try: if end_user_id: query = """ MATCH (n:MemorySummary) WHERE n.group_id = $group_id RETURN count(n) as count """ result = await self.neo4j_connector.execute_query(query, group_id=end_user_id) else: query = """ MATCH (n:MemorySummary) RETURN count(n) as count """ result = await self.neo4j_connector.execute_query(query) count = result[0]["count"] if result and len(result) > 0 else 0 logger.debug(f"情景记忆数量: {count} (end_user_id={end_user_id})") return count except Exception as e: logger.error(f"获取情景记忆数量时出错: {str(e)}", exc_info=True) return 0 async def get_explicit_memory_count( self, end_user_id: Optional[str] = None ) -> int: """ 获取显性记忆数量 显性记忆 = 情景记忆(MemorySummary)+ 语义记忆(ExtractedEntity with is_explicit_memory=true) Args: end_user_id: 可选的终端用户ID,用于过滤特定用户的节点 Returns: 显性记忆的数量 """ try: # 1. 获取情景记忆数量 episodic_count = await self.get_episodic_memory_count(end_user_id) # 2. 获取语义记忆数量(ExtractedEntity 且 is_explicit_memory = true) if end_user_id: semantic_query = """ MATCH (e:ExtractedEntity) WHERE e.group_id = $group_id AND e.is_explicit_memory = true RETURN count(e) as count """ semantic_result = await self.neo4j_connector.execute_query( semantic_query, group_id=end_user_id ) else: semantic_query = """ MATCH (e:ExtractedEntity) WHERE e.is_explicit_memory = true RETURN count(e) as count """ semantic_result = await self.neo4j_connector.execute_query(semantic_query) semantic_count = semantic_result[0]["count"] if semantic_result and len(semantic_result) > 0 else 0 # 3. 计算总数 explicit_count = episodic_count + semantic_count logger.debug( f"显性记忆数量: {explicit_count} " f"(情景={episodic_count}, 语义={semantic_count}, end_user_id={end_user_id})" ) return explicit_count except Exception as e: logger.error(f"获取显性记忆数量时出错: {str(e)}", exc_info=True) return 0 async def get_emotional_memory_count( self, end_user_id: Optional[str] = None, statement_count_fallback: int = 0 ) -> int: """ 获取情绪记忆数量 通过 EmotionAnalyticsService 获取情绪标签统计总数。 如果获取失败或没有指定 end_user_id,使用 statement_count_fallback 作为后备。 Args: end_user_id: 可选的终端用户ID statement_count_fallback: 后备方案的数量(通常是 statement 节点数量) Returns: 情绪记忆的数量 """ try: if end_user_id: emotion_service = EmotionAnalyticsService() emotion_data = await emotion_service.get_emotion_tags( end_user_id=end_user_id, emotion_type=None, start_date=None, end_date=None, limit=10 ) emotion_count = emotion_data.get("total_count", 0) logger.debug(f"情绪记忆数量: {emotion_count} (end_user_id={end_user_id})") return emotion_count else: # 如果没有指定 end_user_id,使用后备方案 logger.debug(f"情绪记忆数量: {statement_count_fallback} (使用后备方案)") return statement_count_fallback except Exception as e: logger.warning(f"获取情绪记忆数量失败,使用后备方案: {str(e)}") return statement_count_fallback async def get_forget_memory_count( self, end_user_id: Optional[str] = None, forgetting_threshold: float = 0.3 ) -> int: """ 获取遗忘记忆数量 统计激活值低于遗忘阈值的节点数量(low_activation_nodes)。 查询范围包括:Statement、ExtractedEntity、MemorySummary、Chunk 节点。 Args: end_user_id: 可选的终端用户ID,用于过滤特定用户的节点 forgetting_threshold: 遗忘阈值,默认 0.3 Returns: 遗忘记忆的数量(激活值低于阈值的节点数) """ try: # 构建查询语句 query = """ MATCH (n) WHERE (n:Statement OR n:ExtractedEntity OR n:MemorySummary OR n:Chunk) """ if end_user_id: query += " AND n.group_id = $group_id" query += """ RETURN sum(CASE WHEN n.activation_value IS NOT NULL AND n.activation_value < $threshold THEN 1 ELSE 0 END) as low_activation_nodes """ # 设置查询参数 params = {'threshold': forgetting_threshold} if end_user_id: params['group_id'] = end_user_id # 执行查询 result = await self.neo4j_connector.execute_query(query, **params) # 提取结果 forget_count = result[0]['low_activation_nodes'] if result and len(result) > 0 else 0 forget_count = forget_count or 0 # 处理 None 值 logger.debug( f"遗忘记忆数量: {forget_count} " f"(threshold={forgetting_threshold}, end_user_id={end_user_id})" ) return forget_count except Exception as e: logger.error(f"获取遗忘记忆数量时出错: {str(e)}", exc_info=True) return 0