LLMLingua Python 教程：压缩预算、关键信息保留与效果校验

python3.11 -m venv .venv
source .venv/bin/activate
python -m pip install 'llmlingua==0.2.2' 'torch==2.2.2' 'numpy==1.26.4' 'transformers==4.44.2' 'accelerate==0.34.2' 'tiktoken==0.14.0'
python -m pip check

from huggingface_hub import snapshot_download
import tiktoken

snapshot_download(
    'microsoft/llmlingua-2-bert-base-multilingual-cased-meetingbank',
    revision='5f0c82792b7ea14c6484e015b6a072009496b7f2',
    local_dir='./llmlingua-2-bert-base-multilingual-cased-meetingbank',
    allow_patterns=['*.json', '*.txt', '*.safetensors'],
)
tiktoken.get_encoding('cl100k_base')

export HF_HOME="$PWD/.hf-cache"
export TIKTOKEN_CACHE_DIR="$PWD/.tiktoken-cache"
python prepare_model.py

import argparse,hashlib,json,time
from pathlib import Path
from importlib.metadata import version
import torch,tiktoken
from llmlingua import PromptCompressor

CASES=[
 {'id':'en','text':'Policy REFUND-731. Customers may request a refund within 7 days. The fee is USD 19. Do not share passwords. We discussed the policy during a long friendly meeting and repeated the same background several times. The seven-day limit is mandatory, not thirty days.','required':['REFUND-731','within 7 days','USD 19','Do not share passwords']},
 {'id':'zh','text':'政策编号 REFUND-731。购买后七天内可以申请退款，手续费为人民币19元。不得分享密码。会议中大家反复介绍背景和问候语，但这些寒暄不是规则。退款期限是七天，不是三十天。','required':['REFUND-731','七天','人民币19元','不得分享密码']},
]
QUESTION='What are the refund deadline, fee, and password restriction?'

def validate_input(text,rate):
 if not text.strip():raise ValueError('empty text')
 if not 0<rate<=1:raise ValueError('rate must be within (0,1]')

def assess(text,required,budget,encoding):
 missing=[x for x in required if x not in text]
 count=len(encoding.encode(text))
 return dict(payload_tokens=count,budget=budget,missing=missing,
   status='ready_for_review' if not missing and count<=budget else 'blocked')

def main():
 p=argparse.ArgumentParser();p.add_argument('model_directory',type=Path);args=p.parse_args()
 torch.set_num_threads(2)
 enc=tiktoken.get_encoding('cl100k_base')
 compressor=PromptCompressor(model_name=str(args.model_directory),device_map='cpu',
    model_config={'torch_dtype':torch.float32},use_llmlingua2=True)
 rows=[]
 for case in CASES:
  for rate in [0.5,0.8]:
   validate_input(case['text'],rate);started=time.perf_counter()
   result=compressor.compress_prompt_llmlingua2([case['text']],rate=rate)
   compressed=result['compressed_prompt']
   # Count the assembled downstream plain-text payload, not just compressed context.
   payload='Answer only from evidence.\nEvidence:\n'+compressed+'\nQuestion:\n'+QUESTION
   rows.append(dict(id=case['id'],rate=rate,compressed=compressed,
     reported_origin_tokens=result['origin_tokens'],reported_compressed_tokens=result['compressed_tokens'],
     elapsed_ms=round((time.perf_counter()-started)*1000,2),
     assessment=assess(payload,case['required'],140,enc)))
 output=dict(versions={n:version(n) for n in ['llmlingua','torch','transformers','tiktoken']},
   model_sha256=hashlib.sha256((args.model_directory/'model.safetensors').read_bytes()).hexdigest(),
   cases=CASES,rows=rows,note='No downstream model invoked. Substring presence is only a necessary check, not semantic faithfulness.')
 Path('compression-output.json').write_text(json.dumps(output,ensure_ascii=False,indent=2));print('Saved compression-output.json')

if __name__=='__main__':main()

HF_HUB_OFFLINE=1 TRANSFORMERS_OFFLINE=1 python compress_validate.py ./llmlingua-2-bert-base-multilingual-cased-meetingbank

import json
from pathlib import Path
import tiktoken
from compress_validate import CASES, QUESTION, assess, validate_input

out = json.loads(Path('compression-output.json').read_text())
encoding = tiktoken.get_encoding('cl100k_base')
assert len(out['rows']) == 4 and out['model_sha256']
for row in out['rows']:
    case = next(c for c in CASES if c['id'] == row['id'])
    payload = 'Answer only from evidence.\nEvidence:\n' + row['compressed'] + '\nQuestion:\n' + QUESTION
    assert row['assessment'] == assess(payload, case['required'], 140, encoding)
    assert row['assessment']['status'] == 'blocked'

for text, rate in [('', 0.5), ('x', 0), ('x', 1.1)]:
    try:
        validate_input(text, rate)
    except ValueError:
        continue
    raise AssertionError('invalid input accepted')
assert assess('keep KEY', ['KEY'], 0, encoding)['status'] == 'blocked'
assert assess('text', ['MISSING'], 140, encoding)['status'] == 'blocked'
assert assess('keep KEY', ['KEY'], 140, encoding)['status'] == 'ready_for_review'
print('Report and refusal controls passed; semantic review is still required.')

python check_report.py