Cogent-ai
/

cogent-csp-15m

@@ -1,439 +0,0 @@
-import torch
-import torch.nn as nn
-import torch.nn.functional as F
-from safetensors.torch import save_file as safetensors_save_file
-from torchvision import models
-import safetensors
-import os
-# --- 1. 原始模型架構 (從 reconstruct_original_model.py 複製) ---
-class OriginalMoETransformerBlock(nn.Module):
-    def __init__(self, input_dim, hidden_dim, output_dim, num_experts):
-        super().__init__()
-        self.num_experts = num_experts
-        self.input_dim = input_dim
-        self.hidden_dim = hidden_dim
-        self.output_dim = output_dim
-        self.experts_w1 = nn.ModuleList([nn.Linear(input_dim, hidden_dim, bias=False) for _ in range(num_experts)])
-        self.experts_w2 = nn.ModuleList([nn.Linear(hidden_dim, output_dim, bias=False) for _ in range(num_experts)])
-        self.gate = nn.Linear(input_dim, num_experts, bias=False)
-    def forward(self, x):
-        gate_logits = self.gate(x)
-        weights = torch.softmax(gate_logits, dim=-1)
-        expert_outputs = torch.empty(x.shape[0], self.num_experts, self.output_dim, device=x.device)
-        for i in range(self.num_experts):
-            expert_outputs[:, i, :] = self.experts_w2[i](self.experts_w1[i](x))
-        output = torch.sum(expert_outputs * weights.unsqueeze(-1), dim=1)
-        return output
-class OriginalModelReconstructed(nn.Module):
-    def __init__(self, vocab_size, embedding_dim, moe_hidden_dim, num_experts):
-        super().__init__()
-        self.embeddings = nn.Embedding(vocab_size, embedding_dim)
-        self.transformer_block_0 = OriginalMoETransformerBlock(embedding_dim, moe_hidden_dim, embedding_dim, num_experts)
-        self.norm = nn.LayerNorm(embedding_dim)
-        self.output_layer = nn.Linear(embedding_dim, vocab_size)
-    def forward(self, x):
-        x = self.embeddings(x).squeeze(1)
-        x = self.transformer_block_0(x)
-        x = self.norm(x)
-        x = self.output_layer(x)
-        return x
-# --- 2. 記憶共生引擎模組 (Person X Memory Symbiosis Engine) ---
-class MemorySymbiosisEngine(nn.Module):
-    def __init__(self, embedding_dim, memory_slots=10, memory_dim=256):
-        super().__init__()
-        self.memory_slots = memory_slots
-        self.memory_dim = memory_dim
-        self.memory_keys = nn.Parameter(torch.randn(memory_slots, memory_dim))
-        self.memory_values = nn.Parameter(torch.randn(memory_slots, embedding_dim))
-        self.query_projection = nn.Linear(embedding_dim, memory_dim)
-        self.memory_read_fusion = nn.Linear(embedding_dim + embedding_dim, embedding_dim)
-    def forward(self, current_features, user_profile_embedding=None):
-        query = self.query_projection(current_features)
-        attention_scores = torch.matmul(query, self.memory_keys.T)
-        attention_weights = torch.softmax(attention_scores, dim=-1)
-        read_memory = torch.matmul(attention_weights, self.memory_values)
-        fused_with_memory = torch.cat((current_features, read_memory), dim=-1)
-        output_features = self.memory_read_fusion(fused_with_memory)
-        return output_features
-# --- 3. 真實錨定 (Reality Anchor, RA) 技術模組 ---
-class RealityAnchor(nn.Module):
-    def __init__(self, embedding_dim, fact_memory_size=1000, fact_dim=256, threshold=0.7):
-        super().__init__()
-        self.fact_memory_size = fact_memory_size
-        self.fact_dim = fact_dim
-        self.threshold = threshold
-        # 內置的事實記憶庫：這裡用可學習的張量模擬
-        # 實際中，這會是一個預訓練的、經過壓縮的知識表示
-        self.fact_memory_bank = nn.Parameter(torch.randn(fact_memory_size, fact_dim))
-        # 事實注意力機制 (Fact-Attention) 的查詢投影
-        self.fact_query_projection = nn.Linear(embedding_dim, fact_dim)
-        # 調整信號生成模塊 (輕量級神經網絡)
-        self.adjustment_module = nn.Sequential(
-            nn.Linear(embedding_dim + fact_dim, embedding_dim // 2),
-            nn.ReLU(),
-            nn.Linear(embedding_dim // 2, embedding_dim) # 輸出與 embedding_dim 相同維度，用於調整
-        )
-        # 事實一致性評分器 (這裡簡化為一個線性層)
-        self.consistency_scorer = nn.Linear(fact_dim, 1)
-    def forward(self, current_token_embedding, generated_context_embedding):
-        # current_token_embedding: 當前生成詞元的嵌入 (batch_size, embedding_dim)
-        # generated_context_embedding: 當前生成上下文的嵌入 (batch_size, embedding_dim)
-        # 1. 動態事實檢索與融合 (這裡簡化為基於上下文的檢索)
-        # 假設 generated_context_embedding 已經包含了足夠的信息來查詢事實
-        fact_query = self.fact_query_projection(generated_context_embedding) # (batch_size, fact_dim)
-        # 近似最近鄰搜索 (這裡簡化為點積相似度)
-        similarity_scores = torch.matmul(fact_query, self.fact_memory_bank.T) # (batch_size, fact_memory_size)
-        # 獲取最相關的事實 (這裡取 top-1)
-        top_fact_scores, top_fact_indices = torch.topk(similarity_scores, 1, dim=-1)
-        retrieved_fact = self.fact_memory_bank[top_fact_indices.squeeze(-1)] # (batch_size, fact_dim)
-        # 2. 事實一致性評分 (Fact Consistency Scoring)
-        # 衡量當前詞元嵌入與檢索到的事實的一致性
-        # 這裡簡化為直接評分檢索到的事實，實際可能需要更複雜的交互
-        fact_consistency_score = torch.sigmoid(self.consistency_scorer(retrieved_fact)) # (batch_size, 1)
-        # 3. 即時調整生成 (Real-Time Generation Adjustment)
-        adjustment_signal = torch.zeros_like(current_token_embedding) # 初始化調整信號
-        # 如果事實一致性分數低於閾值，則觸發調整
-        # 注意：這裡的閾值判斷是簡化的，實際應用中可能需要更精細的控制
-        needs_adjustment = (fact_consistency_score < self.threshold).squeeze(-1) # (batch_size,)
-        if needs_adjustment.any():
-            # 組合當前詞元嵌入和檢索到的事實來生成調整信號
-            combined_for_adjustment = torch.cat((current_token_embedding[needs_adjustment], retrieved_fact[needs_adjustment]), dim=-1)
-            adjustment_signal[needs_adjustment] = self.adjustment_module(combined_for_adjustment)
-        # 將調整信號應用於當前詞元嵌入
-        adjusted_token_embedding = current_token_embedding + adjustment_signal
-        return adjusted_token_embedding, fact_consistency_score
-# --- 4. 多維度生成協調器 (Multidimensional Generation Orchestrator, MGO) 技術模組 ---
-class MultidimensionalGenerationOrchestrator(nn.Module):
-    def __init__(self, embedding_dim, vocab_size, max_output_length=50, num_intent_types=5):
-        super().__init__()
-        self.embedding_dim = embedding_dim
-        self.vocab_size = vocab_size
-        self.max_output_length = max_output_length
-        self.num_intent_types = num_intent_types
-        # 1. 生成意圖解析與分解 (Generation Intent Parsing and Decomposition)
-        # 這裡簡化為一個線性層來預測意圖類型
-        self.intent_parser = nn.Linear(embedding_dim, num_intent_types)
-        self.intent_embedding = nn.Embedding(num_intent_types, embedding_dim) # 意圖嵌入
-        # 2. 多維度生成規劃 (Multidimensional Generation Planning)
-        # 這裡簡化為根據意圖調整生成參數
-        self.planning_layer = nn.Linear(embedding_dim, 4) # 輸出控制：豐富度、連貫性、長度、風格
-        # 3. 自適應生成引導 (Adaptive Generation Guidance)
-        self.guidance_controller = nn.Linear(embedding_dim * 2 + 4, embedding_dim) # 融合上下文、意圖、規劃參數
-        self.length_controller = nn.Linear(embedding_dim, 1) # 預測生成終止概率
-    def forward(self, context_embedding, current_output_length=0, target_intent_idx=None):
-        # context_embedding: 融合了所有特徵的上下文嵌入 (batch_size, embedding_dim)
-        # 1. 生成意圖解析 (簡化：如果未指定，則從上下文預測)
-        if target_intent_idx is None:
-            intent_logits = self.intent_parser(context_embedding)
-            target_intent_idx = torch.argmax(intent_logits, dim=-1) # (batch_size,)
-        intent_embedding = self.intent_embedding(target_intent_idx) # (batch_size, embedding_dim)
-        # 2. 多維度生成規劃
-        planning_params = torch.sigmoid(self.planning_layer(intent_embedding)) # (batch_size, 4)
-        # planning_params: [content_richness, semantic_coherence, length_target_ratio, style_adaptability]
-        # 3. 自適應生成引導
-        # 組合上下文、意圖嵌入和規劃參數
-        combined_guidance_input = torch.cat([
-            context_embedding,
-            intent_embedding,
-            planning_params
-        ], dim=-1)
-        guidance_signal = self.guidance_controller(combined_guidance_input) # (batch_size, embedding_dim)
-        # 長度控制
-        length_control_signal = torch.sigmoid(self.length_controller(context_embedding)) # (batch_size, 1)
-        # 這個信號可以被用來調整生成終止的概率
-        # 這裡返回引導信號和規劃參數，供外部模型使用來調整生成過程
-        return guidance_signal, planning_params, length_control_signal
-# --- 5. 智能體生態框架接口 (Agent Matrix Intelligent Agent Ecosystem Framework) ---
-class AgentMatrixInterface(nn.Module):
-    def __init__(self, model_core):
-        super().__init__()
-        self.model_core = model_core
-        self.task_mapping = {
-            "analyze_image_text": self._analyze_image_text,
-            "retrieve_memory": self._retrieve_memory,
-            "generate_response": self._generate_response,
-            "generate_anchored_response": self._generate_anchored_response # 新增 RA 命令
-        }
-    def _analyze_image_text(self, text_input, image_input):
-        return self.model_core(text_input, image_input, return_fused_features=True)
-    def _retrieve_memory(self, query_text_input, query_image_input=None):
-        text_features = self.model_core.embeddings(query_text_input)
-        if text_features.dim() == 3:
-            text_features = text_features.squeeze(1)
-        if query_image_input is not None:
-            image_features = self.model_core.vision_encoder(query_image_input)
-            image_features = image_features.view(image_features.size(0), -1)
-            image_features = self.model_core.vision_projection(image_features)
-            current_features = self.model_core.initial_fusion_layer(torch.cat((text_features, image_features), dim=1))
-        else:
-            current_features = text_features
-        return self.model_core.memory_engine(current_features)
-    def _generate_response(self, text_input, image_input):
-        # 這裡調用原始的 forward 方法，不啟用 RA
-        return self.model_core(text_input, image_input, use_ra=False)
-    def _generate_anchored_response(self, text_input, image_input):
-        # 這裡調用啟用 RA 的 forward 方法
-        return self.model_core(text_input, image_input, use_ra=True)
-    def forward(self, command, **kwargs):
-        if command in self.task_mapping:
-            if command.startswith("generate"):
-                # 生成相關的命令現在返回多個值
-                if command == "generate_anchored_response":
-                    logits, consistency_score, planning_params, length_control_signal = self.task_mapping[command](**kwargs)
-                    return logits, consistency_score, planning_params, length_control_signal
-                else:
-                    logits, planning_params, length_control_signal = self.task_mapping[command](**kwargs)
-                    return logits, planning_params, length_control_signal
-            else:
-                return self.task_mapping[command](**kwargs)
-        else:
-            raise ValueError(f"Unknown command: {command}")
-# --- 5. 整合後的完整模型架構 (包含 RA) ---
-class FullyIntegratedModelWithRA(nn.Module):
-    def __init__(self, original_model_path, vocab_size, embedding_dim, moe_hidden_dim, num_experts, visual_feature_dim=256,
-                 memory_slots=10, memory_dim=256, fact_memory_size=1000, fact_dim=256, ra_threshold=0.7):
-        super().__init__()
-        state_dict = {}
-        if original_model_path and os.path.exists(original_model_path):
-            with safetensors.safe_open(original_model_path, framework="pt", device="cpu") as f:
-                for key in f.keys():
-                    state_dict[key] = f.get_tensor(key)
-        original_vocab_size = 5000 # 假設原始模型的詞彙量
-        original_embedding_dim = 64 # 假設原始模型的 embedding_dim
-        # 判斷是否需要重新初始化 OriginalMoETransformerBlock 和 LayerNorm
-        # 如果 embedding_dim 或 moe_hidden_dim 改變，則不載入原始權重，讓其重新初始化
-        if embedding_dim != original_embedding_dim or moe_hidden_dim != 192: # 假設原始 moe_hidden_dim 是 192
-            print("Embedding_dim or moe_hidden_dim changed. OriginalMoETransformerBlock and LayerNorm will be reinitialized.")
-            self.original_model_core = OriginalModelReconstructed(vocab_size, embedding_dim, moe_hidden_dim, num_experts)
-            # 這些層將是可訓練的
-        else:
-            self.original_model_core = OriginalModelReconstructed(original_vocab_size, original_embedding_dim, moe_hidden_dim, num_experts)
-            # 載入原始權重到核心模型 (如果存在)
-            if state_dict:
-                print("Loading original model weights...")
-                self.original_model_core.norm.weight.data = state_dict["gamma"]
-                self.original_model_core.norm.bias.data = state_dict["beta"]
-                for i in range(num_experts):
-                    self.original_model_core.transformer_block_0.experts_w1[i].weight.data = state_dict[f"transformer_block.0.moe.experts.w1.weight"][i].T
-                    self.original_model_core.transformer_block_0.experts_w2[i].weight.data = state_dict[f"transformer_block.0.moe.experts.w2.weight"][i].T
-                self.original_model_core.transformer_block_0.gate.weight.data = state_dict["transformer_block.0.moe.gate.weight"].T
-                print("Original model weights loaded.")
-            for param in self.original_model_core.transformer_block_0.parameters():
-                param.requires_grad = False
-            for param in self.original_model_core.norm.parameters():
-                param.requires_grad = False
-        self.embeddings = nn.Embedding(vocab_size, embedding_dim)
-        self.output_layer = nn.Linear(embedding_dim, vocab_size)
-        self.vision_encoder = models.resnet18(weights=models.ResNet18_Weights.IMAGENET1K_V1)
-        self.vision_encoder = nn.Sequential(*list(self.vision_encoder.children())[:-1])
-        self.vision_projection = nn.Linear(512, visual_feature_dim)
-        for param in self.vision_encoder.parameters():
-            param.requires_grad = False
-        self.initial_fusion_layer = nn.Linear(visual_feature_dim + embedding_dim, embedding_dim)
-        self.memory_engine = MemorySymbiosisEngine(embedding_dim, memory_slots, memory_dim)
-        # 引入 Reality Anchor 模組
-        self.reality_anchor = RealityAnchor(embedding_dim, fact_memory_size, fact_dim, ra_threshold)
-        # 引入 Multidimensional Generation Orchestrator (MGO) 模組
-        self.mgo = MultidimensionalGenerationOrchestrator(embedding_dim, vocab_size)
-    def forward(self, text_input, image_input, user_profile_embedding=None, use_ra=False, return_fused_features=False):
-        text_features = self.embeddings(text_input)
-        if text_features.dim() == 3:
-            text_features = text_features.squeeze(1)
-        image_features = self.vision_encoder(image_input)
-        image_features = image_features.view(image_features.size(0), -1)
-        image_features = self.vision_projection(image_features)
-        fused_initial_features = self.initial_fusion_layer(torch.cat((text_features, image_features), dim=1))
-        if return_fused_features:
-            return fused_initial_features
-        memory_enhanced_features = self.memory_engine(fused_initial_features, user_profile_embedding)
-        # 應用 Reality Anchor (RA) 技術
-        if use_ra:
-            # 在這裡，我們假設 memory_enhanced_features 作為生成上下文的嵌入
-            # 並將其作為 current_token_embedding 傳遞給 RA 進行調            # 實際應用中, RA 可能會更細粒度地作用於每個生成的 token
-            adjusted_features, consistency_score = self.reality_anchor(memory_enhanced_features, memory_enhanced_features) # 簡化: 上下文和當前 token 相同
-            x = self.original_model_core.transformer_block_0(adjusted_features)
-        else:
-            x = self.original_model_core.transformer_block_0(memory_enhanced_features)
-        x = self.original_model_core.norm(x)
-        final_features = x # 將 x 賦值給 final_features, 以便 MGO 處理
-        # 應用 Multidimensional Generation Orchestrator (MGO) 技術
-        guidance_signal, planning_params, length_control_signal = self.mgo(final_features, current_output_length=0, target_intent_idx=None)
-        final_features = final_features + guidance_signal # 疊加引導信號
-        output = self.output_layer(final_features)
-        if use_ra:
-            return output, consistency_score, planning_params, length_control_signal
-        else:
-            return output, planning_params, length_control_signal
-if __name__ == "__main__":
-    original_model_path = "/home/ubuntu/upload/moe_model.safetensors"
-    vocab_size = 10000  # 詞彙量保持不變
-    embedding_dim = 128 # 增加嵌入維度以增加參數
-    moe_hidden_dim = 384 # 增加 MoE 隱藏維度以增加參數
-    num_experts = 16
-    visual_feature_dim = 256
-    memory_slots = 10
-    memory_dim = 256
-    fact_memory_size = 1000
-    fact_dim = 256
-    ra_threshold = 0.7
-    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
-    # 初始化整合模型 (包含 RA)
-    integrated_model_ra = FullyIntegratedModelWithRA(
-        original_model_path=original_model_path,
-        vocab_size=vocab_size,
-        embedding_dim=embedding_dim,
-        moe_hidden_dim=moe_hidden_dim,
-        num_experts=num_experts,
-        visual_feature_dim=visual_feature_dim,
-        memory_slots=memory_slots,
-        memory_dim=memory_dim,
-        fact_memory_size=fact_memory_size,
-        fact_dim=fact_dim,
-        ra_threshold=ra_threshold
-    ).to(device)
-    integrated_model_ra.eval() # 設置為評估模式
-    print("Fully Integrated Model with RA initialized successfully.")
-    # 計算總參數數量
-    total_trainable_params = sum(p.numel() for p in integrated_model_ra.parameters() if p.requires_grad)
-    print(f"Total trainable parameters: {total_trainable_params / 1_000_000:.2f}M")
-    total_all_params = sum(p.numel() for p in integrated_model_ra.parameters())
-    print(f"Total all parameters (including frozen): {total_all_params / 1_000_000:.2f}M")
-    # --- 模擬 Agent Matrix 框架交互 (包含 RA 命令) ---
-    agent_interface_ra = AgentMatrixInterface(integrated_model_ra)
-    print("Agent Matrix Interface with RA initialized.")
-    # 模擬輸入
-    dummy_text_input = torch.tensor([[100]], dtype=torch.long).to(device) # Batch size 1, 1 token
-    dummy_image_input = torch.randn(1, 3, 224, 224).to(device) # Batch size 1, 3 channels, 224x224
-    print("\n--- Simulating Agent Matrix Commands (with RA) ---")
-    # 模擬 'analyze_image_text' 命令
-    try:
-        print("Executing command: analyze_image_text")
-        fused_features = agent_interface_ra(command="analyze_image_text", text_input=dummy_text_input, image_input=dummy_image_input)
-        print(f"Analyzed features shape: {fused_features.shape}")
-    except Exception as e:
-        print(f"Error executing analyze_image_text: {e}")
-    # 模擬 'generate_response' 命令 (不啟用 RA)
-    try:
-        print("Executing command: generate_response (without RA)")
-        output_logits, planning_params, length_control_signal = agent_interface_ra(command="generate_response", text_input=dummy_text_input, image_input=dummy_image_input)
-        print(f"Generated response logits shape: {output_logits.shape}")
-        print(f"MGO Planning Params: {planning_params.tolist()}")
-        print(f"MGO Length Control Signal: {length_control_signal.item():.4f}")
-    except Exception as e:
-        print(f"Error executing generate_response (without RA): {e}")
-    # 模擬 'generate_anchored_response' 命令 (啟用 RA)
-    try:
-        print("Executing command: generate_anchored_response (with RA)")
-        output_logits_ra, consistency_score, planning_params_ra, length_control_signal_ra = agent_interface_ra(command="generate_anchored_response", text_input=dummy_text_input, image_input=dummy_image_input)
-        print(f"Generated anchored response logits shape: {output_logits_ra.shape}")
-        print(f"Fact consistency score: {consistency_score.item():.4f}")
-        print(f"MGO Planning Params (RA): {planning_params_ra.tolist()}")
-        print(f"MGO Length Control Signal (RA): {length_control_signal_ra.item():.4f}")
-    except Exception as e:
-        print(f"Error executing generate_anchored_response (with RA): {e}")
-    # 模擬 'retrieve_memory' 命令
-    try:
-        print("Executing command: retrieve_memory")
-        retrieved_memory = agent_interface_ra(command="retrieve_memory", query_text_input=dummy_text_input, query_image_input=dummy_image_input)
-        print(f"Retrieved memory shape: {retrieved_memory.shape}")
-    except Exception as e:
-        print(f"Error executing retrieve_memory: {e}")
-    # 保存整合後的模型 (包含 RA)
-    state_dict_to_save_ra = integrated_model_ra.state_dict()
-    keys_to_remove_ra = [key for key in state_dict_to_save_ra.keys() if 'vision_encoder' in key]
-    for key in keys_to_remove_ra:
-        del state_dict_to_save_ra[key]
-    for key in state_dict_to_save_ra:
-        if isinstance(state_dict_to_save_ra[key], torch.Tensor):
-            state_dict_to_save_ra[key] = state_dict_to_save_ra[key].contiguous()
-    safetensors_save_file(state_dict_to_save_ra, "fully_integrated_model_with_ra_mgo.safetensors")
-    print("Fully integrated model with RA and MGO saved to fully_integrated_model_with_ra_mgo.safetensors")