自动问答系统开发全记录：从理论到实践的技术探索

QuestionAnswerer (基类)
├── RetrievalBasedQA (检索式问答器)
├── KnowledgeBasedQA (基于知识库的问答器)
└── EnsembleQA (集成问答器)

class QuestionAnswerer:
    """
    自动问答器基类
    """

    def __init__(self):
        """
        初始化自动问答器
        """
        pass

    def preprocess(self, text):
        """
        文本预处理
        
        Args:
            text (str): 原始文本
            
        Returns:
            str: 预处理后的文本
        """
        # 去除多余空格和标点符号
        text = re.sub(r'\s+', ' ', text.strip())
        return text

    def train(self, questions, answers):
        """
        训练自动问答器
        
        Args:
            questions (list): 问题列表
            answers (list): 答案列表
        """
        raise NotImplementedError("子类必须实现train方法")

    def answer(self, question):
        """
        回答问题
        
        Args:
            question (str): 待回答的问题
            
        Returns:
            str: 答案
        """
        raise NotImplementedError("子类必须实现answer方法")

class RetrievalBasedQA(QuestionAnswerer):
    """
    检索式问答器
    通过检索知识库中的相关信息来回答问题
    """

    def __init__(self):
        super().__init__()
        # 知识库，存储问题-答案对
        self.knowledge_base = []
        # 倒排索引，用于快速检索
        self.inverted_index = defaultdict(set)
        # 是否已训练
        self.is_trained = False

    def train(self, questions, answers):
        """
        训练检索式问答器
        """
        if len(questions) != len(answers):
            raise ValueError("问题和答案数量不匹配")
        
        # 构建知识库
        for i, (question, answer) in enumerate(zip(questions, answers)):
            # 预处理问题
            processed_question = self.preprocess(question)
            processed_answer = self.preprocess(answer)
            
            # 存储问题-答案对
            self.knowledge_base.append({
                'id': i,
                'question': processed_question,
                'answer': processed_answer
            })
            
            # 构建倒排索引
            words = jieba.cut(processed_question)
            for word in words:
                if len(word) > 1:  # 过滤单字词
                    self.inverted_index[word].add(i)
        
        self.is_trained = True

    def _calculate_similarity(self, question_words, candidate_question):
        """
        计算问题相似度
        """
        # 对候选问题分词
        candidate_words = set(jieba.cut(candidate_question))
        
        # 计算词汇重叠度
        common_words = set(question_words).intersection(candidate_words)
        union_words = set(question_words).union(candidate_words)
        
        if len(union_words) == 0:
            return 0
        
        # 使用Jaccard相似度
        jaccard_similarity = len(common_words) / len(union_words)
        
        # 考虑词频权重的TF-IDF相似度
        tfidf_score = 0
        for word in common_words:
            # 简化的TF-IDF计算
            tf = question_words.count(word)  # 词频
            # 修复除零错误：确保分母不为零
            idf_denominator = len(self.inverted_index.get(word, []))
            if idf_denominator == 0:
                idf_denominator = 1  # 避免除零错误
            idf = math.log(len(self.knowledge_base) / idf_denominator + 1)  # 逆文档频率
            tfidf_score += tf * idf
        
        # 综合得分
        similarity = 0.7 * jaccard_similarity + 0.3 * tfidf_score
        return similarity

    def answer(self, question):
        """
        使用检索式方法回答问题
        """
        if not self.is_trained:
            raise Exception("模型尚未训练，请先调用train方法")
        
        # 预处理问题
        question = self.preprocess(question)
        
        # 对问题分词
        question_words = list(jieba.cut(question))
        
        # 检索相关问题
        candidate_ids = set()
        for word in question_words:
            if len(word) > 1:  # 过滤单字词
                candidate_ids.update(self.inverted_index.get(word, set()))
        
        # 如果没有找到相关问题，返回默认答案
        if not candidate_ids:
            return "抱歉，我没有找到相关问题的答案。"
        
        # 计算所有候选问题的相似度
        similarities = []
        for candidate_id in candidate_ids:
            candidate = self.knowledge_base[candidate_id]
            similarity = self._calculate_similarity(question_words, candidate['question'])
            similarities.append((candidate_id, similarity))
        
        # 选择相似度最高的问题
        best_match = max(similarities, key=lambda x: x[1])
        
        # 如果相似度太低，返回默认答案
        if best_match[1] < 0.1:
            return "抱歉，我没有找到相关问题的答案。"
        
        # 返回最佳匹配的答案
        best_answer = self.knowledge_base[best_match[0]]['answer']
        return best_answer

class KnowledgeBasedQA(QuestionAnswerer):
    """
    基于知识库的问答器
    通过结构化的知识库来回答问题
    """

    def __init__(self):
        super().__init__()
        # 结构化知识库 (实体-属性-值)
        self.knowledge_graph = defaultdict(lambda: defaultdict(list))
        # 实体词典
        self.entities = set()
        # 属性词典
        self.attributes = set()
        # 是否已训练
        self.is_trained = False

    def train(self, triples):
        """
        训练基于知识库的问答器
        
        Args:
            triples (list): 三元组列表，每个三元组为(entity, attribute, value)
        """
        # 构建知识图谱
        for entity, attribute, value in triples:
            # 预处理
            entity = self.preprocess(entity)
            attribute = self.preprocess(attribute)
            value = self.preprocess(value)
            
            # 存储到知识图谱
            self.knowledge_graph[entity][attribute].append(value)
            
            # 更新词典
            self.entities.add(entity)
            self.attributes.add(attribute)
        
        self.is_trained = True

    def _extract_entities(self, question):
        """
        从问题中提取实体
        """
        # 简单的实体提取方法
        words = jieba.cut(question)
        entities = []
        for word in words:
            if word in self.entities:
                entities.append(word)
        return entities

    def _extract_attributes(self, question):
        """
        从问题中提取属性
        """
        # 简单的属性提取方法
        words = jieba.cut(question)
        attributes = []
        for word in words:
            if word in self.attributes:
                attributes.append(word)
        return attributes

    def answer(self, question):
        """
        使用基于知识库的方法回答问题
        """
        if not self.is_trained:
            raise Exception("模型尚未训练，请先调用train方法")
        
        # 预处理问题
        question = self.preprocess(question)
        
        # 提取实体和属性
        entities = self._extract_entities(question)
        attributes = self._extract_attributes(question)
        
        # 如果没有提取到实体或属性，返回默认答案
        if not entities or not attributes:
            return "抱歉，我没有理解您的问题。"
        
        # 查找答案
        answers = []
        for entity in entities:
            for attribute in attributes:
                if attribute in self.knowledge_graph[entity]:
                    values = self.knowledge_graph[entity][attribute]
                    answers.extend(values)
        
        # 返回答案
        if answers:
            return "，".join(answers)
        else:
            return "抱歉，我没有找到相关问题的答案。"

class EnsembleQA(QuestionAnswerer):
    """
    集成问答器
    结合检索式和基于知识库的方法进行问答
    """

    def __init__(self):
        super().__init__()
        self.retrieval_qa = RetrievalBasedQA()
        self.knowledge_qa = KnowledgeBasedQA()

    def train(self, data):
        """
        训练集成问答器
        
        Args:
            data (dict): 训练数据，包含'retrieval'和'knowledge'两部分
        """
        # 训练检索式问答器
        if 'retrieval' in data:
            questions = data['retrieval']['questions']
            answers = data['retrieval']['answers']
            self.retrieval_qa.train(questions, answers)
        
        # 训练基于知识库的问答器
        if 'knowledge' in data:
            triples = data['knowledge']['triples']
            self.knowledge_qa.train(triples)

    def answer(self, question):
        """
        使用集成方法回答问题
        """
        # 首先尝试基于知识库的方法
        if self.knowledge_qa.is_trained:
            knowledge_answer = self.knowledge_qa.answer(question)
            if not knowledge_answer.startswith("抱歉"):
                return knowledge_answer
        
        # 如果基于知识库的方法没有找到答案，尝试检索式方法
        if self.retrieval_qa.is_trained:
            retrieval_answer = self.retrieval_qa.answer(question)
            return retrieval_answer
        
        return "抱歉，我没有理解您的问题。"

# 构建倒排索引
words = jieba.cut(processed_question)
for word in words:
    if len(word) > 1:  # 过滤单字词
        self.inverted_index[word].add(i)

# 综合得分
similarity = 0.7 * jaccard_similarity + 0.3 * tfidf_score

# 首先尝试基于知识库的方法
if self.knowledge_qa.is_trained:
    knowledge_answer = self.knowledge_qa.answer(question)
    if not knowledge_answer.startswith("抱歉"):
        return knowledge_answer