Comparação de técnicas de NLP para Atlas Search de produtos escaláveis

1
import json
2
import requests
3
4
url = "https://asos2.p.rapidapi.com/categories/list"
5
headers = {
6
   "x-rapidapi-key": "YOUR_RAPIDAPI_KEY",
7
   "x-rapidapi-host": "asos2.p.rapidapi.com",
8
}
9
response = requests.request("GET", url, headers=headers)
10
categories = json.loads(response.text)
11
for category in categories:a
12
   url = f"https://asos2.p.rapidapi.com/products/v2/list/{category['id']}"
13
   response = requests.request("GET", url, headers=headers)
14
   products = json.loads(response.text)
15
   # Store products in GCS
16
   bucket = os.environ.get('GS_BUCKET_NAME')
17
   product_filename = f"products_{datetime.now().strftime('%H%m%S')}.json"
18
   write(bucket, str(date.today()) + "/" + product_filename, products)

1
from pyspark.ml.feature import RegexTokenizer, StopWordsRemover
2
3
# Instantiate and set input and output column parameters
4
regexTokenizer = RegexTokenizer(
5
   inputCol="product_name", outputCol="words", pattern="\\W"
6
)
7
remover = StopWordsRemover(inputCol="words", outputCol="filtered")
8
9
# Transform the data
10
regexTokenized = regexTokenizer.transform(df)
11
removedStopWords = remover.transform(regexTokenized)

1
import json
2
3
from pymongo import MongoClient
4
5
client = MongoClient(
6
   "mongodb://<username>:<password>@cluster0.mongodb.net/test?retryWrites=true&w=majority"
7
)
8
db = client.get_database("product_db")
9
products = db.products
10
11
# Fetch data from GCS
12
for product in gcs_data:
13
   products.insert_one(product)

1
from pyspark.ml.feature import IDF, HashingTF
2
3
hashingTF = HashingTF(inputCol="filtered", outputCol="rawFeatures", numFeatures=20)
4
featurizedData = hashingTF.transform(removedStopWords)
5
6
idf = IDF(inputCol="rawFeatures", outputCol="features")
7
idfModel = idf.fit(featurizedData)
8
rescaledData = idfModel.transform(featurizedData)

1
from pyspark.ml.feature import Word2Vec
2
3
word2Vec = Word2Vec(
4
   vectorSize=300, minCount=0, inputCol="filtered", outputCol="features"
5
)
6
model = word2Vec.fit(removedStopWords)
7
result = model.transform(removedStopWords)

1
# Compute IDF values for each term
2
# Compute term frequency in each document for each term
3
# For each query term, do:
4
# For each document containing the term do:
5
# Compute BM25 Score
6
# Sort documents by score

1
def stemming(words):
2
   ps = PorterStemmer()
3
   result = []
4
5
   for w in words:
6
       root_word = ps.stem(w)
7
       result.append(root_word)
8
9
   return result
10
11
stemming_udf = udf(lambda x: stemming(x), ArrayType(elementType=StringType()))
12
preprocessed_data = preprocessed_data.withColumn(
13
"tokens", stemming_udf("filtered_words")
14
)

1
words = preprocessed_data.select(explode("tokens").alias("word"))
2
total_unique_words = words.select("word").distinct().count()
3
total_words = words.count()

1
id_str = concat_ws("", col("_id").cast(StringType()))
2
3
word_count_df = (
4
   preprocessed_data.select("_id", explode("tokens").alias("word"))
5
   .groupBy("word", "_id")
6
   .agg(size(collect_list("word")).alias("occurrence"))
7
   .groupBy("word")
8
   .agg(
9
       map_from_entries(collect_list(struct(id_str.alias("id"), "occurrence"))).alias(
10
           "documents"
11
       )
12
   )
13
   .withColumnRenamed("word", "_id")
14
).rdd.map(lambda row: (row["_id"], row["documents"]))
15
16
inverted_index = word_count_df.collectAsMap()

1
doc_length_df = (
2
   preprocessed_data.select("_id", size("tokens").alias("length"))
3
   .groupBy("_id")
4
   .agg(sum(col("length")).alias("doc_length"))
5
   .withColumn("_id", concat_ws("", col("_id").cast(StringType())))
6
   .rdd.map(lambda row: (row["_id"], row["doc_length"]))
7
)
8
doc_length = doc_length_df.collectAsMap()
9
10
doc_length["word_counter"] = total_unique_words
11
doc_length["doc_counter"] = total_words
12
13
unique_word_count_df = preprocessed_data.select(
14
   "_id",
15
   size(array_distinct("tokens")).alias("num_unique_words"),
16
).withColumn("_id", concat_ws("", col("_id").cast(StringType()))).rdd.map(lambda row: (row["_id"], row["num_unique_words"]))
17
unique_word = unique_word_count_df.collectAsMap()

1
def bm25(doc_length,
2
  inverted_index, 
3
   unique_word, 
4
   word_counter, 
5
   doc_counter, 
6
   query):
7
   k1 = 1.2
8
   b = 0.75
9
   k2 = 100
10
   N = len(doc_length) - 2
11
   avgdl = doc_length["doc_counter"] / N
12
13
   score = defaultdict(int)
14
   for word in query:
15
       if word in inverted_index:
16
           n = len(inverted_index[word])
17
           idf = math.log((N - n + 0.5) / (n + 0.5))
18
           for doc, freq in inverted_index[word].items():
19
               f = freq
20
               qf = query.count(word)
21
               dl = doc_length[doc]
22
               K = k1 * ((1 - b) + b * (float(dl) / float(avgdl)))
23
               R1 = idf * ((f * (k1 + 1)) / (f + K))
24
               R2 = (qf * (k2 + 1)) / (qf + k2)
25
               R = R1 * R2
26
               score[doc] += R
27
       else:
28
           Continue
29
30
   return score

1
from sentence_transformers import SentenceTransformer
2
3
model = SentenceTransformer('bert-base-nli-mean-tokens')
4
sentence_embeddings = model.encode(data['product_name'])