# 訓練反演模型以從嵌入還原文字
 
import torch
import torch.nn as nn
from sentence_transformers import SentenceTransformer
from transformers import GPT2LMHeadModel, GPT2Tokenizer
 
# 步驟 1：產生訓練資料
# 使用與目標領域相似的語料
embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
 
training_pairs = []
for document in domain_corpus:
    embedding = embedding_model.encode(document)
    training_pairs.append((embedding, document))
 
# 步驟 2：訓練反演模型
# 架構：嵌入向量 -> transformer 解碼器 -> 文字
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, embedding, target_ids=None):
        # 將嵌入投影至解碼器的隱藏維度
        context = self.projection(embedding).unsqueeze(1)
        # 作為文字生成的初始上下文
        outputs = self.decoder(
            inputs_embeds=context,
            labels=target_ids
        )
        return outputs
 
inverter = EmbeddingInverter()
# 以（嵌入、文字）配對進行訓練……
 
# 步驟 3：反演目標嵌入
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}")
 
# 典型的還原品質：
# 原文：「Patient John Smith, DOB 03/15/1985, diagnosed
#            with Type 2 diabetes, prescribed metformin 500mg」
# 還原：「Patient [name], date of birth [date], diagnosed
#             with Type 2 diabetes, prescribed metformin [dose]」
# 主題、醫療細節與用藥得以還原；
# 特定 PII 可能部分還原，視嵌入模型與
# 訓練語料重疊程度而定

# 技巧：透過迭代最佳化尋找能產生目標嵌入的文字
#（無需訓練）
 
from sentence_transformers import SentenceTransformer
import torch
 
model = SentenceTransformer('all-MiniLM-L6-v2')
target_embedding = torch.tensor(stolen_embedding)
 
# 使用語言模型產生候選文字
# 並針對嵌入相似度最佳化
 
from transformers import pipeline
generator = pipeline('text-generation', model='gpt2')
 
best_text = ""
best_similarity = -1
 
# 做法：產生多樣候選並選出最佳匹配
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:
    # 產生多種延續文字
    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)
 
# 這種暴力做法能還原主題與關鍵詞
# 更精細的最佳化（梯度型、束搜尋）
# 可還原更具體的細節

# 技巧：分析向量資料庫的結構
# 以萃取底層語料庫的情報
 
import numpy as np
from sklearn.cluster import DBSCAN
from sklearn.manifold import TSNE
 
# 從向量資料庫載入所有嵌入
all_embeddings = load_all_vectors("target_vector_db")
print(f"Total documents: {len(all_embeddings)}")
 
# 步驟 1：叢集分析——辨識主題群組
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}")
 
# 步驟 2：以已知參考嵌入探測叢集
reference_topics = {
    "medical_records": "Patient diagnosis treatment medication",
    "financial_data": "Revenue profit loss quarterly earnings",
    "personal_info": "Name address phone email social security",
    "legal_documents": "Contract agreement liability indemnification",
    "credentials": "Password API key token secret access"
}
 
embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
 
for topic, probe_text in reference_topics.items():
    probe_embedding = embedding_model.encode(probe_text)
    # 尋找鄰近此探針的文件
    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})")
 
# 輸出可能揭示：
# 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)
 
# 這告訴攻擊者：
# - 知識庫包含病歷（HIPAA 疑慮）
# - 存在財務資料（合規疑慮）
# - 個人資訊廣泛（GDPR 疑慮）
# - 存在部分類似憑證的內容（安全疑慮）
# 全部無需還原任何個別文件

# 技巧：判斷特定個人資料是否
# 存在於向量資料庫中（RAG 的成員推論）
 
# 攻擊者建立包含目標個人
# 已知資訊的探針文件
 
target_person = "Jane Doe, 123 Main Street, Springfield"
 
# 產生探針嵌入
probe = embedding_model.encode(
    f"Information about {target_person}"
)
 
# 向向量資料庫查詢最近鄰
results = vector_db.search(
    vector=probe,
    top_k=10,
    include_scores=True
)
 
# 若排名最前的結果相似度極高（>0.85），
# 則關於此人的文件可能存在於資料庫中
if results[0].score > 0.85:
    print(f"High confidence: data about {target_person} "
          f"exists in the knowledge base")
 
    # 反演最近鄰嵌入以還原細節
    recovered = inverter.generate(results[0].vector)
    print(f"Recovered content: {recovered}")
 
# 此攻擊可自動化以檢查數千位個人，
# 構成大規模隱私探測：
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")