helper.py

from enum import Enum

class TaskCode(Enum):
    FS_ADD_DOC = 'add_doc'
    FS_UPDATE_SAMPLE = 'update_sample'
    FS_UPDATE_PIPELINE = 'update_pipeline'
    CHAT = 'chat'
    CHAT_RESPONSE = 'chat_response'


class ErrorCode(Enum):

    SUCCESS = 0, 'success'
    NOT_A_QUESTION = 1, 'query is not a question'
    NO_TOPIC = 2, 'The question does not have a topic. It might be a meaningless sentence.'  # noqa E501
    UNRELATED = 3, 'Topics unrelated to the knowledge base. Revising the question can improve accuracy..'  # noqa E501
    NO_SEARCH_KEYWORDS = 4, 'Cannot extract keywords.'
    NO_SEARCH_RESULT = 5, 'Cannot retrieve results.'
    BAD_ANSWER = 6, 'Irrelevant answer.'
    SECURITY = 7, 'Reply has a high relevance to prohibited topics.'
    NOT_WORK_TIME = 8, 'Non-working hours. The config.ini file can be modified to adjust this. **In scenarios where speech may pose risks, let the robot operate under human supervision**'  # noqa E501

    PARAMETER_ERROR = 9, "HTTP interface parameter error. Query cannot be empty; the format of history is list of lists, like [['question1', 'reply1'], ['question2'], ['reply2']]"  # noqa E501
    PARAMETER_MISS = 10, 'Missing key in http json input parameters.'

    WORK_IN_PROGRESS = 11, 'not finish'
    FAILED = 12, 'fail'
    BAD_PARAMETER = 13, 'bad parameter'
    INTERNAL_ERROR = 14, 'internal error'
    SEARCH_FAIL = 15, 'Search fail, please check TOKEN and quota'
    NOT_FIND_RELATED_DOCS = 16, 'No relevant documents found, the following answer is generated directly by LLM.'
    NON_COMPLIANCE_QUESTION = 17, 'Non-compliance question, refusing to answer.'
    NO_WEB_SEARCH_RESULT = 18, 'Can not fetch result from web.'

    def __new__(cls, value, description):
        """Create new instance of ErrorCode."""
        obj = object.__new__(cls)
        obj._value_ = value
        obj.description = description
        return obj

    def __int__(self):
        return self.value

    def describe(self):
        return self.description

    @classmethod
    def format(cls, code):
        if isinstance(code, cls):
            return {'code': int(code), 'message': code.describe()}
        raise TypeError(f'Expected type {cls}, got {type(code)}')


class LogManager:

    def __init__(self, log_path):
        self.log_path = log_path
        self.log_content_list = []

    def log(self, operation, outcome=''):
        self.log_content_list.append((operation, outcome))

    def __del__(self):
        try:
            with open(self.log_path, 'a', encoding='utf8') as file:
                for operation, outcome in self.log_content_list:
                    file.write(f'{operation}: {outcome}\n')
                file.write('\n')
        except Exception as e:
            print(e)