LLM-output watermerken voor herkomst

import hashlib
import numpy as np
from typing import Optional
 
class ProvenanceWatermark:
    """
    Multi-bit-watermerkschema dat herkomstmetadata codeert
    in LLM-output via vertekende tokenselectie.
    """
 
    def __init__(
        self,
        secret_key: bytes,
        gamma: float = 0.5,
        delta: float = 1.5,
        window_size: int = 4,
    ):
        self.secret_key = secret_key
        self.gamma = gamma
        self.delta = delta
        self.window_size = window_size
 
    def encode_provenance(
        self,
        model_id: int,
        deployment_id: int,
        user_hash: int,
        timestamp: int,
    ) -> bytes:
        """Verpak herkomstmetadata in een binaire payload."""
        import struct
        payload = struct.pack(
            ">HHIH",
            model_id & 0xFFFF,
            deployment_id & 0xFFFF,
            user_hash & 0xFFFFFFFF,
            timestamp & 0xFFFF,
        )
        return payload
 
    def get_bit_assignment(
        self,
        prev_tokens: list[int],
        vocab_size: int,
        bit_value: int,
    ) -> set:
        """
        Partitioneer de woordenschat op basis van het te coderen bit.
        Voor bit=1, verteken richting de groene lijst.
        Voor bit=0, verteken richting de rode lijst (of geen vertekening).
        """
        seed = hashlib.sha256(
            self.secret_key
            + bytes(prev_tokens[-self.window_size:])
        ).digest()
 
        rng = np.random.RandomState(int.from_bytes(seed[:4], "big"))
        green_count = int(vocab_size * self.gamma)
        green_list = set(rng.choice(vocab_size, green_count, replace=False))
 
        if bit_value == 1:
            return green_list  # Verteken richting deze tokens
        else:
            return set()  # Geen vertekening (natuurlijke verdeling)
 
    def apply_watermark(
        self,
        logits: np.ndarray,
        prev_tokens: list[int],
        payload: bytes,
        token_position: int,
    ) -> np.ndarray:
        """Pas watermerkvertekening toe op logits op basis van het huidige payload-bit."""
        # Bepaal welk bit op deze positie gecodeerd moet worden
        bit_index = token_position % (len(payload) * 8)
        byte_index = bit_index // 8
        bit_offset = bit_index % 8
        current_bit = (payload[byte_index] >> (7 - bit_offset)) & 1
 
        green_list = self.get_bit_assignment(
            prev_tokens, len(logits), current_bit
        )
 
        modified = logits.copy()
        for token_id in green_list:
            modified[token_id] += self.delta
 
        return modified

class ProvenanceVerifier:
    """Verifieer en decodeer herkomstwatermerken uit tekst."""
 
    def __init__(self, secret_key: bytes, gamma: float = 0.5, window_size: int = 4):
        self.secret_key = secret_key
        self.gamma = gamma
        self.window_size = window_size
 
    def extract_provenance(
        self,
        token_ids: list[int],
        vocab_size: int,
        payload_bits: int = 80,
    ) -> dict:
        """
        Extraheer gecodeerde herkomst uit tekst met watermerk.
        Gebruikt meerderheidsstemming over meerdere herhalingen van de payload.
        """
        bit_votes = {i: [] for i in range(payload_bits)}
 
        for pos in range(self.window_size, len(token_ids)):
            prev_tokens = token_ids[pos - self.window_size:pos]
            bit_index = (pos - self.window_size) % payload_bits
 
            # Reconstrueer de groene lijst voor deze positie
            seed = hashlib.sha256(
                self.secret_key + bytes(prev_tokens)
            ).digest()
            rng = np.random.RandomState(int.from_bytes(seed[:4], "big"))
            green_count = int(vocab_size * self.gamma)
            green_list = set(rng.choice(vocab_size, green_count, replace=False))
 
            # Stem: zit dit token in de groene lijst?
            is_green = token_ids[pos] in green_list
            bit_votes[bit_index].append(1 if is_green else 0)
 
        # Meerderheidsstemming voor elk bit
        decoded_bits = []
        confidence_scores = []
        for i in range(payload_bits):
            votes = bit_votes[i]
            if votes:
                avg = np.mean(votes)
                decoded_bits.append(1 if avg > self.gamma + 0.05 else 0)
                confidence_scores.append(abs(avg - self.gamma))
            else:
                decoded_bits.append(0)
                confidence_scores.append(0)
 
        # Reconstrueer de payload
        payload_bytes = self._bits_to_bytes(decoded_bits)
 
        return {
            "payload": payload_bytes,
            "provenance": self._decode_payload(payload_bytes),
            "mean_confidence": np.mean(confidence_scores),
            "min_confidence": np.min(confidence_scores),
            "reliable": np.mean(confidence_scores) > 0.1,
        }
 
    def _bits_to_bytes(self, bits: list[int]) -> bytes:
        result = bytearray()
        for i in range(0, len(bits), 8):
            byte = 0
            for j in range(8):
                if i + j < len(bits):
                    byte = (byte << 1) | bits[i + j]
            result.append(byte)
        return bytes(result)
 
    def _decode_payload(self, payload: bytes) -> dict:
        import struct
        try:
            model_id, deployment_id, user_hash, timestamp = struct.unpack(
                ">HHIH", payload[:10]
            )
            return {
                "model_id": model_id,
                "deployment_id": deployment_id,
                "user_hash": user_hash,
                "timestamp": timestamp,
            }
        except struct.error:
            return {"error": "Payload decode failed"}

# Integratie met vLLM-servering
from vllm import LLM, SamplingParams
 
class WatermarkedLLM:
    """Wrapper die herkomstwatermerken toepast tijdens inferentie."""
 
    def __init__(self, model_name: str, watermark_key: bytes):
        self.llm = LLM(model=model_name)
        self.watermark = ProvenanceWatermark(secret_key=watermark_key)
 
    def generate(
        self,
        prompt: str,
        user_id: str,
        deployment_id: int,
        max_tokens: int = 512,
    ) -> dict:
        provenance_payload = self.watermark.encode_provenance(
            model_id=1,
            deployment_id=deployment_id,
            user_hash=hash(user_id) & 0xFFFFFFFF,
            timestamp=int(time.time()) & 0xFFFF,
        )
 
        params = SamplingParams(
            max_tokens=max_tokens,
            logits_processors=[
                lambda token_ids, logits: self.watermark.apply_watermark(
                    logits, token_ids, provenance_payload, len(token_ids)
                )
            ],
        )
 
        outputs = self.llm.generate([prompt], params)
        return {
            "text": outputs[0].outputs[0].text,
            "watermark_payload": provenance_payload.hex(),
        }

def analyze_robustness_capacity_tradeoff(
    text_length_tokens: int,
    payload_bits: int,
    gamma: float = 0.5,
    target_confidence: float = 0.99,
):
    """
    Bereken de relatie tussen watermerkcapaciteit,
    tekstlengte en detectiebetrouwbaarheid.
 
    Meer payload-bits vereisen meer tekst voor betrouwbare extractie.
    """
    from scipy import stats
 
    # Elk payload-bit krijgt text_length / payload_bits stemmen
    votes_per_bit = text_length_tokens / payload_bits
 
    # Onder de watermerkverdeling is de groene fractie = gamma + shift
    # Voor betrouwbare detectie hebben we voldoende stemmen per bit nodig
    min_votes_needed = stats.norm.ppf(target_confidence) ** 2 / (0.1 ** 2)
 
    return {
        "text_length": text_length_tokens,
        "payload_bits": payload_bits,
        "votes_per_bit": votes_per_bit,
        "min_votes_needed": min_votes_needed,
        "sufficient_text": votes_per_bit >= min_votes_needed,
        "min_text_for_payload": int(min_votes_needed * payload_bits),
    }
 
# Voorbeeld: 80-bit herkomst-payload
# analyze_robustness_capacity_tradeoff(1000, 80)
# → votes_per_bit: 12.5, mogelijk onvoldoende
# analyze_robustness_capacity_tradeoff(5000, 80)
# → votes_per_bit: 62.5, waarschijnlijk voldoende

class ComplianceWatermarkService:
    """
    Watermerkdienst ontworpen om te voldoen aan regelgevende compliancevereisten
    voor de herkomst van door AI gegenereerde inhoud.
    """
 
    def __init__(self, config: dict):
        self.watermark = ProvenanceWatermark(
            secret_key=config["watermark_key"],
            delta=config.get("delta", 1.5),
        )
        self.audit_log = config["audit_log_backend"]
 
    def generate_with_provenance(
        self,
        llm,
        prompt: str,
        user_context: dict,
    ) -> dict:
        """Genereer output met watermerk en volledige audit trail."""
 
        # Maak een herkomstrecord
        provenance = {
            "request_id": generate_uuid(),
            "timestamp": datetime.utcnow().isoformat(),
            "model_id": llm.model_id,
            "deployment": llm.deployment_id,
            "user_id_hash": hash_user_id(user_context["user_id"]),
            "safety_policy_version": llm.safety_policy_version,
        }
 
        # Genereer met watermerk
        output = llm.generate(
            prompt=prompt,
            watermark_payload=self.watermark.encode_provenance(**provenance),
        )
 
        # Log naar onveranderlijke audit store
        self.audit_log.write({
            **provenance,
            "prompt_hash": hashlib.sha256(prompt.encode()).hexdigest(),
            "output_hash": hashlib.sha256(output["text"].encode()).hexdigest(),
            "output_length_tokens": output["token_count"],
        })
 
        return {
            "text": output["text"],
            "provenance_id": provenance["request_id"],
            "compliance_record": True,
        }

class WatermarkKeyManager:
    """Beheer de levenscyclus van watermerksleutels, inclusief rotatie en intrekking."""
 
    def __init__(self, key_store):
        self.key_store = key_store
 
    def rotate_key(self, deployment_id: str) -> dict:
        """
        Roteer de watermerksleutel voor een deployment.
        De oude sleutel moet behouden blijven voor verificatie van eerder
        van een watermerk voorziene inhoud.
        """
        import secrets
 
        new_key = secrets.token_bytes(32)
        old_key = self.key_store.get_current_key(deployment_id)
 
        # Archiveer de oude sleutel met vervalmetadata
        self.key_store.archive_key(
            deployment_id=deployment_id,
            key=old_key,
            retired_at=datetime.utcnow(),
            verify_until=datetime.utcnow() + timedelta(days=365),
        )
 
        # Activeer de nieuwe sleutel
        self.key_store.set_current_key(
            deployment_id=deployment_id,
            key=new_key,
            activated_at=datetime.utcnow(),
        )
 
        return {
            "deployment_id": deployment_id,
            "key_rotated": True,
            "old_key_archived": True,
        }