如何生成Python函数代码的持久哈希以实现磁盘缓存？

核心需求 #

实现磁盘缓存机制：当函数参数和函数代码逻辑未发生变化时，直接读取缓存结果跳过重复计算；需精准识别函数代码的逻辑变更（哪怕是细微的运算差异），且支持Jupyter Notebook中定义的函数。

实现思路 #

要达成目标，需分四步：提取并标准化函数代码、计算代码哈希、计算参数哈希、结合两者生成唯一缓存键并实现磁盘读写。

1. 提取并标准化函数代码 #

直接哈希源码会受空格、换行、注释等格式差异影响，因此采用**抽象语法树（AST）**来标准化代码：AST仅保留代码逻辑结构，忽略格式细节，确保逻辑相同的代码生成一致的哈希，逻辑不同的代码哈希必然不同。

对于Jupyter中的函数，inspect.getsource多数情况下能直接获取源码；少数特殊场景可回退到函数的__source__属性。

2. 计算函数代码哈希 #

将标准化后的AST转换为字符串，再用SHA256算法计算哈希值，得到唯一标识函数逻辑的哈希字符串。

3. 计算参数哈希 #

针对Pandas Series，使用pd.util.hash_pandas_object计算其哈希和；其他类型参数通过repr序列化后哈希，确保参数内容变化时哈希值改变。

4. 实现磁盘缓存装饰器 #

结合函数代码哈希与参数哈希生成唯一缓存键，以此键作为缓存文件名，用joblib实现磁盘上的结果读写。

完整代码实现 #

import inspect
import ast
import hashlib
import pandas as pd
import joblib
from pathlib import Path

def get_function_ast_hash(func):
    # 解包装饰器，获取原始函数本体
    func = inspect.unwrap(func)
    # 获取函数源码，兼容Jupyter环境
    try:
        source = inspect.getsource(func)
    except TypeError:
        source = func.__source__
    
    # 解析为AST并标准化（去除位置信息）
    tree = ast.parse(source)
    ast.fix_missing_locations(tree)
    # 将AST转为无格式差异的字符串
    ast_str = ast.dump(tree, annotate_fields=False, include_attributes=False)
    
    # 计算SHA256哈希
    hash_obj = hashlib.sha256(ast_str.encode('utf-8'))
    return hash_obj.hexdigest()

def get_params_hash(*args, **kwargs):
    args_hashes = []
    # 处理位置参数
    for arg in args:
        if isinstance(arg, pd.Series):
            # Pandas Series用哈希值的总和作为标识
            args_hashes.append(str(pd.util.hash_pandas_object(arg).sum()))
        else:
            # 其他类型通过repr序列化后哈希
            args_hashes.append(hashlib.sha256(repr(arg).encode()).hexdigest())
    
    # 处理关键字参数（按键排序确保顺序不影响哈希）
    kwargs_hashes = []
    for k, v in sorted(kwargs.items()):
        if isinstance(v, pd.Series):
            kwargs_hashes.append(f"{k}:{str(pd.util.hash_pandas_object(v).sum())}")
        else:
            kwargs_hashes.append(f"{k}:{hashlib.sha256(repr(v).encode()).hexdigest()}")
    
    # 合并所有参数哈希并生成总哈希
    combined = "|".join(args_hashes + kwargs_hashes)
    return hashlib.sha256(combined.encode()).hexdigest()

def disk_cache(cache_dir="./cache"):
    cache_path = Path(cache_dir)
    cache_path.mkdir(exist_ok=True)
    
    def decorator(func):
        def wrapper(*args, **kwargs):
            # 生成函数代码哈希与参数哈希
            func_hash = get_function_ast_hash(func)
            params_hash = get_params_hash(*args, **kwargs)
            # 生成唯一缓存键
            cache_key = f"{func.__name__}_{func_hash}_{params_hash}"
            cache_file = cache_path / f"{cache_key}.pkl"
            
            # 检查缓存是否存在
            if cache_file.exists():
                return joblib.load(cache_file)
            
            # 计算结果并保存缓存
            result = func(*args, **kwargs)
            joblib.dump(result, cache_file)
            return result
        return wrapper
    return decorator

使用示例 #

# 定义两个逻辑不同的函数
@disk_cache()
def my_task(a: pd.Series, b: pd.Series):
    return a + b

@disk_cache()
def my_task_v2(a: pd.Series, b: pd.Series):
    return a + b - 2

# 测试缓存逻辑
s1 = pd.Series([1,2,3])
s2 = pd.Series([4,5,6])

# 第一次执行：计算并保存缓存
print(my_task(s1, s2))
# 第二次执行：直接加载缓存
print(my_task(s1, s2))

# 逻辑不同的函数会重新计算
print(my_task_v2(s1, s2))

注意事项 #

• Jupyter环境中，修改函数后需重新运行函数定义单元格，否则inspect.getsource会读取旧的源码。
• 若函数依赖全局变量或外部资源，需额外将这些依赖加入参数哈希，否则无法检测依赖变化。
• 大型Series的哈希计算可能耗时，可考虑抽样哈希（需权衡准确性与性能）。
• 定期清理缓存目录，避免磁盘空间占用过大。

内容的提问来源于stack exchange，提问作者Mikko Ohtamaa