Data Provenance & Lineage for LLM Training Data — Mục tiêu: Track sources, timestamps, licenses, consent

Xin chào anh em, Hải đây. Hôm nay mình muốn chia sẻ về một vấn đề mà nhiều team đang phát triển LLM (Large Language Model) đang đau đầu: làm sao để track được nguồn gốc, timestamp, license và consent của dữ liệu training.

Đây không chỉ là vấn đề compliance (tuân thủ) mà còn là survival (sống còn) của dự án. Mình đã chứng kiến nhiều startup bị tố vi phạm bản quyền chỉ vì không lưu lại được cái “receipt” (hóa đơn) cho dữ liệu training.

1. Vấn đề đặt ra

Khi bạn thu thập dữ liệu từ internet để training LLM, bạn đang đối mặt với:

• Nguồn gốc không rõ ràng: Data scraped từ đâu? Khi nào?
• License phức tạp: CC0, CC BY, GPL, Commercial… mỗi cái xài một kiểu
• Consent của người dùng: Dữ liệu có được người dùng đồng ý share không?
• Thay đổi theo thời gian: Nội dung có thể bị xóa, sửa, hoặc license thay đổi

Nếu không giải quyết được những vấn đề này, bạn có thể gặp rủi ro pháp lý khi mô hình bị thương mạiại hóa.

2. Kiến trúc giải pháp

Mình đề xuất kiến trúc tracking data provenance với 4 layer:

┌─────────────────────────────────────────────────────────────┐
│                     Application Layer                       │
│  ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐           │
│  │  API    │ │  Web    │ │  CLI    │ │  SDK    │           │
│  └─────────┘ └─────────┘ └─────────┘ └─────────┘           │
└─────────────────────────────────────────────────────────────┘
                              │
┌─────────────────────────────────────────────────────────────┐
│                     Service Layer                           │
│  ┌─────────────────┐ ┌─────────────────┐                   │
│  │  Provenance     │ │  Consent        │                   │
│  │  Service        │ │  Management     │                   │
│  └─────────────────┘ └─────────────────┘                   │
└─────────────────────────────────────────────────────────────┘
                              │
┌─────────────────────────────────────────────────────────────┐
│                     Data Layer                              │
│  ┌─────────────┐ ┌─────────────┐ ┌─────────────┐           │
│  │  Metadata   │ │  Consent    │ │  Audit Log  │           │
│  │  Store      │ │  Store      │ │  Store      │           │
│  └─────────────┘ └─────────────┘ └─────────────┘           │
└─────────────────────────────────────────────────────────────┘
                              │
┌─────────────────────────────────────────────────────────────┐
│                     Storage Layer                           │
│  ┌─────────────┐ ┌─────────────┐ ┌─────────────┐           │
│  │  PostgreSQL │ │  Redis      │ │  S3/MinIO   │           │
│  └─────────────┘ └─────────────┘ └─────────────┘           │
└─────────────────────────────────────────────────────────────┘

3. Data Model chi tiết

3.1. Source Entity

-- PostgreSQL DDL cho Source Entity
CREATE TABLE source (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    url TEXT NOT NULL,
    domain TEXT NOT NULL,
    title TEXT,
    description TEXT,
    content_hash TEXT NOT NULL, -- SHA-256 của nội dung
    license_type TEXT NOT NULL CHECK (license_type IN (
        'CC0', 'CC_BY', 'CC_BY_SA', 'CC_BY_ND', 'CC_BY_NC', 
        'GPL', 'MIT', 'APACHE', 'COMMERCIAL', 'UNKNOWN'
    )),
    license_text TEXT, -- Nội dung license đầy đủ
    scraped_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(),
    last_verified_at TIMESTAMP WITH TIME ZONE,
    is_active BOOLEAN NOT NULL DEFAULT TRUE,
    consent_status TEXT NOT NULL CHECK (consent_status IN (
        'EXPLICIT_GRANTED', 'IMPLICIT_GRANTED', 'DENIED', 'UNKNOWN'
    )),
    consent_timestamp TIMESTAMP WITH TIME ZONE,
    consent_source TEXT, -- Link đến consent form
    creator_id UUID REFERENCES user(id),
    created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_source_url ON source(url);
CREATE INDEX idx_source_domain ON source(domain);
CREATE INDEX idx_source_scraped_at ON source(scraped_at);
CREATE INDEX idx_source_content_hash ON source(content_hash);

3.2. Dataset Entity

CREATE TABLE dataset (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    name TEXT NOT NULL,
    description TEXT,
    purpose TEXT,
    total_sources INTEGER NOT NULL DEFAULT 0,
    total_size_bytes BIGINT NOT NULL DEFAULT 0,
    created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW()
);

CREATE TABLE dataset_source_mapping (
    dataset_id UUID REFERENCES dataset(id) ON DELETE CASCADE,
    source_id UUID REFERENCES source(id) ON DELETE CASCADE,
    added_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(),
    PRIMARY KEY (dataset_id, source_id)
);

CREATE INDEX idx_dataset_source_mapping ON dataset_source_mapping(dataset_id);

3.3. Consent Management

CREATE TABLE consent_record (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    user_id UUID REFERENCES user(id),
    source_id UUID REFERENCES source(id),
    consent_type TEXT NOT NULL CHECK (consent_type IN (
        'DATA_COLLECTION', 'MODEL_TRAINING', 'COMMERCIAL_USE'
    )),
    status TEXT NOT NULL CHECK (status IN (
        'PENDING', 'GRANTED', 'DENIED', 'REVOKED'
    )),
    granted_at TIMESTAMP WITH TIME ZONE,
    revoked_at TIMESTAMP WITH TIME ZONE,
    ip_address INET,
    user_agent TEXT,
    consent_form_version TEXT,
    metadata JSONB,
    created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_consent_record_user ON consent_record(user_id);
CREATE INDEX idx_consent_record_source ON consent_record(source_id);
CREATE INDEX idx_consent_record_status ON consent_record(status);

4. Tracking System Implementation

4.1. Provenance Service

# provenance_service.py
from datetime import datetime
from typing import List, Dict, Optional
import hashlib
import requests
from sqlalchemy.orm import Session
from models import Source, Dataset, ConsentRecord
from schemas import SourceCreate, ConsentRecordCreate

class ProvenanceService:
    def __init__(self, db: Session):
        self.db = db

    def scrape_and_track(self, url: str, consent_required: bool = True) -> Source:
        """
        Scrape content và track provenance trong cùng 1 transaction
        """
        # Step 1: Fetch content
        response = requests.get(url, timeout=30)
        response.raise_for_status()

        # Step 2: Generate content hash
        content_hash = hashlib.sha256(response.content).hexdigest()

        # Step 3: Detect license (sử dụng https://github.com/osirv/license-detector)
        license_type = self._detect_license(response.text)

        # Step 4: Check consent requirement
        if consent_required:
            consent_status = self._check_consent(url)
        else:
            consent_status = 'IMPLICIT_GRANTED'

        # Step 5: Create source record
        source_data = SourceCreate(
            url=url,
            domain=self._extract_domain(url),
            title=self._extract_title(response.text),
            description=self._extract_description(response.text),
            content_hash=content_hash,
            license_type=license_type,
            consent_status=consent_status,
            scraped_at=datetime.now()
        )

        source = self.create_source(source_data)
        return source

    def create_source(self, source_data: SourceCreate) -> Source:
        db_source = Source(**source_data.dict())
        self.db.add(db_source)
        self.db.commit()
        self.db.refresh(db_source)
        return db_source

    def _detect_license(self, content: str) -> str:
        """
        Heuristic license detection
        """
        if "Creative Commons" in content:
            if "Attribution" in content:
                return "CC_BY"
            elif "NonCommercial" in content:
                return "CC_BY_NC"
            else:
                return "CC_BY_SA"
        elif "MIT License" in content:
            return "MIT"
        elif "Apache" in content:
            return "APACHE"
        elif "GNU" in content:
            return "GPL"
        else:
            return "UNKNOWN"

    def _check_consent(self, url: str) -> str:
        """
        Check consent qua API hoặc database
        """
        # Implement theo business logic của bạn
        # Có thể check qua external API, database, hoặc file config
        return 'EXPLICIT_GRANTED' if self._is_consent_granted(url) else 'UNKNOWN'

    def _is_consent_granted(self, url: str) -> bool:
        # Dummy implementation
        return True

4.2. Audit Log System

# audit_log.py
from enum import Enum
from datetime import datetime
from typing import Dict, Any
import json
from sqlalchemy import Column, Integer, String, JSON, DateTime, Enum as SqlEnum
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class AuditAction(Enum):
    SOURCE_CREATED = "SOURCE_CREATED"
    SOURCE_UPDATED = "SOURCE_UPDATED"
    CONSENT_GRANTED = "CONSENT_GRANTED"
    CONSENT_REVOKED = "CONSENT_REVOKED"
    DATA_DELETED = "DATA_DELETED"
    LICENSE_CHANGED = "LICENSE_CHANGED"

class AuditLog(Base):
    __tablename__ = 'audit_log'

    id = Column(Integer, primary_key=True, autoincrement=True)
    entity_type = Column(String(50), nullable=False)
    entity_id = Column(String(100), nullable=False)
    action = Column(SqlEnum(AuditAction), nullable=False)
    old_values = Column(JSON)
    new_values = Column(JSON)
    user_id = Column(String(100))
    ip_address = Column(String(45))
    user_agent = Column(String(500))
    metadata = Column(JSON)
    created_at = Column(DateTime, default=datetime.utcnow)

    def to_dict(self):
        return {
            "id": self.id,
            "entity_type": self.entity_type,
            "entity_id": self.entity_id,
            "action": self.action.value,
            "old_values": self.old_values,
            "new_values": self.new_values,
            "user_id": self.user_id,
            "ip_address": self.ip_address,
            "user_agent": self.user_agent,
            "metadata": self.metadata,
            "created_at": self.created_at.isoformat()
        }

class AuditLogger:
    def __init__(self, db_session):
        self.db = db_session

    def log(self, action: AuditAction, entity_type: str, entity_id: str,
            old_values: Dict[str, Any] = None, new_values: Dict[str, Any] = None,
            user_id: str = None, ip_address: str = None, user_agent: str = None,
            metadata: Dict[str, Any] = None):
        """
        Log audit record
        """
        audit_log = AuditLog(
            entity_type=entity_type,
            entity_id=entity_id,
            action=action,
            old_values=old_values,
            new_values=new_values,
            user_id=user_id,
            ip_address=ip_address,
            user_agent=user_agent,
            metadata=metadata
        )

        self.db.add(audit_log)
        self.db.commit()
        return audit_log

5. Performance & Scalability Considerations

5.1. Caching Strategy

# caching.py
import redis
from typing import Optional, Any
from datetime import timedelta

class ProvenanceCache:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client

    def get_source_metadata(self, url: str) -> Optional[Dict[str, Any]]:
        """
        Lấy metadata từ cache
        """
        cache_key = f"source_metadata:{hashlib.md5(url.encode()).hexdigest()}"
        cached_data = self.redis.get(cache_key)

        if cached_data:
            return json.loads(cached_data)

        return None

    def set_source_metadata(self, url: str, metadata: Dict[str, Any],
                          ttl: Optional[timedelta] = None):
        """
        Set metadata vào cache
        """
        cache_key = f"source_metadata:{hashlib.md5(url.encode()).hexdigest()}"
        ttl_seconds = int(ttl.total_seconds()) if ttl else 3600 * 24 * 30

        self.redis.setex(
            cache_key,
            ttl_seconds,
            json.dumps(metadata, default=str)
        )

    def get_consent_status(self, url: str) -> Optional[str]:
        """
        Lấy consent status từ cache
        """
        cache_key = f"consent_status:{hashlib.md5(url.encode()).hexdigest()}"
        return self.redis.get(cache_key)

    def set_consent_status(self, url: str, status: str,
                         ttl: Optional[timedelta] = None):
        """
        Set consent status vào cache
        """
        cache_key = f"consent_status:{hashlib.md5(url.encode()).hexdigest()}"
        ttl_seconds = int(ttl.total_seconds()) if ttl else 3600 * 24 * 7

        self.redis.setex(cache_key, ttl_seconds, status)

5.2. Batch Processing

# batch_processor.py
from typing import List
from models import Source
from sqlalchemy.orm import Session
from concurrent.futures import ThreadPoolExecutor
import time

class BatchProcessor:
    def __init__(self, db: Session, max_workers: int = 10):
        self.db = db
        self.max_workers = max_workers

    def process_sources_in_batch(self, source_urls: List[str],
                                batch_size: int = 100) -> List[Source]:
        """
        Xử lý sources theo batch để tối ưu hiệu năng
        """
        results = []
        total = len(source_urls)

        for i in range(0, total, batch_size):
            batch = source_urls[i:i + batch_size]
            batch_results = self._process_batch(batch)
            results.extend(batch_results)

            # Log tiến trình
            print(f"Processed batch {i//batch_size + 1}/{total//batch_size + 1}")
            time.sleep(0.1)  # Throttling để tránh bị block

        return results

    def _process_batch(self, batch_urls: List[str]) -> List[Source]:
        """
        Xử lý 1 batch sources
        """
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = [
                executor.submit(self._process_single_source, url)
                for url in batch_urls
            ]

            results = []
            for future in futures:
                try:
                    result = future.result()
                    if result:
                        results.append(result)
                except Exception as e:
                    print(f"Error processing source: {e}")

            return results

    def _process_single_source(self, url: str) -> Optional[Source]:
        """
        Xử lý 1 source
        """
        try:
            # Check cache trước
            cached = self._check_cache(url)
            if cached:
                return cached

            # Process source
            source_service = ProvenanceService(self.db)
            source = source_service.scrape_and_track(url)

            # Cache kết quả
            self._cache_result(url, source)

            return source
        except Exception as e:
            print(f"Failed to process {url}: {e}")
            return None

    def _check_cache(self, url: str) -> Optional[Source]:
        # Implement cache checking logic
        return None

    def _cache_result(self, url: str, source: Source):
        # Implement cache setting logic
        pass

6. Monitoring & Alerting

6.1. Key Metrics

# monitoring.py
from prometheus_client import Counter, Histogram, Gauge

# Metrics
SOURCES_PROCESSED = Counter(
    'sources_processed_total',
    'Total number of sources processed',
    ['status', 'license_type']
)

CONSENT_CHECKS = Counter(
    'consent_checks_total',
    'Total number of consent checks',
    ['result', 'consent_type']
)

PROCESSING_TIME = Histogram(
    'source_processing_seconds',
    'Time spent processing sources',
    ['step']
)

ACTIVE_SOURCES = Gauge(
    'active_sources',
    'Number of active sources in the system'
)

def track_processing_time(func):
    """
    Decorator để track processing time
    """
    def wrapper(*args, **kwargs):
        with PROCESSING_TIME.labels(step=func.__name__).time():
            return func(*args, **kwargs)

    return wrapper

6.2. Alerting Rules

# alerting_rules.yml
groups:
- name: provenance_alerts
  rules:
  - alert: HighConsentRejectionRate
    expr: rate(consent_checks_total{result="DENIED"}[5m]) > 0.1
    for: 10m
    labels:
      severity: warning
    annotations:
      summary: "High consent rejection rate detected"
      description: "More than 10% of consent checks are being denied in the last 5 minutes"

  - alert: ProcessingLatencyHigh
    expr: histogram_quantile(0.95, rate(source_processing_seconds_bucket[5m])) > 30
    for: 5m
    labels:
      severity: critical
    annotations:
      summary: "Processing latency too high"
      description: "95th percentile processing time is above 30 seconds"

  - alert: SourceVolumeSpike
    expr: increase(sources_processed_total[1h]) > 1000
    for: 5m
    labels:
      severity: info
    annotations:
      summary: "Spike in source processing volume"
      description: "More than 1000 sources processed in the last hour"

7. Use Case: Real-world Scenario

7.1. Scenario: Training Data Collection for LLM

Giả sử bạn đang thu thập dữ liệu từ 10,000 websites để training LLM:

# training_data_collector.py
from typing import List
from models import Dataset
from provenance_service import ProvenanceService
from audit_log import AuditLogger
from monitoring import SOURCES_PROCESSED

class TrainingDataCollector:
    def __init__(self, db: Session, audit_logger: AuditLogger):
        self.db = db
        self.provenance_service = ProvenanceService(db)
        self.audit_logger = audit_logger

    def collect_training_data(self, website_urls: List[str],
                            dataset_name: str, consent_required: bool = True):
        """
        Thu thập dữ liệu training với tracking đầy đủ
        """
        # Step 1: Create dataset
        dataset = Dataset(
            name=dataset_name,
            description=f"Training data collected from {len(website_urls)} websites",
            purpose="LLM training"
        )
        self.db.add(dataset)
        self.db.commit()

        # Step 2: Process sources
        for url in website_urls:
            try:
                source = self.provenance_service.scrape_and_track(
                    url=url,
                    consent_required=consent_required
                )

                # Map source to dataset
                self.db.execute("""
                    INSERT INTO dataset_source_mapping 
                    (dataset_id, source_id, added_at)
                    VALUES (:dataset_id, :source_id, NOW())
                """, {
                    "dataset_id": dataset.id,
                    "source_id": source.id
                })

                # Track metrics
                SOURCES_PROCESSED.labels(
                    status="success",
                    license_type=source.license_type
                ).inc()

                print(f"Processed {url} - License: {source.license_type}")

            except Exception as e:
                SOURCES_PROCESSED.labels(
                    status="error",
                    license_type="UNKNOWN"
                ).inc()
                print(f"Error processing {url}: {e}")

        self.db.commit()
        return dataset

7.2. Performance Benchmark

Dựa trên testing của mình:

8. Best Practices & Recommendations

8.1. Data Retention Policy

# data_retention.py
from datetime import datetime, timedelta
from sqlalchemy import and_

class DataRetentionManager:
    def __init__(self, db: Session):
        self.db = db

    def cleanup_expired_sources(self, retention_days: int = 365):
        """
        Cleanup sources quá hạn retention
        """
        cutoff_date = datetime.now() - timedelta(days=retention_days)

        # Tìm sources chưa được sử dụng trong 1 năm
        expired_sources = self.db.query(Source).filter(
            and_(
                Source.last_verified_at < cutoff_date,
                Source.is_active == True
            )
        ).all()

        for source in expired_sources:
            source.is_active = False
            source.updated_at = datetime.now()

            # Log audit
            self.audit_logger.log(
                action=AuditAction.SOURCE_UPDATED,
                entity_type="source",
                entity_id=source.id,
                old_values={"is_active": True},
                new_values={"is_active": False}
            )

        self.db.commit()
        return len(expired_sources)

8.2. Compliance Checklist

⚠️ Compliance Checklist
– [ ] All sources have license information tracked
– [ ] Consent records are stored for regulated data
– [ ] Content hashes are calculated for deduplication
– [ ] Audit logs are enabled for all data modifications
– [ ] Data retention policies are implemented
– [ ] Regular compliance audits are scheduled

8.3. Security Considerations

# security.py
from typing import Optional
from sqlalchemy.orm import Session
from models import Source

class SecurityValidator:
    def __init__(self, db: Session):
        self.db = db

    def validate_source_safety(self, source: Source) -> dict:
        """
        Validate source content for security issues
        """
        security_checks = {
            "malware_detected": self._check_malware(source),
            "phishing_detected": self._check_phishing(source),
            "doxxing_detected": self._check_doxxing(source),
            "compliance_risks": self._check_compliance_risks(source)
        }

        return security_checks

    def _check_malware(self, source: Source) -> bool:
        """
        Check for malware indicators
        """
        # Implement malware detection logic
        # Có thể dùng VirusTotal API hoặc ML models
        return False

    def _check_phishing(self, source: Source) -> bool:
        """
        Check for phishing indicators
        """
        # Implement phishing detection logic
        return False

    def _check_doxxing(self, source: Source) -> bool:
        """
        Check for personally identifiable information
        """
        # Implement PII detection logic
        return False

    def _check_compliance_risks(self, source: Source) -> List[str]:
        """
        Check for compliance risks
        """
        risks = []

        # Check license compliance
        if source.license_type == "UNKNOWN":
            risks.append("Unknown license")

        # Check consent compliance
        if source.consent_status != "EXPLICIT_GRANTED":
            risks.append("Missing explicit consent")

        return risks

9. Future Considerations

9.1. Blockchain for Provenance

Trong tương lai, có thể tích hợp blockchain để tạo ra “immutable audit trail”:

# blockchain_integration.py
from web3 import Web3

class BlockchainProvenance:
    def __init__(self, web3_url: str, contract_address: str):
        self.web3 = Web3(Web3.HTTPProvider(web3_url))
        self.contract = self.web3.eth.contract(
            address=contract_address,
            abi=self._get_contract_abi()
        )

    def record_provenance(self, source_id: str, metadata: dict):
        """
        Record provenance on blockchain
        """
        transaction = self.contract.functions.recordProvenance(
            source_id,
            json.dumps(metadata)
        ).buildTransaction({
            'from': self._get_default_account(),
            'gas': 2000000
        })

        signed_txn = self.web3.eth.account.sign_transaction(
            transaction,
            private_key=self._get_private_key()
        )

        self.web3.eth.send_raw_transaction(signed_txn.rawTransaction)

    def verify_provenance(self, source_id: str) -> dict:
        """
        Verify provenance from blockchain
        """
        return self.contract.functions.getProvenance(source_id).call()

9.2. AI-powered License Detection

# ai_license_detector.py
from typing import List
import torch
from transformers import pipeline

class AILicenseDetector:
    def __init__(self):
        self.ner_model = pipeline(
            "ner",
            model="dbmdz/bert-large-cased-finetuned-conll03-english",
            aggregation_strategy="simple"
        )
        self.classifier = self._load_license_classifier()

    def detect_license(self, content: str) -> str:
        """
        AI-powered license detection
        """
        # Step 1: Extract entities
        entities = self.ner_model(content)

        # Step 2: Classify license type
        license_type = self.classifier.classify(content)

        # Step 3: Confidence scoring
        confidence = self._calculate_confidence(entities, license_type)

        return {
            "license_type": license_type,
            "confidence": confidence,
            "entities": entities
        }

    def _calculate_confidence(self, entities: List[dict], license_type: str) -> float:
        # Implement confidence calculation logic
        return 0.85

10. Kết luận

Data provenance và lineage cho LLM training data không chỉ là vấn đề kỹ thuật mà còn là vấn đề pháp lý và đạo đức. Một hệ thống tracking tốt sẽ giúp bạn:

1. Giảm rủi ro pháp lý khi bị kiện vi phạm bản quyền
2. Tăng transparency cho khách hàng và đối tác
3. Dễ dàng audit khi có yêu cầu từ cơ quan quản lý
4. Tối ưu dữ liệu qua việc detect duplicate và outdated content

Key Takeaways:

• Implement tracking layer riêng biệt với metadata store
• Use content hashing để detect duplicate data
• Automate consent checking khi có thể
• Enable comprehensive audit logging
• Plan for scalability từ ngày đầu với batch processing và caching

Nếu anh em đang xây dựng hệ thống LLM, hãy bắt đầu nghĩ về provenance ngay từ bây giờ. Tin mình đi, trả giá cho việc không track source sau này sẽ đắt hơn nhiều.

Câu hỏi thảo luận: Anh em đã từng gặp vấn đề gì với data provenance trong dự án AI/ML chưa? Share ở comment để cùng học hỏi nhé!

Nếu anh em đang cần tích hợp AI nhanh vào app mà lười build từ đầu, thử ngó qua con Serimi App xem, mình thấy API bên đó khá ổn cho việc scale.