File size: 3,700 Bytes
929a283
93452e4
929a283
 
93452e4
929a283
25ba997
 
 
 
929a283
796b144
929a283
0bdc9aa
796b144
 
0bdc9aa
796b144
929a283
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
0bdc9aa
 
 
 
 
929a283
 
d702381
929a283
 
d702381
929a283
0bdc9aa
929a283
 
 
 
 
 
 
 
 
0bdc9aa
 
 
 
929a283
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
0bdc9aa
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
929a283
 
3d70771
929a283
 
 
 
 
97c8253
 
 
3d70771
3d00632
0bdc9aa
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
import gradio as gr
import faiss
import numpy as np
import openai
from sentence_transformers import SentenceTransformer
from nltk.tokenize import sent_tokenize
import nltk

# Download the required NLTK data
nltk.download('punkt')


# Load the Ubuntu manual from a .txt file
try:
    with open("ubuntu_manual.txt", "r", encoding="utf-8") as file:
        full_text = file.read()
except FileNotFoundError:
    raise FileNotFoundError("The file ubuntu_manual.txt was not found.")

# Function to chunk the text into smaller pieces
def chunk_text(text, chunk_size=500):  # Larger chunks
    sentences = sent_tokenize(text)
    chunks = []
    current_chunk = []

    for sentence in sentences:
        if len(current_chunk) + len(sentence.split()) <= chunk_size:
            current_chunk.append(sentence)
        else:
            chunks.append(" ".join(current_chunk))
            current_chunk = [sentence]

    if current_chunk:
        chunks.append(" ".join(current_chunk))

    return chunks

# Apply chunking to the entire text
manual_chunks = chunk_text(full_text, chunk_size=500)

# Load your FAISS index
try:
    # Load your FAISS index
    index = faiss.read_index("manual_chunked_faiss_index_500.bin")
except Exception as e:
    raise RuntimeError(f"Failed to load FAISS index: {e}")

# Load your embedding model
embedding_model = SentenceTransformer('FridayMaster/fine_tune_embedding')

# OpenAI API key
openai.api_key = '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'


# Function to create embeddings
def embed_text(text_list):
    return np.array(embedding_model.encode(text_list), dtype=np.float32)

# Function to retrieve relevant chunks for a user query
def retrieve_chunks(query, k=5):
    query_embedding = embed_text([query])
    
    # Search the FAISS index
    try:
        distances, indices = index.search(query_embedding, k=k)
    except Exception as e:
        raise RuntimeError(f"FAISS search failed: {e}")
    
    # Check if indices are valid
    if len(indices[0]) == 0:
        return []

    # Ensure indices are within bounds
    valid_indices = [i for i in indices[0] if i < len(manual_chunks)]
    if not valid_indices:
        return []

    # Retrieve relevant chunks
    relevant_chunks = [manual_chunks[i] for i in valid_indices]
    return relevant_chunks

# Function to truncate long inputs
def truncate_input(text, max_length=512):
    tokens = generator_tokenizer.encode(text, truncation=True, max_length=max_length, return_tensors="pt")
    return tokens

# Function to perform RAG: Retrieve chunks and generate a response
def rag_response(query, k=5, max_new_tokens=150):
    try:
        # Step 1: Retrieve relevant chunks
        relevant_chunks = retrieve_chunks(query, k=k)
        
        if not relevant_chunks:
            return "Sorry, I couldn't find relevant information."

        # Step 2: Combine the query with retrieved chunks
        augmented_input = query + "\n" + "\n".join(relevant_chunks)
        
        # Truncate and encode the input
        inputs = truncate_input(augmented_input)
        
        # Generate response
        outputs = generator_model.generate(inputs, max_new_tokens=max_new_tokens)
        generated_text = generator_tokenizer.decode(outputs[0], skip_special_tokens=True)
        
        return generated_text
    except Exception as e:
        return f"An error occurred: {e}"

# Gradio Interface
iface = gr.Interface(
    fn=rag_response,
    inputs="text",
    outputs="text",
    title="RAG Chatbot with FAISS and GPT-3.5",
    description="Ask me anything!"
)

if __name__ == "__main__":
    iface.launch()