有时候我还真不明白，为什么现在大家把AI吹得那么牛逼。因为就是我的使用体验而言，我觉得在写代码的时候，AI真的是挺糟糕的.我用的AI是Deepseek。

理论上无论我用的是什么AI，它们都应该是写代码的高手，无论是写什么类型的代码都应该没什么问题。我这么低端的用户，也不会提出一些什么高端的要求，但事实证明，不知道是我提要求提的不够精准还是怎么着，反正出来的结果我个人感觉基本上没有一次就能符合我的要求。有些时候的确能接近生成我想要的结果，但更多时候那串代码当你复制下来运行的时候会发现根本运行不了。对一个什么都不知道的小白在代码运行不了的时候可以怎么办呢？换一个AI继续提问继续复制粘贴？但如果这个AI有问题，其它AI会不会也有问题呢？为什么人家觉得那个玩意那么神奇，但我自己的使用体验却那么差呢？是不是因为AI已经判断出我这个人是个挑剔的完美主义者，所以故意给我些有瑕疵的东西？每一次都有瑕疵，每一次的瑕疵都说不准在什么地方，幸好绝大多数情况之下，我要的不是AI的全过程，我只是要参考一下它的思路而已。当我要求它给我写某个代码的时候，实际上我更看重的是它用什么途径，而不是它的代码具体是什么。如果它的代码能顺利运行，那当然好，但实际上我感觉一次性就能成功运行的几率可能不足10%。

当我给出一串数字，要求它给我做数据分析。噼里啪啦一大段字很牛逼，一贴进去发现根本出不了结果，仔细看Python的那些错误代码，原来一开始那个屌丝把原始数据都给我整没了一些，因为长度不对，所以整不出结果。我要用Excel的公式实现某些功能，的确噼里啪啦有好多个方案。有些方案一看就直接可以忽略，因为又长又臭，另外一些我感觉有些希望的却发现到Excel里面同样也是出不了结果。至于为什么我也不去纠结了，我直接去研究这些方法的思路到底是什么。我让它用Power Query，也就是M语言给我实现某个功能，结果发现把那堆代码贴到高级编辑器，甚至都无法按确定提交，因为显示语法错误。那个编辑器是一个神经的存在，说语法错误，但定位的那个点实际上没有问题，但那个问题到底在哪里，你得人肉上下文去找。我使用的通常是对半检索法，首先留一半，然后一半的一半。理论上是这么找的，但实际上后面某些功能不要我觉得都无所谓，所以我只要找到那些我需要的就可以了。最终发现，某个用了函数的地方，理论上应该是“=>”。DS给我的代码是“->”。如果这份代码是由图片识别过来的，出现这种等号变成减号，我觉得可以理解，但显然这种事情怎么可能发生？！对一个什么都不懂的人来说，你叫他怎么可能人肉找得出这样的低级错误呢？！那个代码除了那个等号变成了减号以外，最后的结果还有一些无厘头的数据合并，那些数据在前面几个步骤里面已经不复存在了，在最后那个步骤又突然间被提起，真牛逼！

在这种情况下，说AI可以取代人类完成编程，我觉得起码暂时是不行的，尤其是要实现一些高度定制的功能的时候。

本文有 AI 参与编写。

什么是硬件加速？

硬件加速是指利用计算机中的专用硬件（如 GPU、专用编解码芯片）来执行视频编解码任务，而不是仅依赖 CPU 进行软件编码。相比纯软件编码，硬件加速具有以下优势：

• 更快的处理速度：专用硬件针对视频编解码进行了优化，处理速度可以提升数倍
• 更低的 CPU 占用：将负载转移到 GPU 或专用芯片，释放 CPU 资源
• 更低的功耗：硬件编码通常比软件编码更节能，延长笔记本电脑续航时间

硬件加速的权衡

虽然硬件加速很快，但也有一些需要注意的地方：

• 压缩效率略低：硬件编码器为了速度牺牲了一些压缩效率，相同质量下文件可能略大
• 可控性较差：硬件编码器的参数调节选项通常少于软件编码器
• 平台依赖性：不同平台和硬件支持的加速方式不同

主流硬件加速方案

1. VideoToolbox (macOS/iOS)

Apple 的硬件加速框架，支持 macOS 和 iOS 设备。

# 编码器
h264_videotoolbox
hevc_videotoolbox

# 使用示例
ffmpeg -i input.mp4 -c:v h264_videotoolbox -b:v 2M output.mp4

特点：

• 在 Apple Silicon (M1/M2/M3) 芯片上性能出色
• 支持硬件加速的 H.264、HEVC、ProRes 编码
• 低功耗，适合移动设备

2. NVENC (NVIDIA GPU)

NVIDIA GPU 内置的硬件编码器，从 GTX 600 系列开始支持。

# 编码器
h264_nvenc
hevc_nvenc
av1_nvenc  # RTX 40 系列及以上

# 使用示例
ffmpeg -hwaccel cuda -i input.mp4 -c:v h264_nvenc -preset p4 output.mp4

特点：

• 性能强劲，编码质量较好
• 支持多路并行编码
• 新一代显卡支持 AV1 编码

3. QuickSync (Intel 集成显卡)

Intel 集成显卡的硬件编码器，从第二代酷睿开始支持。

# 编码器
h264_qsv
hevc_qsv
av1_qsv  # 12 代及以上

# 使用示例
ffmpeg -hwaccel qsv -i input.mp4 -c:v h264_qsv -preset medium output.mp4

特点：

• 在没有独立显卡的情况下性能不错
• 功耗低
• 新一代处理器支持 AV1 编码

4. AMF (AMD GPU)

AMD GPU 的硬件编码器。

# 编码器
h264_amf
hevc_amf
av1_amf  # RX 7000 系列及以上

# 使用示例
ffmpeg -hwaccel amf -i input.mp4 -c:v h264_amf output.mp4

5. VAAPI (Linux)

Linux 上的通用硬件加速接口，支持 Intel、AMD 等多种硬件。

# 使用示例
ffmpeg -hwaccel vaapi -vaapi_device /dev/dri/renderD128 -i input.mp4 \
  -vf 'format=nv12,hwupload' -c:v h264_vaapi output.mp4

在 Python 中检测和使用硬件加速

下面是一个完整的 Python 实现，可以自动检测系统可用的硬件加速方式并选择最佳方案：

完整实现

import subprocess
import logging
from typing import Optional, Tuple

logger = logging.getLogger(__name__)


def get_hardware_encoder(use_hwaccel: bool = True) -> Tuple[str, Optional[dict]]:
    """
    检测可用的硬件加速器并返回合适的编码器设置。

    Args:
        use_hwaccel: 是否启用硬件加速

    Returns:
        (video_codec, hwaccel_options) 元组
        - video_codec: 编码器名称，如 'h264_videotoolbox'
        - hwaccel_options: 硬件加速选项字典，如 {'hwaccel': 'videotoolbox'}
    """
    if not use_hwaccel:
        return "libx264", None

    try:
        # 检查可用的硬件加速器
        hwaccel_result = subprocess.run(
            ["ffmpeg", "-hwaccels"],
            capture_output=True,
            text=True,
            timeout=5
        )
        hwaccels = hwaccel_result.stdout.lower()

        # 检查可用的编码器
        encoder_result = subprocess.run(
            ["ffmpeg", "-encoders"],
            capture_output=True,
            text=True,
            timeout=5
        )
        encoders = encoder_result.stdout.lower()

        logger.debug(f"Available hardware accelerators: {hwaccels}")
        logger.debug(f"Available encoders: {encoders}")

        def test_encoder(codec: str) -> bool:
            """测试编码器是否真正可用"""
            try:
                result = subprocess.run(
                    [
                        "ffmpeg",
                        "-f", "lavfi",           # 使用虚拟输入
                        "-i", "testsrc=duration=1:size=320x240:rate=1",
                        "-frames:v", "1",        # 只编码一帧
                        "-c:v", codec,           # 指定编码器
                        "-f", "null",            # 输出到空设备
                        "-"
                    ],
                    capture_output=True,
                    text=True,
                    timeout=10
                )
                return result.returncode == 0
            except Exception as e:
                logger.debug(f"Failed to test encoder {codec}: {e}")
                return False

        # 按优先级检测硬件编码器

        # 1. VideoToolbox (macOS/Apple Silicon)
        if "h264_videotoolbox" in encoders and "videotoolbox" in hwaccels:
            if test_encoder("h264_videotoolbox"):
                logger.info("Using VideoToolbox hardware acceleration")
                return "h264_videotoolbox", {"hwaccel": "videotoolbox"}

        # 2. NVIDIA NVENC
        if "h264_nvenc" in encoders:
            if test_encoder("h264_nvenc"):
                logger.info("Using NVIDIA NVENC hardware acceleration")
                if "cuda" in hwaccels:
                    return "h264_nvenc", {"hwaccel": "cuda"}
                return "h264_nvenc", None

        # 3. Intel QuickSync
        if "h264_qsv" in encoders and "qsv" in hwaccels:
            if test_encoder("h264_qsv"):
                logger.info("Using Intel QuickSync hardware acceleration")
                return "h264_qsv", {"hwaccel": "qsv"}

        # 4. AMD AMF
        if "h264_amf" in encoders and "amf" in hwaccels:
            if test_encoder("h264_amf"):
                logger.info("Using AMD AMF hardware acceleration")
                return "h264_amf", {"hwaccel": "amf"}

        # 5. VAAPI (Linux)
        if "h264_vaapi" in encoders and "vaapi" in hwaccels:
            if test_encoder("h264_vaapi"):
                logger.info("Using VAAPI hardware acceleration")
                return "h264_vaapi", {"hwaccel": "vaapi"}

    except Exception as e:
        logger.warning(f"Error checking hardware encoders: {e}")
        logger.info("Falling back to software encoding")

    # 回退到软件编码
    logger.info("Using software encoding (libx264)")
    return "libx264", None


def get_system_info() -> dict:
    """获取系统硬件加速信息"""
    try:
        # 获取 FFmpeg 版本
        version_result = subprocess.run(
            ["ffmpeg", "-version"],
            capture_output=True,
            text=True,
            timeout=5
        )

        # 获取硬件加速列表
        hwaccel_result = subprocess.run(
            ["ffmpeg", "-hwaccels"],
            capture_output=True,
            text=True,
            timeout=5
        )

        # 获取编码器列表（只提取硬件编码器）
        encoder_result = subprocess.run(
            ["ffmpeg", "-encoders"],
            capture_output=True,
            text=True,
            timeout=5
        )

        hw_encoders = []
        for line in encoder_result.stdout.split('\n'):
            if any(hw in line.lower() for hw in ['nvenc', 'qsv', 'videotoolbox', 'amf', 'vaapi']):
                hw_encoders.append(line.strip())

        return {
            "ffmpeg_version": version_result.stdout.split('\n')[0],
            "hwaccels": hwaccel_result.stdout,
            "hw_encoders": hw_encoders
        }
    except Exception as e:
        return {"error": str(e)}

使用示例

1. 检测系统信息

import json

# 获取系统硬件加速信息
info = get_system_info()
print(json.dumps(info, indent=2))

2. 在 ffmpeg-python 中使用

import ffmpeg

def encode_video_with_hwaccel(input_path: str, output_path: str, use_hwaccel: bool = True):
    """使用硬件加速编码视频"""

    # 获取硬件编码器
    vcodec, hw_options = get_hardware_encoder(use_hwaccel)

    # 创建输入流
    if hw_options:
        # 使用硬件加速解码
        stream = ffmpeg.input(input_path, **hw_options)
    else:
        stream = ffmpeg.input(input_path)

    # 配置输出
    stream = ffmpeg.output(
        stream,
        output_path,
        vcodec=vcodec,           # 使用检测到的编码器
        acodec='aac',            # 音频编码器
        video_bitrate='2M',      # 视频比特率
        audio_bitrate='192k',    # 音频比特率
        preset='medium',         # 编码预设（硬件编码器可能忽略此参数）
        **{'crf': '23'}          # 质量参数（硬件编码器可能忽略此参数）
    )

    # 执行编码
    ffmpeg.run(stream, overwrite_output=True)
    print(f"Video encoded successfully using {vcodec}")


# 使用硬件加速
encode_video_with_hwaccel('input.mp4', 'output.mp4', use_hwaccel=True)

# 强制使用软件编码
encode_video_with_hwaccel('input.mp4', 'output_sw.mp4', use_hwaccel=False)

不同平台的硬件加速检测

macOS

def detect_macos_hwaccel():
    """检测 macOS 硬件加速"""
    import platform

    if platform.system() != 'Darwin':
        return None

    # 检测芯片类型
    machine = platform.machine()
    is_apple_silicon = machine == 'arm64'

    # Apple Silicon 性能更好
    if is_apple_silicon:
        return {
            'platform': 'Apple Silicon',
            'recommended_encoder': 'h264_videotoolbox',
            'performance': 'excellent',
            'codecs': ['h264_videotoolbox', 'hevc_videotoolbox', 'prores_videotoolbox']
        }
    else:
        return {
            'platform': 'Intel Mac',
            'recommended_encoder': 'h264_videotoolbox',
            'performance': 'good',
            'codecs': ['h264_videotoolbox', 'hevc_videotoolbox']
        }

Windows

def detect_windows_hwaccel():
    """检测 Windows 硬件加速"""
    import platform

    if platform.system() != 'Windows':
        return None

    available = []

    # 检测 NVIDIA
    try:
        result = subprocess.run(
            ['nvidia-smi', '--query-gpu=name', '--format=csv,noheader'],
            capture_output=True,
            text=True,
            timeout=5
        )
        if result.returncode == 0:
            available.append({
                'type': 'NVIDIA',
                'encoder': 'h264_nvenc',
                'gpu': result.stdout.strip()
            })
    except:
        pass

    # 检测 Intel QuickSync（通过 FFmpeg）
    vcodec, _ = get_hardware_encoder(True)
    if 'qsv' in vcodec:
        available.append({
            'type': 'Intel QuickSync',
            'encoder': 'h264_qsv'
        })

    # 检测 AMD
    if 'amf' in vcodec:
        available.append({
            'type': 'AMD',
            'encoder': 'h264_amf'
        })

    return available

Linux

def detect_linux_hwaccel():
    """检测 Linux 硬件加速"""
    import platform
    import os

    if platform.system() != 'Linux':
        return None

    available = []

    # 检测 VAAPI 设备
    vaapi_devices = [f'/dev/dri/renderD{i}' for i in range(128, 140)]
    for device in vaapi_devices:
        if os.path.exists(device):
            available.append({
                'type': 'VAAPI',
                'device': device,
                'encoder': 'h264_vaapi'
            })
            break

    # 检测 NVIDIA
    try:
        result = subprocess.run(
            ['nvidia-smi'],
            capture_output=True,
            timeout=5
        )
        if result.returncode == 0:
            available.append({
                'type': 'NVIDIA',
                'encoder': 'h264_nvenc'
            })
    except:
        pass

    return available

性能对比和最佳实践

编码速度对比（参考数据）

以编码一个 1080p 60fps 视频为例：

最佳实践

1. 自动检测并回退

   • 始终先尝试硬件加速
   • 检测失败时自动回退到软件编码
   • 记录日志便于调试

2. 选择合适的预设

   # NVENC 预设
   # p1 (fastest) -> p7 (slowest, best quality)
   stream = ffmpeg.output(stream, 'output.mp4', vcodec='h264_nvenc', preset='p4')
   

3. 考虑批量处理

   • 硬件编码器通常支持多路并行
   • NVENC 可以同时处理多个视频流

4. 监控编码质量

   • 硬件编码质量可能不如软件编码
   • 对质量要求高的场景考虑使用软件编码
   • 可以用 VMAF 等指标评估质量

5. 处理兼容性问题

   def safe_encode(input_path, output_path):
       """带错误处理的编码"""
       try:
           # 尝试硬件加速
           encode_video_with_hwaccel(input_path, output_path, use_hwaccel=True)
       except Exception as e:
           logger.warning(f"Hardware encoding failed: {e}")
           logger.info("Retrying with software encoding")
           # 回退到软件编码
           encode_video_with_hwaccel(input_path, output_path, use_hwaccel=False)
   

调试技巧

查看详细的 FFmpeg 输出

def encode_with_debug(input_path, output_path):
    """启用详细日志的编码"""
    vcodec, hw_options = get_hardware_encoder(True)

    stream = ffmpeg.input(input_path, **hw_options) if hw_options else ffmpeg.input(input_path)
    stream = ffmpeg.output(stream, output_path, vcodec=vcodec)

    # 获取完整命令
    cmd = ffmpeg.compile(stream, overwrite_output=True)
    print(f"FFmpeg command: {' '.join(cmd)}")

    # 执行并查看输出
    try:
        ffmpeg.run(stream, overwrite_output=True, capture_stdout=False, capture_stderr=False)
    except ffmpeg.Error as e:
        print(f"stdout: {e.stdout.decode()}")
        print(f"stderr: {e.stderr.decode()}")
        raise

检查硬件支持

# 查看所有硬件加速方式
ffmpeg -hwaccels

# 查看所有编码器
ffmpeg -encoders | grep -E "(nvenc|qsv|videotoolbox|amf|vaapi)"

# 测试特定编码器
ffmpeg -f lavfi -i testsrc=duration=1:size=1920x1080:rate=30 \
  -c:v h264_videotoolbox -f null -

总结

硬件加速是视频处理中的重要优化手段，可以大幅提升处理速度和降低系统负载。通过自动检测和回退机制，我们可以构建一个跨平台的健壮视频处理系统。

关键要点：

• 优先使用硬件加速，但保留软件编码作为回退方案
• 不同平台选择对应的最佳硬件加速方式
• 通过实际测试验证编码器可用性
• 根据场景在速度和质量之间取得平衡

参考资源

• FFmpeg Hardware Acceleration Guide
• NVIDIA NVENC Documentation
• Intel Quick Sync Video
• Apple VideoToolbox

Parquet 文件入门 Python 读写 Parquet 文件实战 大数据存储优化：Parquet 格式解析 Python 数据分析必备：Parquet 文件处理技巧 列式存储揭秘：Parquet 文件与性能优化 使用 Python 和 PyArrow 处理嵌套 Parquet 数据 从 CSV 到 Parquet：Python 数据格式转换教程

什么是 Parquet 文件？

为什么使用 Parquet？

• 列式存储：按列存储数据，提高分析任务的查询性能。
• 压缩：支持高效的压缩技术，减少存储空间。
• 兼容性：可与多种数据处理框架配合使用。
• 模式演进：支持增加或删除列而不破坏现有数据。

安装所需库

pip install pandas pyarrow

在 Python 中读取 Parquet 文件

import pandas as pd
import pyarrow.parquet as pq

# Parquet 文件路径
file_path = "example.parquet"

# 读取 Parquet 文件到 DataFrame
df = pd.read_parquet(file_path)

# 显示前 5 行
print(df.head())

写入 Parquet 文件

import pandas as pd

# 创建示例 DataFrame
data = {
    "name": ["Alice", "Bob", "Charlie"],
    "age": [25, 30, 35],
    "city": ["London", "Paris", "New York"]
}
df = pd.DataFrame(data)

# 保存为 Parquet
df.to_parquet("output.parquet", engine="pyarrow", index=False)

处理嵌套数据

import pyarrow.parquet as pq
from io import BytesIO

# 直接读取 Parquet 文件
table = pq.read_table("example.parquet")
df = table.to_pandas()
print(df.head())

总结

理解 Sigma 函数：因子、乘法性与公式推导

一文看懂 Sigma 函数：因子分解的终极威力！ σ(n) 完全解析：为什么求和函数能“自动”变成乘积？ 数学之美：Sigma 函数的推导、公式与 Python 实现 从几何级数到质因数：Sigma 函数的魔法公式大揭秘 搞懂 σ(n) 的那一天，我看到了数学的秩序 为什么 σ(n) = 乘积？带你走进数论的核心思想 Divisor 终极指南：Sigma 函数推导 + 代码 一篇搞定

可除性符号

质因数分解与因子的结构

关键性质：Sigma 函数是乘法性的

质数幂的 Sigma 公式

示例：计算 σ(12)

Python 实现：高效的 Sigma 函数

def sigma(n: int) -> int:
    """高效计算因子和函数 σ(n)。"""
    total = 1
    x = n

    # 处理质因数 2
    count = 0
    while x % 2 == 0:
        x //= 2
        count += 1
    if count > 0:
        total *= (2 ** (count + 1) - 1) // (2 - 1)

    # 处理奇质数
    p = 3
    while p * p <= x:
        if x % p == 0:
            count = 0
            while x % p == 0:
                x //= p
                count += 1
            total *= (p ** (count + 1) - 1) // (p - 1)
        p += 2

    # 若剩下的是质数
    if x > 1:
        total *= (x**2 - 1) // (x - 1)

    return total

结语

不知道从什么时候开始，我就迷上了数据查询。

一开始只是想实现某个功能，后来发现原来实现同样东西，我用不同方法都可以做到。哪个方法更直观简便一些？我感觉Excel VBA的SQL，Power Query以及Python相比，就数据处理的方便性来说Python是碾压的，但我没有发现Python的巨大优势。问题可能在我交给Python处理的数据太少了，跟其它两个相比体现不出Python的高效。在控制Excel单元格格式方面，Python天生不如office自家的VBA。为什么我会把PQ跟VBA跟Python相比呢？是因为从Office 2019开始，PQ就算是内置的一个功能。VBA里面的SQL天生有缺陷，因为跟真正的数据库SQL相比，那就是个阉割版，有些你觉得明明可以实现的东西，在Excel VBA里好像就真没有直接的解决方案，为什么居然会这样呢？

到现在为止，我依然没有发现Excel VBA的SQL有直接的文本拼接功能。其它数据库的SQL里，那就是一个很简单的函数。Excel VBA的SQL在合并其他数据方面没问题，但一旦遇到需要进行文本拼接。我感觉除非在查询结束以后再做一个字符串的字典，否则无解。或许你会说其实我也可以直接在Excel的函数层面做这个拼接，因为用textjoin函数实际上是能实现那个功能的，但关键是如果数据比较多，既然我都在VBA里完成前面的所有，为什么最后的功能又要回归到函数呢？二者的运算速度不是一个层次的。每当我遇到文本拼接，我知道SQL是撞墙的，所以我就直接想到PQ。

PQ可以做数据分组，做文本拼接直接在高级编辑器里修改就能实现，但关键是实际上可以不用PQ，我不想用那个玩意，使用可能会有点慢还行，如果要进入到里面编写代码，那个小窗口字体无法放大，简直要逼死我这种老花。更作死的是，很多时候提醒我错误，但错误根本不发生在提醒的那个地方，不断的嵌套括号、逗号、又或者不小心带入的中文标点符号都会导致错误，找茬的过程让人挺绝望。我觉得，我还是喜欢在VBA里用SQL，其实无非就是判断加循环。在PQ里我总觉得有些很容易就能做到的事情，但是它非得用一些看上去很复杂的函数去实现。比如要根据A字段去决定B字段的数值是正数还是负数，在SQL里，一句很简单的iif就能实现，但在PQ里，你还得新建一个条件列，把条件写进去，接着把原来那个数值列删掉，再把条件列的名称改成数值列。当然你也可以直接使用replacevalue函数，但据说那个东西的执行效率反倒不如新增一列再删除一列。PQ里的函数非常多，嵌套用起来的方法更是让你眼花缭乱，也正是因为那些杂七杂八的东西太多了，反而让我觉得不如SQL简单干脆。让我很绝望的是，Access的SQL可以直接文本拼接，但Excel里的就不行，虽然二者是同一个版本的office。

当我在一个问题上钻研得越深，我就越能理解到高中时候，我的数学老师说学习数学的几个境界：不懂不会，会而不对，对而不全，全而不好。

Amdahl 定律 vs Gustafson 定律 — 完整教程、推导、应用场景及 Python 绘图

Amdahl 定律 vs Gustafson 定律：完整教程、推导、应用场景及 Python 绘图 理解并行加速：通过代码讲解 Amdahl 定律和 Gustafson 定律 并行计算基础：Amdahl 定律、Gustafson 定律及加速建模 并行加速原理：Amdahl 和 Gustafson 定律完整指南 并行扩展解析：推导并比较 Amdahl 和 Gustafson 定律 Amdahl vs Gustafson：并行加速完整指南（含 Python 代码） 并行性能建模：Amdahl 定律、Gustafson 定律及实际应用 学习并行加速：数学、直觉、应用场景及 Python 可视化 并行计算：必须掌握的两条定律（Amdahl & Gustafson） 工程师的并行加速：Amdahl 定律、Gustafson 定律及 Python 实现 从理论到代码：用 Amdahl 和 Gustafson 建模并行加速 实用并行加速指南：Amdahl 定律、Gustafson 定律及可视化 为什么并行不是无限的：简单解释 Amdahl vs Gustafson 并行加速真相：Amdahl 限制 vs Gustafson 扩展 并行计算神话与现实：Amdahl 和 Gustafson 的教训

引言

• Amdahl 定律 — 固定规模工作负载的性能
• Gustafson 定律 — 可扩展规模工作负载的性能

1. 什么是加速比？

2. Amdahl 定律（固定工作量）

2.1 直觉

• 总工作量保持 不变
• 部分工作是串行的，无法并行化

• f = 串行比例
• 1 - f = 可并行比例

2.2 推导

2.3 当 N → ∞ 时的极限

2.4 Amdahl 定律的实际应用场景

• GPU 内核优化固定张量大小
• 单次请求推理延迟降低
• 视频编码、压缩、排序
• 加速固定批量作业
• 数据库查询加速

3. Gustafson 定律（可扩展工作量）

3.1 直觉

3.2 推导

• f = 串行比例（按规模测量）

3.3 解释

3.4 Gustafson 定律的实际应用场景

• 天气和气候模拟
• 粒子模拟、CFD、有限元分析
• LLM 训练：更多 GPU → 更长序列或更大模型
• 大数据分析（Spark, Dask, Flink）
• 蒙特卡洛模拟

4. Amdahl 定律 vs Gustafson 定律（比较表）

5. 实际应用场景（综合视角）

Amdahl（延迟优化）

• 减少单次 LLM 查询推理时间
• 加速数据库 join 操作
• 固定张量 GPU 内核优化
• 视频编码（相同视频）

Gustafson（吞吐量 / 扩展）

• LLM 训练（扩展至更多 GPU）
• 高分辨率天气模型模拟
• 大数据 ETL 扩展
• 科学 HPC 工作负载

6. Python 绘图脚本（显示两条定律）

图示解读

• Amdahl 曲线迅速趋于平缓——受串行部分限制。
• Gustafson 曲线几乎线性上升——适用于可扩展工作负载。
• 串行比例 f 越高，两种模型差距越大。

结论

• Amdahl 定律 → 固定规模工作负载 → 收益递减
• Gustafson 定律 → 可扩展工作负载 → 近似线性加速
• 结合使用理解硬件极限与算法特性
• Python 工具使可视化直观易懂

[caption id="attachment_70386" align="alignnone" width="2017"] Q Learning强化学习算法(机器学习/人工智能)[/caption] 强化学习（Reinforcement Learning, RL）是一种让智能体/Agent通过与环境交互、试错学习来获得最优行为策略的机器学习方法。本文用一个简单的 Q-learning 迷宫示例，帮助你快速理解强化学习的基本原理。

强化学习入门：从试错中学习的艺术 Reinforcement Learning 101: The Art of Learning by Trial and Error 深度解析强化学习：Q-Learning算法详解 Deep Dive into Reinforcement Learning: Understanding the Q-Learning Algorithm 机器如何学会自己做决定？强化学习告诉你答案 How Do Machines Learn to Make Their Own Decisions? Reinforcement Learning Explained 从奖励中学习：人工智能的“试错智慧” Learning from Rewards: The Trial-and-Error Intelligence Behind AI

一、什么是强化学习？

• Agent（智能体）：做决策、执行动作的主体
• Environment（环境）：智能体所处的世界
• State（状态）：当前环境的描述
• Action（动作）：智能体可采取的操作
• Reward（奖励）：环境反馈，用来衡量动作的好坏

二、Q-Learning 原理

• [math] \alpha [/math]：学习率（Learning Rate）
• [math] \gamma [/math]：折扣因子（Discount Factor）
• [math] r [/math]：奖励（Reward）
• [math] s' [/math]：下一状态（Next State）

三、迷宫环境设计

• 0：空地
• -1：墙
• 1：出口（目标）

四、完整 Python 实现代码

import numpy as np
import random

# 1️⃣ 定义迷宫
maze = np.array([
    [0,  0,  0, -1,  1],
    [0, -1,  0, -1,  0],
    [0,  0,  0,  0,  0]
])

n_rows, n_cols = maze.shape
actions = ['up', 'down', 'left', 'right']
Q = np.zeros((n_rows, n_cols, len(actions)))

# 2️⃣ 超参数
alpha = 0.1
gamma = 0.9
epsilon = 0.1
episodes = 500

# 3️⃣ 辅助函数
def is_valid(state):
    r, c = state
    return 0 <= r < n_rows and 0 <= c < n_cols and maze[r, c] != -1

def next_state(state, action):
    r, c = state
    if action == 'up': r -= 1
    elif action == 'down': r += 1
    elif action == 'left': c -= 1
    elif action == 'right': c += 1
    return (r, c)

def get_reward(state):
    r, c = state
    if maze[r, c] == 1: return 10
    elif maze[r, c] == -1: return -1
    return -0.1

# 4️⃣ 训练循环
for episode in range(episodes):
    state = (2, 0)
    done = False

    while not done:
        if random.uniform(0, 1) < epsilon:
            action_idx = random.randint(0, len(actions)-1)
        else:
            action_idx = np.argmax(Q[state[0], state[1]])

        action = actions[action_idx]
        next_s = next_state(state, action)

        if not is_valid(next_s):
            reward = -1
            next_s = state
        else:
            reward = get_reward(next_s)

        Q[state[0], state[1], action_idx] += alpha * (
            reward + gamma * np.max(Q[next_s[0], next_s[1]]) - Q[state[0], state[1], action_idx]
        )

        state = next_s
        if maze[state[0], state[1]] == 1:
            done = True

print("✅ 训练完成！")

# 5️⃣ 查看学到的路径
state = (2, 0)
path = [state]

while maze[state[0], state[1]] != 1:
    action_idx = np.argmax(Q[state[0], state[1]])
    next_s = next_state(state, actions[action_idx])
    if not is_valid(next_s) or next_s in path:
        break
    state = next_s
    path.append(state)

print("🗺️ 学到的路径:", path)

五、运行结果

六、总结

• Q 表：保存每个状态-动作的价值
• ε-greedy 策略：平衡探索与利用
• 奖励函数设计：引导智能体形成目标导向行为
• 强化学习思想：通过试错和奖励反馈不断改进策略

本文介绍一例Ta-Lib模块的安装报错以及解决办法

网络连接问题导致webdiver-manager无法正常

本文介绍一例由Windows的“应用执行别名”引起的环境变量冲突

编译环境与云函数实例的运行环境不同，导致编译后的`numpy.core._multiarray_umath`不兼容。

Pandas会直接复制原DataFrame并返回，而不是按照apply指定的函数方法进行处理

重复添加Handler导致重复输出内容

本文以Chrome浏览器为例，介绍在Selenium中保持网站登录状态的方法。

　　本文介绍下载与Chrome浏览器的版本相匹配的ChromeDriver的两种方法。

　　本文介绍如何从Python官网中找到旧版本Python安装包的下载链接。

　　本文介绍在Windows下通过为特定的Python项目创建VS Code工作空间，从而实现为不同的Python项目自动选择不同的解释器。

如果被注释的代码存在转义冲突，则会引起报错。可以在注释符号之前添加一个`r`避免转义。

本文介绍通过Python自带的email模块解析电子邮件的基本实现方法。