Kh0128/Aphasia_Classification__Lang
0
1# -*- coding: utf-8 -*-2"""3Aphasia classification inference (cleaned).4- Respects model_dir argument5- Correctly parses durations like ["word", 300] and [start, end]6- Removes duplicate load_state_dict7- Adds predict_from_chajson(json_path, ...) helper8"""9 10import json as json11import os12import math13from dataclasses import dataclass14from typing import Dict, List, Optional, Tuple15from collections import defaultdict16 17 18import numpy as np19import torch20import torch.nn as nn21import torch.nn.functional as F22import pandas as pd23from transformers import AutoTokenizer, AutoModel24 25 26# =========================27# Model definition (unchanged shape)28# =========================29 30@dataclass31class ModelConfig:32 model_name: str = "microsoft/BiomedNLP-PubMedBERT-base-uncased-abstract-fulltext"33 max_length: int = 51234 hidden_size: int = 76835 pos_vocab_size: int = 15036 pos_emb_dim: int = 6437 grammar_dim: int = 338 grammar_hidden_dim: int = 6439 duration_hidden_dim: int = 12840 prosody_dim: int = 3241 num_attention_heads: int = 842 attention_dropout: float = 0.343 classifier_hidden_dims: List[int] = None44 dropout_rate: float = 0.345 def __post_init__(self):46 if self.classifier_hidden_dims is None:47 self.classifier_hidden_dims = [512, 256]48 49class StablePositionalEncoding(nn.Module):50 def __init__(self, d_model: int, max_len: int = 5000):51 super().__init__()52 self.d_model = d_model53 pe = torch.zeros(max_len, d_model)54 position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1)55 div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model))56 pe[:, 0::2] = torch.sin(position * div_term)57 pe[:, 1::2] = torch.cos(position * div_term)58 self.register_buffer('pe', pe.unsqueeze(0))59 self.learnable_pe = nn.Parameter(torch.randn(max_len, d_model) * 0.01)60 def forward(self, x):61 seq_len = x.size(1)62 sinusoidal = self.pe[:, :seq_len, :].to(x.device)63 learnable = self.learnable_pe[:seq_len, :].unsqueeze(0).expand(x.size(0), -1, -1)64 return x + 0.1 * (sinusoidal + learnable)65 66class StableMultiHeadAttention(nn.Module):67 def __init__(self, feature_dim: int, num_heads: int = 4, dropout: float = 0.3):68 super().__init__()69 self.num_heads = num_heads70 self.feature_dim = feature_dim71 self.head_dim = feature_dim // num_heads72 assert feature_dim % num_heads == 073 self.query = nn.Linear(feature_dim, feature_dim)74 self.key = nn.Linear(feature_dim, feature_dim)75 self.value = nn.Linear(feature_dim, feature_dim)76 self.dropout = nn.Dropout(dropout)77 self.output_proj = nn.Linear(feature_dim, feature_dim)78 self.layer_norm = nn.LayerNorm(feature_dim)79 def forward(self, x, mask=None):80 b, t, _ = x.size()81 Q = self.query(x).view(b, t, self.num_heads, self.head_dim).transpose(1, 2)82 K = self.key(x).view(b, t, self.num_heads, self.head_dim).transpose(1, 2)83 V = self.value(x).view(b, t, self.num_heads, self.head_dim).transpose(1, 2)84 scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(self.head_dim)85 if mask is not None:86 if mask.dim() == 2:87 mask = mask.unsqueeze(1).unsqueeze(1)88 scores.masked_fill_(mask == 0, -1e9)89 attn = F.softmax(scores, dim=-1)90 attn = self.dropout(attn)91 ctx = torch.matmul(attn, V)92 ctx = ctx.transpose(1, 2).contiguous().view(b, t, self.feature_dim)93 out = self.output_proj(ctx)94 return self.layer_norm(out + x)95 96class StableLinguisticFeatureExtractor(nn.Module):97 def __init__(self, config: ModelConfig):98 super().__init__()99 self.config = config100 self.pos_embedding = nn.Embedding(config.pos_vocab_size, config.pos_emb_dim, padding_idx=0)101 self.pos_attention = StableMultiHeadAttention(config.pos_emb_dim, num_heads=4)102 self.grammar_projection = nn.Sequential(103 nn.Linear(config.grammar_dim, config.grammar_hidden_dim),104 nn.Tanh(),105 nn.LayerNorm(config.grammar_hidden_dim),106 nn.Dropout(config.dropout_rate * 0.3)107 )108 self.duration_projection = nn.Sequential(109 nn.Linear(1, config.duration_hidden_dim),110 nn.Tanh(),111 nn.LayerNorm(config.duration_hidden_dim)112 )113 self.prosody_projection = nn.Sequential(114 nn.Linear(config.prosody_dim, config.prosody_dim),115 nn.ReLU(),116 nn.LayerNorm(config.prosody_dim)117 )118 total_feature_dim = (config.pos_emb_dim + config.grammar_hidden_dim +119 config.duration_hidden_dim + config.prosody_dim)120 self.feature_fusion = nn.Sequential(121 nn.Linear(total_feature_dim, total_feature_dim // 2),122 nn.Tanh(),123 nn.LayerNorm(total_feature_dim // 2),124 nn.Dropout(config.dropout_rate)125 )126 def forward(self, pos_ids, grammar_ids, durations, prosody_features, attention_mask):127 b, t = pos_ids.size()128 pos_ids = pos_ids.clamp(0, self.config.pos_vocab_size - 1)129 pos_emb = self.pos_embedding(pos_ids)130 pos_feat = self.pos_attention(pos_emb, attention_mask)131 gra_feat = self.grammar_projection(grammar_ids.float())132 dur_feat = self.duration_projection(durations.unsqueeze(-1).float())133 pro_feat = self.prosody_projection(prosody_features.float())134 combined = torch.cat([pos_feat, gra_feat, dur_feat, pro_feat], dim=-1)135 fused = self.feature_fusion(combined)136 mask_exp = attention_mask.unsqueeze(-1).float()137 pooled = torch.sum(fused * mask_exp, dim=1) / torch.sum(mask_exp, dim=1)138 return pooled139 140class StableAphasiaClassifier(nn.Module):141 def __init__(self, config: ModelConfig, num_labels: int):142 super().__init__()143 self.config = config144 self.num_labels = num_labels145 self.bert = AutoModel.from_pretrained(config.model_name)146 self.bert_config = self.bert.config147 self.positional_encoder = StablePositionalEncoding(d_model=self.bert_config.hidden_size,148 max_len=config.max_length)149 self.linguistic_extractor = StableLinguisticFeatureExtractor(config)150 bert_dim = self.bert_config.hidden_size151 lingu_dim = (config.pos_emb_dim + config.grammar_hidden_dim +152 config.duration_hidden_dim + config.prosody_dim) // 2153 self.feature_fusion = nn.Sequential(154 nn.Linear(bert_dim + lingu_dim, bert_dim),155 nn.LayerNorm(bert_dim),156 nn.Tanh(),157 nn.Dropout(config.dropout_rate)158 )159 self.classifier = self._build_classifier(bert_dim, num_labels)160 self.severity_head = nn.Sequential(nn.Linear(bert_dim, 4), nn.Softmax(dim=-1))161 self.fluency_head = nn.Sequential(nn.Linear(bert_dim, 1), nn.Sigmoid())162 def _build_classifier(self, input_dim: int, num_labels: int):163 layers, cur = [], input_dim164 for h in self.config.classifier_hidden_dims:165 layers += [nn.Linear(cur, h), nn.LayerNorm(h), nn.Tanh(), nn.Dropout(self.config.dropout_rate)]166 cur = h167 layers.append(nn.Linear(cur, num_labels))168 return nn.Sequential(*layers)169 def _attention_pooling(self, seq_out, attn_mask):170 attn_w = torch.softmax(torch.sum(seq_out, dim=-1, keepdim=True), dim=1)171 attn_w = attn_w * attn_mask.unsqueeze(-1).float()172 attn_w = attn_w / (torch.sum(attn_w, dim=1, keepdim=True) + 1e-9)173 return torch.sum(seq_out * attn_w, dim=1)174 def forward(self, input_ids, attention_mask, labels=None,175 word_pos_ids=None, word_grammar_ids=None, word_durations=None,176 prosody_features=None, **kwargs):177 bert_out = self.bert(input_ids=input_ids, attention_mask=attention_mask)178 seq_out = bert_out.last_hidden_state179 pos_enh = self.positional_encoder(seq_out)180 pooled = self._attention_pooling(pos_enh, attention_mask)181 if all(x is not None for x in [word_pos_ids, word_grammar_ids, word_durations]):182 if prosody_features is None:183 b, t = input_ids.size()184 prosody_features = torch.zeros(b, t, self.config.prosody_dim, device=input_ids.device)185 ling = self.linguistic_extractor(word_pos_ids, word_grammar_ids, word_durations,186 prosody_features, attention_mask)187 else:188 ling = torch.zeros(input_ids.size(0),189 (self.config.pos_emb_dim + self.config.grammar_hidden_dim +190 self.config.duration_hidden_dim + self.config.prosody_dim) // 2,191 device=input_ids.device)192 fused = self.feature_fusion(torch.cat([pooled, ling], dim=1))193 logits = self.classifier(fused)194 severity_pred = self.severity_head(fused)195 fluency_pred = self.fluency_head(fused)196 return {"logits": logits, "severity_pred": severity_pred, "fluency_pred": fluency_pred, "loss": None}197 198 199# =========================200# Inference system (fixed wiring)201# =========================202 203class AphasiaInferenceSystem:204 """失語症分類推理系統"""205 206 def __init__(self, model_dir: str):207 self.model_dir = model_dir # <— honor the argument208 self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")209 210 # Descriptions (unchanged)211 self.aphasia_descriptions = {212 "BROCA": {"name": "Broca's Aphasia (Non-fluent)", "description":213 "Characterized by limited speech output, difficulty with grammar and sentence formation, but relatively preserved comprehension. Speech is typically effortful and halting.",214 "features": ["Non-fluent speech", "Preserved comprehension", "Grammar difficulties", "Word-finding problems"]},215 "TRANSMOTOR": {"name": "Trans-cortical Motor Aphasia", "description":216 "Similar to Broca's aphasia but with preserved repetition abilities. Speech is non-fluent with good comprehension.",217 "features": ["Non-fluent speech", "Good repetition", "Preserved comprehension", "Grammar difficulties"]},218 "NOTAPHASICBYWAB": {"name": "Not Aphasic by WAB", "description":219 "Individuals who do not meet the criteria for aphasia according to the Western Aphasia Battery assessment.",220 "features": ["Normal language function", "No significant language impairment", "Good comprehension", "Fluent speech"]},221 "CONDUCTION": {"name": "Conduction Aphasia", "description":222 "Characterized by fluent speech with good comprehension but severely impaired repetition. Often involves phonemic paraphasias.",223 "features": ["Fluent speech", "Good comprehension", "Poor repetition", "Phonemic errors"]},224 "WERNICKE": {"name": "Wernicke's Aphasia (Fluent)", "description":225 "Fluent but often meaningless speech with poor comprehension. Speech may contain neologisms and jargon.",226 "features": ["Fluent speech", "Poor comprehension", "Jargon speech", "Neologisms"]},227 "ANOMIC": {"name": "Anomic Aphasia", "description":228 "Primarily characterized by word-finding difficulties with otherwise relatively preserved language abilities.",229 "features": ["Word-finding difficulties", "Good comprehension", "Fluent speech", "Circumlocution"]},230 "GLOBAL": {"name": "Global Aphasia", "description":231 "Severe impairment in all language modalities - comprehension, production, repetition, and naming.",232 "features": ["Severe comprehension deficit", "Non-fluent speech", "Poor repetition", "Severe naming difficulties"]},233 "ISOLATION": {"name": "Isolation Syndrome", "description":234 "Rare condition with preserved repetition but severely impaired comprehension and spontaneous speech.",235 "features": ["Good repetition", "Poor comprehension", "Limited spontaneous speech", "Echolalia"]},236 "TRANSSENSORY": {"name": "Trans-cortical Sensory Aphasia", "description":237 "Fluent speech with good repetition but impaired comprehension, similar to Wernicke's but with preserved repetition.",238 "features": ["Fluent speech", "Good repetition", "Poor comprehension", "Semantic errors"]}239 }240 241 self.load_configuration()242 self.load_model()243 print(f"推理系統初始化完成，使用設備: {self.device}")244 245 def load_configuration(self):246 cfg_path = os.path.join(self.model_dir, "config.json")247 if os.path.exists(cfg_path):248 with open(cfg_path, "r", encoding="utf-8") as f:249 cfg = json.load(f)250 self.aphasia_types_mapping = cfg.get("aphasia_types_mapping", {251 "BROCA": 0, "TRANSMOTOR": 1, "NOTAPHASICBYWAB": 2,252 "CONDUCTION": 3, "WERNICKE": 4, "ANOMIC": 5,253 "GLOBAL": 6, "ISOLATION": 7, "TRANSSENSORY": 8254 })255 self.num_labels = cfg.get("num_labels", 9)256 self.model_name = cfg.get("model_name", "microsoft/BiomedNLP-PubMedBERT-base-uncased-abstract-fulltext")257 else:258 self.aphasia_types_mapping = {259 "BROCA": 0, "TRANSMOTOR": 1, "NOTAPHASICBYWAB": 2,260 "CONDUCTION": 3, "WERNICKE": 4, "ANOMIC": 5,261 "GLOBAL": 6, "ISOLATION": 7, "TRANSSENSORY": 8262 }263 self.num_labels = 9264 self.model_name = "microsoft/BiomedNLP-PubMedBERT-base-uncased-abstract-fulltext"265 self.id_to_aphasia_type = {v: k for k, v in self.aphasia_types_mapping.items()}266 267 def load_model(self):268 self.tokenizer = AutoTokenizer.from_pretrained(self.model_dir, use_fast=True)269 # pad token fix270 if self.tokenizer.pad_token is None:271 if self.tokenizer.eos_token is not None:272 self.tokenizer.pad_token = self.tokenizer.eos_token273 elif self.tokenizer.unk_token is not None:274 self.tokenizer.pad_token = self.tokenizer.unk_token275 else:276 self.tokenizer.add_special_tokens({"pad_token": "[PAD]"})277 # optional added tokens278 add_path = os.path.join(self.model_dir, "added_tokens.json")279 if os.path.exists(add_path):280 with open(add_path, "r", encoding="utf-8") as f:281 data = json.load(f)282 tokens = list(data.keys()) if isinstance(data, dict) else data283 if tokens:284 self.tokenizer.add_tokens(tokens)285 286 self.config = ModelConfig()287 self.config.model_name = self.model_name288 289 self.model = StableAphasiaClassifier(self.config, self.num_labels)290 self.model.bert.resize_token_embeddings(len(self.tokenizer))291 292 model_path = os.path.join(self.model_dir, "pytorch_model.bin")293 if not os.path.exists(model_path):294 raise FileNotFoundError(f"模型權重文件不存在: {model_path}")295 state = torch.load(model_path, map_location=self.device)296 self.model.load_state_dict(state) # (once)297 298 self.model.to(self.device)299 self.model.eval()300 301 # ---------- helpers ----------302 303 def _dur_to_float(self, d) -> float:304 """Robustly parse duration from various shapes:305 - number306 - ["word", ms]307 - [start, end]308 - {"dur": ms} (future-proof)309 """310 if isinstance(d, (int, float)):311 return float(d)312 if isinstance(d, list):313 if len(d) == 2:314 # ["word", 300] or [start, end]315 a, b = d[0], d[1]316 # case 1: word + ms317 if isinstance(a, str) and isinstance(b, (int, float)):318 return float(b)319 # case 2: start, end320 if isinstance(a, (int, float)) and isinstance(b, (int, float)):321 return float(b) - float(a)322 if isinstance(d, dict):323 for k in ("dur", "duration", "ms"):324 if k in d and isinstance(d[k], (int, float)):325 return float(d[k])326 return 0.0327 328 def _extract_prosodic_features(self, durations, tokens):329 vals = []330 for d in durations:331 vals.append(self._dur_to_float(d))332 vals = [v for v in vals if v > 0]333 if not vals:334 return [0.0] * self.config.prosody_dim335 features = [336 float(np.mean(vals)),337 float(np.std(vals)),338 float(np.median(vals)),339 float(len([v for v in vals if v > (np.mean(vals) * 1.5)])),340 ]341 while len(features) < self.config.prosody_dim:342 features.append(0.0)343 return features[:self.config.prosody_dim]344 345 def _align_features(self, tokens, pos_ids, grammar_ids, durations, encoded):346 # map subtoken -> original token index347 subtoken_to_token = []348 for idx, tok in enumerate(tokens):349 subtoks = self.tokenizer.tokenize(tok)350 subtoken_to_token.extend([idx] * max(1, len(subtoks)))351 352 aligned_pos = [0] # [CLS]353 aligned_grammar = [[0, 0, 0]] # [CLS]354 aligned_durations = [0.0] # [CLS]355 356 # reserve last slot for [SEP]357 max_body = self.config.max_length - 2358 for st_idx in range(max_body):359 if st_idx < len(subtoken_to_token):360 orig = subtoken_to_token[st_idx]361 aligned_pos.append(pos_ids[orig] if orig < len(pos_ids) else 0)362 aligned_grammar.append(grammar_ids[orig] if orig < len(grammar_ids) else [0, 0, 0])363 aligned_durations.append(self._dur_to_float(durations[orig]) if orig < len(durations) else 0.0)364 else:365 aligned_pos.append(0)366 aligned_grammar.append([0, 0, 0])367 aligned_durations.append(0.0)368 369 aligned_pos.append(0) # [SEP]370 aligned_grammar.append([0, 0, 0]) # [SEP]371 aligned_durations.append(0.0) # [SEP]372 return aligned_pos, aligned_grammar, aligned_durations373 374 def preprocess_sentence(self, sentence_data: dict) -> Optional[dict]:375 all_tokens, all_pos, all_grammar, all_durations = [], [], [], []376 for d_idx, dialogue in enumerate(sentence_data.get("dialogues", [])):377 if d_idx > 0:378 all_tokens.append("[DIALOGUE]")379 all_pos.append(0)380 all_grammar.append([0, 0, 0])381 all_durations.append(0.0)382 for par in dialogue.get("PAR", []):383 if "tokens" in par and par["tokens"]:384 toks = par["tokens"]385 pos_ids = par.get("word_pos_ids", [0] * len(toks))386 gra_ids = par.get("word_grammar_ids", [[0, 0, 0]] * len(toks))387 durs = par.get("word_durations", [0.0] * len(toks))388 all_tokens.extend(toks)389 all_pos.extend(pos_ids)390 all_grammar.extend(gra_ids)391 all_durations.extend(durs)392 if not all_tokens:393 return None394 395 text = " ".join(all_tokens)396 enc = self.tokenizer(text, max_length=self.config.max_length, padding="max_length",397 truncation=True, return_tensors="pt")398 aligned_pos, aligned_gra, aligned_dur = self._align_features(399 all_tokens, all_pos, all_grammar, all_durations, enc400 )401 prosody = self._extract_prosodic_features(all_durations, all_tokens)402 prosody_tensor = torch.tensor(prosody).unsqueeze(0).repeat(self.config.max_length, 1)403 404 return {405 "input_ids": enc["input_ids"].squeeze(0),406 "attention_mask": enc["attention_mask"].squeeze(0),407 "word_pos_ids": torch.tensor(aligned_pos, dtype=torch.long),408 "word_grammar_ids": torch.tensor(aligned_gra, dtype=torch.long),409 "word_durations": torch.tensor(aligned_dur, dtype=torch.float),410 "prosody_features": prosody_tensor.float(),411 "sentence_id": sentence_data.get("sentence_id", "unknown"),412 "original_tokens": all_tokens,413 "text": text414 }415 416 def predict_single(self, sentence_data: dict) -> dict:417 proc = self.preprocess_sentence(sentence_data)418 if proc is None:419 return {"error": "無法處理輸入數據", "sentence_id": sentence_data.get("sentence_id", "unknown")}420 inp = {421 "input_ids": proc["input_ids"].unsqueeze(0).to(self.device),422 "attention_mask": proc["attention_mask"].unsqueeze(0).to(self.device),423 "word_pos_ids": proc["word_pos_ids"].unsqueeze(0).to(self.device),424 "word_grammar_ids": proc["word_grammar_ids"].unsqueeze(0).to(self.device),425 "word_durations": proc["word_durations"].unsqueeze(0).to(self.device),426 "prosody_features": proc["prosody_features"].unsqueeze(0).to(self.device),427 }428 with torch.no_grad():429 out = self.model(**inp)430 logits = out["logits"]431 probs = F.softmax(logits, dim=1).cpu().numpy()[0]432 pred_id = int(np.argmax(probs))433 sev = out["severity_pred"].cpu().numpy()[0]434 flu = float(out["fluency_pred"].cpu().numpy()[0][0])435 436 pred_type = self.id_to_aphasia_type[pred_id]437 conf = float(probs[pred_id])438 439 dist = {}440 for a_type, t_id in self.aphasia_types_mapping.items():441 dist[a_type] = {"probability": float(probs[t_id]), "percentage": f"{probs[t_id]*100:.2f}%"}442 443 sorted_dist = dict(sorted(dist.items(), key=lambda x: x[1]["probability"], reverse=True))444 return {445 "sentence_id": proc["sentence_id"],446 "input_text": proc["text"],447 "original_tokens": proc["original_tokens"],448 "prediction": {449 "predicted_class": pred_type,450 "confidence": conf,451 "confidence_percentage": f"{conf*100:.2f}%"452 },453 "class_description": self.aphasia_descriptions.get(pred_type, {454 "name": pred_type, "description": "Description not available", "features": []455 }),456 "probability_distribution": sorted_dist,457 "additional_predictions": {458 "severity_distribution": {459 "level_0": float(sev[0]), "level_1": float(sev[1]),460 "level_2": float(sev[2]), "level_3": float(sev[3])461 },462 "predicted_severity_level": int(np.argmax(sev)),463 "fluency_score": flu,464 "fluency_rating": "High" if flu > 0.7 else ("Medium" if flu > 0.4 else "Low"),465 }466 }467 468 def predict_batch(self, input_file: str, output_file: Optional[str] = None) -> Dict:469 with open(input_file, "r", encoding="utf-8") as f:470 data = json.load(f)471 sentences = data.get("sentences", [])472 results = []473 print(f"開始處理 {len(sentences)} 個句子...")474 for i, s in enumerate(sentences):475 print(f"處理第 {i+1}/{len(sentences)} 個句子...")476 results.append(self.predict_single(s))477 summary = self._generate_summary(results)478 final = {"summary": summary, "total_sentences": len(results), "predictions": results}479 if output_file:480 with open(output_file, "w", encoding="utf-8") as f:481 json.dump(final, f, ensure_ascii=False, indent=2)482 print(f"結果已保存到: {output_file}")483 return final484 485 def _generate_summary(self, results: List[dict]) -> dict:486 if not results:487 return {}488 class_counts = defaultdict(int)489 confs, flus = [], []490 sev_counts = defaultdict(int)491 for r in results:492 if "error" in r:493 continue494 c = r["prediction"]["predicted_class"]495 class_counts[c] += 1496 confs.append(r["prediction"]["confidence"])497 flus.append(r["additional_predictions"]["fluency_score"])498 sev_counts[r["additional_predictions"]["predicted_severity_level"]] += 1499 avg_conf = float(np.mean(confs)) if confs else 0.0500 avg_flu = float(np.mean(flus)) if flus else 0.0501 return {502 "classification_distribution": dict(class_counts),503 "classification_percentages": {k: f"{v/len(results)*100:.1f}%" for k, v in class_counts.items()},504 "average_confidence": f"{avg_conf:.3f}",505 "average_fluency_score": f"{avg_flu:.3f}",506 "severity_distribution": dict(sev_counts),507 "confidence_statistics": {} if not confs else {508 "mean": f"{np.mean(confs):.3f}",509 "std": f"{np.std(confs):.3f}",510 "min": f"{np.min(confs):.3f}",511 "max": f"{np.max(confs):.3f}",512 },513 "most_common_prediction": max(class_counts.items(), key=lambda x: x[1])[0] if class_counts else "None",514 }515 516 def generate_detailed_report(self, results: List[dict], output_dir: str = "./inference_results"):517 os.makedirs(output_dir, exist_ok=True)518 rows = []519 for r in results:520 if "error" in r:521 continue522 row = {523 "sentence_id": r["sentence_id"],524 "predicted_class": r["prediction"]["predicted_class"],525 "confidence": r["prediction"]["confidence"],526 "class_name": r["class_description"]["name"],527 "severity_level": r["additional_predictions"]["predicted_severity_level"],528 "fluency_score": r["additional_predictions"]["fluency_score"],529 "fluency_rating": r["additional_predictions"]["fluency_rating"],530 "input_text": r["input_text"],531 }532 for a_type, info in r["probability_distribution"].items():533 row[f"prob_{a_type}"] = info["probability"]534 rows.append(row)535 if not rows:536 return None537 df = pd.DataFrame(rows)538 df.to_csv(os.path.join(output_dir, "detailed_predictions.csv"), index=False, encoding="utf-8")539 summary_stats = {540 "total_predictions": int(len(rows)),541 "class_distribution": df["predicted_class"].value_counts().to_dict(),542 "average_confidence": float(df["confidence"].mean()),543 "confidence_std": float(df["confidence"].std()),544 "average_fluency": float(df["fluency_score"].mean()),545 "fluency_std": float(df["fluency_score"].std()),546 "severity_distribution": df["severity_level"].value_counts().to_dict(),547 }548 with open(os.path.join(output_dir, "summary_statistics.json"), "w", encoding="utf-8") as f:549 json.dump(summary_stats, f, ensure_ascii=False, indent=2)550 print(f"詳細報告已生成並保存到: {output_dir}")551 return df552 553 554# =========================555# Convenience: run directly or from pipeline556# =========================557 558def predict_from_chajson(model_dir: str, chajson_path: str, output_file: Optional[str] = None) -> Dict:559 """560 Convenience entry:561 - Accepts the JSON produced by cha_json.py562 - If it contains 'sentences', runs per-sentence like before563 - If it only contains 'text_all', creates a single pseudo-sentence564 """565 with open(chajson_path, "r", encoding="utf-8") as f:566 data = json.load(f)567 568 inf = AphasiaInferenceSystem(model_dir)569 570 # If there are sentences, use the full path571 if data.get("sentences"):572 return inf.predict_batch(chajson_path, output_file=output_file)573 574 # Else, fall back to a single synthetic sentence using text_all575 text_all = data.get("text_all", "")576 fake = {577 "sentences": [{578 "sentence_id": "S1",579 "dialogues": [{580 "INV": [],581 "PAR": [{"tokens": text_all.split(),582 "word_pos_ids": [0]*len(text_all.split()),583 "word_grammar_ids": [[0,0,0]]*len(text_all.split()),584 "word_durations": [0.0]*len(text_all.split())}]585 }]586 }]587 }588 tmp_path = chajson_path + "._synthetic.json"589 with open(tmp_path, "w", encoding="utf-8") as f:590 json.dump(fake, f, ensure_ascii=False, indent=2)591 out = inf.predict_batch(tmp_path, output_file=output_file)592 try:593 os.remove(tmp_path)594 except Exception:595 pass596 return out597 598def format_result(pred: dict, style: str = "json") -> str:599 """Back-compat formatter. 'pred' is the dict returned by predict_*."""600 if style == "json":601 return json.dumps(pred, ensure_ascii=False, indent=2)602 # simple text summary603 if isinstance(pred, dict) and "summary" in pred:604 s = pred["summary"]605 lines = [606 f"Total sentences: {pred.get('total_sentences', 0)}",607 f"Avg confidence: {s.get('average_confidence', 'N/A')}",608 f"Avg fluency: {s.get('average_fluency_score', 'N/A')}",609 f"Most common: {s.get('most_common_prediction', 'N/A')}",610 ]611 return "\n".join(lines)612 return str(pred)613 614 615# ---------- CLI ----------616 617def main():618 import argparse619 p = argparse.ArgumentParser(description="失語症分類推理系統")620 p.add_argument("--model_dir", type=str, required=False, default="./adaptive_aphasia_model",621 help="訓練好的模型目錄路徑")622 p.add_argument("--input_file", type=str, required=True,623 help="輸入JSON文件（cha_json 的輸出）")624 p.add_argument("--output_file", type=str, default="./aphasia_predictions.json",625 help="輸出JSON文件路徑")626 p.add_argument("--report_dir", type=str, default="./inference_results",627 help="詳細報告輸出目錄")628 p.add_argument("--generate_report", action="store_true",629 help="是否生成詳細的CSV報告")630 args = p.parse_args()631 632 try:633 print("正在初始化推理系統...")634 sys = AphasiaInferenceSystem(args.model_dir)635 636 print("開始執行批次預測...")637 results = sys.predict_batch(args.input_file, args.output_file)638 639 if args.generate_report:640 print("生成詳細報告...")641 sys.generate_detailed_report(results["predictions"], args.report_dir)642 643 print("\n=== 預測摘要 ===")644 s = results["summary"]645 print(f"總句子數: {results['total_sentences']}")646 print(f"平均信心度: {s.get('average_confidence', 'N/A')}")647 print(f"平均流利度: {s.get('average_fluency_score', 'N/A')}")648 print(f"最常見預測: {s.get('most_common_prediction', 'N/A')}")649 print("\n類別分佈:")650 for name, count in s.get("classification_distribution", {}).items():651 pct = s.get("classification_percentages", {}).get(name, "0%")652 print(f" {name}: {count} ({pct})")653 print(f"\n結果已保存到: {args.output_file}")654 except Exception as e:655 print(f"錯誤: {str(e)}")656 import traceback; traceback.print_exc()657 658if __name__ == "__main__":659 main()660