1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

graph TD
    A[User Query] --> B[Source Embedding]
    B --> C[vector2Trans Encoder]
    C --> D[Transformed Query]
    D --> E[Target Organization Vector DB]
    E --> F[Retrieved Documents]
    F --> G[LLM Generation]
    
    H[Organization A Corpus] --> I[vector2Trans Transform]
    I --> J[Vector DB A]
    
    K[Organization B Corpus] --> L[vector2Trans Transform]
    L --> M[Vector DB B]
    
    J --> E
    M --> E

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

import numpy as np
from scipy.stats import ortho_group
from typing import Tuple

class Vector2Trans:
    """
    Trans-RAG의 vector2Trans 변환기 구현
    각 조직은 고유한 변환 파라미터를 보유
    """
    
    def __init__(self, dim: int, noise_scale: float = 0.01):
        self.dim = dim
        self.noise_scale = noise_scale
        
        # 직교 회전 행렬 생성 (Haar measure)
        self.rotation = ortho_group.rvs(dim)
        
        # 랜덤 스케일링 (각 차원에 대해)
        self.scaling = np.random.uniform(0.5, 2.0, size=dim)
        
        # 바이어스 벡터
        self.bias = np.random.randn(dim)
        
        # 노이즈 스케일
        self.noise_scale = noise_scale
    
    def transform(self, vectors: np.ndarray) -> np.ndarray:
        """
        원본 벡터를 조직 고유의 vector space로 변환
        
        Args:
            vectors: (N, dim) 형태의 원본 embedding 벡터
        
        Returns:
            변환된 벡터 (N, dim)
        """
        # Step 1: Rotation
        rotated = np.dot(vectors, self.rotation.T)
        
        # Step 2: Scaling
        scaled = rotated * self.scaling
        
        # Step 3: Translation + Noise
        noise = np.random.normal(0, self.noise_scale, size=scaled.shape)
        transformed = scaled + self.bias + noise
        
        return transformed
    
    def transform_query(
        self, 
        query_vector: np.ndarray, 
        target_key: Tuple
    ) -> np.ndarray:
        """
        쿼리를 타겟 조직의 vector space로 변환
        
        Args:
            query_vector: (dim,) 형태의 쿼리 embedding
            target_key: 타겟 조직의 공개 변환 키
        
        Returns:
            변환된 쿼리 벡터
        """
        target_rot, target_scale, target_bias = target_key
        
        # 소스 공간에서 역변환 후 타겟 공간으로
        intermediate = query_vector.reshape(1, -1)
        
        # 타겟 조직의 변환 적용
        rotated = np.dot(intermediate, target_rot.T)
        scaled = rotated * target_scale
        transformed = scaled + target_bias
        
        return transformed.flatten()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

# 사용 예시
if __name__ == "__main__":
    dim = 768  # BERT-base embedding dimension
    org_a = Vector2Trans(dim, noise_scale=0.01)
    org_b = Vector2Trans(dim, noise_scale=0.01)
    
    # 조직 A의 문서 embeddings
    doc_embeddings = np.random.randn(1000, dim)
    
    # 조직 A의 vector space로 변환하여 저장
    transformed_docs = org_a.transform(doc_embeddings)
    
    # 쿼리 처리 (조직 B에서 조직 A로 검색)
    query = np.random.randn(dim)
    
    # 조직 A의 공개 키 (rotation, scaling, bias만)
    org_a_public_key = (org_a.rotation, org_a.scaling, org_a.bias)
    
    # 쿼리를 조직 A의 공간으로 변환
    transformed_query = org_b.transform_query(query, org_a_public_key)
    
    # 검색 수행 (코사인 유사도)
    similarities = np.dot(
        transformed_docs, 
        transformed_query
    ) / (
        np.linalg.norm(transformed_docs, axis=1) * 
        np.linalg.norm(transformed_query)
    )
    
    top_k_idx = np.argsort(similarities)[-5:][::-1]
    print(f"Top-5 검색 결과 인덱스: {top_k_idx}")
    print(f"최고 유사도: {similarities[top_k_idx[0]]:.4f}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

# 변환 키 관리를 위한 간단 예시
from dataclasses import dataclass
import hashlib
import os

@dataclass
class OrgKeyPair:
    """조직별 변환 키 쌍"""
    org_id: str
    private_rotation: np.ndarray
    private_scaling: np.ndarray
    private_bias: np.ndarray
    
    def get_public_key(self) -> bytes:
        """공개 키 해시 생성 (키 자체는 직접 교환)"""
        key_data = (
            self.private_rotation.tobytes() +
            self.private_scaling.tobytes() +
            self.private_bias.tobytes()
        )
        return hashlib.sha256(key_data).digest()
    
    def rotate_key(self):
        """주기적 키 로테이션"""
        dim = self.private_rotation.shape[0]
        self.private_rotation = ortho_group.rvs(dim)
        self.private_scaling = np.random.uniform(0.5, 2.0, size=dim)
        self.private