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

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

Дублювання контактів у CRM — хронічна проблема: один клієнт існує як "Іванов І.І.", "Іван Іванов" та "[email protected]" як три окремі записи. AI-система автоматично знаходить дублі та пропонує або автоматично виконує злиття.

Підходи до детекції дублів

На основі правил: явні правила поточки (однаковий email = точно дубль). Висока точність, низький recall — пропускає нечіткі збіги.

На основі ML (entity resolution): модель передбачає ймовірність того, що два записи — один об'єкт. Враховує опечатки, скорочення, транслітерацію.

На основі embedding: перетворюємо кожен контакт на вектор і шукаємо найближчих сусідів. Масштабується швидко.

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 від дублійованих розсилань).