Разработка AI-системы дедупликации контактов и компаний в CRM

Разработка AI-системы дедупликации контактов и компаний в CRM

Дублирование контактов в CRM — хроническая проблема: один клиент существует как "Иванов И.И.", "Иван Иванов" и "[email protected]" как три разные записи. AI-система автоматически находит дубли и предлагает или автоматически выполняет слияние.

Подходы к детекции дублей

Rule-based: явные правила совпадения (одинаковый email = точно дубль). Высокая точность, низкий recall — пропускает нечёткие совпадения.

ML-based (entity resolution): модель предсказывает вероятность того, что две записи — один объект. Учитывает опечатки, сокращения, транслитерацию.

Embedding-based: превращаем каждый контакт в вектор и ищем ближайших соседей. Быстро масштабируется.

ML-модель дедупликации

import pandas as pd
import dedupe
from dedupe import Dedupe

class ContactDeduplicator:
    def __init__(self):
        self.deduper = None

    def setup_fields(self):
        """Описание полей для dedupe"""
        fields = [
            dedupe.variables.String('first_name'),
            dedupe.variables.String('last_name'),
            dedupe.variables.String('email', has_missing=True),
            dedupe.variables.String('phone', has_missing=True),
            dedupe.variables.String('company'),
            dedupe.variables.String('job_title', has_missing=True),
        ]
        return dedupe.Dedupe(fields)

    def train(self, records: dict, training_file: str = None):
        """Обучение на помеченных парах (match/not-match)"""
        self.deduper = self.setup_fields()

        if training_file and os.path.exists(training_file):
            with open(training_file) as f:
                self.deduper.prepare_training(records, f)
        else:
            self.deduper.prepare_training(records)
            # Активное обучение: разметка примерных пар
            dedupe.console_label(self.deduper)
            with open(training_file, 'w') as f:
                self.deduper.write_training(f)

        self.deduper.train()

    def find_duplicates(self, records: dict,
                         threshold: float = 0.5) -> list[tuple]:
        """Поиск дублей с вероятностями"""
        clustered_dupes = self.deduper.partition(records, threshold)

        duplicate_groups = []
        for (cluster_id, record_ids, scores) in clustered_dupes:
            if len(record_ids) > 1:
                duplicate_groups.append({
                    'records': list(record_ids),
                    'scores': list(scores),
                    'max_score': max(scores)
                })

        return sorted(duplicate_groups, key=lambda x: x['max_score'], reverse=True)

Нечёткое сравнение строк

from rapidfuzz import fuzz, process

def compute_similarity(record1: dict, record2: dict) -> float:
    scores = []

    # Email: точное или domain совпадение
    if record1.get('email') and record2.get('email'):
        if record1['email'].lower() == record2['email'].lower():
            return 1.0  # Точное совпадение email — определённо дубль
        email1_domain = record1['email'].split('@')[1]
        email2_domain = record2['email'].split('@')[1]
        if email1_domain == email2_domain:
            scores.append(0.5)  # Один домен — похожи

    # Имя: нечёткое совпадение
    name1 = f"{record1.get('first_name', '')} {record1.get('last_name', '')}"
    name2 = f"{record2.get('first_name', '')} {record2.get('last_name', '')}"
    name_score = fuzz.token_sort_ratio(name1, name2) / 100
    scores.append(name_score * 0.4)

    # Телефон: нормализация и сравнение
    phone1 = re.sub(r'\D', '', record1.get('phone', ''))
    phone2 = re.sub(r'\D', '', record2.get('phone', ''))
    if phone1 and phone2:
        if phone1[-10:] == phone2[-10:]:  # Последние 10 цифр
            scores.append(0.9)

    # Компания
    if record1.get('company') and record2.get('company'):
        company_score = fuzz.token_set_ratio(
            record1['company'], record2['company']
        ) / 100
        scores.append(company_score * 0.2)

    return sum(scores) / len(scores) if scores else 0.0

Стратегия слияния записей

def merge_duplicates(records: list[dict]) -> dict:
    """Слияние группы дублей в одну запись"""
    merged = {}
    field_priority = ['email', 'phone', 'first_name', 'last_name', 'company']

    for field in field_priority:
        values = [r.get(field) for r in records if r.get(field)]
        if not values:
            continue
        # Берём наиболее часто встречающееся значение
        merged[field] = max(set(values), key=values.count)

    # Для created_at берём самую раннюю дату
    dates = [r.get('created_at') for r in records if r.get('created_at')]
    if dates:
        merged['created_at'] = min(dates)

    # Объединяем теги и метки
    all_tags = []
    for r in records:
        all_tags.extend(r.get('tags', []))
    merged['tags'] = list(set(all_tags))

    merged['merged_from'] = [r['id'] for r in records]
    return merged

Типичный результат внедрения: выявление 10-25% дублей в зрелой CRM-базе, сокращение базы на 8-15%, улучшение точности email-маркетинга (снижение unsubscribe rate от дублированных рассылок).