| | 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}") |
