大模型应用：算力分层治理：基于大模型算力四层匹配体系的优化方案.72

# 所需依赖：安装pytorch（需支持CUDA）
import torch
import time

# 1. 检查GPU是否可用（确保有NVIDIA GPU并安装了CUDA）
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"使用设备：{device}")
if device.type == "cuda":
    print(f"GPU型号：{torch.cuda.get_device_name(0)}")

# 2. 定义大模型常见的矩阵运算（GEMM）
batch_size = 128
seq_len = 512
hidden_dim = 4096  # 大模型常见的隐藏层维度

# 3. 生成两种精度的矩阵（FP32 vs FP16）
mat_fp32_1 = torch.randn(batch_size, seq_len, hidden_dim, dtype=torch.float32).to(device)
mat_fp32_2 = torch.randn(batch_size, hidden_dim, seq_len, dtype=torch.float32).to(device)

mat_fp16_1 = mat_fp32_1.half()  # 转换为FP16
mat_fp16_2 = mat_fp32_2.half()

# 4. 计算两种精度下的矩阵乘法耗时（验证计算效率匹配）
def calculate_matmul_time(mat1, mat2, repeat=10):
    """计算矩阵乘法耗时，重复多次取平均值"""
    torch.cuda.synchronize()  # 等待GPU之前的任务完成
    start_time = time.time()
    for _ in range(repeat):
        result = torch.bmm(mat1, mat2)  # 批量矩阵乘法（大模型核心运算）
    torch.cuda.synchronize()  # 等待GPU任务完成
    end_time = time.time()
    avg_time = (end_time - start_time) / repeat
    return avg_time, result.dtype

# 5. 执行计算并输出结果
fp32_time, fp32_dtype = calculate_matmul_time(mat_fp32_1, mat_fp32_2)
fp16_time, fp16_dtype = calculate_matmul_time(mat_fp16_1, mat_fp16_2)

# 6. 打印对比结果
print(f"\n=== 计算效率对比 ===")
print(f"{fp32_dtype} 平均耗时：{fp32_time:.4f} 秒")
print(f"{fp16_dtype} 平均耗时：{fp16_time:.4f} 秒")
print(f"FP16 比 FP32 快 {fp32_time / fp16_time:.2f} 倍")

import torch
from transformers import GPT2Model, GPT2Config

# 1. 检查GPU是否可用
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"使用设备：{device}")
if device.type == "cuda":
    gpu_name = torch.cuda.get_device_name(0)
    gpu_memory = torch.cuda.get_device_properties(0).total_memory / (1024 ** 3)
    print(f"GPU型号：{gpu_name}")
    print(f"GPU显存容量：{gpu_memory:.2f} GB")

# 2. 定义一个超大规模的GPT2模型（故意超出单卡显存容量，模拟OOM场景）
config = GPT2Config(
    vocab_size=50257,
    n_positions=1024,
    n_embd=8192,  # 增大隐藏层维度，提升模型参数规模
    n_layer=48,   # 增大层数，提升模型参数规模
    n_head=64,
)

try:
    # 3. 尝试加载FP32精度模型（不做显存匹配，预期出现OOM错误）
    print("\n=== 尝试加载FP32精度模型（不做显存匹配）===")
    model_fp32 = GPT2Model(config).to(device)
    print("FP32模型加载成功（未出现OOM，说明你的GPU显存足够大）")
except RuntimeError as e:
    if "out of memory" in str(e).lower():
        print("出现OOM错误！FP32模型参数超出显存容量，不匹配！")
    else:
        raise e

# 4. 采用量化技术（INT8）进行显存容量匹配，加载模型
print("\n=== 采用INT8量化技术（做显存匹配）加载模型 ===")
try:
    # 加载模型并转换为INT8精度（减少显存占用，实现容量匹配）
    model_int8 = GPT2Model(config).to(device).to(torch.int8)
    # 验证模型是否可用
    dummy_input = torch.randint(0, 50257, (1, 1024)).to(device)
    output = model_int8(dummy_input)
    print("INT8模型加载成功！显存容量匹配完成，无OOM错误")
    print(f"模型输出形状：{output.last_hidden_state.shape}")
except Exception as e:
    print(f"INT8模型加载异常：{e}")

import torch
import torch.distributed as dist
import time
import os

def init_distributed():
    """初始化分布式环境（多卡通信必备）"""
    dist.init_process_group(backend="nccl")  # NCCL是NVIDIA多卡通信的最优后端
    local_rank = int(os.environ.get("LOCAL_RANK", 0))
    torch.cuda.set_device(local_rank)
    return local_rank

def naive_allreduce(tensor):
    """朴素AllReduce算法（Naive AllReduce）"""
    world_size = dist.get_world_size()
    # 1. 所有进程将张量发送给其他所有进程
    for rank in range(world_size):
        if rank != dist.get_rank():
            dist.send(tensor.clone(), dst=rank)
    # 2. 所有进程接收其他进程的张量并求和
    for rank in range(world_size):
        if rank != dist.get_rank():
            recv_tensor = torch.zeros_like(tensor)
            dist.recv(recv_tensor, src=rank)
            tensor += recv_tensor
    # 3. 归一化
    tensor /= world_size
    return tensor

def ring_allreduce(tensor):
    """环形AllReduce算法（Ring AllReduce）"""
    world_size = dist.get_world_size()
    rank = dist.get_rank()
    tensor_size = tensor.numel()
    chunk_size = tensor_size // world_size

    # 1. 分块（将张量拆分为world_size个块）
    chunks = torch.split(tensor, chunk_size, dim=0)
    result = torch.cat(chunks, dim=0).clone()

    # 2. 环形数据交换与求和
    for i in range(world_size - 1):
        # 发送给下一个进程（rank+1），接收从上一个进程（rank-1）
        send_rank = (rank + 1) % world_size
        recv_rank = (rank - 1) % world_size

        # 发送当前负责的块
        send_chunk = chunks[(rank + i) % world_size].clone()
        dist.send(send_chunk, dst=send_rank)

        # 接收其他进程的块并求和
        recv_chunk = torch.zeros_like(send_chunk)
        dist.recv(recv_chunk, src=recv_rank)
        chunks[(rank + i + 1) % world_size] += recv_chunk

    # 3. 聚合所有块
    for i in range(world_size):
        result[i * chunk_size: (i + 1) * chunk_size] = chunks[i]

    # 4. 归一化
    result /= world_size
    return result

def main():
    # 1. 初始化分布式环境
    local_rank = init_distributed()
    device = torch.device(f"cuda:{local_rank}")
    rank = dist.get_rank()
    world_size = dist.get_world_size()

    # 2. 生成模拟梯度数据（大模型训练中的梯度张量）
    tensor_size = (1024 * 1024 * 8,)  # 约32MB（FP32），可增大规模观察差异
    tensor = torch.randn(tensor_size, dtype=torch.float32).to(device)

    # 3. 验证两种AllReduce算法的耗时
    repeat = 100

    # 3.1 Naive AllReduce
    dist.barrier()  # 同步所有进程
    start_time = time.time()
    for _ in range(repeat):
        naive_result = naive_allreduce(tensor.clone())
    dist.barrier()
    naive_time = (time.time() - start_time) / repeat

    # 3.2 Ring AllReduce
    dist.barrier()
    start_time = time.time()
    for _ in range(repeat):
        ring_result = ring_allreduce(tensor.clone())
    dist.barrier()
    ring_time = (time.time() - start_time) / repeat

    # 4. 输出结果（仅在rank 0进程打印）
    if rank == 0:
        print(f"\n=== 多卡通信算法效率对比（{world_size}卡）===")
        print(f"Naive AllReduce 平均耗时：{naive_time:.6f} 秒")
        print(f"Ring AllReduce 平均耗时：{ring_time:.6f} 秒")
        print(f"Ring AllReduce 比 Naive AllReduce 快 {naive_time / ring_time:.2f} 倍")

    # 5. 销毁分布式环境
    dist.destroy_process_group()

if __name__ == "__main__":
    main()

from dataclasses import dataclass
from typing import List, Optional

# 1. 定义数据结构：算力资源与业务场景
@dataclass
class ComputeResource:
    """算力资源类"""
    name: str  # 资源名称
    type: str  # 资源类型（优质/普通）
    scale: int  # 卡数规模
    speed: float  # 推理/训练速度（单位：任务/秒）
    is_available: bool = True  # 是否可用

@dataclass
class BusinessTask:
    """业务任务类"""
    name: str  # 任务名称
    scene: str  # 业务场景（实时生成/批量推理/模型微调）
    priority: int  # 优先级（1-10，10最高）
    required_scale: int  # 所需卡数规模

# 2. 初始化算力资源池
def init_compute_resources() -> List[ComputeResource]:
    """初始化算力资源池"""
    return [
        ComputeResource("A100-8卡（NVLink）", "优质", 8, 100.0),
        ComputeResource("A100-4卡（NVLink）", "优质", 4, 50.0),
        ComputeResource("RTX 3090-8卡（PCIe）", "普通", 8, 30.0),
        ComputeResource("RTX 3090-4卡（PCIe）", "普通", 4, 15.0),
    ]

# 3. 实现业务-算力匹配调度逻辑
class BusinessComputeMatcher:
    """业务-算力匹配调度器"""
    def __init__(self, resources: List[ComputeResource]):
        self.resources = resources

    def match_resource(self, task: BusinessTask) -> Optional[ComputeResource]:
        """为业务任务匹配最优算力资源"""
        # 步骤1：筛选可用且满足卡数要求的资源
        candidate_resources = [
            res for res in self.resources
            if res.is_available and res.scale >= task.required_scale
        ]
        if not candidate_resources:
            print(f"任务【{task.name}】：无可用算力资源满足需求！")
            return None

        # 步骤2：根据业务场景筛选最优资源类型
        scene_resource_map = {
            "实时生成": ["优质", 4],  # 优先优质、4卡规模
            "批量推理": ["普通", 8],  # 优先普通、8卡规模
            "模型微调": ["优质", 8],  # 优先优质、8卡规模
        }

        preferred_type, preferred_scale = scene_resource_map.get(task.scene, ["普通", 4])

        # 步骤3：优先匹配场景偏好的资源
        preferred_resources = [
            res for res in candidate_resources
            if res.type == preferred_type and res.scale >= preferred_scale
        ]

        if preferred_resources:
            # 按速度排序，选择最快的资源
            best_resource = max(preferred_resources, key=lambda x: x.speed)
        else:
            # 无偏好资源，选择最快的可用资源
            best_resource = max(candidate_resources, key=lambda x: x.speed)

        # 步骤4：占用资源（标记为不可用）
        best_resource.is_available = False
        print(f"任务【{task.name}】：匹配到最优算力资源【{best_resource.name}】")
        print(f"  业务场景：{task.scene} | 优先级：{task.priority} | 处理速度：{best_resource.speed} 任务/秒")
        return best_resource

    def release_resource(self, resource: ComputeResource):
        """释放算力资源（标记为可用）"""
        resource.is_available = True
        print(f"算力资源【{resource.name}】已释放，当前可用")

# 4. 测试调度器
def main():
    # 初始化算力资源池
    resources = init_compute_resources()
    matcher = BusinessComputeMatcher(resources)

    # 定义测试业务任务
    tasks = [
        BusinessTask("用户实时对话", "实时生成", 10, 4),
        BusinessTask("文档批量总结", "批量推理", 5, 8),
        BusinessTask("金融行业模型微调", "模型微调", 8, 8),
        BusinessTask("商品文案实时生成", "实时生成", 9, 4),
    ]

    # 调度任务
    print("=== 大模型业务-算力匹配调度开始 ===")
    allocated_resources = []
    for task in tasks:
        print("-" * 60)
        resource = matcher.match_resource(task)
        if resource:
            allocated_resources.append((task, resource))

    # 模拟任务完成，释放资源
    print("\n" + "-" * 60)
    print("=== 业务任务完成，释放算力资源 ===")
    for task, resource in allocated_resources:
        matcher.release_resource(resource)

if __name__ == "__main__":
    main()