#!/usr/bin/env python3 """ 企業多任務 LLM 評估腳本 測試四大能力: 客服FAQ | 文件問答 | 工單分類 | 資訊抽取 """ import json import torch from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig from peft import PeftModel MODEL_ID = "Qwen/Qwen2.5-7B-Instruct" ADAPTER_ID = "Justin-lee/Qwen2.5-7B-Enterprise-ZH" def load_model(use_adapter=True): """Load model with optional LoRA adapter.""" print(f"📦 Loading {'fine-tuned' if use_adapter else 'base'} model...") bnb_config = BitsAndBytesConfig( load_in_4bit=True, bnb_4bit_quant_type="nf4", bnb_4bit_use_double_quant=True, bnb_4bit_compute_dtype=torch.bfloat16, ) tokenizer = AutoTokenizer.from_pretrained(MODEL_ID, trust_remote_code=True) model = AutoModelForCausalLM.from_pretrained( MODEL_ID, quantization_config=bnb_config, device_map="auto", trust_remote_code=True, torch_dtype=torch.bfloat16, ) if use_adapter: model = PeftModel.from_pretrained(model, ADAPTER_ID) print(f" ✅ LoRA adapter loaded from {ADAPTER_ID}") return model, tokenizer def generate(model, tokenizer, messages, max_new_tokens=512): """Generate response from messages.""" text = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True) inputs = tokenizer(text, return_tensors="pt").to(model.device) with torch.no_grad(): outputs = model.generate( **inputs, max_new_tokens=max_new_tokens, temperature=0.3, top_p=0.9, do_sample=True, pad_token_id=tokenizer.eos_token_id, ) response = tokenizer.decode(outputs[0][inputs.input_ids.shape[1]:], skip_special_tokens=True) return response.strip() # ── Test Cases ── EVAL_CASES = { "客服FAQ": [ { "system": "你是一個專業的企業客服助手。請根據用戶的問題，提供準確、簡潔、有禮貌的回答。", "user": "我昨天下的訂單想取消，還來得及嗎？訂單號 ORD-20240420。", "expected_keywords": ["取消", "訂單", "狀態", "發貨"], }, { "system": "你是一個專業的企業客服助手。請根據用戶的問題，提供準確、簡潔、有禮貌的回答。", "user": "你們的會員制度是怎樣的？有什麼優惠？", "expected_keywords": ["會員", "優惠", "等級"], }, { "system": "你是一個專業的企業客服助手。請根據用戶的問題，提供準確、簡潔、有禮貌的回答。", "user": "商品收到有破損，怎麼處理？", "expected_keywords": ["退", "換", "照片", "客服"], }, ], "文件問答": [ { "system": "你是一個文件分析助手。請仔細閱讀提供的文件內容，僅根據文件中的資訊回答問題。", "user": "請根據以下文件回答問題。\n\n【文件內容】\n公司年假制度：入職滿1年的員工享有5天年假，滿3年享有10天，滿5年享有15天。年假需提前3個工作日申請，由直屬主管審批。未使用的年假不可轉入下年度，但可折算為加班費。特殊情況（如家庭重大事件）可申請額外3天事假。\n\n【問題】\n工作3年的員工有幾天年假？年假可以保留到明年嗎？", "expected_keywords": ["10", "不可", "折算", "加班費"], }, { "system": "你是一個文件分析助手。請仔細閱讀提供的文件內容，僅根據文件中的資訊回答問題。", "user": "請根據以下文件回答問題。\n\n【文件內容】\n退款政策：1. 未發貨訂單：可立即取消並全額退款。2. 已發貨未簽收：需要拒收後由物流退回，退回運費由公司承擔。3. 已簽收：7天內可申請退貨退款，退回運費由買家承擔。4. 特殊商品（定製品、食品、內衣）不支持退貨。退款將原路返回至付款帳戶。\n\n【問題】\n已經簽收的訂單退貨，運費誰出？定製品可以退嗎？", "expected_keywords": ["買家", "不支持", "定製"], }, ], "工單分類": [ { "system": "你是一個工單分類與分流助手。請根據用戶描述的問題，將其分類到最合適的處理類別。", "user": "請將以下客戶訊息分類到合適的處理部門。\n可選部門：售後服務、物流配送、帳號問題、付款財務、產品諮詢、投訴建議、技術支援、合作洽談\n\n客戶訊息：我買的藍牙耳機左耳沒聲音了，買了才兩個禮拜。", "expected_keywords": ["售後", "保固", "故障"], }, { "system": "你是一個工單分類與分流助手。請根據用戶描述的問題，將其分類到最合適的處理類別。", "user": "請將以下客戶訊息分類到合適的處理部門。\n可選部門：售後服務、物流配送、帳號問題、付款財務、產品諮詢、投訴建議、技術支援、合作洽談\n\n客戶訊息：付款一直顯示失敗，我用了三張不同的信用卡都不行。", "expected_keywords": ["付款", "財務"], }, ], "資訊抽取": [ { "system": "你是一個資訊抽取助手。請從文本中準確抽取指定類型的實體資訊。", "user": "請從以下文本中抽取所有關鍵資訊（人名、日期、金額、地址、聯繫方式）：\n\n「客戶王大明先生於2024年4月18日來電，反映其於3月25日在台北市大安區忠孝東路四段200號門市購買的筆記型電腦（金額NT$35,800）出現螢幕閃爍問題。客戶要求維修或更換，聯繫電話：0912-345-678，Email: wang.dm@gmail.com。」", "expected_keywords": ["王大明", "2024年4月18日", "35,800", "忠孝東路", "0912"], }, { "system": "你是一個資訊抽取助手。請從文本中準確抽取指定類型的實體資訊。", "user": "請從以下合約條款中抽取關鍵條件：\n\n「甲方（承租人）李小華需於每月5日前支付租金新台幣28,000元至乙方指定帳戶。租賃期間自2024年5月1日起至2025年4月30日止，共12個月。押金為兩個月租金共56,000元，於退租時無息退還。如逾期付租超過15日，乙方有權終止合約。」", "expected_keywords": ["李小華", "28,000", "2024年5月1日", "2025年4月30日", "56,000", "15日"], }, ], } def evaluate(model, tokenizer): """Run evaluation on all task categories.""" print("\n" + "="*70) print("📊 Enterprise Multi-Task LLM Evaluation") print("="*70) results = {} for task_name, cases in EVAL_CASES.items(): print(f"\n{'─'*60}") print(f"📋 Task: {task_name} ({len(cases)} cases)") print(f"{'─'*60}") task_scores = [] for i, case in enumerate(cases): messages = [ {"role": "system", "content": case["system"]}, {"role": "user", "content": case["user"]}, ] response = generate(model, tokenizer, messages) # Check keyword coverage keywords = case["expected_keywords"] hits = sum(1 for kw in keywords if kw in response) score = hits / len(keywords) if keywords else 1.0 task_scores.append(score) print(f"\n Case {i+1}:") print(f" Q: {case['user'][:80]}...") print(f" A: {response[:200]}...") print(f" Keywords: {hits}/{len(keywords)} ({score:.0%})") avg_score = sum(task_scores) / len(task_scores) if task_scores else 0 results[task_name] = avg_score print(f"\n 📈 {task_name} Average: {avg_score:.1%}") # Summary print("\n" + "="*70) print("📊 Overall Results") print("="*70) overall = sum(results.values()) / len(results) for task, score in results.items(): bar = "█" * int(score * 20) + "░" * (20 - int(score * 20)) print(f" {task:10s} {bar} {score:.1%}") print(f"\n Overall Average: {overall:.1%}") print("="*70) return results if __name__ == "__main__": import sys use_adapter = "--base" not in sys.argv model, tokenizer = load_model(use_adapter=use_adapter) results = evaluate(model, tokenizer) # Save results with open("eval_results.json", "w", encoding="utf-8") as f: json.dump(results, f, ensure_ascii=False, indent=2) print(f"\n📝 Results saved to eval_results.json")