Update app.py

app.py

@@ -1,64 +1,137 @@
+import os
+import json
+import re
+import math
+import nltk
+from collections import defaultdict, Counter
+from nltk.tokenize import word_tokenize
+from nltk.stem import PorterStemmer, WordNetLemmatizer
 import gradio as gr
-from huggingface_hub import InferenceClient
 
-"""
-For more information on `huggingface_hub` Inference API support, please check the docs: https://huggingface.co/docs/huggingface_hub/v0.22.2/en/guides/inference
-"""
-client = InferenceClient("HuggingFaceH4/zephyr-7b-beta")
+nltk.data.path.append("./nltk_data")
 
+with open("docs.json", "r", encoding="utf-8") as f:
+    docs_ds = json.load(f)
 
-def respond(
-    message,
-    history: list[tuple[str, str]],
-    system_message,
-    max_tokens,
-    temperature,
-    top_p,
-):
-    messages = [{"role": "system", "content": system_message}]
+with open("queries.json", "r", encoding="utf-8") as f:
+    queries_ds = json.load(f)
 
-    for val in history:
-        if val[0]:
-            messages.append({"role": "user", "content": val[0]})
-        if val[1]:
-            messages.append({"role": "assistant", "content": val[1]})
+documents = {int(doc["doc_id"]): doc["text"] for doc in docs_ds}
+queries = {int(q["query_id"]): q["text"] for q in queries_ds}
 
-    messages.append({"role": "user", "content": message})
+stop_words = {"a", "is", "the", "of", "all", "and", "to", "can", "be", "as", "once", "for", "at", "am", "are", "has", "have", "had", "up", "his", "her", "in", "on", "no", "we", "do"}
 
-    response = ""
+inverted_index = defaultdict(set)
+positional_index = defaultdict(lambda: defaultdict(list))
+tf_idf_vectors = defaultdict(dict)
+idf_scores = {}
 
-    for message in client.chat_completion(
-        messages,
-        max_tokens=max_tokens,
-        stream=True,
-        temperature=temperature,
-        top_p=top_p,
-    ):
-        token = message.choices[0].delta.content
+def process_documents(documents):
+    stemmer = PorterStemmer()
+    lemmatizer = WordNetLemmatizer()
+    doc_freq = defaultdict(int)
+    term_freqs = {}
+    for doc_id, text in documents.items():
+        words = word_tokenize(text.lower())
+        filtered_words = [lemmatizer.lemmatize(w) for w in words if w.isalnum() and w not in stop_words]
+        term_counts = Counter(filtered_words)
+        term_freqs[doc_id] = term_counts
+        for pos, word in enumerate(filtered_words):
+            stemmed = stemmer.stem(word)
+            inverted_index[stemmed].add(doc_id)
+            positional_index[stemmed][doc_id].append(pos)
+        for word in set(filtered_words):
+            doc_freq[word] += 1
+    total_docs = len(documents)
+    for word, df in doc_freq.items():
+        idf_scores[word] = math.log(total_docs / df)
+    for doc_id, term_counts in term_freqs.items():
+        tf_idf_vectors[doc_id] = {word: count * idf_scores[word] for word, count in term_counts.items()}
 
-        response += token
-        yield response
+def execute_boolean_query(query, documents):
+    query = query.lower()
+    tokens = query.split()
+    stemmer = PorterStemmer()
+    operators = {'and', 'or', 'not'}
+    term_stack = []
+    operator_stack = []
+    for token in tokens:
+        if token in operators:
+            operator_stack.append(token)
+        else:
+            stemmed_word = stemmer.stem(token)
+            term_set = inverted_index.get(stemmed_word, set())
+            term_stack.append(term_set)
+    while 'not' in operator_stack:
+        idx = operator_stack.index('not')
+        term_stack[idx] = set(documents.keys()) - term_stack[idx]
+        operator_stack.pop(idx)
+    while operator_stack:
+        op = operator_stack.pop(0)
+        left = term_stack.pop(0)
+        right = term_stack.pop(0)
+        if op == 'and':
+            term_stack.insert(0, left & right)
+        elif op == 'or':
+            term_stack.insert(0, left | right)
+    return sorted(term_stack[0]) if term_stack else []
 
+def execute_proximity_query(query):
+    match = re.match(r'(\w+)\s+(\w+)\s*/\s*(\d+)', query)
+    if not match:
+        return []
+    word1, word2, k = match.groups()
+    k = int(k)
+    stemmer = PorterStemmer()
+    word1 = stemmer.stem(word1.lower())
+    word2 = stemmer.stem(word2.lower())
+    result_docs = set()
+    if word1 in positional_index and word2 in positional_index:
+        for doc_id in positional_index[word1]:
+            if doc_id in positional_index[word2]:
+                positions1 = positional_index[word1][doc_id]
+                positions2 = positional_index[word2][doc_id]
+                if any(0 < abs(p1 - p2) <= k for p1 in positions1 for p2 in positions2):
+                    result_docs.add(doc_id)
+    return sorted(result_docs)
 
-"""
-For information on how to customize the ChatInterface, peruse the gradio docs: https://www.gradio.app/docs/chatinterface
-"""
-demo = gr.ChatInterface(
-    respond,
-    additional_inputs=[
-        gr.Textbox(value="You are a friendly Chatbot.", label="System message"),
-        gr.Slider(minimum=1, maximum=2048, value=512, step=1, label="Max new tokens"),
-        gr.Slider(minimum=0.1, maximum=4.0, value=0.7, step=0.1, label="Temperature"),
-        gr.Slider(
-            minimum=0.1,
-            maximum=1.0,
-            value=0.95,
-            step=0.05,
-            label="Top-p (nucleus sampling)",
-        ),
-    ],
-)
+def evaluate_cosine_similarity_score(vec1, vec2):
+    common = set(vec1.keys()) & set(vec2.keys())
+    dot_product = sum(vec1[k] * vec2[k] for k in common)
+    norm1 = math.sqrt(sum(v**2 for v in vec1.values()))
+    norm2 = math.sqrt(sum(v**2 for v in vec2.values()))
+    if norm1 == 0 or norm2 == 0:
+        return 0.0
+    return dot_product / (norm1 * norm2)
 
+def process_query(user_input_query):
+    lemmatizer = WordNetLemmatizer()
+    tokens = word_tokenize(user_input_query.lower())
+    filtered = [lemmatizer.lemmatize(w) for w in tokens if w.isalnum() and w not in stop_words]
+    query_counts = Counter(filtered)
+    return {w: query_counts[w] * idf_scores.get(w, 0) for w in query_counts}
 
-if __name__ == "__main__":
-    demo.launch()
+def execute_vsm_query(user_input_query, alpha=0.001):
+    query_vector = process_query(user_input_query)
+    scores = {}
+    for doc_id, doc_vector in tf_idf_vectors.items():
+        sim = evaluate_cosine_similarity_score(query_vector, doc_vector)
+        if sim >= alpha:
+            scores[doc_id] = sim
+    return sorted(scores, key=scores.get, reverse=True)
+
+def chat(query, method):
+    if not query:
+        return "Query cannot be empty"
+    if method == "Boolean":
+        result = execute_boolean_query(query, documents)
+    elif method == "Proximity":
+        result = execute_proximity_query(query)
+    else:
+        result = execute_vsm_query(query)
+    return f"Result-set: {result}"
+
+process_documents(documents)
+
+demo = gr.Interface(fn=chat, inputs=["text", gr.Radio(["Boolean", "Proximity", "Vector Space Model"], label="Model")], outputs="text")
+demo.launch()