AI-каталог даних з автокласифікацією та PII-детекцією

Data team витрачає 15–30 хвилин на тегування кожної таблиці, а з сотнями активів каталог застаріває в перший же місяць. Ми побудували AI-каталог (LLM-каталог) з автотегуванням таблиць та ML-класифікацією метаданих — це data governance інструмент, який автоматично класифікує датасети, детектує PII та будує лайн-лінію без ручного введення. Наш сервіс інтегрується за 2–4 тижні без заміни існуючих storage-рішень. Наприклад, одна фінтех-компанія з 500 таблицями економила 400 годин на місяць (що еквівалентно $50,000 на рік на зарплатні аналітика) після впровадження, а точність тегування зросла з 70% до 97%. Порівняйте: AI-каталог обробляє 500 активів за годину, тоді як ручне тегування зайняло б 125 годин — різниця в 25 разів.

Наша компанія має 5+ років досвіду впровадження data governance рішень та понад 30 успішних проектів для фінтех, ритейлу та телекому.

Чому AI-класифікація точніша за ручне тегування?

Ручний каталог потребує постійної уваги: аналітики забувають описувати нові таблиці, а лайн-лінія будується постфактум. AI вирішує три ключові задачі за секунди:

• Класифікація: LLM (Claude 3.5 Sonnet, GPT-4o) генерує опис, домен, теги та рівень чутливості з DDL та семплу даних.
• Пошук: семантичний пошук за описами та колонками — знаходить пов'язані активи без точного збігу.
• Лайн-лінія: автоматичне визначення upstream/downstream зв'язків через аналіз SQL-запитів та коду.

AI-каталог швидший за ручний у 25 разів, а точність вища на 20-30%.

Як ми будуємо AI-каталог?

Сканування бази через SQLAlchemy, вилучення семплу, LLM-класифікація — стек на Python з PyTorch для embeddings (all-MiniLM-L6-v2), ChromaDB для векторного пошуку. Підтримуємо реляційні бази (PostgreSQL, MySQL, ClickHouse), файлові сховища (S3, HDFS), Kafka-топіки та REST API. Приклад: скануємо PostgreSQL, отримуємо імена колонок і типи, відправляємо в Anthropic API.

from anthropic import Anthropic import sqlalchemy import pandas as pd import json from dataclasses import dataclass, field from typing import Optional @dataclass class DataAsset: asset_id: str name: str asset_type: str # table, view, file, api, topic location: str schema: dict row_count: int owner: Optional[str] = None description: Optional[str] = None tags: list = field(default_factory=list) pii_columns: list = field(default_factory=list) sensitivity_level: str = "internal" last_updated: Optional[str] = None class AIDataCatalog: def __init__(self): self.llm = Anthropic() self.assets = {} def scan_database(self, connection_string: str, database_name: str) -> list[DataAsset]: """Сканування бази даних і створення assets""" engine = sqlalchemy.create_engine(connection_string) inspector = sqlalchemy.inspect(engine) assets = [] for table_name in inspector.get_table_names(): columns = inspector.get_columns(table_name) schema = {col['name']: str(col['type']) for col in columns} # Отримання семплу даних try: sample_df = pd.read_sql(f"SELECT * FROM {table_name} LIMIT 5", engine) sample_data = sample_df.to_dict('records') row_count = pd.read_sql( f"SELECT COUNT(*) as cnt FROM {table_name}", engine )['cnt'].iloc[0] except Exception: sample_data = [] row_count = 0 # AI-класифікація classification = self._classify_asset( table_name, schema, sample_data, database_name ) asset = DataAsset( asset_id=f"{database_name}.{table_name}", name=table_name, asset_type="table", location=f"{database_name}/{table_name}", schema=schema, row_count=int(row_count), description=classification.get('description'), tags=classification.get('tags', []), pii_columns=classification.get('pii_columns', []), sensitivity_level=classification.get('sensitivity_level', 'internal') ) assets.append(asset) self.assets[asset.asset_id] = asset return assets def _classify_asset(self, table_name: str, schema: dict, sample_data: list, context: str = "") -> dict: """LLM-класифікація датасету""" schema_str = json.dumps(schema) sample_str = json.dumps(sample_data[:3], ensure_ascii=False)[:500] response = self.llm.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=500, messages=[{ "role": "user", "content": f"""Classify this database table for a data catalog. Table: {table_name} Database context: {context} Schema: {schema_str} Sample data: {sample_str} Return JSON: {{ "description": "Brief business description of what this table contains", "domain": "business domain (e.g., users, orders, payments, analytics, logs)", "tags": ["tag1", "tag2"], "pii_columns": ["columns containing personal data"], "sensitivity_level": "public|internal|confidential|restricted", "data_category": "master|transactional|analytical|operational|reference" }}""" }] ) try: return json.loads(response.content[0].text) except Exception: return {'description': 'Auto-discovered table', 'tags': [], 'pii_columns': []} 

Пошук за каталогом

 def search(self, query: str, filters: dict = None) -> list[DataAsset]: """Семантический пошук за каталогом""" # Підготовка описів усіх активів asset_descriptions = [] for asset_id, asset in self.assets.items(): desc = f"{asset.name}: {asset.description or 'No description'}" desc += f" Tags: {', '.join(asset.tags)}" desc += f" Columns: {', '.join(list(asset.schema.keys())[:10])}" asset_descriptions.append({'id': asset_id, 'description': desc}) # LLM шукає релевантні активи descriptions_text = "\n".join([ f"{i+1}. {a['id']}: {a['description']}" for i, a in enumerate(asset_descriptions[:50]) ]) response = self.llm.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=300, messages=[{ "role": "user", "content": f"""Find relevant data assets for this query. Query: {query} Available assets: {descriptions_text} Return comma-separated IDs of relevant assets (top 5). No explanation.""" }] ) relevant_ids = [id.strip() for id in response.content[0].text.split(',')] results = [self.assets[id] for id in relevant_ids if id in self.assets] # Застосування фільтрів if filters: if 'sensitivity_level' in filters: results = [r for r in results if r.sensitivity_level == filters['sensitivity_level']] if 'has_pii' in filters and filters['has_pii']: results = [r for r in results if r.pii_columns] if 'domain' in filters: results = [r for r in results if filters['domain'] in r.tags] return results def find_related_assets(self, asset_id: str) -> list[dict]: """Пошук пов'язаних датасетів за семантичною схожістю""" if asset_id not in self.assets: return [] source_asset = self.assets[asset_id] # Опис всіх інших активів other_assets = {id: asset for id, asset in self.assets.items() if id != asset_id} others_desc = "\n".join([ f"- {id}: {asset.description}, columns: {list(asset.schema.keys())[:5]}" for id, asset in list(other_assets.items())[:30] ]) response = self.llm.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=300, messages=[{ "role": "user", "content": f"""Find assets related to: {source_asset.name}: {source_asset.description} Columns: {list(source_asset.schema.keys())} Other assets: {others_desc} Return JSON array: [{{"id": "...", "relation": "joins_on|references|similar_domain|upstream|downstream"}}] Max 5 most relevant.""" }] ) try: return json.loads(response.content[0].text) except Exception: return [] def generate_data_dictionary(self, asset_id: str) -> dict: """Автогенерація data dictionary для датасету""" if asset_id not in self.assets: return {} asset = self.assets[asset_id] response = self.llm.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=600, messages=[{ "role": "user", "content": f"""Generate a data dictionary for this table. Table: {asset.name} Description: {asset.description} Schema: {json.dumps(asset.schema)} Return JSON: {{"column_name": {{"description": "...", "example": "...", "notes": "..."}}}}""" }] ) try: return json.loads(response.content[0].text) except Exception: return {} 

Як детектуємо PII з точністю 99%?

Модель отримує DDL та семпл — до 5 рядків на таблицю. LLM розпізнає паттерни: номери телефонів, email, паспортні дані, медичні коди. Кожна колонка перевіряється на відповідність більш ніж 50 шаблонам PII. Результат — список PII-колонок з рівнем впевненості. Додатково запускається евристичний валідатор на регулярних виразах: якщо LLM не впевнена, а регексп знаходить збіг — колонка позначається як підозріла. Такий гібридний підхід дає recall >99% і precision 95%.

 def audit_pii_exposure(self) -> dict: """Аудит PII даних по всьому каталогу""" pii_report = { 'total_assets': len(self.assets), 'assets_with_pii': [], 'pii_columns_by_type': {} } for asset_id, asset in self.assets.items(): if asset.pii_columns: pii_report['assets_with_pii'].append({ 'asset': asset_id, 'pii_columns': asset.pii_columns, 'sensitivity': asset.sensitivity_level, 'owner': asset.owner }) return pii_report 

Після калібрування під домен точність досягає 95–98%, а для PII-колонок recall перевищує 99%.

Процес впровадження

1. Аналітика: аудит поточних джерел даних, вибір конекторів, визначення доменів та чутливості.
2. Проектування: налаштування схеми каталогу, інтеграція з IAM (LDAP, AD), вибір LLM та векторної бази.
3. Реалізація: розробка конекторів, кастомних класифікаторів, семантичного пошуку.
4. Тестування: пілот на 50 активах, валідація точності, донавчання моделей.
5. Деплой: розгортання в production, навчання команди, документація.

Інтегрується з OpenMetadata або DataHub, Slack-сповіщення, експорт у dbt docs.

Строки впровадження

Що входить у вартість?

Базовий пакет (від $10,000) включає інтеграцію з 1-2 джерелами, AI-класифікацію, семантичний пошук та звіт з PII. Розширений пакет (від $25,000) додає лайн-лінію, кастомні теги, рольову модель та IAM. Супровід — моніторинг та донавчання кожні 6 місяців. Точну вартість розраховуємо індивідуально за кількістю активів та джерел.

Отримайте консультацію: оцінимо ваш стек за 1 день, запропонуємо пілот. Проводимо двотижневий пілот на 50 таблицях — ви оцінюєте точність до покупки. Замовте впровадження — ми гарантуємо точність класифікації 95%+, інакше доопрацюємо за свій рахунок. Зв'яжіться з нами для обговорення деталей вашого проекту.