AI-система дедупликации и очистки данных

Реализация AI-дедупликации и очистки данных

Мы интегрируем AI-дедупликацию, которая находит неочевидные дубликаты: «Иван Иванов» и «I. Ivanov», «ООО Ромашка» и «ООО "Ромашка"». Fuzzy matching + ML-классификация находит такие пары, а LLM добавляет семантическое понимание — какую запись считать эталонной и как объединить атрибуты. Типичная ситуация: в CRM скопились дубликаты клиентов — разные менеджеры вносили одни и те же контакты с опечатками и в разных форматах. В результате рассылки уходят дважды, а аналитика показывает искажённую картину воронки. Результат — один чистовой датасет без потери информации. Закажите внедрение под ключ, чтобы забыть о «грязных» данных и сэкономить до 80% времени на рутине.

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

Как AI находит дубликаты?

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

• Блокировка — быстрая группировка записей по первым 3 символам нормализованного поля.
• Fuzzy matching на 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] для быстрого fuzzy matching, 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]

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

Data Engineering для ML: пайплайны, разметка и качество данных

«У нас много данных» — фраза, которая на деле часто означает «у нас много сырых логов в S3, которые никто не трогал два года». Перед тем как обучить модель, нужно понять, что вообще есть: какова структура, есть ли дубли, как часто меняется схема, насколько репрезентативна выборка.

Data Engineering для ML — не просто ETL. Это построение воспроизводимой инфраструктуры данных, которая делает обучение моделей надёжным, а переобучение — предсказуемым. По опыту нашей команды (8 лет в дата-инжиниринге, более 30 проектов в ML) каждая вторая проблема в продакшене связана не с архитектурой модели, а с качеством данных.

ETЛ-пайплайны для ML: чем отличаются от BI

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

Инструменты. Apache Spark (Wikipedia) для больших объёмов (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% качества модели на валидации.

Инкрементальные пайплайны. Модель переобучается еженедельно на новых данных. Нужен пайплайн, который инкрементально добавляет новые записи к обучающей выборке, не перегружая всё с нуля. 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.

Синтетические данные. Когда реальных данных мало или получить их дорого. Для 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.

Что входит в проект по дата-инжинирингу для ML

Мы предоставляем полный цикл:

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

Как мы строим пайплайн: пошагово

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

Почему стоит доверить это нам

Мы занимаемся дата-инжинирингом и ML с 2016 года. За это время реализовали более 40 проектов — от построения пайплайнов для NLP-моделей до разметки датасетов для компьютерного зрения. Гарантируем воспроизводимость пайплайнов и полную прозрачность процессов. В каждом проекте используем инструменты с открытым исходным кодом, чтобы вы не были привязаны к вендору.

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