企业官网建设流程全解析

患者主索引的数据库设计：多源患者数据的去重、合并与关联

一、同名同姓的噩梦：当"张伟"在数据库里出现了17次

国内某区域医疗信息平台接入了市属8家医院的数据，建成后发现一个尴尬的问题：名叫"张伟"的患者有17条记录，分布在不同的医院系统中。这17条记录可能属于8个不同的人，也可能属于4个人——每个人在不同医院看病时被录入为不同的记录。

这就是患者主索引（Enterprise Master Patient Index, EMPI）要解决的核心问题：在多个异构系统中，识别哪些患者记录属于同一个人，并将它们关联起来。

EMP问题的难度在于"隐式匹配"。传统的主数据管理（MDM）基于唯一标识符（身份证号）做匹配，但在医疗场景中：

• 身份证号可能缺失（儿童、急诊无身份患者）
• 身份证号可能录错（16位旧号vs18位新号、最后一位校验码错误）
• 跨系统标识符完全不同（A医院用住院号、B医院用门诊号）

因此EMPI需要基于多字段的模糊匹配算法。

二、概率匹配算法：Fellegi-Sunter模型的工程落地

Fellegi-Sunter模型的底层思想是：对于每对待匹配记录(A, B)，计算它们在每个字段上"一致"或"不一致"的似然比：

总匹配权重 = Σ(字段i一致时的正权重) + Σ(字段i不一致时的负权重) 正权重 = log(P(字段i一致 | A和B是同一人) / P(字段i一致 | A和B是不同人)) 负权重 = log(P(字段i不一致 | A和B是同一人) / P(字段i不一致 | A和B是不同人))

实际工程简化。姓名用Jaro-Winkler距离（0-1），生日精确度按年/月/日分级加权，身份证号全匹配给极高权重。

三、EMPI的数据表设计与增量匹配实现

-- EMPI主表：全局统一的患者标识 CREATE TABLE empi_patient ( global_id BIGINT AUTO_INCREMENT PRIMARY KEY, empi_id CHAR(32) NOT NULL UNIQUE, -- MD5生成的全局ID master_name VARCHAR(128), -- 主姓名（审核确认） master_gender CHAR(1), master_birth DATE, master_id_card VARCHAR(32), master_phone VARCHAR(20), match_confidence DECIMAL(4,3) DEFAULT 1.000, -- 匹配置信度 record_status ENUM('ACTIVE','MERGED','SPLIT','PENDING') DEFAULT 'ACTIVE', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, INDEX idx_name (master_name), INDEX idx_idcard (master_id_card), INDEX idx_birth (master_birth) ); -- 源记录关联表：记录每个源系统中的患者标识 CREATE TABLE empi_source_record ( id BIGINT AUTO_INCREMENT PRIMARY KEY, global_id BIGINT NOT NULL, -- 关联到EMPI患者 source_system VARCHAR(32) NOT NULL, -- 来源系统：HIS_A/HIS_B source_id VARCHAR(128) NOT NULL, -- 源系统中的患者ID original_name VARCHAR(128), original_gender CHAR(1), original_birth DATE, original_id_card VARCHAR(32), original_phone VARCHAR(20), raw_data JSON, -- 原始数据全量保存 match_score DECIMAL(4,3), created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE KEY uk_source (source_system, source_id), INDEX idx_global (global_id) ); -- 匹配候选队列表：待人工审核 CREATE TABLE empi_match_queue ( id BIGINT AUTO_INCREMENT PRIMARY KEY, source_record_id BIGINT NOT NULL, candidate_global_id BIGINT NOT NULL, match_score DECIMAL(5,4) NOT NULL, field_scores JSON, -- 各字段匹配分 decision ENUM('AUTO_MERGE','MANUAL_REVIEW','REJECT') DEFAULT 'MANUAL_REVIEW', reviewer_id VARCHAR(64), review_decision ENUM('MERGE','SPLIT','REJECT'), reviewed_at TIMESTAMP, FOREIGN KEY (source_record_id) REFERENCES empi_source_record(id) );

增量匹配的核心代码：

from jellyfish import jaro_winkler_similarity import hashlib class EMPIMatcher: # 字段权重（通过历史数据训练得到） WEIGHTS = { 'id_card': 12.0, # 身份证号全匹配 'name': 5.0, # 姓名匹配 'birth': 4.0, # 生日匹配（精确到日） 'birth_year': 2.0, # 生日匹配（仅年份） 'gender': 1.0, # 性别匹配 'phone': 6.0, # 手机号匹配 'address': 3.0, # 地址匹配 } THRESHOLD_AUTO_MERGE = 15.0 # 自动合并阈值 THRESHOLD_MANUAL = 8.0 # 人工审核阈值 def match_patient(self, new_record: dict, existing_records: list) -> dict: """将新患者记录与已有EMPI记录做匹配""" best_score = 0 best_match = None field_details = {} for existing in existing_records: score = 0.0 details = {} # 身份证号精确匹配 if (new_record.get('id_card') and existing.get('id_card') and new_record['id_card'] == existing['id_card']): score += self.WEIGHTS['id_card'] details['id_card'] = 'exact_match' # 姓名模糊匹配 name_sim = jaro_winkler_similarity( str(new_record.get('name', '')), str(existing.get('name', '')) ) if name_sim > 0.85: name_score = self.WEIGHTS['name'] * name_sim score += name_score details['name'] = f'similarity:{name_sim:.3f}' # 生日匹配 if (new_record.get('birth') and existing.get('birth')): if new_record['birth'] == existing['birth']: score += self.WEIGHTS['birth'] details['birth'] = 'exact_match' elif (str(new_record['birth'])[:4] == str(existing['birth'])[:4]): score += self.WEIGHTS['birth_year'] details['birth'] = 'year_match' # 性别匹配 if new_record.get('gender') == existing.get('gender'): score += self.WEIGHTS['gender'] details['gender'] = 'match' # 手机号匹配 if (new_record.get('phone') and existing.get('phone') and new_record['phone'] == existing['phone']): score += self.WEIGHTS['phone'] details['phone'] = 'exact_match' if score > best_score: best_score = score best_match = existing field_details = details # 决策 if best_score >= self.THRESHOLD_AUTO_MERGE: decision = 'AUTO_MERGE' elif best_score >= self.THRESHOLD_MANUAL: decision = 'MANUAL_REVIEW' else: decision = 'CREATE_NEW' return { 'decision': decision, 'best_match_global_id': best_match.get('global_id') if best_match else None, 'score': best_score, 'field_scores': field_details } def create_or_merge(self, source_record: dict, match_result: dict): """执行创建或合并操作""" if match_result['decision'] == 'CREATE_NEW': global_id = self._create_new_empi(source_record) elif match_result['decision'] == 'AUTO_MERGE': global_id = match_result['best_match_global_id'] self._link_to_empi(source_record, global_id, match_result) else: # 人工审核 global_id = None self._enqueue_review(source_record, match_result) return global_id

四、EMPI的四个边界条件与工程陷阱

边界一：新生儿和儿童的特殊性。新生儿没有身份证号、没有手机号、甚至名字都可能只是"张某之子"。匹配几乎只能靠母亲ID+出生日期+出生医院。这类患者需要特殊的匹配规则和更低的自动合并阈值。

边界二：双胞胎的误合并。同卵双胞胎的姓名可能相似、生日完全相同、地址相同、早期身份证号甚至只差最后一位。EMPI很可能会将两人错误合并。缓解方案是引入更多区分性字段（血型、过敏史等），并设置"可能为双胞胎"的标记。

边界三：已合并记录的"再拆分"。人工审核后发现之前的合并是错误的，需要将一个EMPI记录拆分为两个。这是最复杂的操作——所有关联到该global_id的检查报告、医嘱、处方都需要重新分配。技术上是"创建新global_id → 迁移部分源记录 → 同步下游系统"的回滚链。

边界四：实时性与批量的平衡。患者在挂号时需要实时匹配（<200ms），但全量数据的去重处理可以在T+1的批处理中完成。架构上需要设计"实时快速匹配（仅查身份证+手机号）→ 离线全字段模糊匹配（发现新匹配对，进入审核）"的两阶段流水线。

五、总结

患者主索引（EMPI）的实现不是一次性的"全量去重"项目，而是一个持续运行的"增量匹配"引擎。Fellegi-Sunter模型提供了概率论的理论基础，但工程落地的关键在于：分块策略减少候选对数量、多级阈值分流自动化与人工审核、Schema设计中保留足够的审计字段。

EMPI的正确率直接影响所有下游业务——临床决策支持需要聚合患者全部就诊记录，科研统计需要准确的去重基数，医保结算更是对患者身份的精确性有硬性要求。在这个意义上，EMPI不是数据库优化的副产品，而是医疗数据中台的承重墙。

本文属于「行业场景与项目复盘」系列，深入解析医疗场景下患者主索引的概率匹配算法与数据库设计实践。