31.10.2024 в 12:31

GenAI в области очистки данных: первые шаги

В этой статье мы рассмотрим, как Generative AI (GenAI) может ускорить очистку данных, в частности, для очистки адресов электронной почты и дней рождения. Мы решаем обычные проблемы использования регулярных выражений и показываем, как GenAI может вмешаться, чтобы сделать все проще и быстрее. Вы найдете фрагменты кода для генерации поддельных данных, добавления некоторых ошибок и использования Amazon Bedrock для интеллектуального исправления ошибок. Результаты? GenAI отлично справился со всеми исправлениями дней рождения, но некоторые электронные письма споткнулись из-за специальных символов.

Создание образца: очистка адресов электронной почты и дней рождения

Мы создадим образцы из внутренней системы. Они будут содержать такие данные, как электронные письма и дни рождения наших клиентов. К сожалению, наша система ввода не проверяет данные, которые клиенты вводят в нашу систему. Поэтому клиенты иногда отправляют неверные данные в нашу CRM.

Производитель данных: создание данных для целей оценки

Наша система будет имитирована этим скриптом Python. Это создаст допустимые или ожидаемые входные данные с точностью 100%. После этого мы внесем типичные ошибки и опечатки с вероятностью 50% в записи нашего набора данных. Начнем с писем.

src/dataproducer.py

def generate_random_email(self):
    first_name = random.choice(self.first_names)
    last_name = random.choice(self.last_names)
    domains = ['example.com', 'test.com', 'sample.com', 'demo.com']
    email = f"{first_name}.{last_name}@{random.choice(domains)}"
    return email.lower()

def introduce_typo_email(self, email: str):
    if random.random() < 0.5:
        email_list = list(email)
        index = random.randint(0, len(email) - 1)
        action = random.choice(['replace', 'add', 'remove', 'case_change'])
        if action == 'replace':
            email_list[index] = random.choice('abcdefghijklmnopqrstuvwxyz')
        elif action == 'add':
            email_list.insert(index, random.choice('abcdefghijklmnopqrstuvwxyz'))
        elif action == 'remove' and len(email) > 1:
            email_list.pop(index)
        elif action == 'case_change':
            email_list[index] = email_list[index].upper() if email_list[index].islower() else email_list[index].lower()
        return ''.join(email_list)
    return email

Функция generate_random_email() создает случайные электронные письма на основе предопределенного набора немецких имен и фамилий и четырех разных поставщиков электронной почты. Функция introduce_typo_email() вносит ошибку с вероятностью 50%, заменяя, добавляя или удаляя букву из электронного письма. Буквы также можно менять на заглавные.

Второе поле данных — это день рождения наших клиентов.

def generate_correct_birthday(self):
        start_date = datetime(1900, 1, 1)
        end_date = datetime(2023, 12, 31)
        random_date = start_date + timedelta(days=random.randint(0, (end_date - start_date).days))
        return random_date.strftime("%d.%m.%Y")

def introduce_typo_birthday(self, birthday):
    if random.random() < 0.5:
        birthday = birthday.lstrip('0')
    if random.random() < 0.5:    
        birthday = birthday.replace('.', random.choice([' ', '/', '-']))
    if random.random() < 0.5:
        if "19" in birthday or "20" in birthday:
            year = birthday.split('.')[-1]
            if year.startswith("19") and random.choice([True, False]):
                year = year[2:]
            elif year.startswith("20") and random.choice([True, False]):
                year = year[2:]
            birthday = '.'.join(birthday.split('.')[:-1] + [year])
    return birthday

Функция generate_correct_birthday() создает дни рождения в формате ДД.ММ.ГГГГ. Каждая дата будет находиться в диапазоне от 01.01.1900 до 31.12.2023. Опечатки, которые мы вносим, могут включать изменение разделителя с "." на "/" или "-" или даже удаление любого разделителя. Мы удалим начальные нули или "19" и "20".

Наконец, мы запишем набор данных через CSV. Для этого поста в блоге мы ограничим количество строк до 100.

def create_pandas_dataframe(self):
    emails_correct = [self.generate_random_email() for _ in range(self.rows)]
    emails_typo = [self.introduce_typo_email(email=email) for email in emails_correct]
    birthdays_correct = [self.generate_correct_birthday() for _ in range(self.rows)]
    birthdays_typo = [self.introduce_typo_birthday(birthday=birthday) for birthday in birthdays_correct]
    data = {
        'EmailCorrect': emails_correct,
        'EmailTypo': emails_typo,
        'BirthdayCorrect': birthdays_correct,
        'BirthdayTypo': birthdays_typo,
    }
    return pd.DataFrame(data)

def write_as_csv(self, dataframe: pd.DataFrame):
    dataframe.to_csv(path_or_buf='data/out.csv', index=False)

Выполняем фрагменты из файла src/datacleaner.py.

src/datacleaner.py

def create_data():
    logging.info("Creating DataFrame")
    data_producer = dp.DataProducer(
        rows=100,
        first_names=['Max', 'Sophie', 'Leon', 'Marie', 'Paul', 'Emma', 'Lukas', 'Hannah', 'Tim', 'Anna'],
        last_names=['Müller', 'Schmidt', 'Schneider', 'Fischer', 'Weber', 'Meyer', 'Wagner', 'Becker', 'Hoffmann', 'Schulz'],
    )
    df = data_producer.create_pandas_dataframe()
    data_producer.write_as_csv(dataframe=df)
    logging.info("Wrote down DataFrame as csv")

if __name__ == "__main__":
    create_data()

Очистка данных на основе GenAI: вызов Claude Haiku для динамической оценки RegEX во время выполнения

Чтобы реализовать динамическое RegEX на основе GenAI во время выполнения, нам нужно обработать четыре вещи. Во-первых, нам нужно создать приглашение, содержащее соответствующую информацию. Во-вторых, нам нужно подготовить данные, чтобы их могла обработать наша модель GenAI. Наконец, нам нужно вызвать конечную точку Amazon Bedrock. Наконец, результат из конечной точки должен соответствовать и быть передан в наш DataFrame.

Запрос, извлечение и запуск реализации

Следующий фрагмент показывает, как создавать динамические приглашения для различных полей данных. Мы пометили инструкции и данные тегами XML, как предложено на сайте.

def prompt_builder(field_name: str, pattern: str, data: list[str]):
    prompt = f"""<instructions>You are in charge of ensuring data quality. 
Check the following data fields. They represent {field_name} 
from our CRM of our German customers. Fix all fields if they show typos or other mistakes. Ensure your answers follow this pattern: {pattern}.
Return only a comma-separated list and keep the order of the input data.</instructions> <data>{data}</data>"""

    return prompt

Идея состоит в том, чтобы создать ролевой подход, который гарантирует, что задача будет сформулирована четко и прямо, а также будет точным указание ожидаемого результата и, в нашем случае, также порядка.

Список, разделенный запятыми, является подготовкой к интеграции результатов в наш существующий DataFrame. Для этого реализован следующий код:

def process_output(bedrock_response: dict, json_parsable: bool = False):
    output = bedrock_response['body'].read()
    output = output.decode('utf-8')
    output = json.loads(output)['content'][0]['text']
    if json_parsable:
        output = json.loads(output)
    return [elem.strip() for elem in output.split(",")]

Важно отметить, что результат выражается на естественном языке. Вот почему нужно немного обработать вывод, чтобы извлечь элементы из текста. Но обработка относительно проста, так как она включает удаление начальных пробелов перед каждым словом ("05.12.1988").

Теперь, когда мы объявили обработку ввода и вывода, нам осталось только вызвать конечную точку:

def invoke_endpoint(bedrock_client: boto3.client, prompt: str, top_k: int, max_tokens: int, top_p: int, temperature: int):
    body = json.dumps(
        {
            'messages': [{'role': 'user', 'content': [{'type': 'text', 'text': prompt}]}],
            'anthropic_version': 'bedrock-2023-05-31',
            'max_tokens': max_tokens,
            'temperature': temperature,
            'top_p': top_p,
            'top_k': top_k,
        }
    )
    modelId = 'anthropic.claude-3-haiku-20240307-v1:0'
    accept = "*/*"
    contentType = "application/json"
    response = bedrock_client.invoke_model(
        body=body,
        modelId=modelId,
        accept=accept,
        contentType=contentType,
    )
    return response

И вуаля, окончательный сценарий:

src/datacleaner.py

def invoke_endpoint(bedrock_client: boto3.client, prompt: str, top_k: int, max_tokens: int, top_p: int, temperature: int):
    body = json.dumps(
        {
            'messages': [{'role': 'user', 'content': [{'type': 'text', 'text': prompt}]}],
            'anthropic_version': 'bedrock-2023-05-31',
            'max_tokens': max_tokens,
            'temperature': temperature,
            'top_p': top_p,
            'top_k': top_k,
        }
    )
    modelId = 'anthropic.claude-3-haiku-20240307-v1:0'
    accept = "*/*"
    contentType = "application/json"
    response = bedrock_client.invoke_model(
        body=body,
        modelId=modelId,
        accept=accept,
        contentType=contentType,
    )
    return response

def prompt_builder(field_name: str, pattern: str, data: list[str]):
    prompt = f"""<instructions>You are in charge of ensuring data quality. 
Check the following data fields. They represent {field_name} 
from our CRM of our German customers. Fix all fields if they show typos or other mistakes. Ensure your answers follow this pattern: {pattern}.
Return only a comma-separated list and keep the order of the input data.</instructions> <data>{data}</data>"""

    return prompt

def process_output(bedrock_response: dict, json_parsable: bool = False):
    output = bedrock_response['body'].read()
    output = output.decode('utf-8')
    output = json.loads(output)['content'][0]['text']
    if json_parsable:
        output = json.loads(output)
    return [elem.strip() for elem in output.split(",")]

def gen_ai_processing(create_data: bool, aws_region: str, data_path: str):
    config = Config(
        region_name=aws_region,
    )

    logging.info("Creating or reading data ...")
    if create_data:
        create_data()
        logging.info("Created data ...")
    else:
        df = pd.read_csv(filepath_or_buffer=data_path)
        logging.info("Read data ...")

    bedrock_client = boto3.client("bedrock-runtime", config=config)

    logging.info("Running Bedrock inference for data cleansing")
    for entry in [
        {
            "FieldName": "Email",
            "Pattern": "first_name.last_name@company.com",
            "ColumnTypo": "EmailTypo",
            "ColumnCorrect": "EmailCorrect"
        },
        {
            "FieldName": "Birthday",
            "Pattern": "DD.MM.YYYY",
            "ColumnTypo": "BirthdayTypo",
            "ColumnCorrect": "BirthdayCorrect"
        }
    ]:
        prompt = prompt_builder(
            field_name=entry["FieldName"],
            pattern=entry["Pattern"],
            data=df[entry["ColumnTypo"]].tolist()
        )

        logging.info(f"Running Bedrock inference for {entry['FieldName']}")

        llm_res = invoke_endpoint(
            bedrock_client=bedrock_client,
            prompt=prompt,
            temperature=0.2, 
            top_k=100, 
            top_p=0.2, 
            max_tokens=1024
        )

        processed_output = process_output(bedrock_response=llm_res, json_parsable=False)

        df[f"{entry['FieldName']}Bedrock"] = processed_output

        comparison_column = np.where(df[f"{entry['FieldName']}Bedrock"] == df[entry["ColumnCorrect"]], True, False)
        df[f"CompareBedrockOrg{entry['FieldName']}"] = comparison_column

    df.to_csv(path_or_buf='data/bedrock_out.csv', index=False)  

if __name__ == "__main__":
    gen_ai_processing(
        create_data=False,
        aws_region="eu-central-1",
        data_path="data/out.csv",
    )

Анализ результатов

Теперь пришло время провести анализ нашей динамической реализации RegEX с помощью GenAI. Поскольку мы знаем Ground Truth, мы можем просто сравнить с ней результаты Bedrock. Изначально у нас есть 100 записей для дней рождения и адресов электронной почты. Скрипт создал 45 адресов электронной почты и 31 день рождения с опечатками.

С помощью Bedrock все дни рождения можно исправить:

INFO:root:   BirthdayCorrect BirthdayTypo BirthdayBedrock  CompareBedrockOrgBirthday
0       06.06.2006    6/06/2006      06.06.2006                       True
5       22.12.1957   22 12 1957      22.12.1957                       True
6       23.03.1985     23.03.85      23.03.1985                       True
11      22.07.2010   22.07.2010      22.07.2010                       True
14      28.03.1970   28/03/1970      28.03.1970                       True

При этом 5 писем исправить не удалось:

INFO:root:                EmailCorrect                 EmailTypo              EmailBedrock  CompareBedrockOrgEmail
0       tim.weber@sample.com       ti.weber@sample.com       ti.weber@sample.com                   False
5      marie.müller@test.com     marie.müller@test.com    marie.mueller@test.com                   False
6   sophie.hoffmann@demo.com   sophie.offmann@demo.com   sophie.offmann@demo.com                   False
11    hannah.müller@test.com   hannah.müller@testf.com   hannah.mueller@test.com                   False
14   paul.müller@example.com  payul.müller@example.com  paul.mueller@example.com                   False

Однако, если присмотреться, можно увидеть, что Bedrock исправил 3 из 5 ошибок правильно, поскольку электронные письма не могут содержать немецких спецсимволов. Таким образом, два имени («Тим» и «Хоффманн») можно было исправить.

Заключительные слова

Набор данных небольшой. В более крупных настройках фрагментация будет иметь решающее значение, поскольку Bedrock имеет строгие ограничения для пакетных подходов. Кроме того, примеры относительно просты. Более сложные поля данных могут показывать более низкие показатели успеха. Воспроизведение показало очень хорошие результаты в пяти раундах. Однако в более крупных наборах это необходимо оценить.

#Python #Автоматизация

Войти на сайт