יישום דה-דופליקציה וניקוי נתונים באמצעות AI

הטמעת דה-דופליקציה וניקוי נתונים באמצעות AI

תרחיש טיפוסי: ה-CRM שלך צבר רשומות לקוחות כפולות—מנהלים שונים הזינו את אותם אנשי קשר עם שגיאות הקלדה ובפורמטים שונים. כתוצאה מכך, מיילים נשלחים פעמיים, והאנליטיקה מציגה משפך מעוות. אנו משלבים דה-דופליקציה מבוססת AI שמוצאת כפילויות לא ברורות: 'איוון איבנוב' ו-'I. Ivanov', 'LLC Romashka' ו-'Romashka Ltd'. התאמה מטושטשת בשילוב עם סיווג ML מזהה זוגות כאלה, ו-LLM מוסיף הבנה סמנטית—איזו רשומה נחשבת קנונית וכיצד למזג מאפיינים. התוצאה היא מערך נתונים נקי אחד ללא אובדן מידע. לפי מחקר של Gartner, איכות נתונים ירודה עולה לארגונים בממוצע 12.9 מיליון דולר בשנה. דה-דופליקציה מבוססת AI יכולה להפחית משמעותית את ההפסדים הללו. עם ניסיון של למעלה מ-5 שנים ויותר מ-20 פרויקטים של ניקוי נתונים, הצוות שלנו של מומחי AI מוסמכים מבטיח תוצאות באיכות גבוהה. הטמעת דה-דופליקציה מבוססת AI יכולה לחסוך לחברה שלך עד 50,000 דולר בשנה בעלויות ניקוי נתונים ידני.

כיצד פועלת דה-דופליקציה מבוססת AI

SQL קלאסי DISTINCT חסר תועלת כאשר הכפילויות אינן זהות. אנו משתמשים בגישה רב-שכבתית:

• חסימה: קיבוץ מהיר של רשומות לפי 3 התווים הראשונים של שדה מנורמל.
• התאמה מטושטשת ב-Python: מספר מדדי דמיון (token sort, partial ratio) עם משקלים ובונוס להתאמת אימייל/טלפון.
• קיבוץ כפילויות: אלגוריתם Union-Find ממזג זוגות שנמצאו לקבוצות כפילויות.
• פתרון קונפליקטים באמצעות LLM: בחירת הרשומה הקנונית לפי כלל 'הרשומה השלמה ביותר' או באמצעות LLM למקרים מורכבים; prompts עם דוגמאות בודדות משפרים את הדיוק.

לדה-דופליקציה סמנטית, אנו משתמשים ב-embeddings (1536 ממדים) ומשווים וקטורי רשומות באמצעות דמיון קוסינוס. צינור עיבוד זה מטפל במיליון רשומות תוך 15–30 דקות. בסף של 0.85, הדיוק מגיע ל-~92% והרקול ל-~88%. פתרון קונפליקטים באמצעות LLM מיושם רק על 5–10% מהזוגות (השאר ממוזגים אוטומטית), מה ששומר על עלות עיבוד AI סבירה.

הנה דוגמה ליישום:

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 {}

ניקוי נתונים עם LLM

מודלי LLM יכולים גם לתקנן כתובות ושמות חברות. הנה דוגמה:

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]

יתרונות ותוצאות

בעיות שאנו פותרים

• תעתיק ושגיאות הקלדה: 'איוון איבנוב' לעומת 'Ivan Ivanov' — יחס מטושטש 0.74, אך עם בונוס אימייל/טלפון הזוג הופך לכפילות.
• פורמטי כתובות שונים: 'ул. Ленина, д. 5' ו-'Lenina 5' — ה-LLM מתקנן לסכמה אחידה: מדינה, עיר, רחוב, בניין.
• וריאציות של צורה משפטית: 'ООО Ромашка' ו-'Romashka Ltd' — אנו מחלצים שמות טהורים וצורות משפטיות.

השוואת גישות

תרחישי כפילויות טיפוסיים

אנו משתמשים ב-Python + Pandas לעיבוד, rapidfuzz להתאמה מטושטשת מהירה, Scikit-learn לסיווג זוגות ML, ו-Anthropic Claude 3.5 לפתרון קונפליקטים באמצעות LLM. אנו מטמיעים את הפתרון בצינור ה-MLOps שלך באמצעות Docker או Apache Airflow.

מהניסיון שלנו: לקוח עם 500K רשומות CRM איבד לידים עקב כפילויות. לאחר הטמעת המערכת, ההמרה עלתה ב-15%, וזמן העיבוד ירד מ-3 ימים לשעה אחת. אוטומציה של איכות נתונים עם AI מפחיתה עבודה ידנית ב-80%.

דה-דופליקציה מבוססת AI מעבדת מיליון רשומות פי 40 מהר יותר מעבודה ידנית.

תהליך העבודה שלנו והתוצרים

תהליך העבודה שלנו

1. ביקורת נתונים — ניתוח איכות, זיהוי דפוסי כפילויות.
2. הגדרת כללים — ספים, חסימה, כוונון prompts ל-LLM.
3. פיתוח צינור עיבוד — אינטגרציה באמצעות API או ETL.
4. תיעוד והדרכה — תיאור תהליך, Metabase, הדרכת צוות.
5. תמיכה לאחר השחרור — שבועיים של ניטור.

מה כלול

• מערך נתונים נקי ללא כפילויות, תוך שמירה על כל הרשומות הייחודיות.
• דוח מפורט של הכפילויות שנמצאו ומוזגו.
• תיעוד על כללי הדה-דופליקציה והגדרת צינור העיבוד.
• גישה לאינטגרציה באמצעות API או ETL.
• הדרכה לצוות שלך.
• שבועיים של תמיכה לאחר השחרור.

לוח זמנים ותמחור

לוח זמנים: 2 עד 6 שבועות בהתאם לנפח הנתונים ומורכבות הכללים. התמחור נקבע באופן אישי: אנו מעריכים את מספר הרשומות, השדות והצורך בחידודי LLM. בקש ייעוץ לפרויקט שלך—ניתוח נתונים חינמי כלול.

מלכודות נפוצות בדה-דופליקציה ידנית

• שימוש רק בהתאמה מדויקת.
• התעלמות מתעתיק ושגיאות הקלדה.
• ללא נורמליזציה לפני השוואה.
• בחירת רשומה קנונית אקראית במקום השלמה ביותר.

דה-דופליקציה מטושטשת של מיליון רשומות: 15–30 דקות (תלוי בגודל הבלוק הממוצע). דיוק זיהוי כפילויות בסף=0.85: precision ~92%, recall ~88%. פתרון קונפליקטים באמצעות LLM מיושם רק על 5–10% מהזוגות (השאר ממוזגים אוטומטית), מה ששומר על עלות עיבוד AI סבירה.