| 20 | # match with external knowledge for in-context learning |
| 21 | |
| 22 | class KnowledgeExtraction(): |
| 23 | |
| 24 | def __init__(self, file_path, topk=3, keyword_matching_func=bm25): |
| 25 | |
| 26 | # select an attribute in the jsons to embed |
| 27 | self.names = {"matched_attr": "cause_name"} |
| 28 | self.cause_name = self.names["matched_attr"] |
| 29 | |
| 30 | nltk.download('stopwords') |
| 31 | nltk.download('punkt') |
| 32 | nltk.download('averaged_perceptron_tagger') |
| 33 | nltk.download('wordnet') |
| 34 | self.wnl = WordNetLemmatizer() |
| 35 | self.keyword_matching_func = keyword_matching_func |
| 36 | |
| 37 | self.topk = topk |
| 38 | |
| 39 | self.corpus, self.preprocessed_corpus, self.matched_attr, self.stop_words = self.knowledge_load(file_path) |
| 40 | |
| 41 | def knowledge_load(self, file_path): |
| 42 | |
| 43 | # file_path = "/bmtools/tools/db_diag/root_causes_dbmind.jsonl" |
| 44 | with open(str(os.getcwd()) + file_path, 'r') as f: |
| 45 | data = json.load(f) |
| 46 | self.corpus = [example["desc"] for example in data] |
| 47 | self.matched_attr = [example[self.names["matched_attr"]] for example in data] |
| 48 | self.stop_words = set(stopwords.words('english')) |
| 49 | |
| 50 | self.preprocessed_corpus = [] |
| 51 | for c in self.corpus: |
| 52 | word_tokens = word_tokenize(c) |
| 53 | self.preprocessed_corpus.append([self.wnl.lemmatize(w,pos='n') for w in word_tokens if not w in self.stop_words]) # remove useless words and standardize words |
| 54 | |
| 55 | return self.corpus, self.preprocessed_corpus, self.matched_attr, self.stop_words |
| 56 | |
| 57 | def match(self, detailed_metrics): |
| 58 | |
| 59 | metrics_str = [] |
| 60 | for metrics in detailed_metrics.keys(): |
| 61 | metrics = metrics.replace("_"," ") |
| 62 | word_tokens = word_tokenize(metrics) |
| 63 | metrics_str.extend([self.wnl.lemmatize(w,pos='n') for w in word_tokens if not w in self.stop_words]) |
| 64 | metrics_str = list(set(metrics_str)) |
| 65 | |
| 66 | best_index = self.keyword_matching_func(self.topk, metrics_str, self.preprocessed_corpus) |
| 67 | best_docs = [self.corpus[b] for b in best_index] |
| 68 | best_names = [self.matched_attr[b] for b in best_index] |
| 69 | docs_str = "" |
| 70 | print("Best docs: ", best_docs) |
| 71 | for i, docs in enumerate(best_docs): |
| 72 | docs_str = docs_str + "{}: ".format(best_names[i]) + docs + "\n\n" |
| 73 | print("docs_str: ", docs_str) |
| 74 | |
| 75 | return docs_str |
| 76 | |
| 77 | |
| 78 | if __name__ == "__main__": |
no outgoing calls
no test coverage detected