Digital_Human/knowledge_base.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
双知识库智能问答系统
- 知识库A：存储介绍性语句
- 知识库B：存储问答对数据
"""

import time
from logger import logger


class KnowledgeBase:
    """双知识库管理类"""
    
    def __init__(self):
        """初始化双知识库系统"""
        # 知识库A：介绍性语句
        self.knowledge_a = {
            "full": [
                "欢迎所有新",
                "进直播间的朋友！",
                "今天给大家带来",
                "咱们本地超好用的生活服务平台，",
                "吃喝玩乐全都有!",
                "家人们晚上好！",
                "不管你是想点外卖、找团购、还是找生活服务，",
                "今天一个 App 全部搞定！",
                "新进来的宝宝停留一下！今天不带货，",
                "只给大家送福利，",
                "教你怎么用一个 App 省心省钱！",
                "欢迎大家！咱们直播间不讲虚的，",
                "全是你家门口的实惠、方便、好用的生活服务！",
                "哈喽各位！",
                "今天专门给同城的朋友准备的福利场，",
                "本地生活一键搞定，千万别走开！",  
                "欢迎使用本地生活App！",
                "我们为您提供美食外卖、休闲娱乐、",
                "生活服务和附近团购，随时为您推荐最合适的本地服务！",
                "本地生活App超实用！",
                "外卖半小时极速达、休闲娱乐折扣多，",
                "还有上门家政、生鲜配送等一站式生活服务！",
                "选本地生活App准没错！覆盖全城的团购优惠、",
                "24小时外卖服务，让您的居家生活更便捷省心！",
                "本地生活App上新啦！新人注册立减15元，",
                "美食外卖、生鲜果蔬、鲜花绿植都能一键下单！",
                "本地生活App，您的专属生活管家！家政保洁、",
                "家电维修、跑腿代办，各类生活服务一键预约！",
                "不用再对比啦！本地生活App团购低至3折，",
                "火锅、烤肉、奶茶全覆盖，每周还有专属特价！",
                "本地生活App夏季特惠！冰镇饮品、生鲜水果外卖满减，在家就能享受清凉一夏！",
                "上班族福利！本地生活App工作餐专区，营养搭配、准时送达，解决你的午餐烦恼！",
                "咱们这个本地生活 App，外卖、美食、休闲娱乐、家政服务，打开就能用，不用到处找！",
                "想吃火锅、烤肉、奶茶？打开 App，附近所有优惠一目了然，比你自己找便宜太多！",
                "不想做饭？外卖半小时就能到，早餐、午餐、晚餐、下午茶、夜宵，24 小时都能点！",
                "家里需要保洁、维修、跑腿？不用到处打电话，App 一键预约，上门服务超省心！",
                "生鲜果蔬、日用品、鲜花蛋糕，足不出户，给你送到家，新鲜又快捷！",
                "周末想放松，看电影、唱歌、足疗，所有门店优惠全都在这一个 App 里！",
                "附近的团购天天有，低至三折，没有任何套路，到店直接用！",
                "不管你在哪个片区，周边的优质商家全覆盖，真正的本地生活小管家！",
                "操作特别简单，老人小孩都会用，点一点，服务直接上门！",
                "一个 App 顶十个平台，外卖、团购、家政、生鲜、跑腿，全部搞定！"
            ],
            "short": [
                "本地生活App，便民服务一键搞定！",
                "本地生活App，外卖团购超划算！",
                "本地生活App，您的专属生活助手！",
                "本地生活App，新人立减15元！"
            ]
        }
        
        # 知识库B：问答对数据
        self.knowledge_b = {
            "外卖服务": {
                "如何点外卖": "打开本地生活App，点击外卖入口，选择您喜欢的餐厅和菜品，确认地址和支付方式后下单即可。",
                "外卖配送时间多长": "一般情况下，外卖配送时间为30-45分钟，具体时间取决于餐厅出餐速度和距离远近。",
                "外卖可以退款吗": "可以，在订单状态为待接单或待取餐时，您可以在订单详情页面申请退款。",
                "如何查看外卖配送进度": "在订单详情页面，您可以实时查看外卖配送进度和骑手位置。"
            },
            "团购服务": {
                "如何使用团购券": "在本地生活App上购买团购券后，到店消费时出示核销码即可使用。",
                "团购券可以退款吗": "未使用的团购券在有效期内可以申请退款，具体退款规则请查看商品详情。",
                "如何查找附近的团购": "在本地生活App首页点击团购入口，系统会自动推荐附近的团购优惠。",
                "团购有哪些优惠": "本地生活App每日更新团购优惠，包括美食、休闲娱乐、生活服务等多个品类，低至3折起。"
            },
            "生活服务": {
                "如何预约家政服务": "打开本地生活App，点击生活服务入口，选择家政服务类型，预约时间和服务人员即可。",
                "家政服务有哪些类型": "本地生活App提供保洁、做饭、带娃、家电维修等多种家政服务。",
                "如何预约上门维修": "在生活服务入口选择维修服务，填写维修内容和预约时间，师傅会上门服务。",
                "生活服务可以开发票吗": "可以，在服务完成后，您可以在订单详情页面申请开具发票。"
            },
            "会员福利": {
                "如何成为会员": "打开本地生活App，点击我的页面，选择会员中心，按照提示开通会员即可。",
                "会员有哪些权益": "会员可以享受免费配送券、团购折上折、专属特价等多项权益。",
                "会员积分如何使用": "会员积分可以在积分商城兑换商品或抵扣订单金额。",
                "会员日有什么活动": "每周五为会员日，全场团购5折起，外卖满减叠加，省钱超划算！"
            },
            "新人福利": {
                "新人注册有什么优惠": "本地生活App新人注册立减15元，美食外卖、生鲜果蔬、鲜花绿植都能一键下单！",
                "新人注册立减多少元": "新人注册立减15元，可以在美食外卖、生鲜果蔬、鲜花绿植等品类使用。",
                "新人福利有哪些": "新人注册可享受立减15元优惠，还有新人专享特价商品和免费配送券等多重福利。",
                "如何领取新人优惠": "下载本地生活App并注册账号，系统会自动发放15元新人优惠券到您的账户。"
            },
            "常见问题": {
                "如何修改收货地址": "在我的页面点击地址管理，添加或修改收货地址即可。",
                "如何联系客服": "在我的页面点击客服中心，可以通过在线客服或电话联系客服。",
                "如何查看历史订单": "在我的页面点击订单管理，可以查看所有历史订单。",
                "如何使用优惠券": "在结算页面，选择可用的优惠券即可享受折扣。"
            }
        }
        
        logger.info("双知识库系统初始化完成")
    
    def query_knowledge_a(self, play_type: str = "full") -> list:
        """
        获取知识库A的介绍性内容
        :param play_type: 播放类型，"full"（完整版）或"short"（简短版）
        :return: 介绍性内容列表
        """
        if play_type not in self.knowledge_a:
            logger.error(f"无效的播放类型：{play_type}，默认使用完整版")
            play_type = "full"
        return self.knowledge_a[play_type]
    
    def query_knowledge_b(self, question: str) -> tuple:
        """
        查询知识库B获取答案
        :param question: 用户问题
        :return: (answer, match_score) 答案和匹配度分数
        """
        start_time = time.time()
        logger.info(f"收到知识库B查询请求: {question}")
        
        best_match = None
        best_score = 0
        best_answer = None
        
        # 遍历知识库B中的所有类别
        for category, qna_pairs in self.knowledge_b.items():
            # 遍历每个问答对
            for q, a in qna_pairs.items():
                # 计算匹配度
                score = self._calculate_match_score(question, q)
                if score > best_score:
                    best_score = score
                    best_match = q
                    best_answer = a
        
        # 记录响应时间
        response_time = (time.time() - start_time) * 1000
        logger.info(f"知识库B查询响应时间: {response_time:.2f}ms")
        
        if best_score >= 0.85:
            logger.info(f"找到匹配度为 {best_score:.2f} 的答案: {best_answer}")
            return best_answer, best_score
        else:
            logger.info(f"未找到匹配度≥85%的答案，最高匹配度为 {best_score:.2f}")
            return None, best_score
    
    def _calculate_match_score(self, question: str, keyword: str) -> float:
        """
        计算问题与关键词的匹配度
        :param question: 用户问题
        :param keyword: 关键词
        :return: 匹配度分数（0-1）
        """
        # 转换为小写进行比较
        question = question.lower()
        keyword = keyword.lower()
        
        # 精确匹配
        if keyword in question:
            return 1.0
        
        # 计算字符级匹配
        question_chars = set(question)
        keyword_chars = set(keyword)
        common_chars = question_chars & keyword_chars
        
        # 计算字符匹配率
        if len(keyword_chars) == 0:
            return 0
        
        char_match_rate = len(common_chars) / len(keyword_chars)
        
        # 计算连续字符匹配
        max_consecutive_match = 0
        current_consecutive = 0
        
        for char in question:
            if char in keyword_chars:
                current_consecutive += 1
                max_consecutive_match = max(max_consecutive_match, current_consecutive)
            else:
                current_consecutive = 0
        
        # 连续匹配加分
        consecutive_bonus = min(max_consecutive_match / len(keyword), 0.3)
        
        # 计算最终匹配度
        match_score = char_match_rate * 0.7 + consecutive_bonus
        
        return min(match_score, 1.0)


# 创建全局知识库实例
knowledge_base = KnowledgeBase()

# 便捷函数
def query_knowledge_a(play_type: str = "full") -> list:
    """便捷查询知识库A函数"""
    return knowledge_base.query_knowledge_a(play_type)

def query_knowledge_b(question: str) -> tuple:
    """便捷查询知识库B函数"""
    return knowledge_base.query_knowledge_b(question)