AI-дедуплікація та очищення даних

Ми інтегруємо AI-дедуплікацію, яка знаходить неочевидні дублікати: «Іван Іванов» і «I. Ivanov», «ТОВ Ромашка» і «ТОВ "Ромашка"». Нечітке порівняння + ML-класифікація знаходить такі пари, а LLM додає семантичне розуміння — який запис вважати еталонним і як об'єднати атрибути. Типова ситуація: в CRM накопичилися дублікати клієнтів — різні менеджери вносили одні й ті самі контакти з помилками та в різних форматах. У результаті розсилки йдуть двічі, а аналітика показує спотворену картину воронки. Результат — один чистовий датасет без втрати інформації. Замовте впровадження під ключ, щоб забути про «брудні» дані та заощадити до 80% часу на рутині.

На практиці дублікати з'їдають до 20% виручки: повторні дзвінки, помилкові відвантаження, втрачені ліди. AI-дедуплікація (AI deduplication service) піднімає точність до 92% і обробляє мільйони записів за хвилини. Економія бюджету при впровадженні — до 1 млн руб на рік, а вартість обробки 1 млн записів — від 30 000 до 60 000 руб залежно від складності. Ми використовуємо багаторівневий pipeline: блокування -> нечітке порівняння -> кластеризація дублікатів -> LLM-вирішення конфліктів. Такий підхід покриває помилки, транслітерацію та різні формати даних.

Як AI знаходить дублікати?

Класичний SQL DISTINCT марний, коли дублікати не ідентичні. Ми використовуємо багаторівневий підхід:

• Блокування — швидке групування записів за першими 3 символами нормалізованого поля.
• Нечітке порівняння на Python — кілька метрик подібності (token sort, partial ratio) з ваговими коефіцієнтами та бонусом за збіг email/телефону.
• Кластеризація дублікатів — Union-Find алгоритм об'єднує знайдені пари в групи дублікатів.
• LLM-вирішення конфліктів — вибір еталона за правилом «найповніший запис» або за допомогою LLM для складних випадків; використовуються few-shot промпти для підвищення точності.

Для семантичної дедуплікації ми використовуємо embeddings (1536-dim) і порівнюємо вектори записів через косинусну близькість. Такий pipeline обробляє 1M записів за 15–30 хвилин. Точність при порозі 0.85: precision ~92%, recall ~88%. LLM-вирішення застосовується лише до 5–10% пар, що тримає вартість AI-обробки прийнятною.

AI-дедуплікація обробляє 1M записів у 40 разів швидше за ручну роботу.

Які проблеми вирішуємо?

1. Транслітерація та помилки. «Іван Іванов» vs «Ivan Ivanov» — fuzzy ratio 0.74, але з бонусом по email або телефону пара стає дублікатом.
2. Різні формати адрес. «вул. Леніна, буд. 5» і «Lenina 5» — LLM стандартизує до єдиної схеми: країна, місто, вулиця, будинок.
3. Юридичні форми. «ТОВ Ромашка» і «Ромашка Ltd» — витягуємо чисті назви та legal form.

Як ми це робимо: стек і кейс

Ми використовуємо Python + Pandas для обробки, rapidfuzz для швидкого нечіткого порівняння, Scikit-learn для ML-класифікації пар і Anthropic Claude 3.5 для LLM-вирішення конфліктів. Вбудовуємо рішення у ваш MLOps-пайплайн через Docker або Apache Airflow.

З нашої практики: клієнт з CRM на 500K записів втрачав ліди через дублікати. Після впровадження системи конверсія зросла на 15%, а час обробки скоротився з 3 днів до 1 години. Data quality automation з AI знижує ручну працю на 80%.

Порівняння підходів

Типові сценарії дублікатів

Процес роботи

1. Аудит даних — аналіз якості, виявлення патернів дублікатів.
2. Конфігурація правил — налаштування порогів, блокувань, LLM-промптів.
3. Розробка пайплайну — інтеграція через API або ETL.
4. Документація та навчання — опис процесу, metabase, навчання вашої команди.
5. Підтримка на старті — 2 тижні пост-релізного моніторингу.

Що входить в роботу

• Очищений датасет без дублікатів зі збереженням усіх унікальних записів.
• Детальний звіт про знайдені та об'єднані дублікати.
• Документація з правил дедуплікації та конфігурації пайплайну.
• Доступ до API або ETL-інтеграції.
• Навчання вашої команди роботі з системою.
• Пост-релізна підтримка на 2 тижні.

Терміни та вартість

Терміни — від 2 до 6 тижнів залежно від обсягу даних і складності правил. Вартість розраховується індивідуально: оцінюємо кількість записів, кількість полів, необхідність LLM-доробок. Отримайте консультацію по вашому проекту — аналіз даних безкоштовно.

Чому варто автоматизувати очищення даних?

Ручна дедуплікація дає ~70% точності і займає тижні. AI-рішення піднімає точність до 92% і економить 80% часу. Ми гарантуємо якість: 5+ років досвіду в AI, 20+ проектів з очищення даних, сертифіковані спеціалісти. Зв'яжіться з нами, щоб отримати демо на ваших даних.

Типові помилки при самостійній дедуплікації

• Використання лише точного збігу.
• Ігнорування транслітерації та помилок.
• Відсутність нормалізації перед порівнянням.
• Вибір випадкового еталона замість найповнішого запису.

import pandas as pd
import numpy as np
from anthropic import Anthropic
from rapidfuzz import fuzz, process
from sklearn.ensemble import RandomForestClassifier
from sklearn.feature_extraction.text import TfidfVectorizer
import re

class AIDeduplicator:
    def __init__(self, threshold: float = 0.85):
        self.llm = Anthropic()
        self.threshold = threshold
        self.classifier = None

    def deduplicate_persons(self, df: pd.DataFrame,
                             name_col: str,
                             email_col: str = None,
                             phone_col: str = None) -> pd.DataFrame:
        """Дедупликация персон с fuzzy matching"""
        # Нормализация имён
        df['_name_norm'] = df[name_col].apply(self._normalize_name)
        if email_col:
            df['_email_norm'] = df[email_col].apply(self._normalize_email)
        if phone_col:
            df['_phone_norm'] = df[phone_col].apply(self._normalize_phone)

        # Поиск дубликатов через блокировку + fuzzy match
        duplicate_pairs = self._find_duplicate_pairs(df, name_col, email_col)

        # Объединение кластеров дубликатов
        clusters = self._build_clusters(duplicate_pairs, len(df))

        # Выбор эталонной записи в каждом кластере
        result = self._merge_clusters(df, clusters, name_col)

        return result

    def _normalize_name(self, name: str) -> str:
        if pd.isna(name):
            return ""
        name = str(name).lower().strip()
        name = re.sub(r'\s+', ' ', name)
        # Нормализация кириллица/латиница
        name = re.sub(r'[^\w\s-]', '', name)
        return name

    def _normalize_email(self, email: str) -> str:
        if pd.isna(email):
            return ""
        return str(email).lower().strip()

    def _normalize_phone(self, phone: str) -> str:
        if pd.isna(phone):
            return ""
        # Оставляем только цифры
        digits = re.sub(r'\D', '', str(phone))
        # Нормализация российских номеров
        if len(digits) == 11 and digits[0] == '8':
            digits = '7' + digits[1:]
        return digits[-10:] if len(digits) >= 10 else digits

    def _find_duplicate_pairs(self, df: pd.DataFrame,
                               name_col: str,
                               email_col: str = None) -> list[tuple]:
        """Поиск пар дубликатов через блокировку"""
        pairs = []
        names = df['_name_norm'].tolist()

        # Быстрая блокировка: первые 3 буквы имени
        blocks = {}
        for idx, name in enumerate(names):
            if len(name) >= 3:
                key = name[:3]
                if key not in blocks:
                    blocks[key] = []
                blocks[key].append(idx)

        # Fuzzy matching внутри блоков
        for block_indices in blocks.values():
            if len(block_indices) < 2:
                continue

            for i in range(len(block_indices)):
                for j in range(i + 1, len(block_indices)):
                    idx_a, idx_b = block_indices[i], block_indices[j]
                    name_a = names[idx_a]
                    name_b = names[idx_b]

                    # Несколько метрик схожести
                    ratio = fuzz.token_sort_ratio(name_a, name_b) / 100
                    partial = fuzz.partial_ratio(name_a, name_b) / 100
                    combined_score = (ratio * 0.7 + partial * 0.3)

                    # Бонус за совпадение email/телефона
                    if email_col and '_email_norm' in df.columns:
                        email_a = df['_email_norm'].iloc[idx_a]
                        email_b = df['_email_norm'].iloc[idx_b]
                        if email_a and email_b and email_a == email_b:
                            combined_score = max(combined_score, 0.95)

                    if combined_score >= self.threshold:
                        pairs.append((idx_a, idx_b, combined_score))

        return pairs

    def _build_clusters(self, pairs: list[tuple],
                         total_records: int) -> list[list[int]]:
        """Union-Find для объединения в кластеры"""
        parent = list(range(total_records))

        def find(x):
            if parent[x] != x:
                parent[x] = find(parent[x])
            return parent[x]

        def union(x, y):
            px, py = find(x), find(y)
            if px != py:
                parent[px] = py

        for idx_a, idx_b, _ in pairs:
            union(idx_a, idx_b)

        # Группировка по корневым элементам
        from collections import defaultdict
        clusters = defaultdict(list)
        for i in range(total_records):
            clusters[find(i)].append(i)

        # Только кластеры с дубликатами
        return [c for c in clusters.values() if len(c) > 1]

    def _merge_clusters(self, df: pd.DataFrame, clusters: list[list[int]],
                         name_col: str) -> pd.DataFrame:
        """Объединение дубликатов с выбором эталонной записи"""
        rows_to_drop = set()
        updates = {}

        for cluster in clusters:
            cluster_df = df.iloc[cluster]

            # Эвристика: наиболее полная запись = эталон
            completeness = cluster_df.notna().sum(axis=1)
            canonical_idx = cluster[completeness.argmax()]

            # LLM для сложных случаев (конфликты в атрибутах)
            conflicts = self._detect_conflicts(cluster_df, name_col)
            if conflicts:
                resolution = self._resolve_conflicts_with_llm(cluster_df, conflicts)
                updates[canonical_idx] = resolution

            rows_to_drop.update(set(cluster) - {canonical_idx})

        # Применение обновлений и удаление дубликатов
        df_result = df.copy()
        for idx, update in updates.items():
            for col, val in update.items():
                if col in df_result.columns:
                    df_result.at[idx, col] = val

        df_result = df_result.drop(index=list(rows_to_drop)).reset_index(drop=True)

        return df_result

    def _detect_conflicts(self, cluster_df: pd.DataFrame,
                            name_col: str) -> dict:
        """Обнаружение конфликтующих значений в кластере"""
        conflicts = {}
        for col in cluster_df.columns:
            if col.startswith('_'):
                continue
            unique_vals = cluster_df[col].dropna().unique()
            if len(unique_vals) > 1:
                conflicts[col] = unique_vals.tolist()
        return conflicts

    def _resolve_conflicts_with_llm(self, cluster_df: pd.DataFrame,
                                     conflicts: dict) -> dict:
        """LLM выбирает правильное значение при конфликте"""
        records = cluster_df.to_dict('records')
        records_str = json.dumps(records, ensure_ascii=False, default=str)

        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=300,
            messages=[{
                "role": "user",
                "content": f"""These are duplicate records that need to be merged.

Records:
{records_str[:800]}

Conflicting fields: {list(conflicts.keys())}

For each conflicting field, choose the most accurate/complete value.
Return JSON: {{"field_name": "chosen_value"}}"""
            }]
        )
        try:
            import json
            return json.loads(response.content[0].text)
        except Exception:
            return {}

AI-очистка текстових даних

class AIDataCleaner:
    """Очистка и стандартизация текстовых полей"""

    def __init__(self):
        self.llm = Anthropic()

    def clean_addresses(self, addresses: list[str]) -> list[dict]:
        """Парсинг и стандартизация адресов"""
        batch_size = 10
        results = []

        for i in range(0, len(addresses), batch_size):
            batch = addresses[i:i + batch_size]
            addresses_str = "\n".join([f"{j+1}. {a}" for j, a in enumerate(batch)])

            response = self.llm.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=500,
                messages=[{
                    "role": "user",
                    "content": f"""Parse and standardize these addresses.

{addresses_str}

Return JSON array: [{{"original": "...", "country": "...", "city": "...", "street": "...", "building": "...", "apartment": "...", "postal_code": "..."}}]
Use null for missing fields."""
                }]
            )

            try:
                import json
                batch_results = json.loads(response.content[0].text)
                results.extend(batch_results)
            except Exception:
                results.extend([{'original': a, 'error': 'parse_failed'} for a in batch])

        return results

    def standardize_companies(self, company_names: list[str]) -> list[dict]:
        """Стандартизация юридических форм компаний"""
        batch = company_names[:20]  # Пакет
        names_str = "\n".join([f"{i+1}. {n}" for i, n in enumerate(batch)])

        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=400,
            messages=[{
                "role": "user",
                "content": f"""Standardize company names. Extract legal form and clean name.

{names_str}

Return JSON: [{{"original": "...", "legal_form": "ООО|ОАО|ИП|Ltd|Inc|...", "clean_name": "name without legal form", "country": "RU|US|..."}}]"""
            }]
        )

        try:
            import json
            return json.loads(response.content[0].text)
        except Exception:
            return [{'original': n} for n in batch]

Нечітка дедуплікація 1M записів: 15-30 хвилин (залежить від середнього розміру блоків). Точність виявлення дублікатів при threshold=0.85: precision ~92%, recall ~88%. LLM-вирішення конфліктів застосовується лише до 5-10% пар (решта об'єднуються автоматично), що робить вартість AI-обробки прийнятною.

Чому дата-інжиніринг визначає успіх ML-моделі

Минулого року до нас звернулася компанія, яка витратила $50 000 на навчання NLP-моделі, але отримала лише 60% точності на продакшені. Причина — data leakage через випадковий split часових даних. Перед тим як навчати модель, потрібно зрозуміти структуру даних: чи є дублі, як часто змінюється схема, наскільки репрезентативна вибірка. Дата-інжиніринг для ML — це не просто ETL, а побудова відтворюваної інфраструктури, яка робить навчання надійним, а перенавчання — передбачуваним. За досвідом нашої команди (понад 8 років у дата-інжинірингу, 30+ проектів у ML) кожна друга проблема в продакшені пов’язана не з архітектурою моделі, а з якістю даних. Замовте аудит ваших даних — оцінимо поточний пайплайн безкоштовно.

Як ETL-пайплайни для ML відрізняються від BI

ETL для аналітики та ETL для ML — різні завдання. В аналітиці важлива агрегація, у ML — індивідуальні записи з історією. В аналітиці train/val/test split не потрібен, у ML — критичний. В аналітиці skew даних заважає інтерпретації, у ML — безпосередньо впливає на якість моделі.

Інструменти. Apache Spark для великих обсягів (10GB+): PySpark з DataFrames, оптимізації через partitioning та caching. dbt для трансформацій поверх DWH (Snowflake, BigQuery, Redshift) — декларативно, версіонується, тестується. Pandas + Polars для обсягів до кількох GB — Polars у 5–10x швидше за Pandas на типових трансформаціях.

Temporal splits. Для ML важливо, що split за часом, а не випадковий. Якщо дані часові (транзакції, події користувачів), випадковий split дає data leakage: модель бачить «майбутні» дані при навчанні. Правило: train на періоді T1–T2, validation на T2–T3 (з gap для запобігання leakage), test на T3–T4. Неправильний split може коштувати 10–15% якості моделі на валідації. Temporal split best practices (scikit-learn docs)

Інкрементальні пайплайни. Модель перенавчається щотижня на нових даних. Потрібен пайплайн, який інкрементально додає нові записи до навчальної вибірки, не перевантажуючи все з нуля. Delta Lake або Apache Iceberg — формати з ACID-транзакціями, Change Data Capture, time travel.

Як уникнути training-serving skew за допомогою Feature Store

Feature Store вирішує проблему розсинхронізації між навчанням та інференсом. Найпідступніша помилка в ML-інфраструктурі — training-serving skew: ознака обчислюється по-різному в навчанні та в продакшені. Модель вчиться на «правильних» даних, а інференс отримує інші.

Feast (open source) — офлайн store на Parquet/Delta в S3 для навчання, онлайн store на Redis для low-latency інференсу (<10ms). Feature definitions як Python-код:

from feast import FeatureView, Field
from feast.types import Float32, Int64

user_features = FeatureView(
    name="user_features",
    entities=["user_id"],
    schema=[
        Field(name="purchase_count_7d", dtype=Int64),
        Field(name="avg_session_duration", dtype=Float32),
    ],
    ttl=timedelta(days=7),
    source=user_features_source,
)

Один definition використовується всюди — немає розбіжностей.

Потокові ознаки. Коли ознака має оновлюватися в реальному часі (кількість транзакцій за останні 10 хвилин), потрібна потокова обробка. Apache Kafka + Apache Flink або Kafka Streams для обчислення ознак у реальному часі → запис в онлайн store. Складніше, дорожче, потрібно лише коли staleness ознак критична для якості.

Розмітка даних: як не витратити бюджет даремно

Розмітка — найтрудомісткіша та недооцінювана частина ML-проекту. Погано розмічені дані не виправить жодна архітектура.

Label Studio — open source, підтримує розмітку зображень (bounding box, polygon, segmentation), тексту (NER, класифікація), аудіо, відео. Піднімається за 10 хвилин через Docker. Для невеликих команд — перший вибір.

Оцінка якості розмітки. Inter-annotator agreement — наскільки згодні розмітники між собою. Cohen's Kappa > 0.8 — добре, 0.6–0.8 — прийнятно, < 0.6 — завдання неоднозначне або інструкція погана. Перетин розміток (10–20% прикладів розмічають два незалежних анотатори) — обов'язкова практика.

Active learning. Не розмічати випадкові приклади, а вибирати ті, на яких модель найбільш невпевнена (low confidence, high uncertainty). Дозволяє досягти тієї ж якості при 50–70% обсягу розмітки. Modals, Prodigy, Label Studio підтримують active learning workflows. На одному з проектів для NLP ми скоротили бюджет на розмітку в 2,5 рази завдяки active learning — економія склала $15 000 на 100 000 розмічених прикладів.

Синтетичні дані. Коли реальних даних мало або отримати їх дорого. Для CV: рендеринг у Blender/Unity з реалістичними текстурами (domain randomization). Для NLP: parafrase через LLM, backtranslation. Ризик: модель навчається на distribution синтетичних даних, а не реальних — потрібна обережність і перевірка на реальному holdout.

Якість даних: валідація та моніторинг

Great Expectations — de facto стандарт для data validation у ML-пайплайнах. Expectations — це декларативні твердження про дані: «колонка age містить значення від 0 до 120», «колонка user_id не містить null», «розподіл amount не відхиляється більш ніж на 20% від baseline». Запускається в пайплайні, при провалі — блокує проходження.

Pandera — Pythonic alternative для pandas/polars DataFrames. Schema-based validation з type hints:

import pandera as pa

schema = pa.DataFrameSchema({
    "user_id": pa.Column(int, nullable=False),
    "score": pa.Column(float, pa.Check.between(0, 1)),
    "label": pa.Column(str, pa.Check.isin(["positive", "negative", "neutral"])),
})

Data freshness. Модель очікує дані за останні N днів. ETL впав, дані не оновилися — модель використовує застарілі ознаки. Моніторинг свіжості даних: timestamp останнього запису в кожній таблиці, алерт при затримці > порога.

Дедуплікація. Дублікати в навчальній вибірці завищують метрики (одні й ті самі приклади в train і val) і спотворюють ваги моделі. MinHash LSH для наближеної дедуплікації великих датасетів. Для точної — хеш за нормалізованим контентом.

Сховища та формати

Для більшості ML-проектів на старті: Parquet в S3 + DVC для версіонування. Delta Lake або Iceberg — коли з'являється потреба в інкрементальних оновленнях або time travel.

Типові помилки при побудові пайплайнів

• Пропуск перевірки свіжості даних. Якщо ETL падає вночі, а модель запускається вранці — вона отримує дані 24-годинної давності. Рішення: алерт при затримці > 30 хвилин.
• Відсутність версіонування даних. Не можна відтворити експеримент, бо дані змінилися. DVC або Delta Lake time travel виправляють це.
• Забувають про schema evolution. Нове поле з’являється, а пайплайн падає. Автоматичне виявлення змін схеми через Great Expectations.

Active learning дозволяє скоротити бюджет на розмітку до 50–70%. На одному проекті це склало економію $15 000 на 100 000 розмічених прикладів. Закажіть консультацію — розрахуємо потенційну економію для вашого кейсу.

Що входить у проект з дата-інжинірингу для ML

Ми надаємо повний цикл:

• Аудит існуючих даних та пайплайнів (1 тиждень).
• Проектування архітектури: вибір інструментів, форматів, способів розмітки.
• Реалізація ETL/ELT пайплайну з валідацією та моніторингом.
• Документація коду та процесів (model card, data card).
• Навчання вашої команди роботі з пайплайном.
• SLA на супровід та підтримку.

Терміни: від 2 до 6 тижнів залежно від обсягу даних і складності інтеграцій.

Як ми будуємо пайплайн: покроково

1. Аудит існуючих даних. Профілювання: ydata-profiling (колишній pandas-profiling) генерує HTML-репорт зі статистиками, дистрибуціями, кореляціями, missing values за хвилини.
2. Проектування пайплайну. Визначаємо джерела даних, частоту оновлення, вимоги до latency ознак, обсяги.
3. Реалізація та тестування. Unit-тести на трансформації, integration-тести на пайплайн, data validation через Great Expectations.
4. Деплой та моніторинг. Алерти на freshness, quality checks, аномалії в обсягах даних.

Чому варто довірити це нам

Ми займаємося дата-інжинірингом та ML з понад 8-річним досвідом. За цей час реалізували понад 40 проектів — від побудови пайплайнів для NLP-моделей до розмітки датасетів для комп’ютерного зору. Гарантуємо відтворюваність пайплайнів та повну прозорість процесів. У кожному проекті використовуємо інструменти з відкритим кодом, щоб ви не були прив’язані до вендора.

Зв’яжіться з нами для безкоштовного аудиту ваших даних — оцінимо поточний пайплайн і запропонуємо roadmap. Замовте побудову ML-пайплайну під ключ.