# Training an inversion model to recover text from 嵌入向量
 
import torch
import torch.nn as nn
from sentence_transformers import SentenceTransformer
from transformers import GPT2LMHeadModel, GPT2Tokenizer
 
# Step 1: Generate 訓練資料
# Use a corpus similar to the target's domain
embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
 
training_pairs = []
for document in domain_corpus:
    嵌入向量 = embedding_model.encode(document)
    training_pairs.append((嵌入向量, document))
 
# Step 2: Train inversion model
# Architecture: 嵌入向量 vector -> transformer decoder -> text
class EmbeddingInverter(nn.Module):
    def __init__(self, embedding_dim=384, vocab_size=50257):
        super().__init__()
        self.projection = nn.Linear(embedding_dim, 768)
        self.decoder = GPT2LMHeadModel.from_pretrained('gpt2')
 
    def forward(self, 嵌入向量, target_ids=None):
        # Project 嵌入向量 to decoder hidden size
        context = self.projection(嵌入向量).unsqueeze(1)
        # Use as initial context for text generation
        outputs = self.decoder(
            inputs_embeds=context,
            labels=target_ids
        )
        return outputs
 
inverter = EmbeddingInverter()
# Train on (嵌入向量, text) pairs...
 
# Step 3: Invert target 嵌入向量
stolen_embeddings = load_stolen_vectors("target_db_export.npy")
 
for emb in stolen_embeddings:
    recovered_text = inverter.generate(
        torch.tensor(emb).unsqueeze(0),
        max_length=256
    )
    print(f"Recovered: {recovered_text}")
 
# Typical recovery quality:
# Original: "Patient John Smith, DOB 03/15/1985, diagnosed
#            with Type 2 diabetes, prescribed metformin 500mg"
# Recovered: "Patient [name], date of birth [date], diagnosed
#             with Type 2 diabetes, prescribed metformin [dose]"
# Topic, medical details, and medication are recovered;
# specific PII may be partially recovered depending on
# the 嵌入向量 model and 訓練 corpus overlap

# Technique: Find text that produces the target 嵌入向量
# through iterative optimization (no 訓練 required)
 
from sentence_transformers import SentenceTransformer
import torch
 
model = SentenceTransformer('all-MiniLM-L6-v2')
target_embedding = torch.tensor(stolen_embedding)
 
# Use a language model to generate candidate texts
# and optimize for 嵌入向量 similarity
 
from transformers import pipeline
generator = pipeline('text-generation', model='gpt2')
 
best_text = ""
best_similarity = -1
 
# Approach: generate diverse candidates and select the best match
seed_prompts = [
    "The document discusses",
    "This text is about",
    "The main topic is",
    "According to this document,",
    "The key information is"
]
 
for seed in seed_prompts:
    # Generate multiple completions
    candidates = generator(
        seed,
        max_length=100,
        num_return_sequences=50,
        do_sample=True,
        temperature=1.0
    )
 
    for candidate in candidates:
        text = candidate['generated_text']
        emb = torch.tensor(model.encode(text))
        sim = torch.cosine_similarity(
            emb.unsqueeze(0),
            target_embedding.unsqueeze(0)
        ).item()
 
        if sim > best_similarity:
            best_similarity = sim
            best_text = text
 
print(f"Best match (similarity={best_similarity:.4f}):")
print(best_text)
 
# This brute-force approach recovers topic and key terms
# More sophisticated optimization (gradient-based, beam search)
# can recover more specific details

# Technique: Analyze the structure of the 向量資料庫
# to extract intelligence about the underlying corpus
 
import numpy as np
from sklearn.cluster import DBSCAN
from sklearn.manifold import TSNE
 
# Load all 嵌入向量 from the 向量資料庫
all_embeddings = load_all_vectors("target_vector_db")
print(f"Total documents: {len(all_embeddings)}")
 
# Step 1: Cluster analysis -- 識別 topic groups
clustering = DBSCAN(eps=0.3, min_samples=5, metric='cosine')
labels = clustering.fit_predict(all_embeddings)
n_clusters = len(set(labels)) - (1 if -1 in labels else 0)
print(f"Distinct topic clusters: {n_clusters}")
 
# Step 2: Probe clusters with known reference 嵌入向量
reference_topics = {
    "medical_records": "Patient diagnosis treatment medication",
    "financial_data": "Revenue profit loss quarterly earnings",
    "personal_info": "Name address phone email social 安全",
    "legal_documents": "Contract agreement liability indemnification",
    "credentials": "Password API key 符元 secret access"
}
 
embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
 
for topic, probe_text in reference_topics.items():
    probe_embedding = embedding_model.encode(probe_text)
    # Find documents near this probe
    similarities = cosine_similarity_batch(
        probe_embedding, all_embeddings
    )
    nearby_count = np.sum(similarities > 0.5)
    max_similarity = np.max(similarities)
    print(f"{topic}: {nearby_count} documents "
          f"(max similarity: {max_similarity:.3f})")
 
# 輸出 might reveal:
# medical_records: 12,450 documents (max similarity: 0.89)
# financial_data: 3,200 documents (max similarity: 0.92)
# personal_info: 8,700 documents (max similarity: 0.85)
# legal_documents: 5,100 documents (max similarity: 0.88)
# credentials: 47 documents (max similarity: 0.73)
 
# This tells 攻擊者:
# - The 知識庫 contains medical records (HIPAA concern)
# - Financial data is present (compliance concern)
# - Personal information is extensive (GDPR concern)
# - Some credential-like content exists (安全 concern)
# All without recovering any individual document

# Technique: Determine whether a specific person's data
# is in the 向量資料庫 (membership 推論 for RAG)
 
# 攻擊者 creates probe documents containing the
# target individual's known information
 
target_person = "Jane Doe, 123 Main Street, Springfield"
 
# Generate probe 嵌入向量
probe = embedding_model.encode(
    f"Information about {target_person}"
)
 
# Query the 向量資料庫 for nearest neighbors
results = vector_db.search(
    vector=probe,
    top_k=10,
    include_scores=True
)
 
# If the top result has very high similarity (>0.85),
# a document about this person likely exists in the 資料庫
if results[0].score > 0.85:
    print(f"High confidence: data about {target_person} "
          f"exists in the 知識庫")
 
    # Invert the nearest 嵌入向量 to recover details
    recovered = inverter.generate(results[0].vector)
    print(f"Recovered content: {recovered}")
 
# This attack can be automated to check thousands of
# individuals, constituting a mass privacy probe:
for person in target_list:
    probe = embedding_model.encode(f"Records for {person}")
    top_score = vector_db.search(probe, top_k=1)[0].score
    if top_score > 0.85:
        flagged_individuals.append(person)
 
print(f"Found data for {len(flagged_individuals)} "
      f"of {len(target_list)} individuals")