add bge_m3,flag_reranker,m3_base.

Author: wangwenxing-mt

pytorch/Embedding/bge_m3/README.md

@@ -16,3 +16,8 @@ pip install -r requirements.txt
 ```shell
 python test_bge_m3.py
 ```
+
+4. performance
+```shell
+python perf_bge_m3_big.py
+```

pytorch/Embedding/bge_m3/perf_bge_m3_big.py

@@ -0,0 +1,110 @@
+from FlagEmbedding import BGEM3FlagModel
+import torch
+import torch_musa
+import time
+import numpy as np
+import concurrent.futures
+import random
+import string
+from tqdm import tqdm
+
+# 生成1024 tokens的长文本（约1500-1800字符）
+def generate_long_text(target_tokens=1024):
+    """生成符合目标token长度的随机文本"""
+    words = []
+    current_tokens = 0
+    while current_tokens < target_tokens:
+        word_len = random.randint(3, 10)
+        words.append(''.join(random.choices(string.ascii_letters, k=word_len)))
+        current_tokens += 1
+    return " ".join(words)
+
+def process_batch(model, batch_sentences, max_length):
+    """处理单个批次并返回结果和耗时"""
+    start_time = time.time()
+    embeddings = model.encode(batch_sentences, max_length=max_length)['dense_vecs']
+    end_time = time.time()
+    return embeddings, end_time - start_time
+
+def warmup_model(model, batch_size=32, max_length=512, iterations=10):
+    """执行模型预热以消除冷启动影响"""
+    warmup_sentences = ["Warmup sentence " * 20] * batch_size  # 模拟长文本
+    for _ in range(iterations):
+        model.encode(warmup_sentences, max_length=max_length)['dense_vecs']
+
+if __name__ == '__main__':
+    # 初始化模型
+    model = BGEM3FlagModel('./bge-m3', use_fp16=True, device='musa:0')
+    
+    # ===== 关键优化1：生成1024 tokens的长文本 =====
+    print("=== Generating 1024-token texts ===")
+    long_query = generate_long_text(1024)
+    long_passage = generate_long_text(1024)
+    
+    # ===== 关键优化2：添加长文本预热 =====
+    print("\n=== Starting model warm-up with long texts ===")
+    warmup_model(model, batch_size=32, max_length=1024, iterations=10)
+    print("=== Warm-up completed ===\n")
+    
+    # ===== 准备30个批次的并行任务 =====
+    batch_pairs_list = []
+    for _ in range(30):
+        batch = []
+        for _ in range(32):  # 每个批次32个样本
+            q = generate_long_text(1024) if random.random() > 0.5 else long_query
+            p = generate_long_text(1024) if random.random() > 0.5 else long_passage
+            batch.append(q)
+            batch.append(p)
+        batch_pairs_list.append(batch)
+    
+    # ===== 并行执行30个批次 =====
+    print("=== Starting 30 parallel batch processing ===")
+    total_tokens = 0
+    for batch in batch_pairs_list:
+        total_tokens += sum(len(text.split()) for text in batch)
+    
+    start_time = time.time()
+    batch_results = []
+    batch_times = []
+    
+    # 使用线程池并行处理
+    with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor:
+        futures = [executor.submit(process_batch, model, batch, 1024) for batch in batch_pairs_list]
+        
+        for future in tqdm(concurrent.futures.as_completed(futures), total=len(futures)):
+            embeddings, batch_time = future.result()
+            batch_results.append(embeddings)
+            batch_times.append(batch_time)
+    
+    end_time = time.time()
+    total_time = end_time - start_time
+    
+    # ===== 性能指标计算 =====
+    # 1. 吞吐量指标
+    throughput_batches = len(batch_results) / total_time
+    throughput_tokens = total_tokens / total_time
+    
+    # 2. 延迟指标
+    avg_batch_time = sum(batch_times) / len(batch_times)
+    max_batch_time = max(batch_times)
+    min_batch_time = min(batch_times)
+    
+    # ===== 性能报告 =====
+    print("\n===== Performance Report =====")
+    print(f"Total batches processed: {len(batch_results)}")
+    print(f"Total tokens processed: {total_tokens}")
+    print(f"Total processing time: {total_time:.2f} seconds")
+    print("\n--- Throughput ---")
+    print(f"Throughput (batches/sec): {throughput_batches:.2f}")
+    print(f"Throughput (tokens/sec): {throughput_tokens:.2f}")
+    print("\n--- Latency ---")
+    print(f"Avg batch time: {avg_batch_time:.4f} sec")
+    print(f"Max batch time: {max_batch_time:.4f} sec")
+    print(f"Min batch time: {min_batch_time:.4f} sec")
+    print("=============================")
+    
+    # 示例相似度计算
+    embeddings_1 = batch_results[0][0:2]  # 取第一个批次的前两个查询
+    embeddings_2 = batch_results[0][2:4]  # 取第一个批次的前两个段落
+    similarity = np.dot(embeddings_1, embeddings_2.T)
+    print(f"\nSample similarity matrix:\n{similarity}")

pytorch/Embedding/flag_reranker/READE.md

@@ -0,0 +1,18 @@
+0. Start docker
+启动命令可参考: [README.md](../../README.md)
+
+1. Prerequisites
+```shell
+pip install -r requirements.txt
+
+pip install -U huggingface_hub
+```
+2. export env
+```shell
+export export HF_ENDPOINT=https://hf-mirror.com
+```
+
+3. Test
+```shell
+python perf_flag_reranker.py
+```

pytorch/Embedding/flag_reranker/perf_flag_reranker.py

@@ -0,0 +1,107 @@
+from FlagEmbedding import FlagReranker
+import time
+import numpy as np
+import concurrent.futures
+import random
+import string
+
+# 生成长文本（1024 tokens约1500-1800字符）
+def generate_long_text(target_tokens=1024):
+    words = []
+    current_tokens = 0
+    while current_tokens < target_tokens:
+        word_len = random.randint(3, 10)
+        words.append(''.join(random.choices(string.ascii_letters, k=word_len)))
+        current_tokens += 1
+    return " ".join(words)
+
+def process_batch(reranker, batch_pairs):
+    """处理单个批次并返回结果和耗时"""
+    start_time = time.perf_counter()
+    scores = reranker.compute_score(batch_pairs)
+    end_time = time.perf_counter()
+    return scores, end_time - start_time
+
+def main():
+    # 加载模型（FP16精度 + MUSA设备加速）
+    reranker = FlagReranker('BAAI/bge-reranker-large', 
+                           use_fp16=True, 
+                           device="musa")
+
+    # ===== 长文本优化：生成1024 tokens的输入 =====
+    print("=== Generating 1024-token texts ===")
+    long_query = generate_long_text(1024)
+    long_passage = generate_long_text(1024)
+    
+    # ===== 关键优化：添加模型预热（使用长文本）===== [1,6](@ref)
+    print("=== Starting model warm-up with long texts ===")
+    warmup_pairs = [[long_query, long_passage]] * 16
+    for _ in range(5):
+        reranker.compute_score(warmup_pairs)
+    print("=== Warm-up completed ===\n")
+
+    # 单次长文本推理测试
+    start_time = time.perf_counter()
+    score = reranker.compute_score([long_query, long_passage])
+    latency = (time.perf_counter() - start_time) * 1000
+    print(f"Long text score: {str(score)} | Latency: {latency:.2f} ms")
+
+    # 准备批量数据（30个并行任务）
+    batch_pairs_list = []
+    for _ in range(30):
+        pairs = []
+        for _ in range(64):  # 每个任务64个样本
+            q = generate_long_text(1024) if random.random() > 0.5 else long_query
+            p = generate_long_text(1024) if random.random() > 0.5 else long_passage
+            pairs.append([q, p])
+        batch_pairs_list.append(pairs)
+
+    # ===== 并行执行30个任务 =====
+    print("\n=== Starting 30 parallel batch processing ===")
+    total_tokens = sum(
+        sum(len(q.split()) + len(p.split()) for q, p in pairs) 
+        for pairs in batch_pairs_list
+    )
+    batch_times = [] 
+    start_time = time.perf_counter()
+    with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor:  # 控制并发数
+        futures = [executor.submit(process_batch, reranker, pairs) for pairs in batch_pairs_list]
+        
+        batch_results = []
+        for future in concurrent.futures.as_completed(futures):
+            scores, batch_time = future.result()
+            batch_results.append(scores)
+            batch_times.append(batch_time)
+    
+    total_time = time.perf_counter() - start_time
+
+    # 性能统计
+    total_pairs = 30 * 64  # 30任务 * 每任务64对
+    throughput_pairs = total_pairs / total_time
+    throughput_tokens = total_tokens / total_time
+
+    avg_batch_time = sum(batch_times) / len(batch_times)
+    max_batch_time = max(batch_times)
+    min_batch_time = min(batch_times)
+
+    print("\n===== Performance Report =====")
+    print(f"Total batches processed: {len(batch_results)}")
+    print(f"Total pairs processed: {total_pairs}")
+    print(f"Total tokens processed: {total_tokens}")
+    print(f"Total processing time: {total_time:.2f} seconds")
+
+    print("\n--- Throughput ---")
+    print(f"Throughput: {throughput_pairs:.2f} pairs/sec")
+    print(f"Token throughput: {throughput_tokens:.2f} tokens/sec")
+    
+    print("\n--- Latency ---")
+    print(f"Average batch time: {avg_batch_time:.4f} sec")
+    print(f"Max batch time: {max_batch_time:.4f} sec")
+    print(f"Min batch time: {min_batch_time:.4f} sec")
+
+    print("=============================")
+
+
+
+if __name__ == "__main__":
+    main()

pytorch/Embedding/flag_reranker/requirements.txt

@@ -0,0 +1,5 @@
+FlagEmbedding
+accelerate==1.0.1
+transformers==4.44.0
+peft
+

pytorch/Embedding/m3_base/README.md

@@ -0,0 +1,18 @@
+0. Start docker
+启动命令可参考: [README.md](../../README.md)
+
+1. Prerequisites
+```shell
+pip install -r requirements.txt
+
+pip install -U huggingface_hub
+```
+2. export env
+```shell
+export export HF_ENDPOINT=https://hf-mirror.com
+```
+
+3. Test
+```shell
+python perf_m3_base.py
+```