在医疗AI快速发展的今天如何高效处理海量的非结构化医疗文本数据成为技术落地的关键瓶颈。临床笔记、电子健康记录、医学文献等资料蕴含着宝贵的医疗信息但其中的专业术语、多变缩写和复杂句式让传统NLP方法力不从心。本文将完整拆解基于多模态AI的医疗术语抽取和文献分析实战方案从核心概念到代码实现帮助开发者快速构建可落地的医疗文本处理能力。1. 多模态AI在医疗文本处理中的核心价值1.1 医疗文本数据的独特挑战医疗领域文本数据具有高度的专业性和复杂性。临床文档通常包含大量的医学术语、药物名称、疾病编码、实验室指标等专业词汇同时存在大量的缩写、简写和机构特定的表达方式。这些文本数据虽然信息密度高但标准化程度低给自动化处理带来巨大挑战。传统的单一模态文本处理方法在医疗场景下表现有限无法充分理解文本背后的医学上下文和语义关联。而多模态AI通过融合文本、图像、知识图谱等多种信息源能够更准确地理解医疗文本的深层含义。1.2 多模态AI的技术优势多模态AI技术将不同类型的医疗数据进行联合分析实现信息互补。例如在分析放射学报告时可以同时考虑文本描述和对应的医学影像在处理电子健康记录时可以结合结构化数据和非结构化文本。这种融合分析能够显著提升医疗实体识别、关系抽取和语义理解的准确性。在实际应用中多模态AI模型能够识别文本中隐含的临床意图理解医学术语之间的复杂关系甚至发现潜在的诊断模式和治疗方案。这种能力对于医疗文献综述、临床决策支持和医学研究都具有重要价值。2. 环境准备与工具选型2.1 基础开发环境配置医疗文本处理项目需要稳定的Python环境和相关的深度学习框架。建议使用Python 3.8及以上版本并配置合适的虚拟环境管理工具。# 创建并激活虚拟环境 python -m venv medical_ai_env source medical_ai_env/bin/activate # Linux/Mac # medical_ai_env\Scripts\activate # Windows # 安装核心依赖 pip install torch1.9.0 pip install transformers4.20.0 pip install spacy3.4.0 pip install scikit-learn1.0.0 pip install pandas1.4.0 pip install numpy1.21.02.2 专业医疗NLP工具链针对医疗领域的特殊需求需要选择经过医学文本训练的专用模型和工具。以下是推荐的工具组合# 安装医疗专用NLP库 pip install medspacy pip install scispacy pip install biobert-pytorch # 下载医疗语言模型 python -m spacy download en_core_sci_md python -m spacy download en_ner_bc5cdr_md2.3 多模态处理框架选择对于需要结合文本和图像的多模态任务建议使用基于Transformer的现代架构# 多模态处理框架 pip install transformers[sentencepiece] pip install torchvision pip install Pillow pip install opencv-python # 医疗多模态专用库 pip install monai pip install torchxrayvision3. 医疗术语抽取核心技术实现3.1 医疗实体识别基础模型医疗实体识别是术语抽取的基础需要识别文本中的疾病、药物、症状、检查等关键信息。以下是基于BERT的医疗NER实现import torch from transformers import AutoTokenizer, AutoModelForTokenClassification import medspacy from medspacy.ner import TargetRule from medspacy.visualization import visualize_ent class MedicalEntityRecognizer: def __init__(self, model_nameemilyalsentzer/Bio_ClinicalBERT): self.tokenizer AutoTokenizer.from_pretrained(model_name) self.model AutoModelForTokenClassification.from_pretrained(model_name) self.label_map { 0: O, 1: B-DISEASE, 2: I-DISEASE, 3: B-DRUG, 4: I-DRUG, 5: B-SYMPTOM, 6: I-SYMPTOM, 7: B-TEST, 8: I-TEST } def extract_entities(self, text): # 分词和编码 inputs self.tokenizer(text, return_tensorspt, truncationTrue, max_length512) # 模型预测 with torch.no_grad(): outputs self.model(**inputs) predictions torch.argmax(outputs.logits, dim2) # 实体解码 tokens self.tokenizer.convert_ids_to_tokens(inputs[input_ids][0]) entities [] current_entity current_label for token, prediction in zip(tokens, predictions[0]): label self.label_map[prediction.item()] if label.startswith(B-): if current_entity: entities.append((current_entity, current_label)) current_entity token current_label label[2:] elif label.startswith(I-) and current_entity: current_entity token else: if current_entity: entities.append((current_entity, current_label)) current_entity current_label return entities # 使用示例 recognizer MedicalEntityRecognizer() sample_text 患者表现为持续性咳嗽和发热胸部X光显示肺炎迹象建议使用阿莫西林治疗。 entities recognizer.extract_entities(sample_text) print(识别到的医疗实体:, entities)3.2 基于规则的后处理优化医疗文本中存在大量缩写和变体表达需要结合规则方法进行后处理import re from collections import defaultdict class MedicalTermProcessor: def __init__(self): self.abbreviation_map { CAD: 冠状动脉疾病, MI: 心肌梗死, COPD: 慢性阻塞性肺疾病, HTN: 高血压, DM: 糖尿病 } self.synonym_map { 心肌梗塞: 心肌梗死, 心梗: 心肌梗死, 高血压病: 高血压 } def normalize_terms(self, entities): normalized [] for entity, label in entities: # 处理缩写 if entity in self.abbreviation_map: entity self.abbreviation_map[entity] # 处理同义词 if entity in self.synonym_map: entity self.synonym_map[entity] # 清理特殊字符 entity re.sub(r[^\w\s], , entity) normalized.append((entity.strip(), label)) return normalized def group_by_category(self, entities): categorized defaultdict(list) for entity, label in entities: categorized[label].append(entity) return dict(categorized) # 使用示例 processor MedicalTermProcessor() normalized_entities processor.normalize_terms(entities) categorized_terms processor.group_by_category(normalized_entities) print(标准化后的实体:, normalized_entities) print(按类别分组:, categorized_terms)4. 文献综述信息抽取实战4.1 医学文献结构解析医学文献通常具有标准化的结构包括摘要、引言、方法、结果、讨论等部分。针对不同部分需要采用不同的抽取策略import re from typing import Dict, List class LiteratureParser: def __init__(self): self.section_patterns { abstract: rabstract|摘要, introduction: rintroduction|引言|背景, methods: rmethods|方法|材料与方法, results: rresults|结果, discussion: rdiscussion|讨论, conclusion: rconclusion|结论 } def parse_sections(self, text: str) - Dict[str, str]: sections {} lines text.split(\n) current_section header current_content [] for line in lines: line line.strip() if not line: continue # 检测章节标题 section_found False for section_name, pattern in self.section_patterns.items(): if re.search(pattern, line.lower()): if current_content: sections[current_section] .join(current_content) current_section section_name current_content [] section_found True break if not section_found: current_content.append(line) # 处理最后一部分 if current_content: sections[current_section] .join(current_content) return sections def extract_key_claims(self, text: str) - List[str]: 从文献中提取主要结论 claims [] # 匹配结论性语句模式 patterns [ r结果表明[^。]。, r我们的研究发现[^。]。, r结论显示[^。]。, r[^。]*显著[^。]*。, r[^。]*相关[^。]*。 ] for pattern in patterns: matches re.findall(pattern, text) claims.extend(matches) return claims # 使用示例 parser LiteratureParser() sample_literature 摘要本研究旨在评估新型抗病毒药物治疗COVID-19的效果。 方法采用随机对照试验设计纳入200例患者。 结果治疗组有效率显著高于对照组(P0.05)。 讨论该药物显示出良好的临床应用前景。 sections parser.parse_sections(sample_literature) claims parser.extract_key_claims(sections.get(results, )) print(文献章节解析:, sections) print(主要研究结论:, claims)4.2 多模态文献信息融合对于包含图表数据的医学文献需要结合文本和视觉信息进行综合分析import cv2 import pytesseract from PIL import Image import numpy as np class MultimodalLiteratureAnalyzer: def __init__(self): self.text_analyzer LiteratureParser() self.entity_recognizer MedicalEntityRecognizer() def extract_text_from_image(self, image_path: str) - str: 从文献图表中提取文本信息 try: image cv2.imread(image_path) gray cv2.cvtColor(image, cv2.COLOR_BGR2GRAY) text pytesseract.image_to_string(gray, langengchi_sim) return text except Exception as e: print(f图像文本提取失败: {e}) return def analyze_table_data(self, image_path: str) - Dict: 分析文献中的表格数据 # 简单的表格检测和解析 image cv2.imread(image_path) gray cv2.cvtColor(image, cv2.COLOR_BGR2GRAY) # 边缘检测 edges cv2.Canny(gray, 50, 150) # 查找表格轮廓 contours, _ cv2.findContours(edges, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) table_data { has_table: len(contours) 0, contour_count: len(contours), extracted_text: self.extract_text_from_image(image_path) } return table_data def integrated_analysis(self, text_content: str, image_paths: List[str]) - Dict: 综合文本和图像的多模态分析 analysis_result { text_analysis: self.text_analyzer.parse_sections(text_content), image_analysis: [], integrated_entities: [] } # 分析文本中的实体 text_entities self.entity_recognizer.extract_entities(text_content) analysis_result[integrated_entities].extend(text_entities) # 分析图像内容 for img_path in image_paths: img_analysis self.analyze_table_data(img_path) analysis_result[image_analysis].append(img_analysis) # 从图像文本中提取实体 if img_analysis[extracted_text]: img_entities self.entity_recognizer.extract_entities( img_analysis[extracted_text]) analysis_result[integrated_entities].extend(img_entities) return analysis_result # 使用示例需要实际图像文件 analyzer MultimodalLiteratureAnalyzer() # 假设有文献文本和对应的图表文件 analysis_result analyzer.integrated_analysis(sample_literature, []) print(多模态分析结果:, analysis_result)5. 医疗术语关系抽取与知识图谱构建5.1 实体关系识别医疗术语之间的关系抽取对于构建知识图谱至关重要import networkx as nx from transformers import pipeline class RelationExtractor: def __init__(self): self.relation_classifier pipeline( text-classification, modelbvanaken/clinical-relation-extraction ) self.relation_patterns { treatment: [r治疗, r用药, r处方], symptom: [r表现为, r症状, r体征], cause: [r导致, r引起, r因为], diagnosis: [r诊断, r确诊, r判断为] } def extract_relations(self, text: str, entities: List) - List: 从文本中提取实体间关系 relations [] # 基于规则的关系提取 for i, (ent1, label1) in enumerate(entities): for j, (ent2, label2) in enumerate(entities): if i j: continue # 查找两个实体之间的文本片段 ent1_start text.find(ent1) ent2_start text.find(ent2) if ent1_start -1 or ent2_start -1: continue # 确定上下文窗口 start min(ent1_start, ent2_start) end max(ent1_start len(ent1), ent2_start len(ent2)) context text[max(0, start-50):min(len(text), end50)] # 基于模式匹配的关系分类 relation_type self.classify_relation_by_pattern(context) if relation_type: relations.append({ subject: ent1, object: ent2, relation: relation_type, context: context }) return relations def classify_relation_by_pattern(self, text: str) - str: 基于模式匹配的关系分类 for relation_type, patterns in self.relation_patterns.items(): for pattern in patterns: if re.search(pattern, text): return relation_type return None # 使用示例 relation_extractor RelationExtractor() sample_clinical_text 患者因高血压导致头痛使用硝苯地平治疗后症状缓解。 entities recognizer.extract_entities(sample_clinical_text) relations relation_extractor.extract_relations(sample_clinical_text, entities) print(提取的实体关系:, relations)5.2 医疗知识图谱构建基于抽取的实体和关系构建医疗知识图谱class MedicalKnowledgeGraph: def __init__(self): self.graph nx.DiGraph() self.entity_counter 0 def add_entity(self, entity: str, label: str) - int: 添加实体到知识图谱 if entity not in self.graph: self.graph.add_node(entity, labellabel, idself.entity_counter) self.entity_counter 1 return self.entity_counter - 1 def add_relation(self, subject: str, object: str, relation: str, context: str ): 添加关系到知识图谱 if subject not in self.graph: self.add_entity(subject, UNKNOWN) if object not in self.graph: self.add_entity(object, UNKNOWN) self.graph.add_edge(subject, object, relationrelation, contextcontext) def build_from_extractions(self, entities: List, relations: List): 从抽取结果构建知识图谱 for entity, label in entities: self.add_entity(entity, label) for relation in relations: self.add_relation( relation[subject], relation[object], relation[relation], relation[context] ) def query_relations(self, entity: str, relation_type: str None) - List: 查询特定实体的关系 results [] if entity not in self.graph: return results for neighbor in self.graph.neighbors(entity): edge_data self.graph.get_edge_data(entity, neighbor) if relation_type is None or edge_data[relation] relation_type: results.append({ target: neighbor, relation: edge_data[relation], context: edge_data.get(context, ) }) return results def visualize_subgraph(self, central_entity: str, depth: int 2): 可视化以特定实体为中心的子图 if central_entity not in self.graph: print(f实体 {central_entity} 不存在于知识图谱中) return # 获取指定深度的子图 subgraph_nodes set([central_entity]) current_depth_nodes [central_entity] for _ in range(depth): next_depth_nodes [] for node in current_depth_nodes: neighbors list(self.graph.neighbors(node)) \ list(self.graph.predecessors(node)) for neighbor in neighbors: if neighbor not in subgraph_nodes: subgraph_nodes.add(neighbor) next_depth_nodes.append(neighbor) current_depth_nodes next_depth_nodes subgraph self.graph.subgraph(subgraph_nodes) print(f\n以 {central_entity} 为中心的知识子图 (深度{depth}):) for node in subgraph.nodes(): print(f实体: {node} (类型: {subgraph.nodes[node].get(label, UNKNOWN)})) relations self.query_relations(node) for rel in relations: print(f - {rel[relation]} - {rel[target]}) # 使用示例 kg MedicalKnowledgeGraph() kg.build_from_extractions(entities, relations) # 查询特定实体的关系 hypertension_relations kg.query_relations(高血压) print(高血压相关关系:, hypertension_relations) # 可视化知识子图 kg.visualize_subgraph(高血压, depth2)6. 性能优化与大规模处理6.1 批量处理与并行计算医疗文献处理通常涉及大量数据需要优化处理性能import concurrent.futures from tqdm import tqdm import pandas as pd class BatchProcessor: def __init__(self, max_workers: int 4): self.max_workers max_workers self.entity_recognizer MedicalEntityRecognizer() self.relation_extractor RelationExtractor() def process_single_document(self, document: Dict) - Dict: 处理单个文档 try: text document.get(text, ) entities self.entity_recognizer.extract_entities(text) relations self.relation_extractor.extract_relations(text, entities) return { doc_id: document.get(id), entities: entities, relations: relations, entity_count: len(entities), relation_count: len(relations), status: success } except Exception as e: return { doc_id: document.get(id), entities: [], relations: [], error: str(e), status: failed } def process_batch(self, documents: List[Dict]) - pd.DataFrame: 批量处理文档 results [] with concurrent.futures.ThreadPoolExecutor(max_workersself.max_workers) as executor: future_to_doc { executor.submit(self.process_single_document, doc): doc for doc in documents } for future in tqdm(concurrent.futures.as_completed(future_to_doc), totallen(documents), desc处理进度): result future.result() results.append(result) return pd.DataFrame(results) def analyze_batch_results(self, results_df: pd.DataFrame) - Dict: 分析批量处理结果 success_count len(results_df[results_df[status] success]) failed_count len(results_df[results_df[status] failed]) total_entities results_df[entity_count].sum() total_relations results_df[relation_count].sum() avg_entities_per_doc total_entities / success_count if success_count 0 else 0 avg_relations_per_doc total_relations / success_count if success_count 0 else 0 return { total_documents: len(results_df), successful_documents: success_count, failed_documents: failed_count, success_rate: success_count / len(results_df) * 100, total_entities_extracted: total_entities, total_relations_extracted: total_relations, average_entities_per_doc: avg_entities_per_doc, average_relations_per_doc: avg_relations_per_doc } # 使用示例 documents [ {id: doc1, text: 患者高血压伴有头痛使用降压药物治疗。}, {id: doc2, text: 糖尿病患者的血糖控制需要胰岛素治疗。}, {id: doc3, text: 冠心病患者需要定期进行心电图检查。} ] batch_processor BatchProcessor(max_workers2) results_df batch_processor.process_batch(documents) analysis batch_processor.analyze_batch_results(results_df) print(批量处理结果分析:, analysis) print(详细结果:) print(results_df)6.2 内存优化与模型缓存处理大规模医疗文本时需要注意内存使用优化import gc import psutil import os from transformers import AutoModel, AutoTokenizer class OptimizedMedicalProcessor: def __init__(self, model_path: str None): self.model_path model_path self.model None self.tokenizer None self._load_model() def _load_model(self): 延迟加载模型以节省内存 if self.model is None: print(正在加载医疗NLP模型...) self.tokenizer AutoTokenizer.from_pretrained( self.model_path or emilyalsentzer/Bio_ClinicalBERT) self.model AutoModel.from_pretrained( self.model_path or emilyalsentzer/Bio_ClinicalBERT) def _cleanup_memory(self): 清理内存 if self.model is not None: del self.model self.model None gc.collect() def process_large_corpus(self, corpus: List[str], batch_size: int 32) - List: 处理大规模语料库 results [] for i in range(0, len(corpus), batch_size): batch corpus[i:ibatch_size] batch_results self._process_batch(batch) results.extend(batch_results) # 监控内存使用 memory_usage psutil.Process(os.getpid()).memory_info().rss / 1024 / 1024 print(f处理批次 {i//batch_size 1}, 内存使用: {memory_usage:.2f} MB) # 定期清理内存 if (i // batch_size) % 10 0: self._cleanup_memory() self._load_model() # 重新加载模型 return results def _process_batch(self, batch: List[str]) - List: 处理单个批次 self._load_model() # 确保模型已加载 batch_results [] for text in batch: try: # 简化的处理逻辑 inputs self.tokenizer(text, return_tensorspt, truncationTrue, max_length512) outputs self.model(**inputs) # 实际应用中这里会有更复杂的处理逻辑 batch_results.append({text: text, status: processed}) except Exception as e: batch_results.append({text: text, status: error, error: str(e)}) return batch_results # 使用示例 optimized_processor OptimizedMedicalProcessor() large_corpus [医疗文本示例] * 100 # 模拟大规模语料 results optimized_processor.process_large_corpus(large_corpus, batch_size10) print(f处理完成共处理 {len(results)} 个文档)7. 常见问题与解决方案7.1 医疗术语识别准确率提升医疗术语识别中的常见问题包括缩写识别、同义词处理和边界检测class TermRecognitionOptimizer: def __init__(self): self.common_abbreviations self._load_medical_abbreviations() self.symptom_patterns self._load_symptom_patterns() def _load_medical_abbreviations(self) - Dict: 加载医疗缩写词典 return { CAD: 冠状动脉疾病, MI: 心肌梗死, CHF: 充血性心力衰竭, COPD: 慢性阻塞性肺疾病, DM: 糖尿病, HTN: 高血压, CVA: 脑血管意外, PUD: 消化性溃疡疾病 } def _load_symptom_patterns(self) - List: 加载症状描述模式 return [ r[患表表现现为有]现[^\u4e00-\u9fa5]{0,10}[症状体征不适], r主诉[^]{0,20}[疼痛咳嗽发热], r[伴随合并]有[^]{0,15}[症状] ] def enhance_entity_recognition(self, text: str, raw_entities: List) - List: 增强实体识别结果 enhanced_entities [] # 处理缩写 for entity, label in raw_entities: enhanced_entity self._resolve_abbreviations(entity) if enhanced_entity ! entity: enhanced_entities.append((enhanced_entity, label)) enhanced_entities.append((entity, label)) # 基于模式的实体补充 pattern_entities self._extract_pattern_based_entities(text) enhanced_entities.extend(pattern_entities) return list(set(enhanced_entities)) # 去重 def _resolve_abbreviations(self, entity: str) - str: 解析医疗缩写 return self.common_abbreviations.get(entity.upper(), entity) def _extract_pattern_based_entities(self, text: str) - List: 基于模式提取实体 entities [] for pattern in self.symptom_patterns: matches re.finditer(pattern, text) for match in matches: entities.append((match.group(), SYMPTOM)) return entities # 使用示例 optimizer TermRecognitionOptimizer() sample_text 患者CAD病史主诉胸痛伴有呼吸困难。 raw_entities recognizer.extract_entities(sample_text) enhanced_entities optimizer.enhance_entity_recognition(sample_text, raw_entities) print(原始实体:, raw_entities) print(增强后实体:, enhanced_entities)7.2 多模态数据对齐问题处理文本和图像数据时的对齐挑战class MultimodalAlignment: def __init__(self): self.reference_patterns { figure: r图\s*\d[:]\s*[^。], table: r表\s*\d[:]\s*[^。], section: r\d\.\s*\d\s*[^。] } def align_text_with_figures(self, text: str, figure_captions: List[str]) - Dict: 对齐文本内容和图表 alignment {} # 提取文本中的图表引用 for ref_type, pattern in self.reference_patterns.items(): references re.findall(pattern, text) alignment[ref_type] references # 简单的基于顺序的对齐 for i, caption in enumerate(figure_captions): alignment[ffigure_{i1}] { caption: caption, references: self._find_references_to_figure(text, i1), position_in_text: self._find_figure_position(text, i1) } return alignment def _find_references_to_figure(self, text: str, figure_num: int) - List: 查找对特定图表的引用 pattern fr图\s*{figure_num}[^\\d] return re.findall(pattern, text) def _find_figure_position(self, text: str, figure_num: int) - int: 查找图表在文本中的位置 pattern fr图\s*{figure_num} match re.search(pattern, text) return match.start() if match else -1 # 使用示例 alignment_tool MultimodalAlignment() text_content 如图1所示患者心电图显示异常。图2展示了治疗方案对比。 figure_captions [图1患者心电图, 图2治疗方案对比] alignment_result alignment_tool.align_text_with_figures(text_content, figure_captions) print(多模态对齐结果:, alignment_result)8. 生产环境部署最佳实践8.1 模型服务化部署将医疗文本处理能力封装为API服务from flask import Flask, request, jsonify import logging from healthcheck import HealthCheck app Flask(__name__) health HealthCheck() # 设置日志 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class MedicalTextService: def __init__(self): self.entity_recognizer MedicalEntityRecognizer() self.relation_extractor RelationExtractor() self.kg_builder MedicalKnowledgeGraph() def process_medical_text(self, text: str) - Dict: 处理医疗文本的完整流程 try: # 实体识别 entities self.entity_recognizer.extract_entities(text) # 关系抽取 relations self.relation_extractor.extract_relations(text, entities) # 知识图谱构建 self.kg_builder.build_from_extractions(entities, relations) return { status: success, entities: entities, relations: relations, entity_count: len(entities), relation_count: len(relations) } except Exception as e: logger.error(f文本处理失败: {e}) return {status: error, message: str(e)} # 初始化服务 medical_service MedicalTextService() # 健康检查端点 health.add_check(lambda: True, basic_health_check) app.add_url_rule(/health, healthcheck, view_funclambda: health.run()) app.route(/api/process, methods[POST]) def process_text(): 处理医疗文本的API端点 try: data request.get_json() text data.get(text, ) if not text: return jsonify({error: 缺少文本内容}), 400 result medical_service.process_medical_text(text) return jsonify(result) except Exception as e: logger.error(fAPI处理错误: {e}) return jsonify({error: 内部服务器错误}), 500 app.route(/api/entities/entity_name, methods[GET]) def get_entity_relations(entity_name: str): 查询实体关系的API端点 try: relations medical_service.kg_builder.query_relations(entity_name) return jsonify({ entity: entity_name, relations: relations }) except Exception as e: logger.error(f实体查询错误: {e}) return jsonify({error: 查询失败}), 500 if __name__ __main__: app.run(host0.0.0.0, port5000, debugFalse)8.2 监控与性能优化生产环境中的监控和性能保障import time from prometheus_client import Counter, Histogram, generate_latest from flask import Response # 定义监控指标 REQUEST_COUNT Counter(medical_api_requests_total, Total API requests, [endpoint, status]) REQUEST_DURATION Histogram(medical_api_request_duration_seconds, API request duration, [endpoint]) app.before_request def before_request(): request.start_time time.time() app.after_request def after_request(response): # 记录请求指标 endpoint request.endpoint or unknown status_code str(response.status_code) REQUEST_COUNT.labels(endpointendpoint, statusstatus_code).inc() if hasattr(request, start_time): duration time.time() - request.start_time REQUEST_DURATION.labels(endpointendpoint).observe(duration) return response app.route(/metrics) def metrics(): Prometheus指标端点 return Response(generate_latest(), mimetypetext/plain) class PerformanceMonitor: def __init__(self): self.performance_data {} def record_processing_time(self, operation: str, duration: float): 记录处理时间 if operation not in self.performance_data: self.performance_data[operation] [] self.performance_data[operation].append(duration) def get_performance_stats(self) - Dict: 获取性能统计 stats {} for operation, durations in self.performance_data.items(): if durations: stats[operation] { count: len(durations), avg_duration: sum(durations) / len(durations), max_duration: max(durations), min_duration: min(durations) } return stats # 使用示例 performance_monitor PerformanceMonitor() # 在关键操作处添加性能监控 def monitored_processing(text: str) - Dict: start_time time.time() result medical_service.process_medical_text(text) duration time.time() - start_time performance_monitor.record_processing_time(text_processing, duration) return result # 定期输出性能报告 import threading import time def periodic_performance_report(interval: int 300): 定期输出性能报告 while True: time.sleep(interval) stats performance_monitor.get_performance_stats() logger.info(性能统计报告:) for operation, stat in stats.items(): logger.info(f{operation}: {stat}) # 启动性能监控线程 monitor_thread threading.Thread(targetperiodic_performance_report, daemonTrue) mon