154 lines
5.6 KiB
Python
154 lines
5.6 KiB
Python
import json
|
|
import os
|
|
from typing import List, Dict, Optional, Tuple
|
|
from loguru import logger
|
|
from .match_service import get_match_service
|
|
|
|
class StaticQAService:
|
|
def __init__(self, config_path: str = None):
|
|
if config_path is None:
|
|
backend_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
config_path = os.path.join(backend_path, 'config', 'static_qa.json')
|
|
|
|
self.config_path = config_path
|
|
self.match_service = get_match_service()
|
|
self._qa_pairs: List[Dict] = []
|
|
self._flattened_pairs: List[Tuple[str, Dict]] = []
|
|
self._load_qa_data()
|
|
|
|
def _load_qa_data(self):
|
|
try:
|
|
if not os.path.exists(self.config_path):
|
|
logger.warning(f'[StaticQA] 配置文件不存在: {self.config_path}')
|
|
self._qa_pairs = []
|
|
self._flattened_pairs = []
|
|
return
|
|
|
|
with open(self.config_path, 'r', encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
|
|
self._qa_pairs = data.get('qa_pairs', [])
|
|
self._build_flattened_pairs()
|
|
|
|
logger.info(f'[StaticQA] 加载了 {len(self._qa_pairs)} 个问答对，{len(self._flattened_pairs)} 个问题变体')
|
|
|
|
except Exception as e:
|
|
logger.error(f'[StaticQA] 加载配置文件失败: {e}')
|
|
self._qa_pairs = []
|
|
self._flattened_pairs = []
|
|
|
|
def _build_flattened_pairs(self):
|
|
self._flattened_pairs = []
|
|
|
|
for qa_pair in self._qa_pairs:
|
|
main_question = qa_pair.get('question', '')
|
|
answer = qa_pair.get('answer', '')
|
|
category = qa_pair.get('category', '')
|
|
priority = qa_pair.get('priority', 0)
|
|
|
|
if not main_question or not answer:
|
|
continue
|
|
|
|
qa_data = {
|
|
'question': main_question,
|
|
'answer': answer,
|
|
'category': category,
|
|
'priority': priority,
|
|
'source': 'static_qa'
|
|
}
|
|
|
|
self._flattened_pairs.append((main_question, qa_data))
|
|
|
|
sub_questions = qa_pair.get('sub_questions', [])
|
|
for sub_q in sub_questions:
|
|
if sub_q:
|
|
sub_qa_data = qa_data.copy()
|
|
sub_qa_data['is_sub_question'] = True
|
|
self._flattened_pairs.append((sub_q, sub_qa_data))
|
|
|
|
variations = qa_pair.get('variations', [])
|
|
for var in variations:
|
|
if var:
|
|
var_qa_data = qa_data.copy()
|
|
var_qa_data['is_variation'] = True
|
|
self._flattened_pairs.append((var, var_qa_data))
|
|
|
|
def reload(self):
|
|
self._load_qa_data()
|
|
|
|
def find_match(self, question: str, threshold: float = 0.70) -> Tuple[Optional[Dict], float]:
|
|
if not question or not self._flattened_pairs:
|
|
return None, 0.0
|
|
|
|
decomposed_questions = self.match_service.decompose_question(question)
|
|
|
|
if len(decomposed_questions) > 1:
|
|
logger.info(f'[StaticQA] 问题分解为 {len(decomposed_questions)} 个子问题: {decomposed_questions}')
|
|
|
|
best_match = None
|
|
best_similarity = 0.0
|
|
|
|
for sub_q in decomposed_questions:
|
|
match, similarity = self.match_service.find_best_match(
|
|
sub_q,
|
|
self._flattened_pairs,
|
|
threshold=threshold
|
|
)
|
|
|
|
if match and similarity > best_similarity:
|
|
best_match = match
|
|
best_similarity = similarity
|
|
|
|
if best_match:
|
|
logger.info(f'[StaticQA] 找到匹配 (相似度={best_similarity:.2f}): {best_match.get("question")}')
|
|
|
|
return best_match, best_similarity
|
|
|
|
def find_all_matches(self, question: str, threshold: float = 0.70, max_results: int = 5) -> List[Tuple[Dict, float]]:
|
|
if not question or not self._flattened_pairs:
|
|
return []
|
|
|
|
decomposed_questions = self.match_service.decompose_question(question)
|
|
|
|
all_matches = []
|
|
|
|
for sub_q in decomposed_questions:
|
|
for candidate_text, candidate_data in self._flattened_pairs:
|
|
similarity = self.match_service.calculate_similarity(sub_q, candidate_text)
|
|
|
|
if similarity >= threshold:
|
|
all_matches.append((candidate_data, similarity))
|
|
|
|
all_matches.sort(key=lambda x: (x[1], x[0].get('priority', 0)), reverse=True)
|
|
|
|
return all_matches[:max_results]
|
|
|
|
def get_all_categories(self) -> List[str]:
|
|
categories = set()
|
|
for qa_pair in self._qa_pairs:
|
|
category = qa_pair.get('category', '')
|
|
if category:
|
|
categories.add(category)
|
|
return sorted(list(categories))
|
|
|
|
def get_qa_by_category(self, category: str) -> List[Dict]:
|
|
result = []
|
|
for qa_pair in self._qa_pairs:
|
|
if qa_pair.get('category') == category:
|
|
result.append(qa_pair)
|
|
return result
|
|
|
|
def get_qa_count(self) -> int:
|
|
return len(self._qa_pairs)
|
|
|
|
def get_question_count(self) -> int:
|
|
return len(self._flattened_pairs)
|
|
|
|
_static_qa_service_instance = None
|
|
|
|
def get_static_qa_service(config_path: str = None) -> StaticQAService:
|
|
global _static_qa_service_instance
|
|
if _static_qa_service_instance is None:
|
|
_static_qa_service_instance = StaticQAService(config_path)
|
|
return _static_qa_service_instance
|