1013 lines
36 KiB
Python
1013 lines
36 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
eval.py - 评估模型在给定文本上的表现
|
||
|
||
使用方法:
|
||
python eval.py --text-file path/to/text.txt [--checkpoint path/to/model.pt] [--num-samples N]
|
||
|
||
功能:
|
||
1. 读取文本文件，限制长度300，随机抽取
|
||
2. 参考dataset.py将文本内容转化为模型可接受数据，包含part1/part2/part3/part4/槽位历史等信息
|
||
3. 使用模型推理，参考test.py
|
||
4. 错误的打印，对于方差极小，概率和猜没什么差别的，重点标记
|
||
"""
|
||
|
||
import argparse
|
||
import random
|
||
import re
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Dict, List, Tuple, Optional
|
||
|
||
import numpy as np
|
||
import torch
|
||
import torch.nn.functional as F
|
||
from modelscope import AutoTokenizer
|
||
from pypinyin import lazy_pinyin
|
||
from pypinyin.contrib.tone_convert import to_initials
|
||
|
||
# 添加src目录到路径
|
||
sys.path.append("src")
|
||
|
||
from src.model.model import InputMethodEngine
|
||
from src.model.query import QueryEngine
|
||
from src.model.dataset import text_to_pinyin_ids
|
||
|
||
_HANZI_RE = re.compile(r"[\u4e00-\u9fff]+")
|
||
|
||
|
||
class TextEvaluator:
|
||
def __init__(
|
||
self,
|
||
checkpoint_path: str,
|
||
device: str = "cuda" if torch.cuda.is_available() else "cpu",
|
||
):
|
||
self.device = torch.device(device)
|
||
self.checkpoint_path = checkpoint_path
|
||
|
||
# 加载组件
|
||
self.load_model()
|
||
self.load_tokenizer()
|
||
self.load_query_engine()
|
||
|
||
# 拼音风格权重（与dataset.py一致）
|
||
self.py_style_weight = np.array([9, 2, 1]) / sum([9, 2, 1])
|
||
|
||
print(f"✅ 评估器初始化完成 (设备: {self.device})")
|
||
|
||
# 调试：运行一个固定样本测试
|
||
import os
|
||
|
||
if os.environ.get("EVAL_DEBUG"):
|
||
self._debug_sample()
|
||
|
||
def load_model(self):
|
||
"""加载训练好的模型"""
|
||
self.model = InputMethodEngine(pinyin_vocab_size=30, compile=False)
|
||
checkpoint = torch.load(self.checkpoint_path, map_location="cpu")
|
||
if "model_state_dict" in checkpoint:
|
||
self.model.load_state_dict(checkpoint["model_state_dict"])
|
||
else:
|
||
self.model.load_state_dict(checkpoint)
|
||
self.model.eval()
|
||
self.model.to(self.device)
|
||
print(
|
||
f"✅ 模型加载完成，参数量: {sum(p.numel() for p in self.model.parameters()):,}"
|
||
)
|
||
print(f"✅ 模型词汇表大小: {self.model.vocab_size}")
|
||
|
||
def load_tokenizer(self):
|
||
"""加载tokenizer"""
|
||
tokenizer_path = (
|
||
Path(__file__).parent / "src" / "model" / "assets" / "tokenizer"
|
||
)
|
||
self.tokenizer = AutoTokenizer.from_pretrained(str(tokenizer_path))
|
||
print(f"✅ Tokenizer加载完成，词汇表大小: {self.tokenizer.vocab_size}")
|
||
|
||
def load_query_engine(self):
|
||
"""加载查询引擎用于字符-ID转换"""
|
||
try:
|
||
self.query_engine = QueryEngine()
|
||
stats_path = (
|
||
Path(__file__).parent
|
||
/ "src"
|
||
/ "model"
|
||
/ "assets"
|
||
/ "pinyin_char_statistics.json"
|
||
)
|
||
if stats_path.exists():
|
||
self.query_engine.load(stats_path)
|
||
print(
|
||
f"✅ 查询引擎加载完成，字符对数量: {len(self.query_engine._id_to_info)}"
|
||
)
|
||
else:
|
||
print(f"⚠️ 统计文件不存在: {stats_path}")
|
||
self.query_engine = None
|
||
except Exception as e:
|
||
print(f"⚠️ 无法加载查询引擎: {e}")
|
||
self.query_engine = None
|
||
|
||
def _debug_sample(self):
|
||
"""调试：运行与test.py相同的固定样本"""
|
||
print("\n🧪 调试：运行固定样本（与test.py相同）")
|
||
|
||
# 复制test.py中的样本
|
||
part1 = "他是一名大学生，在上海读"
|
||
part2 = "dayi"
|
||
pinyin_ids = text_to_pinyin_ids(part2)
|
||
len_py = len(pinyin_ids)
|
||
if len_py < 24:
|
||
pinyin_ids.extend([0] * (24 - len_py))
|
||
else:
|
||
pinyin_ids = pinyin_ids[:24]
|
||
pinyin_ids = torch.tensor(pinyin_ids, dtype=torch.long).unsqueeze(0)
|
||
masked_labels = [15, 4, 0, 0, 0, 0, 0, 0]
|
||
part3 = "。"
|
||
part4 = "可行|特别|伤害"
|
||
|
||
encoded = self.tokenizer(
|
||
f"{part4}|{part1}",
|
||
part3,
|
||
max_length=128,
|
||
padding="max_length",
|
||
truncation=True,
|
||
return_tensors="pt",
|
||
return_token_type_ids=True,
|
||
)
|
||
|
||
sample = {
|
||
"input_ids": torch.stack([encoded["input_ids"].squeeze(0)]),
|
||
"token_type_ids": torch.stack([encoded["token_type_ids"].squeeze(0)]),
|
||
"attention_mask": torch.stack([encoded["attention_mask"].squeeze(0)]),
|
||
"history_slot_ids": torch.tensor(masked_labels, dtype=torch.long).unsqueeze(
|
||
0
|
||
),
|
||
"prefix": f"{part4}^{part1}",
|
||
"suffix": part3,
|
||
"pinyin": part2,
|
||
"pinyin_ids": pinyin_ids,
|
||
}
|
||
|
||
# 推理
|
||
logits, probs = self.inference(sample)
|
||
analysis = self.analyze_probability_distribution(probs)
|
||
|
||
print(f"📊 调试样本结果:")
|
||
print(f" 最高概率: {analysis['max_prob']:.6f}")
|
||
print(f" 平均概率: {analysis['mean_prob']:.6f}")
|
||
print(f" 标准差: {analysis['std_prob']:.6f}")
|
||
print(f" 低方差: {analysis['low_variance']}")
|
||
|
||
# 打印top-5
|
||
print(f" Top-5预测:")
|
||
for j in range(5):
|
||
idx = analysis["top_indices"][j]
|
||
prob = analysis["top_probs"][j]
|
||
char_info = (
|
||
self.query_engine.query_by_id(idx) if self.query_engine else None
|
||
)
|
||
char = char_info.char if char_info else f"[ID:{idx}]"
|
||
print(f" {j + 1}. {char} (ID: {idx}) - 概率: {prob:.6f}")
|
||
|
||
def generate_pinyin(self, text: str) -> List[str]:
|
||
"""
|
||
流式处理单条文本，转换为拼音列表。
|
||
参考dataset.py中的generate_pinyin方法。
|
||
"""
|
||
if not text:
|
||
return []
|
||
|
||
text_len = len(text)
|
||
result: List[str] = [""] * text_len
|
||
|
||
# 遍历所有连续汉字片段
|
||
for match in _HANZI_RE.finditer(text):
|
||
start_idx = match.start()
|
||
hanzi_segment = match.group()
|
||
|
||
pinyin_list = lazy_pinyin(hanzi_segment)
|
||
|
||
if len(pinyin_list) != len(hanzi_segment):
|
||
pinyin_list = [lazy_pinyin(c)[0] for c in hanzi_segment]
|
||
|
||
for i, py in enumerate(pinyin_list):
|
||
result[start_idx + i] = py
|
||
|
||
# 填充非汉字字符
|
||
for i, char in enumerate(text):
|
||
if not result[i]:
|
||
result[i] = char
|
||
|
||
return result
|
||
|
||
def get_mask_pinyin(
|
||
self, text: str, pinyin_list: List[str]
|
||
) -> Tuple[int, List[str]]:
|
||
"""
|
||
生成需要预测汉字对应的拼音，并进行加强。
|
||
参考dataset.py中的get_mask_pinyin方法。
|
||
"""
|
||
mask_pinyin = []
|
||
for i in range(len(text)):
|
||
if not self.query_engine or not self.query_engine.is_chinese_char(text[i]):
|
||
break
|
||
else:
|
||
py = np.random.choice(
|
||
(pinyin_list[i], to_initials(pinyin_list[i]), pinyin_list[i][0]),
|
||
p=self.py_style_weight,
|
||
)
|
||
if py == "":
|
||
py = pinyin_list[i][0]
|
||
mask_pinyin.append(py)
|
||
return len(mask_pinyin), mask_pinyin
|
||
|
||
def create_sample_from_text(
|
||
self,
|
||
text: str,
|
||
position: Optional[int] = None,
|
||
force_slot_count: Optional[int] = None,
|
||
) -> Dict:
|
||
"""
|
||
从文本创建单个样本，参考dataset.py的样本生成逻辑。
|
||
|
||
Args:
|
||
text: 输入文本
|
||
position: 预测位置（汉字位置），如果为None则随机选择
|
||
force_slot_count: 强制设置有效槽位数量(0-8)，如果为None则自动计算
|
||
|
||
Returns:
|
||
样本字典，包含模型所需的所有字段
|
||
"""
|
||
# 确保文本长度至少为1
|
||
if not text:
|
||
raise ValueError("文本不能为空")
|
||
|
||
# 生成拼音列表
|
||
pinyin_list = self.generate_pinyin(text)
|
||
|
||
# 找到所有汉字位置
|
||
chinese_positions = [
|
||
i
|
||
for i, char in enumerate(text)
|
||
if self.query_engine and self.query_engine.is_chinese_char(char)
|
||
]
|
||
|
||
if not chinese_positions:
|
||
raise ValueError("文本中没有汉字")
|
||
|
||
# 随机选择预测位置
|
||
if position is None:
|
||
position = random.choice(chinese_positions)
|
||
elif position >= len(text) or position < 0:
|
||
raise ValueError(f"位置 {position} 超出文本范围")
|
||
elif not self.query_engine or not self.query_engine.is_chinese_char(
|
||
text[position]
|
||
):
|
||
raise ValueError(f"位置 {position} 不是汉字")
|
||
|
||
i = position
|
||
|
||
# part1: 光标前文本（最多48个字符）
|
||
if i < 48:
|
||
part1 = text[0:i]
|
||
else:
|
||
part1 = text[i - 48 : i]
|
||
|
||
# part2: 拼音输入（随机长度1-8，高斯分布）
|
||
# 当强制指定历史槽位时，需要确保拼音长度足够
|
||
pinyin_len_probs = [0.05, 0.16, 0.30, 0.20, 0.12, 0.08, 0.05, 0.04]
|
||
min_pinyin_len = (force_slot_count + 1) if force_slot_count is not None else 1
|
||
min_pinyin_len = max(1, min(8, min_pinyin_len))
|
||
|
||
# 检查剩余可用字符数量
|
||
remaining_chars = len(text) - i
|
||
max_valid_len = 0
|
||
for j in range(i, min(i + 8, len(text))):
|
||
if self.query_engine and self.query_engine.is_chinese_char(text[j]):
|
||
max_valid_len += 1
|
||
else:
|
||
break
|
||
|
||
# 调整拼音长度：必须至少有 min_pinyin_len 个有效字符
|
||
if max_valid_len < min_pinyin_len:
|
||
# 重新选择位置，确保有足够字符
|
||
raise ValueError(
|
||
f"位置 {i} 后只有 {max_valid_len} 个有效汉字，不足以测试历史槽位 {force_slot_count}"
|
||
)
|
||
|
||
# 随机选择拼音长度（至少 min_pinyin_len）
|
||
if force_slot_count is not None:
|
||
# 强制模式：随机选择比 min_pinyin_len 大的长度
|
||
valid_lengths = [
|
||
l for l in range(min_pinyin_len, min(max_valid_len + 1, 9))
|
||
]
|
||
if not valid_lengths:
|
||
valid_lengths = [min_pinyin_len]
|
||
# 使用原来的概率分布，但只选择有效长度
|
||
adjusted_probs = [pinyin_len_probs[l - 1] for l in valid_lengths]
|
||
adjusted_probs = np.array(adjusted_probs) / sum(adjusted_probs)
|
||
pinyin_len = np.random.choice(valid_lengths, p=adjusted_probs)
|
||
else:
|
||
pinyin_len = np.random.choice(range(1, 9), p=pinyin_len_probs)
|
||
pinyin_len = min(pinyin_len, max_valid_len)
|
||
|
||
py_end = min(i + pinyin_len, len(text))
|
||
|
||
# 获取拼音
|
||
pinyin_len_actual, part2_pinyin = self.get_mask_pinyin(
|
||
text[i:py_end], pinyin_list[i:py_end]
|
||
)
|
||
|
||
# 添加分隔符
|
||
split_char = np.random.choice(["", "`", "'", "-"], p=[0.9, 0.04, 0.04, 0.02])
|
||
part2 = split_char.join(part2_pinyin)
|
||
|
||
# 转换为拼音ID
|
||
pinyin_ids = text_to_pinyin_ids(part2)
|
||
len_py = len(pinyin_ids)
|
||
if len_py < 24:
|
||
pinyin_ids.extend([0] * (24 - len_py))
|
||
else:
|
||
pinyin_ids = pinyin_ids[:24]
|
||
pinyin_ids_tensor = torch.tensor(pinyin_ids, dtype=torch.long)
|
||
|
||
# part3: 光标后文本（70%概率为空）
|
||
part3 = ""
|
||
if random.random() > 0.7:
|
||
part3 = text[
|
||
i + pinyin_len_actual : i
|
||
+ pinyin_len_actual
|
||
+ np.random.choice(range(1, 17))
|
||
]
|
||
|
||
# part4: 上下文提示（50%概率为空）
|
||
part4 = ""
|
||
if random.random() > 0.5:
|
||
num_strings = random.randint(1, 5)
|
||
string_list = []
|
||
for _ in range(num_strings):
|
||
start_pos = random.randint(0, len(text) - 1)
|
||
x = random.randint(2, 6)
|
||
end_pos = min(start_pos + x + 1, len(text))
|
||
string_list.append(text[start_pos:end_pos])
|
||
part4 = "|".join(string_list)
|
||
|
||
# 标签：预测字符的ID（整个拼音序列的所有字符）
|
||
labels = []
|
||
try:
|
||
labels = [
|
||
self.query_engine.get_char_info_by_char_pinyin(c, p).id
|
||
for c, p in zip(
|
||
text[i : i + pinyin_len_actual],
|
||
pinyin_list[i : i + pinyin_len_actual],
|
||
)
|
||
]
|
||
except (AttributeError, TypeError) as e:
|
||
print(f"⚠️ 获取标签失败: {e}")
|
||
labels = [0] * pinyin_len_actual
|
||
|
||
# 历史槽位：与训练数据生成逻辑一致
|
||
# 当 force_slot_count=k 时，历史槽位为 labels[:k]，预测目标为 labels[k]
|
||
history_slot_ids = []
|
||
predict_idx = 0 # 当前预测的标签索引
|
||
|
||
if force_slot_count is not None:
|
||
force_slot_count = max(0, min(8, force_slot_count))
|
||
# 检查 labels 长度是否足够
|
||
if len(labels) <= force_slot_count:
|
||
raise ValueError(
|
||
f"标签数量 {len(labels)} 不足以支持历史槽位 {force_slot_count}"
|
||
)
|
||
predict_idx = force_slot_count
|
||
history_slot_ids = labels[:force_slot_count]
|
||
else:
|
||
# 正常评估：随机选择历史长度，模拟训练数据分布
|
||
history_weights = [0.2, 0.2, 0.2, 0.9, 1.2, 1.8, 2.5, 3.5, 4.0]
|
||
max_history = min(len(labels) - 1, 8) # 至少留一个字符预测
|
||
if max_history < 0:
|
||
max_history = 0
|
||
valid_history_lens = list(range(0, max_history + 1))
|
||
if valid_history_lens:
|
||
adjusted_probs = [history_weights[h] for h in valid_history_lens]
|
||
adjusted_probs = np.array(adjusted_probs) / sum(adjusted_probs)
|
||
history_len = np.random.choice(valid_history_lens, p=adjusted_probs)
|
||
else:
|
||
history_len = 0
|
||
predict_idx = history_len
|
||
history_slot_ids = labels[:history_len]
|
||
|
||
# 填充到8个槽位
|
||
if len(history_slot_ids) < 8:
|
||
history_slot_ids.extend([0] * (8 - len(history_slot_ids)))
|
||
else:
|
||
history_slot_ids = history_slot_ids[:8]
|
||
|
||
# Tokenize输入
|
||
encoded = self.tokenizer(
|
||
f"{part4}|{part1}",
|
||
part3,
|
||
max_length=128,
|
||
padding="max_length",
|
||
truncation=True,
|
||
return_tensors="pt",
|
||
return_token_type_ids=True,
|
||
)
|
||
|
||
# 计算有效槽位数量（ID > 0 的槽位，与训练数据一致）
|
||
valid_slot_count = sum(1 for slot_id in history_slot_ids if slot_id > 0)
|
||
|
||
# 拼音输入长度（字符数）
|
||
pinyin_input_length = len(part2)
|
||
|
||
# 当前预测的字符和标签
|
||
current_target_char = (
|
||
text[i + predict_idx] if (i + predict_idx) < len(text) else ""
|
||
)
|
||
current_target_label = labels[predict_idx] if predict_idx < len(labels) else 0
|
||
|
||
# 构建样本
|
||
sample = {
|
||
"input_ids": encoded["input_ids"],
|
||
"token_type_ids": encoded["token_type_ids"],
|
||
"attention_mask": encoded["attention_mask"],
|
||
"history_slot_ids": torch.tensor(
|
||
history_slot_ids, dtype=torch.long
|
||
).unsqueeze(0),
|
||
"prefix": f"{part4}^{part1}",
|
||
"suffix": part3,
|
||
"pinyin": part2,
|
||
"pinyin_ids": pinyin_ids_tensor.unsqueeze(0),
|
||
"true_labels": labels, # 完整标签列表（所有字符）
|
||
"predict_idx": predict_idx, # 当前预测的标签索引
|
||
"current_target_label": current_target_label, # 当前预测的标签
|
||
"position": i,
|
||
"target_char": current_target_char, # 当前预测的字符
|
||
"valid_slot_count": valid_slot_count,
|
||
"pinyin_input_length": pinyin_input_length,
|
||
}
|
||
|
||
return sample
|
||
|
||
def inference(self, sample: Dict) -> Tuple[torch.Tensor, torch.Tensor]:
|
||
"""
|
||
执行模型推理。
|
||
|
||
Returns:
|
||
(logits, probs)
|
||
"""
|
||
# 准备输入张量并移动到设备
|
||
input_ids = sample["input_ids"].to(self.device)
|
||
token_type_ids = sample["token_type_ids"].to(self.device)
|
||
attention_mask = sample["attention_mask"].to(self.device)
|
||
pinyin_ids = sample["pinyin_ids"].to(self.device)
|
||
history_slot_ids = sample["history_slot_ids"].to(self.device)
|
||
|
||
with torch.no_grad():
|
||
if self.device.type == "cuda":
|
||
with torch.autocast(device_type="cuda"):
|
||
logits = self.model(
|
||
input_ids,
|
||
token_type_ids,
|
||
attention_mask,
|
||
pinyin_ids,
|
||
history_slot_ids,
|
||
)
|
||
else:
|
||
logits = self.model(
|
||
input_ids,
|
||
token_type_ids,
|
||
attention_mask,
|
||
pinyin_ids,
|
||
history_slot_ids,
|
||
)
|
||
|
||
probs = F.softmax(logits, dim=-1)
|
||
return logits, probs
|
||
|
||
def analyze_probability_distribution(
|
||
self, probs: torch.Tensor, threshold_low_var: float = 0.01
|
||
) -> Dict:
|
||
"""
|
||
分析概率分布，检测低方差情况。
|
||
|
||
Args:
|
||
probs: 概率张量，形状为 (1, vocab_size)
|
||
threshold_low_var: 低方差阈值，最高概率低于此值则标记为低方差
|
||
|
||
Returns:
|
||
分析结果字典
|
||
"""
|
||
probs_np = probs.cpu().numpy().flatten()
|
||
|
||
# 基本统计
|
||
max_prob = np.max(probs_np)
|
||
min_prob = np.min(probs_np)
|
||
mean_prob = np.mean(probs_np)
|
||
std_prob = np.std(probs_np)
|
||
|
||
# 熵
|
||
entropy = -np.sum(probs_np * np.log(probs_np + 1e-12))
|
||
|
||
# 低方差检测
|
||
low_variance = max_prob < threshold_low_var
|
||
|
||
# Top-k 分析
|
||
top_k = 20
|
||
top_indices = np.argsort(probs_np)[-top_k:][::-1]
|
||
top_probs = probs_np[top_indices]
|
||
|
||
# 检查前5个概率是否接近均匀
|
||
top5_probs = top_probs[:5]
|
||
top5_ratio = (
|
||
top5_probs[0] / (top5_probs[1] + 1e-12) if len(top5_probs) > 1 else 1.0
|
||
)
|
||
|
||
analysis = {
|
||
"max_prob": max_prob,
|
||
"min_prob": min_prob,
|
||
"mean_prob": mean_prob,
|
||
"std_prob": std_prob,
|
||
"entropy": entropy,
|
||
"low_variance": low_variance,
|
||
"top_indices": top_indices,
|
||
"top_probs": top_probs,
|
||
"top5_ratio": top5_ratio,
|
||
"vocab_size": len(probs_np),
|
||
}
|
||
|
||
return analysis
|
||
|
||
def evaluate_sample_single_char(
|
||
self, sample: Dict, print_details: bool = True
|
||
) -> Dict:
|
||
"""
|
||
单字符评估（与验证集 trainer.py 评估逻辑一致）。
|
||
|
||
使用 argmax 选择预测字符，与训练时验证逻辑完全一致。
|
||
根据 predict_idx 选择正确的预测目标。
|
||
|
||
Returns:
|
||
评估结果字典
|
||
"""
|
||
logits, probs = self.inference(sample)
|
||
analysis = self.analyze_probability_distribution(probs)
|
||
|
||
true_labels = sample.get("true_labels", [])
|
||
predict_idx = sample.get("predict_idx", 0)
|
||
current_target_label = sample.get(
|
||
"current_target_label", true_labels[predict_idx] if true_labels else 0
|
||
)
|
||
target_char = sample.get("target_char", "")
|
||
|
||
pred_idx = analysis["top_indices"][0]
|
||
correct = pred_idx == current_target_label
|
||
|
||
if print_details:
|
||
valid_slot_count = sample.get("valid_slot_count", 0)
|
||
pinyin_input_length = sample.get("pinyin_input_length", 0)
|
||
|
||
print(f" {target_char}", end="")
|
||
print(f"(ID:{current_target_label})", end="")
|
||
print(
|
||
f" | 拼音:{sample.get('pinyin', '')[:10]}{'...' if len(sample.get('pinyin', '')) > 10 else ''}",
|
||
end="",
|
||
)
|
||
print(f" | 槽位:{valid_slot_count}/8 拼音长:{pinyin_input_length}", end="")
|
||
print(f" | 最高概率:{analysis['max_prob']:.3f}", end="")
|
||
if analysis["low_variance"]:
|
||
print(f" ⚠️", end="")
|
||
|
||
top_strs = []
|
||
for j in range(min(2, len(analysis["top_indices"]))):
|
||
idx = analysis["top_indices"][j]
|
||
prob = analysis["top_probs"][j]
|
||
if idx == 0:
|
||
char = "[空]"
|
||
else:
|
||
char_info = (
|
||
self.query_engine.query_by_id(idx)
|
||
if self.query_engine
|
||
else None
|
||
)
|
||
char = char_info.char if char_info else f"[{idx}]"
|
||
is_correct = idx == current_target_label
|
||
correct_mark = "✓" if is_correct else ""
|
||
top_strs.append(f"{char}{correct_mark}({prob:.2f})")
|
||
print(f" | Top-2: {' '.join(top_strs)}", end="")
|
||
print(f" | 结果:{'✓' if correct else '✗'}")
|
||
|
||
return {
|
||
"sample": sample,
|
||
"analysis": analysis,
|
||
"correct": correct,
|
||
"target_char": target_char,
|
||
"true_label": current_target_label,
|
||
"predicted_label": pred_idx,
|
||
"predict_idx": predict_idx,
|
||
}
|
||
|
||
def evaluate_sample(self, sample: Dict, print_details: bool = True) -> Dict:
|
||
"""
|
||
评估单个样本（兼容旧接口，调用单字符评估）。
|
||
|
||
Returns:
|
||
评估结果字典
|
||
"""
|
||
return self.evaluate_sample_single_char(sample, print_details)
|
||
|
||
def evaluate_sample_with_sequential_confirmation(
|
||
self, sample: Dict, print_details: bool = True, use_oracle_history: bool = True
|
||
) -> Dict:
|
||
"""
|
||
评估样本并模拟用户逐步确认过程（用于多字符预测）。
|
||
|
||
Args:
|
||
sample: 样本字典
|
||
print_details: 是否打印详细信息
|
||
use_oracle_history: True=使用真实标签作为历史（与训练/验证一致），False=使用预测结果
|
||
|
||
对于拼音长度>1的情况，模拟用户逐步选择过程。
|
||
|
||
Returns:
|
||
评估结果字典
|
||
"""
|
||
# 获取真实标签和拼音长度
|
||
true_labels = sample.get("true_labels", [])
|
||
pinyin_len = len(true_labels) if true_labels else 1
|
||
|
||
if pinyin_len <= 1:
|
||
# 单字符预测，直接使用普通评估
|
||
return self.evaluate_sample(sample, print_details)
|
||
|
||
# 多字符预测：模拟逐步确认
|
||
confirmed_history = [] # 已确认的字符ID
|
||
all_predictions = [] # 每一步的预测结果
|
||
all_correct = [] # 每一步是否正确
|
||
|
||
# 复制样本用于逐步推理
|
||
sample_copy = sample.copy()
|
||
|
||
for step in range(pinyin_len):
|
||
# 更新历史槽位：
|
||
# - oracle模式：使用真实标签作为历史（与训练一致）
|
||
# - user模式：使用模型预测作为历史（模拟实际使用）
|
||
if use_oracle_history:
|
||
# 使用真实标签作为历史（与训练数据生成逻辑一致）
|
||
history_slot_ids = true_labels[:step]
|
||
else:
|
||
# 使用预测结果作为历史
|
||
history_slot_ids = confirmed_history[:]
|
||
|
||
if len(history_slot_ids) < 8:
|
||
history_slot_ids.extend([0] * (8 - len(history_slot_ids)))
|
||
else:
|
||
history_slot_ids = history_slot_ids[:8]
|
||
|
||
# 更新样本的历史槽位
|
||
sample_copy["history_slot_ids"] = torch.tensor(
|
||
history_slot_ids, dtype=torch.long
|
||
).unsqueeze(0)
|
||
|
||
# 执行推理
|
||
logits, probs = self.inference(sample_copy)
|
||
analysis = self.analyze_probability_distribution(probs)
|
||
|
||
# 获取Top-1预测
|
||
pred_top_idx = analysis["top_indices"][0]
|
||
pred_top_prob = analysis["top_probs"][0]
|
||
|
||
# 检查是否正确
|
||
correct = False
|
||
if step < len(true_labels):
|
||
correct = pred_top_idx == true_labels[step]
|
||
|
||
# 记录结果
|
||
all_predictions.append(
|
||
{
|
||
"predicted_id": pred_top_idx,
|
||
"probability": pred_top_prob,
|
||
"correct": correct,
|
||
}
|
||
)
|
||
all_correct.append(correct)
|
||
|
||
# 更新已确认历史（用于下一轮）
|
||
if pred_top_idx > 0:
|
||
confirmed_history.append(pred_top_idx)
|
||
|
||
# 计算整体准确率：所有步骤都正确才算正确
|
||
overall_correct = all(all_correct)
|
||
|
||
# 打印结果
|
||
if print_details:
|
||
target_char = sample.get("target_char", "")
|
||
pinyin = sample.get("pinyin", "")
|
||
valid_slot_count = sample.get("valid_slot_count", 0)
|
||
|
||
print(f" {target_char}", end="")
|
||
print(f" | 拼音:{pinyin[:10]}{'...' if len(pinyin) > 10 else ''}", end="")
|
||
print(f" | 槽位:{valid_slot_count}/8 拼音长:{pinyin_len}", end="")
|
||
mode_str = "oracle" if use_oracle_history else "user"
|
||
print(f" | [{mode_str}]", end="")
|
||
|
||
# 显示逐步预测结果
|
||
step_results = []
|
||
for i, pred in enumerate(all_predictions):
|
||
char_info = (
|
||
self.query_engine.query_by_id(pred["predicted_id"])
|
||
if self.query_engine
|
||
else None
|
||
)
|
||
char = char_info.char if char_info else f"[{pred['predicted_id']}]"
|
||
correct_mark = "✓" if pred["correct"] else "✗"
|
||
step_results.append(f"{char}{correct_mark}")
|
||
|
||
print(f" | 逐步:{'→'.join(step_results)}", end="")
|
||
print(f" | 结果:{'✓' if overall_correct else '✗'}")
|
||
|
||
# 返回评估结果
|
||
result = {
|
||
"sample": sample,
|
||
"sequential_predictions": all_predictions,
|
||
"correct": overall_correct,
|
||
"step_correct": all_correct,
|
||
"target_char": sample.get("target_char", ""),
|
||
"true_labels": true_labels,
|
||
"pinyin_len": pinyin_len,
|
||
}
|
||
|
||
return result
|
||
|
||
def evaluate_text(
|
||
self, text: str, num_samples: int = 10, use_sequential: bool = False
|
||
) -> List[Dict]:
|
||
"""
|
||
评估给定文本，生成多个样本进行评估。
|
||
|
||
Args:
|
||
text: 输入文本
|
||
num_samples: 生成的样本数量
|
||
use_sequential: True=使用逐步确认评估，False=使用单字符评估（与验证集一致）
|
||
|
||
Returns:
|
||
评估结果列表
|
||
"""
|
||
# 限制文本长度
|
||
if len(text) > 300:
|
||
start = random.randint(0, len(text) - 300)
|
||
text = text[start : start + 300]
|
||
print(f"📝 文本长度超过300，随机截取300字符: {text[:50]}...")
|
||
else:
|
||
print(f"📝 文本长度: {len(text)} 字符")
|
||
|
||
results = []
|
||
low_variance_samples = []
|
||
|
||
for sample_idx in range(num_samples):
|
||
print(f"\n[样本 {sample_idx + 1}/{num_samples}]", end=" ")
|
||
|
||
try:
|
||
sample = self.create_sample_from_text(text)
|
||
# 默认使用单字符评估（与验证集一致）
|
||
if use_sequential:
|
||
result = self.evaluate_sample_with_sequential_confirmation(
|
||
sample, print_details=True, use_oracle_history=True
|
||
)
|
||
else:
|
||
result = self.evaluate_sample_single_char(
|
||
sample, print_details=True
|
||
)
|
||
results.append(result)
|
||
|
||
except Exception as e:
|
||
print(f"❌ 样本生成或评估失败: {e}")
|
||
import traceback
|
||
|
||
traceback.print_exc()
|
||
continue
|
||
|
||
if results:
|
||
self.print_summary(results, low_variance_samples)
|
||
|
||
return results
|
||
|
||
def evaluate_by_slot_count(
|
||
self, text: str, samples_per_slot: int = 100, use_sequential: bool = False
|
||
) -> Dict[int, Dict]:
|
||
"""
|
||
按槽位数量分别评估，每个槽位数量测试 samples_per_slot 个样本。
|
||
|
||
Args:
|
||
text: 输入文本
|
||
samples_per_slot: 每个槽位数量的测试样本数
|
||
use_sequential: True=使用逐步确认评估，False=使用单字符评估（与验证集一致）
|
||
|
||
Returns:
|
||
按槽位数量分组的评估结果 {slot_count: {results, accuracy, ...}}
|
||
"""
|
||
if len(text) > 300:
|
||
start = random.randint(0, len(text) - 300)
|
||
text = text[start : start + 300]
|
||
|
||
print(f"\n{'=' * 60}")
|
||
mode_str = "逐步确认" if use_sequential else "单字符(与验证集一致)"
|
||
print(f"按槽位数量评估 [{mode_str}] | 每组 {samples_per_slot} 个样本")
|
||
print(f"{'=' * 60}")
|
||
|
||
slot_results = {}
|
||
|
||
for slot_count in range(0, 9):
|
||
results = []
|
||
low_var_count = 0
|
||
failed = 0
|
||
|
||
for sample_idx in range(samples_per_slot):
|
||
try:
|
||
sample = self.create_sample_from_text(
|
||
text, force_slot_count=slot_count
|
||
)
|
||
# 默认使用单字符评估（与验证集一致）
|
||
if use_sequential:
|
||
result = self.evaluate_sample_with_sequential_confirmation(
|
||
sample, print_details=False, use_oracle_history=True
|
||
)
|
||
else:
|
||
result = self.evaluate_sample_single_char(
|
||
sample, print_details=False
|
||
)
|
||
results.append(result)
|
||
# 统计低方差样本
|
||
if "analysis" in result and result["analysis"].get("low_variance"):
|
||
low_var_count += 1
|
||
except Exception:
|
||
failed += 1
|
||
continue
|
||
|
||
if results:
|
||
correct_count = sum(1 for r in results if r["correct"])
|
||
accuracy = correct_count / len(results)
|
||
|
||
# 计算平均步数准确率（仅用于逐步确认模式）
|
||
step_accuracies = []
|
||
if use_sequential:
|
||
for r in results:
|
||
if "step_correct" in r:
|
||
step_correct = r["step_correct"]
|
||
if step_correct:
|
||
step_acc = sum(step_correct) / len(step_correct)
|
||
step_accuracies.append(step_acc)
|
||
|
||
mean_step_accuracy = np.mean(step_accuracies) if step_accuracies else 0
|
||
|
||
slot_results[slot_count] = {
|
||
"results": results,
|
||
"accuracy": accuracy,
|
||
"correct": correct_count,
|
||
"total": len(results),
|
||
"mean_step_accuracy": mean_step_accuracy,
|
||
"low_var_count": low_var_count,
|
||
"failed": failed,
|
||
}
|
||
|
||
bar_len = 30
|
||
filled = int(bar_len * accuracy)
|
||
bar = "█" * filled + "░" * (bar_len - filled)
|
||
if use_sequential:
|
||
print(
|
||
f" 槽位 {slot_count}/8 | {bar} {accuracy:6.1%} "
|
||
f"({correct_count:2d}/{len(results):2d}) "
|
||
f"| 平均步数准确率: {mean_step_accuracy:.1%}"
|
||
)
|
||
else:
|
||
print(
|
||
f" 槽位 {slot_count}/8 | {bar} {accuracy:6.1%} "
|
||
f"({correct_count:2d}/{len(results):2d}) "
|
||
f"| 低方差: {low_var_count}"
|
||
)
|
||
else:
|
||
slot_results[slot_count] = {
|
||
"results": [],
|
||
"accuracy": 0,
|
||
"correct": 0,
|
||
"total": 0,
|
||
"mean_step_accuracy": 0,
|
||
"low_var_count": 0,
|
||
"failed": failed,
|
||
}
|
||
print(f" 槽位 {slot_count}/8 | 全部样本生成失败")
|
||
|
||
print(f"\n{'=' * 60}")
|
||
overall_correct = sum(s["correct"] for s in slot_results.values())
|
||
overall_total = sum(s["total"] for s in slot_results.values())
|
||
if overall_total > 0:
|
||
print(
|
||
f" 总体准确率: {overall_correct / overall_total:.1%} ({overall_correct}/{overall_total})"
|
||
)
|
||
print(f"{'=' * 60}")
|
||
|
||
return slot_results
|
||
|
||
def print_summary(self, results: List[Dict], low_variance_samples: List[Dict]):
|
||
"""打印评估汇总"""
|
||
print(f"\n--- 汇总 ({len(results)}样本) ---")
|
||
|
||
# 计算准确率
|
||
correct_count = sum(1 for r in results if r["correct"])
|
||
accuracy = correct_count / len(results) if results else 0
|
||
print(f"准确率: {accuracy:.1%} ({correct_count}/{len(results)})")
|
||
|
||
# 计算平均步数准确率（对于多字符预测）
|
||
step_accuracies = []
|
||
for r in results:
|
||
if "step_correct" in r:
|
||
step_correct = r["step_correct"]
|
||
if step_correct:
|
||
step_acc = sum(step_correct) / len(step_correct)
|
||
step_accuracies.append(step_acc)
|
||
|
||
if step_accuracies:
|
||
mean_step_accuracy = np.mean(step_accuracies)
|
||
print(f"平均步数准确率: {mean_step_accuracy:.1%}")
|
||
|
||
# 槽位和拼音长度统计
|
||
if results and "sample" in results[0]:
|
||
valid_slots = [r["sample"].get("valid_slot_count", 0) for r in results]
|
||
pinyin_lengths = [r.get("pinyin_len", 1) for r in results]
|
||
avg_slots = np.mean(valid_slots) if valid_slots else 0
|
||
avg_pinyin = np.mean(pinyin_lengths) if pinyin_lengths else 0
|
||
print(f"平均槽位: {avg_slots:.1f}/8, 平均拼音长度: {avg_pinyin:.1f}")
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="评估模型在文本上的表现")
|
||
parser.add_argument("--text-file", type=str, required=True, help="文本文件路径")
|
||
parser.add_argument(
|
||
"--checkpoint",
|
||
type=str,
|
||
default="/home/songsenand/下载/best_model.ptrom",
|
||
help="模型checkpoint路径 (默认: /home/songsenand/下载/best_model.ptrom)",
|
||
)
|
||
parser.add_argument(
|
||
"--num-samples", type=int, default=100, help="评估样本数量 (默认: 100)"
|
||
)
|
||
parser.add_argument(
|
||
"--device",
|
||
type=str,
|
||
default="auto",
|
||
choices=["auto", "cpu", "cuda"],
|
||
help="推理设备 (默认: auto)",
|
||
)
|
||
parser.add_argument(
|
||
"--use-sequential",
|
||
action="store_true",
|
||
help="使用逐步确认评估模式（默认使用单字符评估，与验证集一致）",
|
||
)
|
||
|
||
args = parser.parse_args()
|
||
|
||
# 选择设备
|
||
if args.device == "auto":
|
||
device = "cuda" if torch.cuda.is_available() else "cpu"
|
||
else:
|
||
device = args.device
|
||
|
||
# 读取文本文件
|
||
try:
|
||
with open(args.text_file, "r", encoding="utf-8") as f:
|
||
text = f.read().strip()
|
||
except Exception as e:
|
||
print(f"❌ 无法读取文本文件: {e}")
|
||
sys.exit(1)
|
||
|
||
if not text:
|
||
print("❌ 文本文件为空")
|
||
sys.exit(1)
|
||
|
||
print(f"📄 读取文本文件: {args.text_file}")
|
||
print(f"📏 原始文本长度: {len(text)} 字符")
|
||
print(f"🔧 使用设备: {device}")
|
||
print(f"📦 模型checkpoint: {args.checkpoint}")
|
||
print(f"🔬 评估样本数: {args.num_samples}")
|
||
print(
|
||
f"📊 评估模式: {'逐步确认' if args.use_sequential else '单字符(与验证集一致)'}"
|
||
)
|
||
|
||
# 初始化评估器
|
||
try:
|
||
evaluator = TextEvaluator(args.checkpoint, device)
|
||
except Exception as e:
|
||
print(f"❌ 初始化评估器失败: {e}")
|
||
import traceback
|
||
|
||
traceback.print_exc()
|
||
sys.exit(1)
|
||
|
||
# 执行评估
|
||
evaluator.evaluate_by_slot_count(
|
||
text, samples_per_slot=args.num_samples, use_sequential=args.use_sequential
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|