diff --git a/funasr/utils/kws_utils.py b/funasr/utils/kws_utils.py index 11de3238b8..59447361f0 100644 --- a/funasr/utils/kws_utils.py +++ b/funasr/utils/kws_utils.py @@ -117,6 +117,11 @@ def __init__( keywords_list = self.keywords_str.strip().replace(' ', '').split(',') for keyword in keywords_list: strs, indexs = query_token_set(keyword, token_table, seg_dict) + if len(indexs) == 0: + # An empty entry (e.g. from a trailing comma) would match every + # hypothesis, so skip it instead of reporting a false detection. + logging.warning(f"Skipping keyword {keyword!r}: it has no tokens to match.") + continue self.keywords_token[keyword] = {} self.keywords_token[keyword]['token_id'] = indexs self.keywords_token[keyword]['token_str'] = ''.join('%s ' % str(i) for i in indexs) diff --git a/tests/test_kws_utils.py b/tests/test_kws_utils.py index 79fe4cdc82..12e8fd1a88 100644 --- a/tests/test_kws_utils.py +++ b/tests/test_kws_utils.py @@ -1,7 +1,27 @@ import re import unittest -from funasr.utils.kws_utils import query_token_set, split_mixed_label, symbol_str +import torch + +from funasr.utils.kws_utils import ( + KwsCtcPrefixDecoder, + query_token_set, + split_mixed_label, + symbol_str, +) + + +class _IdentityCtc: + def softmax(self, x): + return x + + +def _posteriors(token_ids, vocab_size=4): + # One frame per entry: token_id gets all the probability mass (0 is blank). + probs = torch.zeros(len(token_ids), vocab_size) + for frame, token_id in enumerate(token_ids): + probs[frame, token_id] = 1.0 + return probs class KwsUtilsTest(unittest.TestCase): @@ -18,6 +38,19 @@ def test_query_token_set_strips_symbols_before_character_lookup(self): self.assertEqual(tokens, ("a", "中")) self.assertEqual(token_ids, (1, 2)) + def test_empty_keyword_entries_do_not_produce_detections(self): + token_list = ["", "", "小", "云"] + for keywords in ("小云小云,", "小云小云, ,", ",小云小云", "小云小云,!!"): + with self.subTest(keywords=keywords): + decoder = KwsCtcPrefixDecoder(_IdentityCtc(), keywords, token_list, {}) + + self.assertEqual(decoder.decode(_posteriors([0, 0, 0, 0])), (False, None, None)) + self.assertEqual(decoder.decode(_posteriors([2, 0, 0, 0])), (False, None, None)) + self.assertEqual( + decoder.decode(_posteriors([2, 3, 0, 2, 3])), (True, "小云小云", 1.0) + ) + self.assertEqual(list(decoder.keywords_token), ["小云小云"]) + if __name__ == "__main__": unittest.main()