# -*- coding: utf-8 -*- """ Aphasia classification inference (cleaned). - Respects model_dir argument - Correctly parses durations like ["word", 300] and [start, end] - Removes duplicate load_state_dict - Adds predict_from_chajson(json_path, ...) helper """ import json as json import os import math from dataclasses import dataclass from typing import Dict, List, Optional, Tuple from collections import defaultdict import numpy as np import torch import torch.nn as nn import torch.nn.functional as F import pandas as pd from transformers import AutoTokenizer, AutoModel # ========================= # Model definition (unchanged shape) # ========================= @dataclass class ModelConfig: model_name: str = "microsoft/BiomedNLP-PubMedBERT-base-uncased-abstract-fulltext" max_length: int = 512 hidden_size: int = 768 pos_vocab_size: int = 150 pos_emb_dim: int = 64 grammar_dim: int = 3 grammar_hidden_dim: int = 64 duration_hidden_dim: int = 128 prosody_dim: int = 32 num_attention_heads: int = 8 attention_dropout: float = 0.3 classifier_hidden_dims: List[int] = None dropout_rate: float = 0.3 def __post_init__(self): if self.classifier_hidden_dims is None: self.classifier_hidden_dims = [512, 256] class StablePositionalEncoding(nn.Module): def __init__(self, d_model: int, max_len: int = 5000): super().__init__() self.d_model = d_model pe = torch.zeros(max_len, d_model) position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1) div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model)) pe[:, 0::2] = torch.sin(position * div_term) pe[:, 1::2] = torch.cos(position * div_term) self.register_buffer('pe', pe.unsqueeze(0)) self.learnable_pe = nn.Parameter(torch.randn(max_len, d_model) * 0.01) def forward(self, x): seq_len = x.size(1) sinusoidal = self.pe[:, :seq_len, :].to(x.device) learnable = self.learnable_pe[:seq_len, :].unsqueeze(0).expand(x.size(0), -1, -1) return x + 0.1 * (sinusoidal + learnable) class StableMultiHeadAttention(nn.Module): def __init__(self, feature_dim: int, num_heads: int = 4, dropout: float = 0.3): super().__init__() self.num_heads = num_heads self.feature_dim = feature_dim self.head_dim = feature_dim // num_heads assert feature_dim % num_heads == 0 self.query = nn.Linear(feature_dim, feature_dim) self.key = nn.Linear(feature_dim, feature_dim) self.value = nn.Linear(feature_dim, feature_dim) self.dropout = nn.Dropout(dropout) self.output_proj = nn.Linear(feature_dim, feature_dim) self.layer_norm = nn.LayerNorm(feature_dim) def forward(self, x, mask=None): b, t, _ = x.size() Q = self.query(x).view(b, t, self.num_heads, self.head_dim).transpose(1, 2) K = self.key(x).view(b, t, self.num_heads, self.head_dim).transpose(1, 2) V = self.value(x).view(b, t, self.num_heads, self.head_dim).transpose(1, 2) scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(self.head_dim) if mask is not None: if mask.dim() == 2: mask = mask.unsqueeze(1).unsqueeze(1) scores.masked_fill_(mask == 0, -1e9) attn = F.softmax(scores, dim=-1) attn = self.dropout(attn) ctx = torch.matmul(attn, V) ctx = ctx.transpose(1, 2).contiguous().view(b, t, self.feature_dim) out = self.output_proj(ctx) return self.layer_norm(out + x) class StableLinguisticFeatureExtractor(nn.Module): def __init__(self, config: ModelConfig): super().__init__() self.config = config self.pos_embedding = nn.Embedding(config.pos_vocab_size, config.pos_emb_dim, padding_idx=0) self.pos_attention = StableMultiHeadAttention(config.pos_emb_dim, num_heads=4) self.grammar_projection = nn.Sequential( nn.Linear(config.grammar_dim, config.grammar_hidden_dim), nn.Tanh(), nn.LayerNorm(config.grammar_hidden_dim), nn.Dropout(config.dropout_rate * 0.3) ) self.duration_projection = nn.Sequential( nn.Linear(1, config.duration_hidden_dim), nn.Tanh(), nn.LayerNorm(config.duration_hidden_dim) ) self.prosody_projection = nn.Sequential( nn.Linear(config.prosody_dim, config.prosody_dim), nn.ReLU(), nn.LayerNorm(config.prosody_dim) ) total_feature_dim = (config.pos_emb_dim + config.grammar_hidden_dim + config.duration_hidden_dim + config.prosody_dim) self.feature_fusion = nn.Sequential( nn.Linear(total_feature_dim, total_feature_dim // 2), nn.Tanh(), nn.LayerNorm(total_feature_dim // 2), nn.Dropout(config.dropout_rate) ) def forward(self, pos_ids, grammar_ids, durations, prosody_features, attention_mask): b, t = pos_ids.size() pos_ids = pos_ids.clamp(0, self.config.pos_vocab_size - 1) pos_emb = self.pos_embedding(pos_ids) pos_feat = self.pos_attention(pos_emb, attention_mask) gra_feat = self.grammar_projection(grammar_ids.float()) dur_feat = self.duration_projection(durations.unsqueeze(-1).float()) pro_feat = self.prosody_projection(prosody_features.float()) combined = torch.cat([pos_feat, gra_feat, dur_feat, pro_feat], dim=-1) fused = self.feature_fusion(combined) mask_exp = attention_mask.unsqueeze(-1).float() pooled = torch.sum(fused * mask_exp, dim=1) / torch.sum(mask_exp, dim=1) return pooled class StableAphasiaClassifier(nn.Module): def __init__(self, config: ModelConfig, num_labels: int): super().__init__() self.config = config self.num_labels = num_labels self.bert = AutoModel.from_pretrained(config.model_name) self.bert_config = self.bert.config self.positional_encoder = StablePositionalEncoding(d_model=self.bert_config.hidden_size, max_len=config.max_length) self.linguistic_extractor = StableLinguisticFeatureExtractor(config) bert_dim = self.bert_config.hidden_size lingu_dim = (config.pos_emb_dim + config.grammar_hidden_dim + config.duration_hidden_dim + config.prosody_dim) // 2 self.feature_fusion = nn.Sequential( nn.Linear(bert_dim + lingu_dim, bert_dim), nn.LayerNorm(bert_dim), nn.Tanh(), nn.Dropout(config.dropout_rate) ) self.classifier = self._build_classifier(bert_dim, num_labels) self.severity_head = nn.Sequential(nn.Linear(bert_dim, 4), nn.Softmax(dim=-1)) self.fluency_head = nn.Sequential(nn.Linear(bert_dim, 1), nn.Sigmoid()) def _build_classifier(self, input_dim: int, num_labels: int): layers, cur = [], input_dim for h in self.config.classifier_hidden_dims: layers += [nn.Linear(cur, h), nn.LayerNorm(h), nn.Tanh(), nn.Dropout(self.config.dropout_rate)] cur = h layers.append(nn.Linear(cur, num_labels)) return nn.Sequential(*layers) def _attention_pooling(self, seq_out, attn_mask): attn_w = torch.softmax(torch.sum(seq_out, dim=-1, keepdim=True), dim=1) attn_w = attn_w * attn_mask.unsqueeze(-1).float() attn_w = attn_w / (torch.sum(attn_w, dim=1, keepdim=True) + 1e-9) return torch.sum(seq_out * attn_w, dim=1) def forward(self, input_ids, attention_mask, labels=None, word_pos_ids=None, word_grammar_ids=None, word_durations=None, prosody_features=None, **kwargs): bert_out = self.bert(input_ids=input_ids, attention_mask=attention_mask) seq_out = bert_out.last_hidden_state pos_enh = self.positional_encoder(seq_out) pooled = self._attention_pooling(pos_enh, attention_mask) if all(x is not None for x in [word_pos_ids, word_grammar_ids, word_durations]): if prosody_features is None: b, t = input_ids.size() prosody_features = torch.zeros(b, t, self.config.prosody_dim, device=input_ids.device) ling = self.linguistic_extractor(word_pos_ids, word_grammar_ids, word_durations, prosody_features, attention_mask) else: ling = torch.zeros(input_ids.size(0), (self.config.pos_emb_dim + self.config.grammar_hidden_dim + self.config.duration_hidden_dim + self.config.prosody_dim) // 2, device=input_ids.device) fused = self.feature_fusion(torch.cat([pooled, ling], dim=1)) logits = self.classifier(fused) severity_pred = self.severity_head(fused) fluency_pred = self.fluency_head(fused) return {"logits": logits, "severity_pred": severity_pred, "fluency_pred": fluency_pred, "loss": None} # ========================= # Inference system (fixed wiring) # ========================= class AphasiaInferenceSystem: """失語症分類推理系統""" def __init__(self, model_dir: str): self.model_dir = model_dir # <— honor the argument self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") # Descriptions (unchanged) self.aphasia_descriptions = { "BROCA": {"name": "Broca's Aphasia (Non-fluent)", "description": "Characterized by limited speech output, difficulty with grammar and sentence formation, but relatively preserved comprehension. Speech is typically effortful and halting.", "features": ["Non-fluent speech", "Preserved comprehension", "Grammar difficulties", "Word-finding problems"]}, "TRANSMOTOR": {"name": "Trans-cortical Motor Aphasia", "description": "Similar to Broca's aphasia but with preserved repetition abilities. Speech is non-fluent with good comprehension.", "features": ["Non-fluent speech", "Good repetition", "Preserved comprehension", "Grammar difficulties"]}, "NOTAPHASICBYWAB": {"name": "Not Aphasic by WAB", "description": "Individuals who do not meet the criteria for aphasia according to the Western Aphasia Battery assessment.", "features": ["Normal language function", "No significant language impairment", "Good comprehension", "Fluent speech"]}, "CONDUCTION": {"name": "Conduction Aphasia", "description": "Characterized by fluent speech with good comprehension but severely impaired repetition. Often involves phonemic paraphasias.", "features": ["Fluent speech", "Good comprehension", "Poor repetition", "Phonemic errors"]}, "WERNICKE": {"name": "Wernicke's Aphasia (Fluent)", "description": "Fluent but often meaningless speech with poor comprehension. Speech may contain neologisms and jargon.", "features": ["Fluent speech", "Poor comprehension", "Jargon speech", "Neologisms"]}, "ANOMIC": {"name": "Anomic Aphasia", "description": "Primarily characterized by word-finding difficulties with otherwise relatively preserved language abilities.", "features": ["Word-finding difficulties", "Good comprehension", "Fluent speech", "Circumlocution"]}, "GLOBAL": {"name": "Global Aphasia", "description": "Severe impairment in all language modalities - comprehension, production, repetition, and naming.", "features": ["Severe comprehension deficit", "Non-fluent speech", "Poor repetition", "Severe naming difficulties"]}, "ISOLATION": {"name": "Isolation Syndrome", "description": "Rare condition with preserved repetition but severely impaired comprehension and spontaneous speech.", "features": ["Good repetition", "Poor comprehension", "Limited spontaneous speech", "Echolalia"]}, "TRANSSENSORY": {"name": "Trans-cortical Sensory Aphasia", "description": "Fluent speech with good repetition but impaired comprehension, similar to Wernicke's but with preserved repetition.", "features": ["Fluent speech", "Good repetition", "Poor comprehension", "Semantic errors"]} } self.load_configuration() self.load_model() print(f"推理系統初始化完成，使用設備: {self.device}") def load_configuration(self): cfg_path = os.path.join(self.model_dir, "config.json") if os.path.exists(cfg_path): with open(cfg_path, "r", encoding="utf-8") as f: cfg = json.load(f) self.aphasia_types_mapping = cfg.get("aphasia_types_mapping", { "BROCA": 0, "TRANSMOTOR": 1, "NOTAPHASICBYWAB": 2, "CONDUCTION": 3, "WERNICKE": 4, "ANOMIC": 5, "GLOBAL": 6, "ISOLATION": 7, "TRANSSENSORY": 8 }) self.num_labels = cfg.get("num_labels", 9) self.model_name = cfg.get("model_name", "microsoft/BiomedNLP-PubMedBERT-base-uncased-abstract-fulltext") else: self.aphasia_types_mapping = { "BROCA": 0, "TRANSMOTOR": 1, "NOTAPHASICBYWAB": 2, "CONDUCTION": 3, "WERNICKE": 4, "ANOMIC": 5, "GLOBAL": 6, "ISOLATION": 7, "TRANSSENSORY": 8 } self.num_labels = 9 self.model_name = "microsoft/BiomedNLP-PubMedBERT-base-uncased-abstract-fulltext" self.id_to_aphasia_type = {v: k for k, v in self.aphasia_types_mapping.items()} def load_model(self): self.tokenizer = AutoTokenizer.from_pretrained(self.model_dir, use_fast=True) # pad token fix if self.tokenizer.pad_token is None: if self.tokenizer.eos_token is not None: self.tokenizer.pad_token = self.tokenizer.eos_token elif self.tokenizer.unk_token is not None: self.tokenizer.pad_token = self.tokenizer.unk_token else: self.tokenizer.add_special_tokens({"pad_token": "[PAD]"}) # optional added tokens add_path = os.path.join(self.model_dir, "added_tokens.json") if os.path.exists(add_path): with open(add_path, "r", encoding="utf-8") as f: data = json.load(f) tokens = list(data.keys()) if isinstance(data, dict) else data if tokens: self.tokenizer.add_tokens(tokens) self.config = ModelConfig() self.config.model_name = self.model_name self.model = StableAphasiaClassifier(self.config, self.num_labels) self.model.bert.resize_token_embeddings(len(self.tokenizer)) model_path = os.path.join(self.model_dir, "pytorch_model.bin") if not os.path.exists(model_path): raise FileNotFoundError(f"模型權重文件不存在: {model_path}") state = torch.load(model_path, map_location=self.device) self.model.load_state_dict(state) # (once) self.model.to(self.device) self.model.eval() # ---------- helpers ---------- def _dur_to_float(self, d) -> float: """Robustly parse duration from various shapes: - number - ["word", ms] - [start, end] - {"dur": ms} (future-proof) """ if isinstance(d, (int, float)): return float(d) if isinstance(d, list): if len(d) == 2: # ["word", 300] or [start, end] a, b = d[0], d[1] # case 1: word + ms if isinstance(a, str) and isinstance(b, (int, float)): return float(b) # case 2: start, end if isinstance(a, (int, float)) and isinstance(b, (int, float)): return float(b) - float(a) if isinstance(d, dict): for k in ("dur", "duration", "ms"): if k in d and isinstance(d[k], (int, float)): return float(d[k]) return 0.0 def _extract_prosodic_features(self, durations, tokens): vals = [] for d in durations: vals.append(self._dur_to_float(d)) vals = [v for v in vals if v > 0] if not vals: return [0.0] * self.config.prosody_dim features = [ float(np.mean(vals)), float(np.std(vals)), float(np.median(vals)), float(len([v for v in vals if v > (np.mean(vals) * 1.5)])), ] while len(features) < self.config.prosody_dim: features.append(0.0) return features[:self.config.prosody_dim] def _align_features(self, tokens, pos_ids, grammar_ids, durations, encoded): # map subtoken -> original token index subtoken_to_token = [] for idx, tok in enumerate(tokens): subtoks = self.tokenizer.tokenize(tok) subtoken_to_token.extend([idx] * max(1, len(subtoks))) aligned_pos = [0] # [CLS] aligned_grammar = [[0, 0, 0]] # [CLS] aligned_durations = [0.0] # [CLS] # reserve last slot for [SEP] max_body = self.config.max_length - 2 for st_idx in range(max_body): if st_idx < len(subtoken_to_token): orig = subtoken_to_token[st_idx] aligned_pos.append(pos_ids[orig] if orig < len(pos_ids) else 0) aligned_grammar.append(grammar_ids[orig] if orig < len(grammar_ids) else [0, 0, 0]) aligned_durations.append(self._dur_to_float(durations[orig]) if orig < len(durations) else 0.0) else: aligned_pos.append(0) aligned_grammar.append([0, 0, 0]) aligned_durations.append(0.0) aligned_pos.append(0) # [SEP] aligned_grammar.append([0, 0, 0]) # [SEP] aligned_durations.append(0.0) # [SEP] return aligned_pos, aligned_grammar, aligned_durations def preprocess_sentence(self, sentence_data: dict) -> Optional[dict]: all_tokens, all_pos, all_grammar, all_durations = [], [], [], [] for d_idx, dialogue in enumerate(sentence_data.get("dialogues", [])): if d_idx > 0: all_tokens.append("[DIALOGUE]") all_pos.append(0) all_grammar.append([0, 0, 0]) all_durations.append(0.0) for par in dialogue.get("PAR", []): if "tokens" in par and par["tokens"]: toks = par["tokens"] pos_ids = par.get("word_pos_ids", [0] * len(toks)) gra_ids = par.get("word_grammar_ids", [[0, 0, 0]] * len(toks)) durs = par.get("word_durations", [0.0] * len(toks)) all_tokens.extend(toks) all_pos.extend(pos_ids) all_grammar.extend(gra_ids) all_durations.extend(durs) if not all_tokens: return None text = " ".join(all_tokens) enc = self.tokenizer(text, max_length=self.config.max_length, padding="max_length", truncation=True, return_tensors="pt") aligned_pos, aligned_gra, aligned_dur = self._align_features( all_tokens, all_pos, all_grammar, all_durations, enc ) prosody = self._extract_prosodic_features(all_durations, all_tokens) prosody_tensor = torch.tensor(prosody).unsqueeze(0).repeat(self.config.max_length, 1) return { "input_ids": enc["input_ids"].squeeze(0), "attention_mask": enc["attention_mask"].squeeze(0), "word_pos_ids": torch.tensor(aligned_pos, dtype=torch.long), "word_grammar_ids": torch.tensor(aligned_gra, dtype=torch.long), "word_durations": torch.tensor(aligned_dur, dtype=torch.float), "prosody_features": prosody_tensor.float(), "sentence_id": sentence_data.get("sentence_id", "unknown"), "original_tokens": all_tokens, "text": text } def predict_single(self, sentence_data: dict) -> dict: proc = self.preprocess_sentence(sentence_data) if proc is None: return {"error": "無法處理輸入數據", "sentence_id": sentence_data.get("sentence_id", "unknown")} inp = { "input_ids": proc["input_ids"].unsqueeze(0).to(self.device), "attention_mask": proc["attention_mask"].unsqueeze(0).to(self.device), "word_pos_ids": proc["word_pos_ids"].unsqueeze(0).to(self.device), "word_grammar_ids": proc["word_grammar_ids"].unsqueeze(0).to(self.device), "word_durations": proc["word_durations"].unsqueeze(0).to(self.device), "prosody_features": proc["prosody_features"].unsqueeze(0).to(self.device), } with torch.no_grad(): out = self.model(**inp) logits = out["logits"] probs = F.softmax(logits, dim=1).cpu().numpy()[0] pred_id = int(np.argmax(probs)) sev = out["severity_pred"].cpu().numpy()[0] flu = float(out["fluency_pred"].cpu().numpy()[0][0]) pred_type = self.id_to_aphasia_type[pred_id] conf = float(probs[pred_id]) dist = {} for a_type, t_id in self.aphasia_types_mapping.items(): dist[a_type] = {"probability": float(probs[t_id]), "percentage": f"{probs[t_id]*100:.2f}%"} sorted_dist = dict(sorted(dist.items(), key=lambda x: x[1]["probability"], reverse=True)) return { "sentence_id": proc["sentence_id"], "input_text": proc["text"], "original_tokens": proc["original_tokens"], "prediction": { "predicted_class": pred_type, "confidence": conf, "confidence_percentage": f"{conf*100:.2f}%" }, "class_description": self.aphasia_descriptions.get(pred_type, { "name": pred_type, "description": "Description not available", "features": [] }), "probability_distribution": sorted_dist, "additional_predictions": { "severity_distribution": { "level_0": float(sev[0]), "level_1": float(sev[1]), "level_2": float(sev[2]), "level_3": float(sev[3]) }, "predicted_severity_level": int(np.argmax(sev)), "fluency_score": flu, "fluency_rating": "High" if flu > 0.7 else ("Medium" if flu > 0.4 else "Low"), } } def predict_batch(self, input_file: str, output_file: Optional[str] = None) -> Dict: with open(input_file, "r", encoding="utf-8") as f: data = json.load(f) sentences = data.get("sentences", []) results = [] print(f"開始處理 {len(sentences)} 個句子...") for i, s in enumerate(sentences): print(f"處理第 {i+1}/{len(sentences)} 個句子...") results.append(self.predict_single(s)) summary = self._generate_summary(results) final = {"summary": summary, "total_sentences": len(results), "predictions": results} if output_file: with open(output_file, "w", encoding="utf-8") as f: json.dump(final, f, ensure_ascii=False, indent=2) print(f"結果已保存到: {output_file}") return final def _generate_summary(self, results: List[dict]) -> dict: if not results: return {} class_counts = defaultdict(int) confs, flus = [], [] sev_counts = defaultdict(int) for r in results: if "error" in r: continue c = r["prediction"]["predicted_class"] class_counts[c] += 1 confs.append(r["prediction"]["confidence"]) flus.append(r["additional_predictions"]["fluency_score"]) sev_counts[r["additional_predictions"]["predicted_severity_level"]] += 1 avg_conf = float(np.mean(confs)) if confs else 0.0 avg_flu = float(np.mean(flus)) if flus else 0.0 return { "classification_distribution": dict(class_counts), "classification_percentages": {k: f"{v/len(results)*100:.1f}%" for k, v in class_counts.items()}, "average_confidence": f"{avg_conf:.3f}", "average_fluency_score": f"{avg_flu:.3f}", "severity_distribution": dict(sev_counts), "confidence_statistics": {} if not confs else { "mean": f"{np.mean(confs):.3f}", "std": f"{np.std(confs):.3f}", "min": f"{np.min(confs):.3f}", "max": f"{np.max(confs):.3f}", }, "most_common_prediction": max(class_counts.items(), key=lambda x: x[1])[0] if class_counts else "None", } def generate_detailed_report(self, results: List[dict], output_dir: str = "./inference_results"): os.makedirs(output_dir, exist_ok=True) rows = [] for r in results: if "error" in r: continue row = { "sentence_id": r["sentence_id"], "predicted_class": r["prediction"]["predicted_class"], "confidence": r["prediction"]["confidence"], "class_name": r["class_description"]["name"], "severity_level": r["additional_predictions"]["predicted_severity_level"], "fluency_score": r["additional_predictions"]["fluency_score"], "fluency_rating": r["additional_predictions"]["fluency_rating"], "input_text": r["input_text"], } for a_type, info in r["probability_distribution"].items(): row[f"prob_{a_type}"] = info["probability"] rows.append(row) if not rows: return None df = pd.DataFrame(rows) df.to_csv(os.path.join(output_dir, "detailed_predictions.csv"), index=False, encoding="utf-8") summary_stats = { "total_predictions": int(len(rows)), "class_distribution": df["predicted_class"].value_counts().to_dict(), "average_confidence": float(df["confidence"].mean()), "confidence_std": float(df["confidence"].std()), "average_fluency": float(df["fluency_score"].mean()), "fluency_std": float(df["fluency_score"].std()), "severity_distribution": df["severity_level"].value_counts().to_dict(), } with open(os.path.join(output_dir, "summary_statistics.json"), "w", encoding="utf-8") as f: json.dump(summary_stats, f, ensure_ascii=False, indent=2) print(f"詳細報告已生成並保存到: {output_dir}") return df # ========================= # Convenience: run directly or from pipeline # ========================= def predict_from_chajson(model_dir: str, chajson_path: str, output_file: Optional[str] = None) -> Dict: """ Convenience entry: - Accepts the JSON produced by cha_json.py - If it contains 'sentences', runs per-sentence like before - If it only contains 'text_all', creates a single pseudo-sentence """ with open(chajson_path, "r", encoding="utf-8") as f: data = json.load(f) inf = AphasiaInferenceSystem(model_dir) # If there are sentences, use the full path if data.get("sentences"): return inf.predict_batch(chajson_path, output_file=output_file) # Else, fall back to a single synthetic sentence using text_all text_all = data.get("text_all", "") fake = { "sentences": [{ "sentence_id": "S1", "dialogues": [{ "INV": [], "PAR": [{"tokens": text_all.split(), "word_pos_ids": [0]*len(text_all.split()), "word_grammar_ids": [[0,0,0]]*len(text_all.split()), "word_durations": [0.0]*len(text_all.split())}] }] }] } tmp_path = chajson_path + "._synthetic.json" with open(tmp_path, "w", encoding="utf-8") as f: json.dump(fake, f, ensure_ascii=False, indent=2) out = inf.predict_batch(tmp_path, output_file=output_file) try: os.remove(tmp_path) except Exception: pass return out def format_result(pred: dict, style: str = "json") -> str: """Back-compat formatter. 'pred' is the dict returned by predict_*.""" if style == "json": return json.dumps(pred, ensure_ascii=False, indent=2) # simple text summary if isinstance(pred, dict) and "summary" in pred: s = pred["summary"] lines = [ f"Total sentences: {pred.get('total_sentences', 0)}", f"Avg confidence: {s.get('average_confidence', 'N/A')}", f"Avg fluency: {s.get('average_fluency_score', 'N/A')}", f"Most common: {s.get('most_common_prediction', 'N/A')}", ] return "\n".join(lines) return str(pred) # ---------- CLI ---------- def main(): import argparse p = argparse.ArgumentParser(description="失語症分類推理系統") p.add_argument("--model_dir", type=str, required=False, default="./adaptive_aphasia_model", help="訓練好的模型目錄路徑") p.add_argument("--input_file", type=str, required=True, help="輸入JSON文件（cha_json 的輸出）") p.add_argument("--output_file", type=str, default="./aphasia_predictions.json", help="輸出JSON文件路徑") p.add_argument("--report_dir", type=str, default="./inference_results", help="詳細報告輸出目錄") p.add_argument("--generate_report", action="store_true", help="是否生成詳細的CSV報告") args = p.parse_args() try: print("正在初始化推理系統...") sys = AphasiaInferenceSystem(args.model_dir) print("開始執行批次預測...") results = sys.predict_batch(args.input_file, args.output_file) if args.generate_report: print("生成詳細報告...") sys.generate_detailed_report(results["predictions"], args.report_dir) print("\n=== 預測摘要 ===") s = results["summary"] print(f"總句子數: {results['total_sentences']}") print(f"平均信心度: {s.get('average_confidence', 'N/A')}") print(f"平均流利度: {s.get('average_fluency_score', 'N/A')}") print(f"最常見預測: {s.get('most_common_prediction', 'N/A')}") print("\n類別分佈:") for name, count in s.get("classification_distribution", {}).items(): pct = s.get("classification_percentages", {}).get(name, "0%") print(f" {name}: {count} ({pct})") print(f"\n結果已保存到: {args.output_file}") except Exception as e: print(f"錯誤: {str(e)}") import traceback; traceback.print_exc() if __name__ == "__main__": main()