|
|
import os
|
|
|
import sys
|
|
|
import torch
|
|
|
from typing import List, Dict
|
|
|
|
|
|
def test_tokenizer():
|
|
|
print("Testing tokenizer...")
|
|
|
from src.tokenization_openpeer import OpenPeerTokenizer
|
|
|
|
|
|
tokenizer = OpenPeerTokenizer()
|
|
|
test_text = "Hello world"
|
|
|
|
|
|
tokens = tokenizer(test_text)
|
|
|
print(f"Input text: {test_text}")
|
|
|
print(f"Tokenized: {tokens}")
|
|
|
decoded = tokenizer.decode(tokens["input_ids"])
|
|
|
print(f"Decoded: {decoded}")
|
|
|
|
|
|
def test_model_config():
|
|
|
print("\nTesting model configuration...")
|
|
|
from src.configuration_openpeer import OpenPeerConfig
|
|
|
|
|
|
config = OpenPeerConfig()
|
|
|
print("Model Configuration:")
|
|
|
print(f"Hidden Size: {config.hidden_size}")
|
|
|
print(f"Number of Layers: {config.num_hidden_layers}")
|
|
|
print(f"Number of Attention Heads: {config.num_attention_heads}")
|
|
|
|
|
|
def test_model_architecture():
|
|
|
print("\nTesting model architecture...")
|
|
|
from src.modeling_openpeer import OpenPeerLLM
|
|
|
from src.configuration_openpeer import OpenPeerConfig
|
|
|
|
|
|
config = OpenPeerConfig()
|
|
|
model = OpenPeerLLM(config)
|
|
|
|
|
|
|
|
|
print("Model Structure:")
|
|
|
for name, param in model.named_parameters():
|
|
|
print(f"{name}: {param.shape}")
|
|
|
|
|
|
def run_inference_test():
|
|
|
print("Initializing OpenPeerLLM...")
|
|
|
from src.modeling_openpeer import OpenPeerLLM
|
|
|
from src.configuration_openpeer import OpenPeerConfig
|
|
|
from src.tokenization_openpeer import OpenPeerTokenizer
|
|
|
|
|
|
config = OpenPeerConfig()
|
|
|
model = OpenPeerLLM(config)
|
|
|
tokenizer = OpenPeerTokenizer()
|
|
|
|
|
|
|
|
|
test_prompts = [
|
|
|
"Explain how decentralized computing works.",
|
|
|
"What are the benefits of peer-to-peer networks?",
|
|
|
"How does distributed machine learning improve model training?"
|
|
|
]
|
|
|
|
|
|
print("\nRunning inference tests...")
|
|
|
for i, prompt in enumerate(test_prompts, 1):
|
|
|
print(f"\nTest {i}:")
|
|
|
print(f"Prompt: {prompt}")
|
|
|
try:
|
|
|
|
|
|
inputs = tokenizer(prompt)
|
|
|
input_ids = torch.tensor([inputs["input_ids"]], dtype=torch.long)
|
|
|
|
|
|
|
|
|
outputs = model(input_ids)
|
|
|
|
|
|
|
|
|
logits = outputs["logits"]
|
|
|
predictions = torch.argmax(logits[0], dim=-1)
|
|
|
response = tokenizer.decode(predictions.tolist())
|
|
|
|
|
|
print(f"Response: {response}")
|
|
|
print("-" * 80)
|
|
|
except Exception as e:
|
|
|
print(f"Error during inference: {str(e)}")
|
|
|
|
|
|
|
|
|
print("\nModel Architecture:")
|
|
|
print(f"Hidden Size: {model.config.hidden_size}")
|
|
|
print(f"Number of Layers: {model.config.num_hidden_layers}")
|
|
|
print(f"Number of Attention Heads: {model.config.num_attention_heads}")
|
|
|
|
|
|
|
|
|
if torch.cuda.is_available():
|
|
|
print("\nGPU Memory Usage:")
|
|
|
print(f"Allocated: {torch.cuda.memory_allocated() / 1024**2:.2f} MB")
|
|
|
print(f"Cached: {torch.cuda.memory_reserved() / 1024**2:.2f} MB")
|
|
|
|
|
|
print("\nTest completed!")
|
|
|
|
|
|
def main():
|
|
|
print("Starting OpenPeerLLM tests...")
|
|
|
print("=" * 80)
|
|
|
|
|
|
try:
|
|
|
test_tokenizer()
|
|
|
except Exception as e:
|
|
|
print(f"Tokenizer test failed: {str(e)}")
|
|
|
|
|
|
try:
|
|
|
test_model_config()
|
|
|
except Exception as e:
|
|
|
print(f"Config test failed: {str(e)}")
|
|
|
|
|
|
try:
|
|
|
test_model_architecture()
|
|
|
except Exception as e:
|
|
|
print(f"Model architecture test failed: {str(e)}")
|
|
|
|
|
|
print("=" * 80)
|
|
|
print("Tests completed!")
|
|
|
|
|
|
try:
|
|
|
run_inference_test()
|
|
|
except Exception as e:
|
|
|
print(f"Inference test failed: {str(e)}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
main() |