lib/llm_2.py

import re
import numpy as np
import requests
import json
import pandas as pd

typedf = pd.read_excel('types.xlsx')
verbose = False
# s = 'Тип обращения: Жалобы > Сектор: Акционерные общества/эмитенты эмиссионных ценных бумаг > Продукт: Дивиденды'
# s = 'Тип обращения: Прочие обращения (запрос, заявление, вопрос) > Сектор: Субъекты страхового дела > Продукт: Прочие обращения (страхование) > Проблема: Иные обращения: Страхование > Обращение может быть типологизировано по иной проблеме: Нет'
level = 'Тип обращения'
level = 'Сектор'
level = 'Продукт'
level = 'Проблема'
level = 'Проблема'

def getType(str, level) :
    type = ''
    t = level + ': [^>]+'
    # t = r'Сектор\: \w+/\w+ >'
    m = re.search(t, str)
    if m :
        substr = str[m.start() : m.end()]
        substr
        m = re.search('\: [\w ]+', substr)
        subsubstr = substr[m.start() + 2 : ]
        # subsubstr
        type = subsubstr.strip()
        
    return type

def getQuestionAnswer(str) :
    q = ''
    a = ''
    l = str.split('>')
    if len(l) == 5 :
        q = l[4][1 :]
        x = q.find(':')
        a = q[x + 2 :]
        q = q[: x]

    return q, a

types = []
sectors = []
products = {}
problems = {}
for index, row in typedf.iterrows() :
    text = row['Путь до вершины']
    text = str(text)
    if text != '' :
        apptype = getType(text, 'Тип обращения')
        sector = getType(text, 'Сектор')
        product = getType(text, 'Продукт')
        problem = getType(text, 'Проблема')
        sector = sector.replace('  ', ' ')
        product = product.replace('  ', ' ')
        problem = problem.replace('  ', ' ')
        
        if apptype == 'Жалобы' and sector != '' and product != '' and problem != '' :
            if sector not in sectors :
                sectors.append(sector)
                
            if sector not in products :
                products[sector] = []
        
            if product not in products[sector] :
                products[sector].append(product)
    
            if sector not in problems :
                problems[sector] = {}
    
            if product not in problems[sector] :
                problems[sector][product] = []
    
            if problem not in problems[sector][product] :
                problems[sector][product].append(problem)

def getCategory(text, categories) :
    found = False
    text = text.lower()
    for category in categories :
        if category.lower() in text :
            found = True
            break

    if found == False :
        category = ''

    return category

def getResponse(prompt) :
    url = "https://muryshev-mixtral-api.hf.space/completion"
    print('Prompt: '+prompt)
    payload = json.dumps({
      "prompt": '[INST]' + prompt + '[/INST]'
    })
    
    headers = {
      'Content-Type': 'application/json'
    }
    
    response = requests.request("POST", url, headers = headers, data = payload)
    result = response.content.decode('utf-8')
    return result

def getCategoryFromLLM(prompt, categories) :
    category = ''
    for j in range(5) :
        result = getResponse(prompt)
        category = getCategory(result, categories)
        if category != '' :
            break
    
        prompt += '.'
        
    return category, result

def getAccuracy(answers, trueanswers) :
    count = 0
    for i in range(len(trueanswers)) :
        if answers[i] == trueanswers[i] :
            count += 1

    return count / len(trueanswers)

def getAnswers(applications, prefix, categories, answers) :
    # print(categories)
    output = []
    for i in range(len(applications)) :
        text = applications[i]
        prompt = prefix + text
        category, response = getCategoryFromLLM(prompt, categories)
        
        answer = ''
        for j in range(len(categories)) :
            if category == categories[j] :
                answer = answers[j]
                break

        brief = response.replace('\n', '')
        if len(brief) > 80 :
            brief = brief[:80] + '...'
            
        if verbose :
            print(i, ':', answer, ' \tLLM output :', brief)
        
        output.append(answer)

    return output

def getSector(application) :

    sectortext = ''
    for j in range(len(sectors)) :
        sectortext += str(j) + '. ' + sectors[j] + '\n'  

    prompt = '''Ты мой помощник. Ты отвечаешь только на РУССКОМ языке. Ты сортируешь заявления клиентов. 
Ты не отвечаешь на вопросы, не комментируешь, не выражаешь эмоций, не выражаешь соображений по теме заявления. 
Ты извлекаешь информацию. Ты не анализируешь. 
Ты выполняешь только эту задачу: ты определяешь категорию заявления. Для этого ты используешь ТОЛЬКО список возможных категорий, который я тебе предоставляю. 
Ты выбираешь только ТУ категорию, которая на сто процентов соответсвует обращению. Проверь свой ответ дважды. 
Ты всегда используешь такой формат ответа: "название категории". 
Если в тексте обращения есть аббревиатуры "МФО", "МФК" или "МКК", ты должен выбрать категорию "Микрофинансовые организации". 
Если в тексте обращения есть аббревиатуры "ОСАГО" или "КАСКО", ты должен выбрать категорию "Субъекты страхового дела". 
Список категорий: 
''' + sectortext + '\nЗаявление: ' + application

    sector, response = getCategoryFromLLM(prompt, sectors)
    
    if verbose :
        print(i, ':', sector)

    return sector, prompt, response

def getProduct(application, sector) :
    product = ''
    # sector = appsectors[i]
    if sector != '' :
        subproducts = products[sector]
    
        producttext = ''
        for j in range(len(subproducts)) :
            producttext += str(j) + '. ' + subproducts[j] + '\n'
        
        prompt = 'Ты мой помощник. Ты отвечаешь только на РУССКОМ языке. Ты сортируешь заявления клиентов. Ты не отвечаешь на вопросы, не комментируешь, \
не выражаешь эмоций, не выражаешь соображений по теме обращения. Ты извлекаешь персональные данные. Ты не анализируешь. \
Ты выполняешь только эту задачу: \
ты определяешь категорию заявления. Для этого ты используешь ТОЛЬКО список возможных категорий, который я тебе предоставляю. \
Ты выбираешь только ТУ категорию, которая на сто процентов соответсвует заявлению. Проверь свой ответ дважды. \
Ты всегда используешь такой формат ответа: "название категории". \n\
Список категорий:\n' + producttext + '\nЗаявление: ' + application

        product, response = getCategoryFromLLM(prompt, subproducts)
        
        if verbose :
            print(product)

    return product, prompt, response

def getProblem(application, sector, product) :
    problem = ''
    if sector != '' and product != '':
        subpproblems = problems[sector][product]
    
        problemtext = ''
        for j in range(len(subpproblems)) :
            problemtext += str(j) + '. ' + subpproblems[j] + '\n'
        
        prompt = 'Ты мой помощник. Ты отвечаешь только на РУССКОМ языке. Ты сортируешь заявления клиентов. Ты не отвечаешь на вопросы, не комментируешь, \
не выражаешь эмоций, не выражаешь соображений по теме обращения. Ты извлекаешь персональные данные. Ты не анализируешь. \
Ты выполняешь только эту задачу: \
ты определяешь категорию заявления. Для этого ты используешь ТОЛЬКО список возможных категорий, который я тебе предоставляю. \
Ты выбираешь только ТУ категорию, которая на сто процентов соответсвует заявлению. Проверь свой ответ дважды. \
Ты всегда используешь такой формат ответа: "название категории". \n\
Список категорий:\n' + problemtext + '\nЗаявление: ' + application
    
        problem, response = getCategoryFromLLM(prompt, subpproblems)
        
        if verbose :
            print(problem)

    return problem, prompt, response

def getAuthor(application) :
    prefix = '''Ты мой помощник. Ты отвечаешь только на РУССКОМ языке. Ты не отвечаешь на вопросы, не комментируешь, 
не выражаешь эмоций, не выражаешь соображений по теме обращения. 
Ты извлекаешь информацию из заявления. Ты отвечаешь на МОЙ вопрос: 
"Кто является заявителем в заявлении?". Ты называешь имя заявителя в формате: "Заявитель: Фамилия Имя Отчество".
Если заявиитель не указан в заявлении, ты отвечаешь: "Заявитель: не указан". 
Ты не комментируешь, не обясняешь, не выражаешь мысли, вообще ничего больше не говоришь. 
Обращение: '''

    prompt = prefix + application
    # prompt = prefix + appdata.loc[i, 'Текст обращения обезличенный']
    response = getResponse(prompt)
    response = response.replace('.', '. ')
    name = 'не указан'
    if name not in response :
        m = re.search(r'Заявитель: [А-Я][а-я][\w\.]+ [А-Я][\w\.]+ [А-Я][\w\.]+', response)
        if m :
            name = response[m.start() + 11 : m.end()]
        else :
            m = re.search(r'Заявитель: [А-Я][а-я][\w]+ [А-Я][а-я][\w]+', response)
            if m :
                name = response[m.start() + 11 : m.end()]

    if verbose :
        print(name, '\n', response[:100].replace('\n', ' '))
        
    return name

def checkContractNumber(application) :
    categories = ['да', 'нет']
    answers = ['да', 'нет']
    
    prefix = '''Ты мой помощник. Ты отвечаешь только на РУССКОМ языке. Ты извлекаешь информацию из заявлений.
Ты не отвечаешь на вопросы, не комментируешь, не выражаешь эмоций, не выражаешь соображений по теме обращения. 
Ты только отвечаешь на МОЙ вопрос: "Имеется ли в заявлении указанный номер договора?". 
Ты отвечаешь либо ТАК "ответ: да, имеется" ЛИБО так "ответ: нет, не имеется". Конец ответа.
Если в заявлении нет слова "договор", ты отвечаешь "ответ: нет, не имеется"
Ты не комментируешь, не объясняешь, не выражаешь мысли, вообще ничего больше не говоришь.
Заявление: '''

    ifcontract = getAnswers([application], prefix, categories, answers)

    return ifcontract[0]

def checkIfIdentified(application) :
    сategories = ['нельзя', 'можно']
    answers = ['нет', 'да']

    prefix = '''Ты мой помощник. Ты отвечаешь только на РУССКОМ языке. 
Ты не отвечаешь на вопросы, не комментируешь, не выражаешь эмоций, не выражаешь соображений по теме жалобы. 
Ты ищешь в заявлении объект жалобы: "Можно ли идентицифировать в заявлении объект жалобы (тот, на кого жалуется заявитель)?". 
Твой ответ ВСЕГДА состоит из ТРЕХ слов: ты отвечаешь либо ТАК "да, можно", ЛИБО так "нет, нельзя". 
Ты не комментируешь, не объясняешь, не выражаешь мысли, вообще ничего больше не говоришь. 
Жалоба: '''
    
    ifidentified = getAnswers([application], prefix, сategories, answers)

    return ifidentified[0]

def checkIfPerson(application) :
    categories = ['физическое лицо', 'юридическое лицо']
    answers = ['Физ.лицо', 'Юр.лицо']
    
    prefix = '''Ты мой помощник. Ты отвечаешь только на РУССКОМ языке. Ты сортируешь "заявления" клиентов. 
Ты не отвечаешь на вопросы, не комментируешь, не выражаешь эмоций, не выражаешь соображений по теме обращения. 
Ты извлекаешь информацию. Ты не анализируешь. 
Ты отвечаешь ТОЛЬКО на мои вопросы. Ты определяешь кем является заявитель: "физическое лицо" или "юридическое лицо". 
Условие: если заявление написано в первом лице (местоимения Я, МНЕ, МНОЮ, МОЕ, МЕНЯ), то это физическое лицо, НО если заявление написано в третьем лице, то это юридическое лицо. 
Ты отвечаешь только так: "Заявитель: юрдическое лицо" или "Заявитель: физическое лицо".
Ты не комментируешь, не обясняешь, не выражаешь мысли, вообще ничего больше не говоришь.
Заявление: '''
    
    ifperson = getAnswers([application], prefix, categories, answers)

    return ifperson[0]

def checkIfcomission(application) :
    categories = ['не касается', 'касается']
    answers = ['нет', 'да']
    
    prefix = '''Ты мой помощник. Ты отвечаешь только на РУССКОМ языке. Ты извлекаешь информацию из заявлений.
Ты не отвечаешь на вопросы, не комментируешь, не выражаешь эмоций, не выражаешь соображений по теме обращения. 
Ты только отвечаешь на МОЙ вопрос: "Касается ли заявление комиссии за обслуживание рублевого счета?". 
Ты отвечаешь либо ТАК "ответ: да, касается" ЛИБО так "ответ: нет, не касается". Конец ответа.
Если в заявлении нет слова "комиссия", ты отвечаешь "ответ: нет, не касается"
Ты не комментируешь, не объясняешь, не выражаешь мысли, вообще ничего больше не говоришь.
Заявление: '''
    
    ifсomission = getAnswers([application], prefix, categories, answers)

    return ifсomission[0]

def getContractData(application) :
    prefix = '''Ты мой помощник. Ты отвечаешь только на РУССКОМ языке. Ты не отвечаешь на вопросы, не комментируешь,
не выражаешь эмоций, не выражаешь соображений по теме обращения.Ты выполняешь только эту задачу:
ты извлекаешь из заявления только *номер ДОГОВОРА* и "дата ДОГОВОРА".
Ты всегда используешь только такой формат: "Номер договора: *номер ДОГОВОРА*, Дата: *дата этого договора*;".
Если указан любой другой номер, но НЕ номер ДОГОВОРА, то ты отвечаешь так: "Номер договора не указан."
Ты должен убедиться, что слово "договор" присутствует рядом с указанным номером и исключить другие документы, такие как счета или заказы, 
например: "В соответствии с Договором № 0001 от 01.01.2022 года...". 
В этом примере номером договора является "0001" и датой договора является "01.01.2022".
Даты договоров должны быть указаны в формате "дд.мм.гггг", где "дд" - это число от 01 до 31, "мм" - число от 01 до 12, 
а "гггг" - четырехзначное число года. Между днями, месяцами и годами должны быть разделители, например, точки или тире.
Ты больше НИЧЕГО не говоришь, не комментируешь, не объясняешь, не добавляешь. 
Заявление: '''

    prompt = prefix + application
    response = getResponse(prompt)
    response = response.replace(';', '\n')
    response = response.replace('\\\\', '')
    l = response.split('\n')
    ll = []
    for s in l :
        s = s.strip()
        if 'Номер договора:' == s[:15] :
            ll.append(s)

    result = '\n'.join(ll)

    if result == '' :
        result = 'не указаны'

    if verbose :
        print(result)
        # print(name, '\n', response.strip()[:100].replace('\n', ' '))
        
    return result
    
def getPersons(application) :
#     prefix = '''Ты мой помощник. Ты отвечаешь только на РУССКОМ языке. Ты сортируешь заявления клиентов. Ты не отвечаешь на вопросы, не комментируешь,
# не выражаешь эмоций, не выражаешь соображений по теме заявления. Ты извлекаешь ВСЕ персональные данные из заявлений. Ты не анализируешь. 
# Ты выполняешь только эту задачу:
# ты читаешь заявление и извлекаешь из заявления все встретившиеся Фамилии Имена Отчества.
# Ты всегда используешь только такой формат: "ФИО пользователя". Перед ответом убедись, что "ФИО" - это человеческие фамилия, имя, отчество.
# Ты больше ничего не говоришь, не комментируешь, не объясняешь, не добавляешь. Отвечай так: "ФИО;"; ЛИБО: "ФИО не указано".
# Заявление: '''

    prefix = '''Ты мой помощник. Ты отвечаешь только на РУССКОМ языке. Ты излекаешь информацию из заявления. Ты не отвечаешь на вопросы, не комментируешь,
не выражаешь эмоций, не выражаешь соображений по теме заявления. Ты извлекаешь персональные данные. Ты не анализируешь. 
Ты выполняешь только эту задачу:
ты читаешь заявление и извлекаешь из заявления все встретившиеся Фамилии Имена Отчества.
Ты отвечаешь в формате: "ФИО: Фамилия Имя Отчество ;" или "ФИО: Фамилия И. О. ;".
Если в заявлении не указаны имена людей, ты отвечаешь "ФИО не указаны".
Перед ответом убедись, что "ФИО" - это человеческие фамилия, имя, отчество.
Ты больше ничего не говоришь, не комментируешь, не объясняешь, не добавляешь. 
Заявление: '''

    prompt = prefix + application
    # prompt = prefix + appdata.loc[i, 'Текст обращения обезличенный']
    response = getResponse(prompt)
    response = response.replace('указаны.', 'указаны')
    response = response.replace('.', '. ')
    response = response.replace(';', '\n')
    # response = response.replace('\\\\', '')
    l = response.split('\n')
    ll = []
    for s in l :
        s = s.strip()
        if 'ФИО: ' == s[:5] and 'ФИО: не указаны' not in s:
            ss = ''
            s = s[5:]
            s = re.sub('\(.+\)', '', s)
            s = s.replace('ч.', 'ч')
            s = s.replace('а.', 'а')
            s = s.replace('Президент Российской Федерации', '')
            s = s.replace(',', '').strip()
            # n = len(s.split(' '))
            # print(n, s.split(' '))
            # if n > 1 and n < 4 :
            #     ll.append(s)
            m = re.search(r'[А-Я][а-я][\w\.]+ [А-Я][\w\.]+ [А-Я][\w\.]+', s)
            if m :
                ss = s[m.start() : m.end()]
            else :
                m = re.search(r'[А-Я][а-я][\w]+ [А-Я][а-я][\w]+', s)
                if m :
                    ss = s[m.start(): m.end()]

            if ss != '' :
                ll.append(ss)

    result = '\n'.join(ll)

    if result == '' :
        result = 'не указаны'
        
    names = result

    if verbose :
        print(names, '\n', response[:100].replace('\n', ' '))
        
    return names
    
def getApplicationInfo(application) :
    sector, sector_prompt, sector_response = getSector(application)
    product, product_prompt, product_response = getProduct(application, sector)
    problem, problem_prompt, problem_response = getProblem(application, sector, product)
    author = getAuthor(application)
    persons = getPersons(application)
    contractdata = getContractData(application)
    
    ifcontract = checkContractNumber(application)
    ifidentified = checkIfIdentified(application)
    ifperson = checkIfPerson(application)
    ifcomission = checkIfcomission(application)
    
    app_info = {}
    app_info['Сектор'] = sector
    app_info['Продукт'] = product
    app_info['Проблема'] = problem
    app_info['Заявитель'] = author
    app_info['Физлица'] = persons
    app_info['Данные договора'] = contractdata
    app_info['Заявитель физическое или юридическое лицо?'] = ifperson
    app_info['Можно ли идентифицировать лицо, на которого пожаловались?'] = ifidentified
    app_info['Указан ли в обращении номер договора?'] = ifcontract
    app_info['Жалоба касается комиссии за обслуживание рублевого счета?'] = ifcomission
    
    debug_info = {
        'sector_prompt': sector_prompt,
        'sector_response': sector_response,
        'product_prompt': product_prompt,
        'product_response': product_response,
        'problem_prompt': problem_prompt,
        'problem_response': problem_response,
    }
    
    if verbose :
        print()
        print(i)
        print('Сектор', sector)
        print('Продукт', product)
        print('Проблема', problem)
        print('Заявитель', author)
        print('Физлица', persons)
        print('Данные договора', contractdata)
        print('Заявитель физическое или юридическое лицо?', ifperson)
        print('Можно ли идентифицировать лицо, на которого пожаловались?', ifidentified)
        print('Указан ли в обращении номер договора?', ifcontract)
        print('Жалоба касается комиссии за обслуживание рублевого счета?', ifcomission)

    return {'application': app_info, 'debug': debug_info}

# application = '''Я, Кристенгоф Оксана Михайловна, заключила кредитный договор с ПАО Камабанк № 59875 и № 456835645645654 
# и договор на выдачу кредитной карты с ПАО "БСТ-Банк" №4567652124534235 и №4561254153412. 
# 12.08.2020 года направила в банки заявление на истребование кредитной документации заказным письмом с уведомлением, 
# но банки мне так ничего и не прислали. Прошу поспособствовать в том,что кредитные организации выслали мне кредитные договоры.
# '''

# verbose = False
# app_info = getApplicationInfo(application)
# print(app_info)

# Результат: 
# {'Сектор': 'Кредитные организации',
#  'Продукт': 'Качество обслуживания',
#  'Проблема': 'Проблемы, связанные с качеством обслуживания',
#  'Заявитель': 'Кристенгоф Оксана Михайловна',
#  'Физлица': 'Кристенгоф Оксана Михайловна',
#  'Данные договора': 'Номер договора: 59875, Дата: 12.08.2020\nНомер договора: 456835645645654, Дата: 12.08.2020',
#  'Заявитель физическое или юридическое лицо?': 'Физ.лицо',
#  'Можно ли идентифицировать лицо, на которого пожаловались?': 'да',
#  'Указан ли в обращении номер договора?': 'да',
#  'Жалоба касается комиссии за обслуживание рублевого счета?': 'нет'}