worker.py

import time
import os
import pickle
from loguru import logger
from .helper import ErrorCode
from .http_client import OpenAPIClient, ClassifyModel, CacheRetriever


SECURITY_TEMAPLTE = '判断以下句子是否涉及政治、辱骂、色情、恐暴、宗教、网络暴力、种族歧视等违禁内容，结果用 0～10 表示，不要解释直接给出得分。判断标准：涉其中任一问题直接得 10 分；完全不涉及得 0 分。直接给得分不要解释：“{}”'  # noqa E501
GENERATE_TEMPLATE = '<Data>{}</Data> \n 回答要求：\n如果你不清楚答案，你需要澄清。\n避免提及你是从 <Data></Data> 获取的知识。\n保持答案与 <Data></Data> 中描述的一致。\n使用 Markdown 语法优化回答格式。\n使用与问题相同的语言回答。问题:"{}"'
MARKDOWN_TEMPLATE = '问题：“{}”  \n请使用markdown格式回答此问题'

COMMON = {
    "<光合组织登记网址>": "https://www.hieco.com.cn/partner?from=timeline",
    "<官网>": "https://www.sugon.com/after_sale/policy?sh=1",
    "<平台联系方式>": "1、访问官网,根据您所在地地址联系平台人员,网址地址：https://www.sugon.com/about/contact;\n2、点击人工客服进行咨询;\n3、请您拨打中科曙光服务热线400-810-0466联系人工进行咨询。",
    "<购买与维修的咨询方法>": "1、确定付费处理,可以微信搜索'sugon中科曙光服务'小程序,选择'在线报修'业务\n2、先了解价格,可以微信搜索'sugon中科曙光服务'小程序,选择'其他咨询'业务\n3、请您拨打中科曙光服务热线400-810-0466",
    "<服务器续保流程>": "1、微信搜索'sugon中科曙光服务'小程序,选择'延保与登记'业务\n2、点击人工客服进行登记\n3、请您拨打中科曙光服务热线400-810-0466根据语音提示选择维保与购买",
    "<XC内外网OS网盘链接>": "【腾讯文档】XC内外网OS网盘链接：https://docs.qq.com/sheet/DTWtXbU1BZHJvWkJm",
    "<W360-G30机器,安装Win7使用的镜像链接>": "W360-G30机器,安装Win7使用的镜像链接:https://pan.baidu.com/s/1SjHqCP6kJ9KzdJEBZDEynw;提取码:x6m4",
    "<麒麟系统搜狗输入法下载链接>": "软件下载链接（百度云盘）:链接:https://pan.baidu.com/s/18Iluvs4BOAfFET0yFMBeLQ,提取码:bhkf",
    "<X660 G45 GPU服务器拆解视频网盘链接>": "链接: https://pan.baidu.com/s/1RkRGh4XY1T2oYftGnjLp4w;提取码: v2qi",
    "<DS800,SANTRICITY存储IBM版本模拟器网盘链接>": "链接:https://pan.baidu.com/s/1euG9HGbPfrVbThEB8BX76g;提取码:o2ya",
    "<E80-D312（X680-G55）风冷整机组装说明下载链接>": "链接:https://pan.baidu.com/s/17KDpm-Z9lp01WGp9sQaQ4w;提取码:0802",
    "<X680 G55 风冷相关资料下载链接>": "链接:https://pan.baidu.com/s/1KQ-hxUIbTWNkc0xzrEQLjg;提取码:0802",
    "<R620 G51刷写EEPROM下载>": "下载链接如下:http://10.2.68.104/tools/bytedance/eeprom/",
    "<X7450A0服务器售后培训文件网盘链接>": "网盘下载:https://pan.baidu.com/s/1tZJIf_IeQLOWsvuOawhslQ?pwd=kgf1;提取码:kgf1",
    "<福昕阅读器补丁链接>": "补丁链接: https://pan.baidu.com/s/1QJQ1kHRplhhFly-vxJquFQ,提取码: aupx1",
    "<W330-H35A_22DB4/W3335HA安装win7网盘链接>": "硬盘链接: https://pan.baidu.com/s/1fDdGPH15mXiw0J-fMmLt6Q提取码: k97i",
    "<X680 G55服务器售后培训资料网盘链接>": "云盘连接下载:链接:https://pan.baidu.com/s/1gaok13DvNddtkmk6Q-qLYg?pwd=xyhb提取码:xyhb",
    "<展厅管理员>": "北京-穆淑娟18001053012\n天津-马书跃15720934870\n昆山-关天琪15304169908\n成都-贾小芳18613216313\n重庆-李子艺17347743273\n安阳-郭永军15824623085\n桐乡-李梦瑶18086537055\n青岛-陶祉伊15318733259",
    "<线上预约展厅>": "北京、天津、昆山、成都、重庆、安阳、桐乡、青岛",
    "<马华>": "联系人:马华，电话:13761751980，邮箱:china@pinbang.com",
    "<梁静>": "联系人:梁静，电话:18917566297，邮箱:ing.liang@omaten.com",
    "<徐斌>": "联系人:徐斌，电话:13671166044，邮箱:244898943@qq.com",
    "<俞晓枫>": "联系人:俞晓枫，电话13750869272，邮箱:857233013@qq.com",
    "<刘广鹏>": "联系人:刘广鹏，电话13321992411，邮箱:liuguangpeng@pinbang.com",
    "<马英伟>": "联系人:马英伟，电话:13260021849，邮箱:13260021849@163.com",
    "<杨洋>": "联系人：杨洋，电话15801203938，邮箱bing523888@163.com",
    "<展会合规要求>": "1.展品内容:展品内容需符合公司合规要求，展示内容需经过法务合规审查。\n2.文字材料内容:文字材料内容需符合公司合规要求，展示内容需经过法务合规审查。\n3.展品标签:展品标签内容需符合公司合规要求。\n4.礼品内容:礼品内容需符合公司合规要求。\n5.视频内容:视频内容需符合公司合规要求，展示内容需经过法务合规审查。\n6.讲解词内容:讲解词内容需符合公司合规要求，展示内容需经过法务合规审查。\n7.现场发放材料:现场发放的材料内容需符合公司合规要求。\n8.展示内容:整体展示内容需要经过法务合规审查。",
    "<展会质量>": "1.了解展会的组织者背景、往届展会的评价以及提供的服务支持，确保展会的专业性和高效性。\n.了解展会的规模、参观人数、行业影响力等因素，以判断展会是否能够提供足够的曝光度和商机。\n3.关注同行业其他竞争对手是否参展，以及他们的展位布置、展示内容等信息，以便制定自己的参展策略。\n4.展会的日期是否与公司的其他重要活动冲突，以及举办地点是否便于客户和合作伙伴的参观。\n5.销售部门会询问展会方提供的宣传渠道和推广服务，以及如何利用这些资源来提升公司及产品的知名度。\n6.记录展会期间的重要领导参观、商机线索、合作洽谈、公司拜访预约等信息，跟进后续商业机会。",
    "<摊位费规则>": "根据展位面积大小，支付相应费用。\n展位照明费:支付展位内的照明服务费。\n展位保安费:支付展位内的保安服务费。\n展位网络使用费:支付展位内网络使用的费用。\n展位电源使用费:支付展位内电源使用的费用。",
    "<展会主题要求>": "展会主题的确定需要符合公司产品和服务业务范围，以确保能够吸引目标客户群体。因此，确定展会主题时，需要考虑以下因素:\n专业性:展会的主题应确保专业性，符合行业特点和目标客户的需求。\n目标客户群体:展会的主题定位应考虑目标客户群体，确保能够吸引他们的兴趣。\n业务重点:展会的主题应突出公司的业务重点和优势，以便更好地推广公司的核心产品或服务。\n行业影响力:展会的主题定位需要考虑行业的最新发展趋势，以凸显公司的行业地位和影响力。\n往届展会经验:可以参考往届展会的主题定位，总结经验教训，以确定本届展会的主题。\n市场部意见:在确定展会主题时，应听取市场部的意见，确保主题符合公司的整体市场战略。\n领导意见:还需要考虑公司领导的意见，以确保展会主题符合公司的战略发展方向。",
    "<办理展商证注意事项>": "人员范围:除公司领导和同事需要办理展商证外，展会运营工作人员也需要办理。\n提前准备:展商证的办理需要提前进行，以确保摄影师、摄像师等工作人员可以提前入场进行布置。\n办理流程:需要熟悉展商证的办理流程，准备好相关材料，如身份证件等。\n数量需求:需要评估所需的展商证数量，避免数量不足或过多的情况。\n有效期限:展商证的有效期限需要注意，避免在展期内过期。\n存放安全:办理完的展商证需要妥善保管，避免丢失或被他人使用。\n使用规范:使用展商证时需要遵守展会相关规定，不得转让给他人使用。\n回收处理:展会结束后，需要及时回收展商证，避免泄露相关信息。",
    "<项目单价要求>": "请注意:无论是否年框供应商，项目单价都不得超过采购部制定的“2024常见活动项目标准单价”，此报价仅可内部使用，严禁外传",
    "<年框供应商细节表格>": "在线表格https://kdocs.cn/l/camwZE63frNw",
    "<年框供应商流程>": "1.需求方发出项目需求（大型项目需比稿）\n2.外协根据项目需求报价，提供需求方“预算单”（按照基准单价报价，如有发现不按单价情况，解除合同不再使用）\n3.需求方确认预算价格，并提交OA市场活动申请\n4.外协现场执行\n5.需求方现场验收，并签署验收单（物料、设备、人员等实际清单）\n6.外协出具结算单（金额与验收单一致，加盖公章）、结案报告、年框合同，作为报销凭证\n7.外协请需求方项目负责人填写“满意度调研表”（如无，会影响年度评价）\n8.需求方项目经理提交报销",
    "<市场活动结案报告内容>": "1.项目简介（时间、地点、参与人数等）；2.最终会议安排；3.活动各环节现场图片；4.费用相关证明材料（如执行人员、物料照片）；5.活动成效汇总；6.活动原始照片/视频网络链接",
    "<展板设计选择>": "1.去OA文档中心查找一些设计模板; 2. 联系专业的活动服务公司来协助设计",
    "<餐费标准>": "一般地区的餐饮费用规定为不超过300元/人（一顿正餐），特殊地区则为不超过400元/人（一顿正餐），特殊地区的具体规定请参照公司的《差旅费管理制度》",
    "":"",
}


def substitution(chunks):
    # 翻译特殊字符
    import re
    new_chunks = []
    for chunk in chunks:
        matchObj = re.split('.*(<.*>).*', chunk, re.M|re.I)
        if len(matchObj) > 1:
            obj = matchObj[1]
            replace_str = COMMON.get(obj)
            if replace_str:
                chunk = chunk.replace(obj, replace_str)
                logger.info(f"{obj} be replaced {replace_str}, after {chunk}")
        new_chunks.append(chunk)
    return new_chunks


class Worker:

    def __init__(self, config):
        self.work_dir = config['default']['work_dir']
        llm_model = config['model']['llm_model']
        local_model = config['model']['local_model']
        llm_service_address = config['model']['llm_service_address']
        cls_model_path = config['model']['cls_model_path']
        local_server_address = config['model']['local_service_address']
        reject_throttle = float(config['feature_database']['reject_throttle'])

        if not llm_service_address:
            raise Exception('llm_service_address is required in config.ini')
        if not cls_model_path:
            raise Exception('cls_model_path is required in config.ini')

        self.max_input_len = int(config['model']['max_input_length'])

        self.retriever = CacheRetriever(
                self.embedding_model_path,
                self.reranker_model_path).get(reject_throttle=reject_throttle,
                                              work_dir=self.work_dir)
        self.openapi_service = OpenAPIClient(llm_service_address, llm_model)
        self.openapi_local_server = OpenAPIClient(local_server_address, local_model)
        self.classify_service = ClassifyModel(cls_model_path)

        self.tasks = {}
        if os.path.exists(self.work_dir + '/tasks_status.pkl'):
            with open(self.work_dir + '/tasks_status.pkl', 'rb') as f:
                self.tasks = pickle.load(f)

    def generate_prompt(self,
                        history_pair,
                        instruction: str,
                        context: str = ''):

        if context is not None and len(context) > 0:
            str_context = str(context)
            if len(str_context) > self.max_input_len:
                str_context = str_context[:self.max_input_len]
            instruction = GENERATE_TEMPLATE.format(str_context, instruction)

        real_history = []
        for pair in history_pair:
            if pair[0] is None or pair[1] is None:
                continue
            if len(pair[0]) < 1 or len(pair[1]) < 1:
                continue
            real_history.append(pair)

        return instruction, real_history

    async def generater(self, content):
        for word in content:
            yield word
            #await asyncio.sleep(0.1)

    async def response_by_common(self, query, history, output_format=False, stream=False):
        if output_format:
            query = MARKDOWN_TEMPLATE.format(query)
        logger.info('Prompt is: {}, History is: {}'.format(query, history))
        response_direct = await self.openapi_service.chat(query, history, stream=stream)
        return response_direct

    def format_rag_result(self, chunks, references, stream=False):
        result = "针对您的问题，我们找到了如下解决方案：\n%s"
        content = ""
        for i, item in enumerate(references):
            if item.endswith(".json"):
                content += " - %s.%s\n" % (i + 1, chunks[i])
            else:
                line = chunks[i]
                if len(line) > 300:
                    line = line[:300] + "..." + '\n'
                    line += "详细内容参见：%s" % item
                content += " - %s.%s\n" % (i + 1, line)
        if stream:
            return self.generater((result % content))
        return result % content

    def response_by_finetune(self, query, history=[]):
        '''微调模型回答'''
        logger.info('Prompt is: {}, History is: {}'.format(query, history))
        response_direct = self.openapi_local_server.chat(query, history)
        return response_direct

    async def produce_response(self, config, query, history, stream=False):
        response = ''
        references = []
        use_template = config.getboolean('default', 'use_template')
        output_format = config.getboolean('default', 'output_format')

        if query is None:
            return ErrorCode.NOT_A_QUESTION, response, references

        logger.info('input: %s' % [query, history])

        # classify
        score = self.classify_service.classfication(query)
        if score > 0.8:
            logger.debug('Start RAG search')
            chunks, references = self.retriever.query(query)

            if len(chunks) == 0:
                logger.debug('Response by finetune model')
                chunks = [self.response_by_finetune(query, history=history)]
            elif use_template:
                logger.debug('Response by template')
                response = self.format_rag_result(chunks, references, stream=stream)
                return ErrorCode.SUCCESS, response, references

            logger.debug('Response with common model')
            new_chunks = substitution(chunks)
            prompt, history = self.generate_prompt(
                instruction=query,
                context=new_chunks,
                history_pair=history)

            logger.debug('prompt: {}'.format(prompt))
            response = await self.response_by_common(prompt, history=history, output_format=False, stream=stream)
            return ErrorCode.SUCCESS, response, references
        else:
            logger.debug('Response by common model')
            response = await self.response_by_common(query, history=history, output_format=output_format, stream=stream)
            return ErrorCode.SUCCESS, response, references