[公开]2026年度北京文旅投诉分转服务和文旅行业投诉大数据平台服务中标公告

020/cog

/README.md
import os
import sys
import numpy as np
from scipy.spatial.distance import cosine
from scipy.sparse.linalg import svds

from cog.utils import get_model_path, get_cache_path, get_save_path, get_logger

logger = get_logger(__name__)

class Embedding(object):

def __init__(self, model_path, cache_path, save_path, use_cache=False, use_save=False, **kwargs):
self.model_path = model_path
self.cache_path = cache_path
self.save_path = save_path
self.use_cache = use_cache
self.use_save = use_save

self.embeddings = None
self.word2idx = None
self.idx2word = None

def _init_embeddings(self):
"""
Load embeddings from model path
"""
if self.use_cache:
self.embeddings, self.word2idx, self.idx2word = self._load_from_cache()
else:
self.embeddings, self.word2idx, self.idx2word = self._load_embeddings()

def _load_from_cache(self):
cache_path = get_cache_path(self.cache_path)
if not os.path.exists(cache_path):
logger.warning(f'{cache_path} does not exist, will download it.')
embeddings, word2idx, idx2word = self._load_embeddings()
np.savez(cache_path, embeddings=embeddings, word2idx=word2idx, idx2word=idx2word)
else:
embeddings, word2idx, idx2word = np.load(cache_path)['embeddings'], np.load(cache_path)['word2idx'], np.load(cache_path)['idx2word']
return embeddings, word2idx, idx2word

def _load_embeddings(self):
if not os.path.exists(self.model_path):
logger.error(f'Error: Model path {self.model_path} does not exist. Please provide a valid model path.')
sys.exit(-1)
embeddings, word2idx, idx2word = self._load_from_file()
return embeddings, word2idx, idx2word

def _load_from_file(self):
embeddings, word2idx, idx2word = [], {}, {}
with open(self.model_path, 'r', encoding='utf-8') as f:
for line in f:
if len(line.strip()) == 0:
continue
tokens = line.strip().split(' ')
word = tokens[0]
vec = np.array([float(v) for v in tokens[1:]])
embeddings.append(vec)
idx2word[len(embeddings) - 1] = word
word2idx[word] = len(embeddings) - 1
embeddings = np.array(embeddings)
return embeddings, word2idx, idx2word

def get_embedding(self, word):
"""
Get embedding for a word
"""
if self.embeddings is None:
self._init_embeddings()
if word not in self.word2idx:
return None
idx = self.word2idx[word]
return self.embeddings[idx]

def get_embeddings(self):
"""
Get embeddings for all words
"""
if self.embeddings is None:
self._init_embeddings()
return self.embeddings

def get_word2idx(self):
"""
Get word2idx mapping
"""
if self.embeddings is None:
self._init_embeddings()
return self.word2idx

def get_idx2word(self):
"""
Get idx2word mapping
"""
if self.embeddings is None:
self._init_embeddings()
return self.idx2word

class Similarity(object):

def __init__(self, emb, metric='cosine', top_k=5):
self.emb = emb
self.metric = metric
self.top_k = top_k

def get_similar_words(self, word, top_k=None):
"""
Get similar words for a word
"""
if top_k is None:
top_k = self.top_k
if word not in self.emb.word2idx:
return None
idx = self.emb.word2idx[word]
emb = self.emb.embeddings[idx]
distances = []
for i in range(self.emb.embeddings.shape[0]):
if i != idx:
distance = cosine(emb, self.emb.embeddings[i])
distances.append(distance)
sorted_indices = sorted(range(len(distances)), key=lambda k: distances[k])[:top_k]
return [self.emb.idx2word[i] for i in sorted_indices]

class Similarity2(Similarity):
def __init__(self, emb, metric='cosine', top_k=5):
super().__init__(emb, metric, top_k)

def get_similar_words(self, word, top_k=None):
if top_k is None:
top_k = self.top_k
if word not in self.emb.word2idx:
return None
idx = self.emb.word2idx[word]
emb = self.emb.embeddings[idx]
distances = []
for i in range(self.emb.embeddings.shape[0]):
if i != idx:
distance = cosine(emb, self.emb.embeddings[i])
distances.append(distance)
sorted_indices = sorted(range(len(distances)), key=lambda k: distances[k], reverse=True)[:top_k]
return [self.emb.idx2word[i] for i in sorted_indices]

class Similarity3(Similarity):
def __init__(self, emb, metric='cosine', top_k=5):
super().__init__(emb, metric, top_k