262 lines
9.5 KiB
Python
262 lines
9.5 KiB
Python
import re
|
||
import jieba
|
||
from typing import List, Tuple, Optional
|
||
from loguru import logger
|
||
|
||
class MatchService:
|
||
def __init__(self):
|
||
self._colloquial_map = {
|
||
'咋': '怎么',
|
||
'咋办': '怎么办',
|
||
'咋样': '怎么样',
|
||
'啥': '什么',
|
||
'啥子': '什么',
|
||
'啥时候': '什么时候',
|
||
'啥地方': '什么地方',
|
||
'哪儿': '哪里',
|
||
'咋个': '怎么',
|
||
'咋整': '怎么办',
|
||
'咋回事': '怎么回事',
|
||
'咋回事儿': '怎么回事',
|
||
'咋样': '怎么样',
|
||
'啥人': '什么人',
|
||
'啥东西': '什么东西',
|
||
'啥玩意': '什么东西',
|
||
'啥情况': '什么情况',
|
||
'咋样': '怎么样',
|
||
'咋办': '怎么办',
|
||
'咋弄': '怎么办',
|
||
'咋整': '怎么办',
|
||
'咋回事': '怎么回事',
|
||
'咋回事儿': '怎么回事',
|
||
'咋样': '怎么样',
|
||
'啥': '什么',
|
||
'啥子': '什么',
|
||
'啥时候': '什么时候',
|
||
'啥地方': '什么地方',
|
||
'啥人': '什么人',
|
||
'啥东西': '什么东西',
|
||
'啥玩意': '什么东西',
|
||
'啥情况': '什么情况',
|
||
'啥事': '什么事',
|
||
'啥事儿': '什么事',
|
||
'啥时候': '什么时候',
|
||
'啥地方': '什么地方',
|
||
'哪儿': '哪里',
|
||
'哪儿个': '哪个',
|
||
'咋': '怎么',
|
||
'咋办': '怎么办',
|
||
'咋样': '怎么样',
|
||
'咋个': '怎么',
|
||
'咋整': '怎么办',
|
||
'咋弄': '怎么办',
|
||
'咋回事': '怎么回事',
|
||
'咋回事儿': '怎么回事',
|
||
'咋样': '怎么样',
|
||
'啥': '什么',
|
||
'啥子': '什么',
|
||
'啥时候': '什么时候',
|
||
'啥地方': '什么地方',
|
||
'啥人': '什么人',
|
||
'啥东西': '什么东西',
|
||
'啥玩意': '什么东西',
|
||
'啥情况': '什么情况',
|
||
'啥事': '什么事',
|
||
'啥事儿': '什么事',
|
||
}
|
||
|
||
self._stop_words = {
|
||
'的', '了', '在', '是', '我', '有', '和', '就', '不', '人', '都', '一',
|
||
'一个', '上', '也', '很', '到', '说', '要', '去', '你', '会', '着', '没有',
|
||
'看', '好', '自己', '这', '那', '吗', '呢', '吧', '啊', '哦', '呀', '嘛',
|
||
'啦', '呗', '哇', '嗯', '哎', '唉', '哼', '哦', '噢', '哎哟', '哎呀',
|
||
'请问', '请问一下', '请问一下', '请问一下', '我想知道', '我想问',
|
||
'帮我', '帮我看', '帮我看一下', '帮我查', '帮我查一下',
|
||
'能不能', '可不可以', '是否', '有没有', '是不是',
|
||
'麻烦', '麻烦问一下', '麻烦您', '麻烦帮我',
|
||
'请问', '请问一下', '请问问',
|
||
'我们', '你们', '他们', '她们', '它们',
|
||
'这个', '那个', '这些', '那些',
|
||
'这种', '那种', '这类', '那类',
|
||
'你好', '您好', '大家好', '嗨', '哈喽', '哈罗',
|
||
'早上好', '上午好', '下午好', '晚上好',
|
||
'hi', 'hello', 'hey', '嗨', '哈喽',
|
||
'怎么样', '如何', '怎样',
|
||
'关于', '对于', '至于',
|
||
'以及', '或者', '还是',
|
||
'因为', '所以', '但是', '不过',
|
||
'如果', '要是', '假如',
|
||
'虽然', '尽管',
|
||
'而且', '并且', '同时',
|
||
'然后', '接着', '之后',
|
||
'首先', '其次', '最后',
|
||
'总之', '总的来说',
|
||
'吗', '呢', '吧', '啊', '哦', '呀', '嘛', '啦', '呗', '哇', '嗯', '哎', '唉', '哼', '噢',
|
||
'哎哟', '哎呀', '哎嘿', '哎呦', '哎呀',
|
||
'呀', '耶', '哟', '喔', '唔', '诶', '咦', '嗯哼', '哼哼', '呵呵', '嘿嘿',
|
||
'嘻嘻', '哈哈', '哇塞', '哇哦', '天哪', '天啊', '哎呀呀',
|
||
'呃', '额', '嗯嗯', '哦哦', '啊啊', '嗯啊', '哦啊',
|
||
}
|
||
|
||
self._question_words = {
|
||
'什么', '怎么', '如何', '为什么', '哪儿', '哪里', '哪个', '谁',
|
||
'多少', '几', '何时', '什么时候', '怎样', '怎么样',
|
||
'是否', '有没有', '是不是', '能不能', '可不可以',
|
||
}
|
||
|
||
jieba.setLogLevel(jieba.logging.INFO)
|
||
|
||
def preprocess_text(self, text: str) -> str:
|
||
if not text:
|
||
return ''
|
||
|
||
text = text.strip()
|
||
text = re.sub(r'[^\u4e00-\u9fa5a-zA-Z0-9，。？！、；：""''（）【】《》\s]', '', text)
|
||
text = re.sub(r'\s+', ' ', text)
|
||
|
||
for colloquial, standard in self._colloquial_map.items():
|
||
text = text.replace(colloquial, standard)
|
||
|
||
text = re.sub(r'[，。？！、；：""''（）【】《》]', ' ', text)
|
||
text = re.sub(r'\s+', ' ', text)
|
||
|
||
return text.strip()
|
||
|
||
def extract_keywords(self, text: str, top_k: int = 10) -> List[str]:
|
||
if not text:
|
||
return []
|
||
|
||
text = self.preprocess_text(text)
|
||
words = jieba.lcut(text)
|
||
|
||
keywords = []
|
||
for word in words:
|
||
word = word.strip()
|
||
if len(word) < 2:
|
||
continue
|
||
if word in self._stop_words:
|
||
continue
|
||
if word.isdigit():
|
||
continue
|
||
keywords.append(word)
|
||
|
||
question_keywords = []
|
||
for word in words:
|
||
if word in self._question_words:
|
||
question_keywords.append(word)
|
||
|
||
keywords = list(set(keywords))
|
||
|
||
word_freq = {}
|
||
for word in words:
|
||
if word in keywords:
|
||
word_freq[word] = word_freq.get(word, 0) + 1
|
||
|
||
keywords.sort(key=lambda x: word_freq.get(x, 0), reverse=True)
|
||
|
||
return keywords[:top_k]
|
||
|
||
def calculate_similarity(self, q1: str, q2: str) -> float:
|
||
if not q1 or not q2:
|
||
return 0.0
|
||
|
||
n1 = self.preprocess_text(q1)
|
||
n2 = self.preprocess_text(q2)
|
||
|
||
if n1 == n2:
|
||
return 1.0
|
||
|
||
kw1 = set(self.extract_keywords(q1))
|
||
kw2 = set(self.extract_keywords(q2))
|
||
|
||
logger.info(f'[MatchDebug] q1={q1} | n1={n1} | kw1={kw1}')
|
||
logger.info(f'[MatchDebug] q2={q2} | n2={n2} | kw2={kw2}')
|
||
|
||
if not kw1 or not kw2:
|
||
return 0.0
|
||
|
||
intersection = len(kw1 & kw2)
|
||
union = len(kw1 | kw2)
|
||
jaccard_sim = intersection / union if union > 0 else 0
|
||
|
||
len_sim = 1 - abs(len(n1) - len(n2)) / max(len(n1), len(n2))
|
||
len_sim = max(0, len_sim)
|
||
|
||
similarity = 0.7 * jaccard_sim + 0.3 * len_sim
|
||
|
||
return similarity
|
||
|
||
def find_best_match(self, query: str, candidates: List[Tuple[str, any]], threshold: float = 0.6) -> Tuple[Optional[any], float]:
|
||
if not query or not candidates:
|
||
return None, 0.0
|
||
|
||
best_match = None
|
||
best_similarity = 0.0
|
||
|
||
for candidate_text, candidate_data in candidates:
|
||
similarity = self.calculate_similarity(query, candidate_text)
|
||
logger.info(f'[MatchDebug] compare | query="{query}" | candidate="{candidate_text}" | sim={similarity:.3f}')
|
||
|
||
if similarity > best_similarity:
|
||
best_similarity = similarity
|
||
best_match = candidate_data
|
||
|
||
logger.info(f'[MatchDebug] best | query="{query}" | best_sim={best_similarity:.3f} | threshold={threshold}')
|
||
|
||
if best_similarity >= threshold:
|
||
return best_match, best_similarity
|
||
|
||
return None, best_similarity
|
||
|
||
def decompose_question(self, question: str) -> List[str]:
|
||
if not question:
|
||
return []
|
||
|
||
question = self.preprocess_text(question)
|
||
|
||
separators = ['？', '?', '。', '.', '；', ';', '，', ',', '和', '以及', '还有', '另外']
|
||
|
||
sub_questions = [question]
|
||
for sep in separators:
|
||
new_sub_questions = []
|
||
for sq in sub_questions:
|
||
parts = sq.split(sep)
|
||
new_sub_questions.extend(parts)
|
||
sub_questions = new_sub_questions
|
||
|
||
sub_questions = [sq.strip() for sq in sub_questions if sq.strip()]
|
||
|
||
unique_sub_questions = []
|
||
seen = set()
|
||
for sq in sub_questions:
|
||
if sq not in seen:
|
||
seen.add(sq)
|
||
unique_sub_questions.append(sq)
|
||
|
||
return unique_sub_questions
|
||
|
||
def calculate_keyword_coverage(self, query: str, candidate: str) -> float:
|
||
if not query or not candidate:
|
||
return 0.0
|
||
|
||
kw1 = set(self.extract_keywords(query))
|
||
kw2 = set(self.extract_keywords(candidate))
|
||
|
||
if not kw1:
|
||
return 0.0
|
||
|
||
if not kw2:
|
||
return 0.0
|
||
|
||
coverage = len(kw1 & kw2) / len(kw1)
|
||
|
||
return coverage
|
||
|
||
_match_service_instance = None
|
||
|
||
def get_match_service() -> MatchService:
|
||
global _match_service_instance
|
||
if _match_service_instance is None:
|
||
_match_service_instance = MatchService()
|
||
return _match_service_instance
|