eval_enterprise.py

#!/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")