|
|
import fasttext
|
|
|
import streamlit as st
|
|
|
import numpy as np
|
|
|
import pandas as pd
|
|
|
from gensim.models import Word2Vec
|
|
|
from sklearn.metrics.pairwise import cosine_similarity
|
|
|
import plotly.express as px
|
|
|
import plotly.graph_objects as go
|
|
|
from collections import Counter
|
|
|
import os
|
|
|
import glob
|
|
|
|
|
|
|
|
|
class UnifiedVectorModel:
|
|
|
def __init__(self, backend_model, model_type="w2v"):
|
|
|
self.model = backend_model
|
|
|
self.model_type = model_type.lower()
|
|
|
|
|
|
if self.model_type == "w2v":
|
|
|
self.wv = backend_model.wv
|
|
|
self.key_to_index = self.wv.key_to_index
|
|
|
self.vector_size = self.wv.vector_size
|
|
|
self._words = set(self.wv.key_to_index.keys())
|
|
|
|
|
|
elif self.model_type == "ft":
|
|
|
|
|
|
self.key_to_index = {word: i for i, word in enumerate(backend_model.get_words())}
|
|
|
self.vector_size = backend_model.get_dimension()
|
|
|
self._words = set(self.key_to_index.keys())
|
|
|
else:
|
|
|
raise ValueError("model_type must be 'w2v' or 'ft'")
|
|
|
|
|
|
def __contains__(self, word):
|
|
|
return word in self._words
|
|
|
|
|
|
def __getitem__(self, word):
|
|
|
if self.model_type == "w2v":
|
|
|
return self.wv[word]
|
|
|
elif self.model_type == "ft":
|
|
|
return self.model.get_word_vector(word)
|
|
|
|
|
|
def most_similar(self, positive=None, negative=None, topn=10):
|
|
|
from sklearn.metrics.pairwise import cosine_similarity
|
|
|
|
|
|
if not positive:
|
|
|
positive = []
|
|
|
if not negative:
|
|
|
negative = []
|
|
|
|
|
|
try:
|
|
|
if self.model_type == "w2v":
|
|
|
return self.wv.most_similar(positive=positive, negative=negative, topn=topn)
|
|
|
|
|
|
elif self.model_type == "ft":
|
|
|
vec = np.zeros(self.vector_size)
|
|
|
for w in positive:
|
|
|
if w in self:
|
|
|
vec += self[w]
|
|
|
else:
|
|
|
continue
|
|
|
for w in negative:
|
|
|
if w in self:
|
|
|
vec -= self[w]
|
|
|
else:
|
|
|
continue
|
|
|
|
|
|
if np.allclose(vec, 0):
|
|
|
return []
|
|
|
|
|
|
words = list(self._words)
|
|
|
vectors = np.array([self[w] for w in words])
|
|
|
|
|
|
sims = cosine_similarity([vec], vectors)[0]
|
|
|
best = np.argsort(sims)[::-1][:topn + len(positive) + len(negative)]
|
|
|
|
|
|
result = []
|
|
|
for i in best:
|
|
|
word = words[i]
|
|
|
if word not in positive and word not in negative:
|
|
|
result.append((word, float(sims[i])))
|
|
|
if len(result) >= topn:
|
|
|
break
|
|
|
return result
|
|
|
|
|
|
except Exception as e:
|
|
|
print(f"Error in most_similar: {e}")
|
|
|
return []
|
|
|
|
|
|
def similar_by_vector(self, vector, topn=10):
|
|
|
from sklearn.metrics.pairwise import cosine_similarity
|
|
|
|
|
|
words = list(self._words)
|
|
|
vectors = np.array([self[w] for w in words])
|
|
|
sims = cosine_similarity([vector], vectors)[0]
|
|
|
best = np.argsort(sims)[::-1][:topn]
|
|
|
|
|
|
return [(words[i], float(sims[i])) for i in best]
|
|
|
|
|
|
def get_words(self):
|
|
|
return list(self._words)
|
|
|
|
|
|
@property
|
|
|
def vectors(self):
|
|
|
if not hasattr(self, '_cached_vectors'):
|
|
|
words = list(self._words)
|
|
|
self._cached_words = words
|
|
|
self._cached_vectors = np.array([self[w] for w in words])
|
|
|
return self._cached_vectors
|
|
|
|
|
|
@property
|
|
|
def index_to_key(self):
|
|
|
if not hasattr(self, '_index_to_key'):
|
|
|
self._index_to_key = list(self._words)
|
|
|
return self._index_to_key
|
|
|
|
|
|
|
|
|
@st.cache_resource
|
|
|
def load_model(model_path):
|
|
|
try:
|
|
|
if model_path.endswith(".model"):
|
|
|
raw_model = Word2Vec.load(model_path)
|
|
|
current_model = UnifiedVectorModel(raw_model, model_type="w2v")
|
|
|
|
|
|
elif model_path.endswith(".bin"):
|
|
|
raw_model = fasttext.load_model(model_path)
|
|
|
current_model = UnifiedVectorModel(raw_model, model_type="ft")
|
|
|
else:
|
|
|
raise ValueError(f"wrong path format")
|
|
|
return current_model
|
|
|
except Exception as e:
|
|
|
st.error(f"error loading model {model_path}: {e}")
|
|
|
return None
|
|
|
|
|
|
|
|
|
MODELS_DIR = "models"
|
|
|
|
|
|
if not os.path.exists(MODELS_DIR):
|
|
|
st.error(f"Folder `{MODELS_DIR}` not found.")
|
|
|
st.stop()
|
|
|
|
|
|
model_files = []
|
|
|
for ext in ["*.bin", "*.model", "*.vec"]:
|
|
|
model_files.extend(glob.glob(os.path.join(MODELS_DIR, ext)))
|
|
|
model_files = [f for f in model_files if os.path.isfile(f)]
|
|
|
model_names = [os.path.basename(f) for f in model_files]
|
|
|
|
|
|
if len(model_names) == 0:
|
|
|
st.error(f"No models in folder `{MODELS_DIR}` (.bin, .model, .vec).")
|
|
|
st.info("Supported formats: Word2Vec (binary/text), FastText.")
|
|
|
st.stop()
|
|
|
|
|
|
selected_model_name = st.sidebar.selectbox(
|
|
|
"Choose pretrained model",
|
|
|
model_names
|
|
|
)
|
|
|
|
|
|
selected_model_path = os.path.join(MODELS_DIR, selected_model_name)
|
|
|
|
|
|
st.sidebar.info(f"loading: `{selected_model_name}`")
|
|
|
|
|
|
model = load_model(selected_model_path)
|
|
|
|
|
|
if model is None:
|
|
|
st.stop()
|
|
|
else:
|
|
|
st.sidebar.success(f"Model '{selected_model_name}' loaded")
|
|
|
st.sidebar.write(f"Voc size: {len(model.key_to_index):,}")
|
|
|
st.sidebar.write(f"Vector size: {model.vector_size}")
|
|
|
|
|
|
def analogy_accuracy(model, file_name):
|
|
|
right = 0
|
|
|
count = 0
|
|
|
results = []
|
|
|
with open(file_name, encoding='utf-8') as file:
|
|
|
for line in file:
|
|
|
words = line.strip().split()
|
|
|
if len(words) != 4:
|
|
|
continue
|
|
|
try:
|
|
|
most_similar = model.most_similar(positive=[words[0], words[2]], negative=[words[1]], topn=10)
|
|
|
predicted = [x[0] for x in most_similar]
|
|
|
correct = words[3]
|
|
|
if correct in predicted:
|
|
|
rank = predicted.index(correct) + 1
|
|
|
right += 1
|
|
|
else:
|
|
|
rank = None
|
|
|
count += 1
|
|
|
results.append({
|
|
|
"query": f"{words[0]} - {words[1]} + {words[2]}",
|
|
|
"target": correct,
|
|
|
"predicted": predicted[0],
|
|
|
"rank": rank,
|
|
|
"in_top10": bool(rank)
|
|
|
})
|
|
|
except KeyError as e:
|
|
|
continue
|
|
|
accuracy = right / count if count > 0 else 0
|
|
|
return accuracy, results
|
|
|
|
|
|
|
|
|
def avg_similarity(model, file_name):
|
|
|
res = []
|
|
|
with open(file_name, encoding='utf-8') as file:
|
|
|
for line in file:
|
|
|
words = line.strip().split()
|
|
|
try:
|
|
|
vectors = [model[word] for word in words]
|
|
|
except KeyError:
|
|
|
continue
|
|
|
sims = cosine_similarity(vectors)
|
|
|
for i in range(len(words) - 1):
|
|
|
for j in range(i + 1, len(words)):
|
|
|
res.append(sims[i][j])
|
|
|
return sum(res) / len(res) if res else 0
|
|
|
|
|
|
|
|
|
def projection(word_vec, axis):
|
|
|
axis_norm = axis / np.linalg.norm(axis)
|
|
|
return np.dot(word_vec, axis_norm)
|
|
|
|
|
|
|
|
|
def get_projection_row(model, axis):
|
|
|
words = list(model.key_to_index.keys())
|
|
|
projections = [(word, projection(model[word], axis)) for word in words]
|
|
|
projections = sorted(projections, key=lambda x: x[1])
|
|
|
return projections
|
|
|
|
|
|
|
|
|
st.title("Vector embeddings")
|
|
|
|
|
|
tab1, tab2, tab3, tab4, tab5 = st.tabs([
|
|
|
"Vector ariphmetics",
|
|
|
"Semantic consistency",
|
|
|
"Semantic axis",
|
|
|
"Distribution analysis",
|
|
|
"Report"
|
|
|
])
|
|
|
|
|
|
with tab1:
|
|
|
st.header("Vector ariphmetics")
|
|
|
expr = st.text_input("Insert expression", value="рубль - россия + сша")
|
|
|
|
|
|
if st.button("Compute"):
|
|
|
words = expr.replace('+', ' + ').replace('-', ' - ').split()
|
|
|
positive, negative = [], []
|
|
|
current = 'pos'
|
|
|
|
|
|
for w in words:
|
|
|
if w == '+':
|
|
|
current = 'pos'
|
|
|
elif w == '-':
|
|
|
current = 'neg'
|
|
|
else:
|
|
|
(positive if current == 'pos' else negative).append(w)
|
|
|
|
|
|
missing = [w for w in positive + negative if w not in model]
|
|
|
if missing:
|
|
|
st.warning(f"Words not found in voc: {', '.join(missing)}")
|
|
|
st.stop()
|
|
|
|
|
|
try:
|
|
|
similar = model.most_similar(
|
|
|
positive=positive,
|
|
|
negative=negative,
|
|
|
topn=10
|
|
|
)
|
|
|
|
|
|
st.write("### Result:")
|
|
|
result_words = [f"{w} ({s:.3f})" for w, s in similar]
|
|
|
st.write("Nearest words: " + ", ".join(result_words))
|
|
|
|
|
|
st.write("### In-between steps")
|
|
|
|
|
|
cum_vec = np.zeros(model.vector_size)
|
|
|
|
|
|
steps_data = []
|
|
|
|
|
|
for i in range(len(positive)):
|
|
|
cum_vec += model[w]
|
|
|
nearest = model.most_similar(positive=positive[:i + 1], topn=1)
|
|
|
steps_data.append({
|
|
|
"step": f"+ {positive[i]}",
|
|
|
"nearest word": nearest[0][0],
|
|
|
"similarity": nearest[0][1]
|
|
|
})
|
|
|
|
|
|
for i in range(len(negative)):
|
|
|
cum_vec -= model[w]
|
|
|
nearest = model.most_similar(positive=positive, negative=negative[:i + 1], topn=1)
|
|
|
steps_data.append({
|
|
|
"step": f"- {negative[i]}",
|
|
|
"nearest word": nearest[0][0],
|
|
|
"similarity": nearest[0][1]
|
|
|
})
|
|
|
|
|
|
df_steps = pd.DataFrame(steps_data)
|
|
|
st.dataframe(df_steps[["step", "nearest word", "similarity"]])
|
|
|
|
|
|
result_word = similar[0][0]
|
|
|
fig = px.scatter(
|
|
|
x=[cum_vec[0]], y=[cum_vec[1]],
|
|
|
text=[result_word],
|
|
|
title="Result (first 2 components)"
|
|
|
)
|
|
|
fig.update_traces(textposition='top center', marker=dict(size=12, color='red'))
|
|
|
st.plotly_chart(fig)
|
|
|
|
|
|
except Exception as e:
|
|
|
st.error(f"Error computing: {e}")
|
|
|
|
|
|
with tab2:
|
|
|
st.header("Similarity calculator")
|
|
|
col1, col2 = st.columns(2)
|
|
|
with col1:
|
|
|
word1 = st.text_input("word 1", value="мужчина")
|
|
|
with col2:
|
|
|
word2 = st.text_input("word 2", value="женщина")
|
|
|
|
|
|
if st.button("Compute similarity"):
|
|
|
try:
|
|
|
v1, v2 = model[word1], model[word2]
|
|
|
sim = cosine_similarity([v1], [v2])[0][0]
|
|
|
st.metric("Cosine similarity", f"{sim:.4f}")
|
|
|
|
|
|
st.write("### Nearest neighbors graph")
|
|
|
neighbors = model.most_similar(word1, topn=5) + model.most_similar(word2, topn=5)
|
|
|
nodes = list(set([word1, word2] + [n[0] for n in neighbors]))
|
|
|
edges = [(word1, n[0]) for n in model.most_similar(word1, topn=5)] + \
|
|
|
[(word2, n[0]) for n in model.most_similar(word2, topn=5)]
|
|
|
|
|
|
G = go.Figure()
|
|
|
pos = np.random.rand(len(nodes), 2) * 2 - 1
|
|
|
node_x = pos[:, 0]
|
|
|
node_y = pos[:, 1]
|
|
|
|
|
|
for edge in edges:
|
|
|
x0, y0 = pos[nodes.index(edge[0])]
|
|
|
x1, y1 = pos[nodes.index(edge[1])]
|
|
|
G.add_trace(go.Scatter(x=[x0, x1], y=[y0, y1], mode='lines', line=dict(width=1, color='gray'), showlegend=False))
|
|
|
|
|
|
G.add_trace(go.Scatter(x=node_x, y=node_y, mode='text+markers',
|
|
|
marker=dict(size=10, color='lightblue'),
|
|
|
text=nodes, textposition="top center"))
|
|
|
G.update_layout(title="Semantic links graph", showlegend=False)
|
|
|
st.plotly_chart(G)
|
|
|
|
|
|
except KeyError as e:
|
|
|
st.error(f"Word not found: {e}")
|
|
|
|
|
|
with tab3:
|
|
|
st.header("Semantic axis projection")
|
|
|
col1, col2 = st.columns(2)
|
|
|
with col1:
|
|
|
pos_axis = st.text_input("positive", value="мужчина")
|
|
|
with col2:
|
|
|
neg_axis = st.text_input("negative", value="женщина")
|
|
|
|
|
|
if st.button("Build axis"):
|
|
|
try:
|
|
|
pos_vec = model[pos_axis]
|
|
|
neg_vec = model[neg_axis]
|
|
|
axis = pos_vec - neg_vec
|
|
|
|
|
|
projections = get_projection_row(model, axis)
|
|
|
top_pos = projections[-10:][::-1]
|
|
|
top_neg = projections[:10]
|
|
|
|
|
|
st.write(f"Axis: **{pos_axis} – {neg_axis}**")
|
|
|
st.write("### Top 10 positive:")
|
|
|
st.write(", ".join([f"{w} ({p:.3f})" for w, p in top_pos]))
|
|
|
|
|
|
st.write("### Top 10 negative:")
|
|
|
st.write(", ".join([f"{w} ({p:.3f})" for w, p in top_neg]))
|
|
|
|
|
|
df_proj = pd.DataFrame(top_pos + top_neg, columns=["word", "projection"])
|
|
|
fig = px.bar(df_proj, x="projection", y="word", orientation='h', title=f"Projection on axis: {pos_axis}–{neg_axis}")
|
|
|
st.plotly_chart(fig)
|
|
|
|
|
|
except KeyError as e:
|
|
|
st.error(f"Error: {e}")
|
|
|
|
|
|
with tab4:
|
|
|
st.header("Distance distribution analysis")
|
|
|
all_vectors = model.vectors
|
|
|
sample = all_vectors[np.random.choice(all_vectors.shape[0], 1000, replace=False)]
|
|
|
|
|
|
dists = cosine_similarity(sample)
|
|
|
np.fill_diagonal(dists, 0)
|
|
|
flat_dists = dists.flatten()
|
|
|
flat_dists = flat_dists[flat_dists > 0]
|
|
|
|
|
|
fig = px.histogram(flat_dists, nbins=50, title="Cosine similarity distribution between random words")
|
|
|
st.plotly_chart(fig)
|
|
|
|
|
|
st.metric("Mean similarity", f"{np.mean(flat_dists):.3f}")
|
|
|
st.metric("Std deviation", f"{np.std(flat_dists):.3f}")
|
|
|
|
|
|
with tab5:
|
|
|
st.header("Report")
|
|
|
|
|
|
st.subheader("1. Analogy rate")
|
|
|
analogies_file = "data/analogy.txt"
|
|
|
if os.path.exists(analogies_file):
|
|
|
acc, results = analogy_accuracy(model, analogies_file)
|
|
|
st.metric("Analogy accuracy (in top 10)", f"{acc:.2%}")
|
|
|
st.dataframe(pd.DataFrame(results))
|
|
|
else:
|
|
|
st.warning("File `analogy.txt` not found.")
|
|
|
|
|
|
st.subheader("2. Average synonyms similarity")
|
|
|
sim_file = "data/synonyms.txt"
|
|
|
if os.path.exists(sim_file):
|
|
|
avg_sim = avg_similarity(model, sim_file)
|
|
|
st.metric("Average similarity", f"{avg_sim:.4f}")
|
|
|
else:
|
|
|
st.warning("File `similarity_words.txt` not found.")
|
|
|
|
|
|
st.subheader("3. Average antonyms similarity")
|
|
|
sim_file = "data/antonyms.txt"
|
|
|
if os.path.exists(sim_file):
|
|
|
avg_sim = avg_similarity(model, sim_file)
|
|
|
st.metric("Average similarity", f"{avg_sim:.4f}")
|
|
|
else:
|
|
|
st.warning("File `similarity_words.txt` not found.")
|
|
|
|
|
|
st.subheader("4. Heatmap for nearest words")
|
|
|
query_words = st.text_input("Enter words", value="мужчина женщина мальчик девочка").split()
|
|
|
if st.button("Build heatmap"):
|
|
|
try:
|
|
|
vectors = [model[w] for w in query_words]
|
|
|
sims = cosine_similarity(vectors)
|
|
|
fig = px.imshow(sims, x=query_words, y=query_words, color_continuous_scale="Blues", title="Similarity heatmap")
|
|
|
st.plotly_chart(fig)
|
|
|
except KeyError as e:
|
|
|
st.error(f"Error: {e}")
|
|
|
|
|
|
st.subheader("5. 2D projection")
|
|
|
sample_words = st.text_input("Input words", value="мужчина женщина мальчик девочка")
|
|
|
word_list = sample_words.split()
|
|
|
if st.button("Show clusters"):
|
|
|
try:
|
|
|
from sklearn.manifold import TSNE
|
|
|
vectors = np.array([model[w] for w in word_list])
|
|
|
tsne = TSNE(n_components=2, perplexity=len(vectors) - 1, random_state=42)
|
|
|
embedded = tsne.fit_transform(vectors)
|
|
|
|
|
|
fig = px.scatter(x=embedded[:, 0], y=embedded[:, 1], text=word_list, title="words projection")
|
|
|
fig.update_traces(textposition='top center')
|
|
|
st.plotly_chart(fig)
|
|
|
except KeyError as e:
|
|
|
st.error(f"Word not found: {e}") |
