SemHash Python 教程：重复检测、阈值校准与误删复核

python3.11 -m venv .venv
source .venv/bin/activate
python -m pip install 'semhash==0.5.0' 'model2vec==0.10.0' 'vicinity==0.4.6'

from huggingface_hub import snapshot_download

snapshot_download(
    'minishlab/potion-base-8M',
    revision='bf8b056651a2c21b8d2565580b8569da283cab23',
    local_dir='./potion-base-8m',
    allow_patterns=['*.json', '*.safetensors', 'README.md'],
)

python download_model.py

import argparse,json,hashlib
from dataclasses import asdict
from importlib.metadata import version
from pathlib import Path
from semhash import SemHash
from model2vec import StaticModel

RECORDS=[
 {'id':'refund-7','text':'Customers can request a refund within seven days.'},
 {'id':'refund-copy','text':'Customers can request a refund within seven days.'},
 {'id':'refund-30','text':'Customers can request a refund within thirty days.'},
 {'id':'no-refund','text':'Customers cannot request a refund.'},
 {'id':'password','text':'Reset your password from the account security page.'},
 {'id':'zh-7','text':'购买后七天内可以申请退款。'},
 {'id':'zh-30','text':'购买后三十天内可以申请退款。'},
]

def validate(records,threshold):
 if not records:raise ValueError('empty input')
 if not 0<=threshold<=1:raise ValueError('threshold outside [0,1]')
 if len({r['id'] for r in records})!=len(records):raise ValueError('duplicate source id')
 if any(not r['text'].strip() for r in records):raise ValueError('empty text')

def report(result):
 return {'selected':[dict(x) for x in result.selected],
  'filtered':[{'record':dict(x.record),'exact':x.exact,
    'duplicate_of':dict(x.duplicate_of),'score':float(x.score)} for x in result.filtered]}

def main():
 p=argparse.ArgumentParser();p.add_argument('model_directory',type=Path);args=p.parse_args()
 validate(RECORDS,0.9)
 model=StaticModel.from_pretrained(str(args.model_directory))
 engine=SemHash.from_records(RECORDS,columns=['text'],model=model,ann_backend='basic')
 results={str(t):report(engine.self_deduplicate(threshold=t)) for t in [0.8,0.9,0.98]}
 output={'versions':{n:version(n) for n in ['semhash','model2vec','vicinity']},
  'model_sha256':hashlib.sha256((args.model_directory/'model.safetensors').read_bytes()).hexdigest(),
  'input':RECORDS,'results':results,
  'note':'Review candidates only; no source files changed or deleted.'}
 print(json.dumps(output,ensure_ascii=False,indent=2))

if __name__=='__main__':main()

HF_HUB_OFFLINE=1 python deduplicate.py ./potion-base-8m > deduplication-output.json

import json
from pathlib import Path
from deduplicate import RECORDS, validate

out = json.loads(Path('deduplication-output.json').read_text())
expected = sorted(row['id'] for row in RECORDS)
for threshold, result in out['results'].items():
    actual = [r['id'] for r in result['selected']]
    actual += [r['record']['id'] for r in result['filtered']]
    assert sorted(actual) == expected, 'lost or duplicated source'
    for item in result['filtered']:
        assert item['duplicate_of']['id'] in expected
        assert item['exact'] or item['score'] >= float(threshold) - 1e-6

assert any(r['record']['id'] == 'refund-30'
           for r in out['results']['0.9']['filtered'])
assert 'refund-30' in {r['id'] for r in out['results']['0.98']['selected']}
for records, threshold in [([], 0.9), ([RECORDS[0], RECORDS[0]], 0.9),
                            ([{'id':'x', 'text':' '}], 0.9), (RECORDS, 1.1)]:
    try:
        validate(records, threshold)
    except ValueError:
        continue
    raise AssertionError('invalid input accepted')
print('Report consistency passed; factual merges still require review.')

python check_report.py