Python实现医疗病案主诊断编码AI辅助系统

1. 用Python实现辅助病案首页主诊断编码：从数据清洗到模型上线（上）

作为一名在医疗信息化领域摸爬滚打多年的技术老兵，我深知病案首页编码这个看似简单的工作背后藏着多少"坑"。记得刚接手某三甲医院编码系统改造时，亲眼目睹编码员对着厚如砖头的ICD-10手册反复翻查，一份出院记录平均要花20分钟才能完成编码。更可怕的是，不同编码员对同一份病历给出的主诊断编码差异率高达15%，直接影响到DRGs分组和医保结算。

这就是为什么我们要用Python打造这个AI辅助编码系统——不是要取代人工，而是要把编码员从繁琐的机械劳动中解放出来，让他们专注于需要专业判断的复杂病例。下面分享的这套方法论，是我们团队经过三年迭代，在6家不同等级医院验证过的实战方案。

1.1 环境准备：5分钟搭建可复现环境

医疗AI项目最怕"在我机器上能跑"的尴尬。经过多次教训，我们现在所有项目都采用容器化环境管理。以下是经过优化的配置方案：

bash复制# 创建项目目录结构（医院IT系统通常有严格的目录规范）
mkdir -p icd_assistant/{data/raw,data/processed,models,notebooks,src/{preprocess,model,eval}}

推荐使用conda创建专属环境（比virtualenv更适合处理医疗影像相关的依赖）：

bash复制conda create -n icd python=3.8 -y
conda activate icd

# 核心依赖（经过多家医院生产环境验证的稳定版本）
pip install pandas==1.3.5 scikit-learn==1.0.2 spacy==3.4.1 
pip install torch==1.12.1+cu113 -f https://download.pytorch.org/whl/torch_stable.html

# 医疗文本处理专用包
pip install medspacy==1.0.1 pyhealth==0.1.6

关键细节：必须锁定版本！我们曾因scikit-learn自动升级到1.2.0导致特征提取结果不一致，差点引发医保审计问题。

安装后执行以下验证脚本确保环境正确：

python复制import torch
print(f"PyTorch可用GPU：{torch.cuda.is_available()}")

import medspacy
nlp = medspacy.load()
print(f"MedSpacy组件加载：{nlp.pipe_names}")

1.2 数据准备与清洗策略

1.2.1 原始数据解析

医疗数据通常来自以下几个系统，需要特殊处理：

1. HIS系统导出数据

   • 格式：通常为CSV或DBF
   • 问题：字符编码混乱（常见GB2312/GBK混用）
   • 解决方案：

     python复制def detect_encoding(file_path):
         with open(file_path, 'rb') as f:
             raw = f.read(10000)
             return chardet.detect(raw)['encoding']
     
     his_data = pd.read_csv('his_export.csv', 
                          encoding=detect_encoding('his_export.csv'),
                          engine='python')
     

2. EMR文本数据

   • 格式：非结构化文本（含HTML/RTF格式）
   • 问题：关键信息隐藏在自由文本中
   • 解决方案：使用医疗专用NLP管道

     python复制clinical_nlp = medspacy.load(enable=["sectionizer"])
     doc = clinical_nlp("主诉：反复胸痛3年，加重1周。现病史：患者...")
     print([section.text for section in doc._.sections])
     

1.2.2 文本清洗标准化

医疗文本清洗需要特别注意保留临床语义：

python复制def clean_medical_text(text):
    # 保留关键标点（如"3.5kg"中的点）
    text = re.sub(r'(?<!\d)[.,;:](?!\d)', ' ', text)
    
    # 处理特殊单位符号
    text = re.sub(r'°C', '摄氏度', text)
    
    # 标准化药物剂量表达
    text = re.sub(r'(\d+)\s*([mg])\s*', r'\1\2 ', text)
    
    # 保护医学缩写（如"T:36.5℃"中的T）
    protected_abbrs = ['T', 'P', 'R', 'BP']
    for abbr in protected_abbrs:
        text = re.sub(fr'\b{abbr}\s*[:=]', f'{abbr}:', text)
    
    return text

踩坑记录：曾因过度清洗导致"Ⅱ型糖尿病"被转成"II型糖尿病"，ICD编码完全错误！

1.2.3 敏感信息脱敏处理

医疗数据脱敏不是简单替换，要考虑后续NLP处理需求：

python复制class MedicalDeidentifier:
    def __init__(self):
        self.patient_id_map = {}
        self.phone_pattern = re.compile(r'(\d{3})-(\d{4})-(\d{4})')
        
    def deidentify(self, text):
        # 保护性替换病历号
        text = re.sub(r'病历号\s*[:：]\s*(\w+)', 
                     lambda m: f'病历号：PAT{len(self.patient_id_map)+1:04d}',
                     text)
        
        # 处理电话号码（保留部分信息用于地域分析）
        text = self.phone_pattern.sub(r'**\1**-****-\3', text)
        
        return text

1.3 医疗实体识别与标准化

1.3.1 构建领域词典

我们采用混合策略提升召回率：

python复制def build_medical_lexicon():
    # 从权威来源加载基础术语
    icd10_terms = pd.read_csv('icd10_terms.csv') 
    drug_lexicon = set(pd.read_excel('drug_dictionary.xlsx')['药品名称'])
    
    # 补充医院特色术语
    with open('hospital_slang.txt', 'r', encoding='utf-8') as f:
        local_terms = {line.strip() for line in f}
    
    # 生成同义词映射
    synonym_map = {
        '心梗': '心肌梗死',
        'DM': '糖尿病',
        # ...其他映射
    }
    
    return {
        'standard_terms': icd10_terms,
        'drug_names': drug_lexicon,
        'local_terms': local_terms,
        'synonyms': synonym_map
    }

1.3.2 增强型实体识别

结合规则和模型提升准确率：

python复制def enhanced_ner(text, lexicon):
    # 规则匹配优先
    rule_matches = []
    for term in lexicon['standard_terms']:
        if term in text:
            rule_matches.append((term, 'ICD_STANDARD'))
    
    # 模型预测
    doc = nlp(text)
    model_matches = [(ent.text, ent.label_) for ent in doc.ents]
    
    # 冲突解决（规则优先）
    all_matches = rule_matches + [
        m for m in model_matches 
        if not any(m[0] in rm[0] for rm in rule_matches)
    ]
    
    return all_matches

1.4 诊断文本结构化处理

1.4.1 关键信息抽取模板

针对不同科室设计定制化模板：

python复制cardiac_template = {
    'symptom': r'(主诉|症状)[:：](.*?)(?=\n\w+[:：]|$)',
    'diagnosis': {
        'pattern': r'(初步诊断|出院诊断)[:：](.*?)(?=\n\w+[:：]|$)',
        'cleaner': lambda s: re.sub(r'[①②③]', '', s)
    },
    'procedure': r'(手术操作)[:：](.*?)(?=\n\w+[:：]|$)'
}

def extract_by_template(text, template):
    results = {}
    for field, pattern in template.items():
        if isinstance(pattern, dict):
            match = re.search(pattern['pattern'], text, re.DOTALL)
            if match:
                cleaned = pattern['cleaner'](match.group(2))
                results[field] = cleaned.strip()
        else:
            match = re.search(pattern, text, re.DOTALL)
            if match:
                results[field] = match.group(2).strip()
    return results

1.4.2 诊断优先级判定算法

基于临床规则的特征提取：

python复制def diagnose_priority(features):
    """
    根据临床特征判断主诊断优先级
    特征包括：
    - 是否手术相关诊断
    - 是否入院主要原因
    - 是否消耗最多资源
    - 是否导致住院时间延长
    """
    score = 0
    if features['is_primary_reason']:
        score += 3
    if features['is_procedure_related']:
        score += 2
    if features['resource_usage'] == 'high':
        score += 2
    if features['los_impact'] == 'significant':
        score += 1
        
    return score

1.5 数据增强与样本平衡

1.5.1 医疗文本语义保持增强

不同于通用NLP，医疗数据增强必须保证临床准确性：

python复制class MedicalAugmenter:
    def __init__(self, terminology_db):
        self.term_db = terminology_db
        
    def synonym_replacement(self, text):
        for term, syns in self.term_db['synonyms'].items():
            if term in text:
                text = text.replace(term, random.choice(syns))
        return text
    
    def context_aware_insert(self, text, entity_type):
        """在适当位置插入相关临床实体"""
        insertion_points = [m.end() for m in re.finditer(r'[.,;]', text)]
        if not insertion_points:
            return text
            
        insert_pos = random.choice(insertion_points)
        candidate_terms = [
            t for t in self.term_db[entity_type] 
            if t not in text
        ]
        if not candidate_terms:
            return text
            
        return (text[:insert_pos] + ' ' + random.choice(candidate_terms) + 
                text[insert_pos:])

1.5.2 罕见病种过采样策略

python复制def rare_disease_oversampling(df, target_col, threshold=50):
    # 计算每个类别的样本量
    class_counts = df[target_col].value_counts()
    
    # 识别罕见病种
    rare_classes = class_counts[class_counts < threshold].index.tolist()
    
    # 对每个罕见类别进行过采样
    augmented_samples = []
    for rare_class in rare_classes:
        class_samples = df[df[target_col] == rare_class]
        num_needed = threshold - len(class_samples)
        
        # 使用SMOTE生成新样本
        smote = SMOTE(sampling_strategy={rare_class: threshold}, 
                      k_neighbors=min(3, len(class_samples)-1))
        X_res, _ = smote.fit_resample(
            class_samples.drop(columns=[target_col]),
            class_samples[target_col]
        )
        
        augmented_samples.append(X_res.tail(num_needed))
    
    return pd.concat([df] + augmented_samples)

1.6 工程化实践要点

1.6.1 医疗数据版本控制

使用DVC管理数据和模型版本：

bash复制# 初始化DVC
dvc init
dvc add data/raw/patient_records.csv
git add data/raw/patient_records.csv.dvc .gitignore

1.6.2 自动化测试策略

医疗AI必须包含临床逻辑测试：

python复制class TestClinicalLogic(unittest.TestCase):
    def test_diagnosis_priority(self):
        test_case = {
            'is_primary_reason': True,
            'is_procedure_related': False,
            'resource_usage': 'medium',
            'los_impact': 'significant'
        }
        self.assertEqual(diagnose_priority(test_case), 4)
        
    def test_deidentification(self):
        deid = MedicalDeidentifier()
        text = "病历号：123456 电话：138-1234-5678"
        processed = deid.deidentify(text)
        self.assertNotIn("123456", processed)
        self.assertIn("PAT0001", processed)

1.6.3 性能优化技巧

处理百万级病历时的内存优化方案：

python复制def batch_process(reader, chunk_size=10000):
    for chunk in reader:
        # 使用迭代器减少内存占用
        processed = (clean_medical_text(text) for text in chunk['text'])
        yield pd.DataFrame({
            'id': chunk['id'],
            'processed_text': list(processed)
        })

# 使用示例
csv_reader = pd.read_csv('large_records.csv', chunksize=10000)
for batch in batch_process(csv_reader):
    save_to_database(batch)

经过以上步骤，我们已经构建了一个健壮的医疗数据处理流水线。在下篇中，我们将深入讲解模型选型、训练策略和部署方案，包括：

1. 如何设计适合ICD编码的层次化模型架构
2. 处理医疗文本长尾分布的特殊训练技巧
3. 符合医院IT环境的低延迟部署方案
4. 持续监控与模型迭代的最佳实践

在实现过程中最深的体会是：医疗AI项目成败的关键，往往不在于模型有多复杂，而在于对临床场景和业务规则的理解深度。我们团队花在业务需求分析上的时间，通常是编码时间的三倍以上。这也解释了为什么很多学术界的SOTA模型在实际医疗场景中表现不佳——它们缺乏对医疗业务流程的尊重和理解。