AI做真实世界研究数据整理、质控与分析效率怎么一起提真实世界研究的数据治理最耗时间的环节往往不在建模或统计而在分析之前。项目启动后研究团队通常会陆续收到 HIS/EMR 导出的结构化表、随访系统数据、检验检查明细、研究登记表、人工补录 Excel以及几轮历史批次文件。它们表面上都在描述同一批研究对象但真正合并时会出现一批反复返工的问题主键不统一、日期口径不一致、字段含义随批次变化、缺失值没有语义、同一指标存在多个单位、人工补录文件列名被改过。这类问题如果没有先在数据底座上处理清楚后续分析效率很难真正提升。AI 可以帮忙做字段匹配提示、质控报告摘要、异常模式归纳但它不能替代研究方案、数据管理计划、统计分析方案和专业人员确认。本文复盘一条偏工程治理的轻量链路用 Python、Pandas、DuckDB、Airflow 和 Great Expectations 的思路把多源数据整理、跨表质控和分析准备串起来。本文仅讨论医疗健康数据工程和研究数据治理不提供诊断、治疗、分诊、用药或风险分层建议。文中所有字段、阈值和规则均为示例不能直接用于临床判断。真实项目应以研究方案、数据管理计划、伦理审批要求、机构规范和专业人员确认结果为准。问题背景RWS 数据为什么容易卡在分析前真实世界研究常见数据来源包括业务系统导出表电子病历或结构化病案数据检验、检查、用药等明细记录随访系统或研究登记系统人工补录 Excel历史批次归档文件。这些数据的主要问题不是“没有数据”而是“数据能不能按同一个口径被解释”。项目复盘中返工点通常集中在几类问题上字段映射靠人工维护例如patient_id、pid、研究编号可能指向同一概念但不同文件批次命名不同。缺失值语义被抹平未采集、不适用、采集失败、未知被统一转成空值后后续分析很难还原原因。日期字段口径混乱登记日期、首次就诊日期、首次随访日期、检测日期混用。单位和编码不统一同一检验项目在不同来源中可能存在不同单位、不同项目代码。分析脚本直接读取原始表每次统计都要重新解释字段含义结果难以复现。跨表关系没有提前校验随访表、检验表里出现主体表不存在的研究对象到了统计阶段才发现。批次差异没有记录历史文件和新导出文件字段发生变化但缺少变更记录。在这种情况下如果直接把 AI 接到原始数据上让模型解释字段、生成统计或总结结论风险比较高。更合适的路径是先把链路拆开接入、标准化、质控、分析准备、可追溯输出。AI 放在低风险辅助环节而不是放在入组、排除、诊疗、风险分层或结局判定这些专业决策环节。技术目标和边界本文目标不是搭建完整临床科研平台而是整理一条可复用的数据处理流水线多源数据 - 原始层 raw - 标准化层 normalized - 质控层 quality_checked - 分析准备层 analysis_ready - 宽表、特征表、质控报告技术选型如下Pandas处理轻量级清洗、字段映射、缺失值归一、日期转换。DuckDB完成多表连接、聚合和中间结果落盘。Great Expectations把质控规则结构化、配置化用于生成可审计检查结果。Airflow编排批处理任务保证流程可重复、可追踪、可失败重跑。AI 辅助用于低风险环节例如字段名候选匹配、质控报告摘要、异常模式文本归纳。边界需要提前写清楚AI 不参与研究对象入组或排除AI 不生成诊疗建议AI 不决定风险分层AI 不替代研究者、数据管理员、统计人员或临床专业人员判断示例字段、示例阈值和示例规则不能直接迁移到真实临床判断中。合规和数据安全边界医疗健康数据治理不能只看技术流程还要把合规边界放到链路设计里。不同国家、地区、机构和项目类型要求不同下面只列出工程实施时常见的控制点具体要求应以本机构制度、伦理审批、数据使用协议和适用法律法规为准。建议在项目启动阶段确认以下事项伦理审批和研究授权研究数据的使用范围、研究目的、数据字段、处理方式和输出形式应与伦理审批或机构审批保持一致。数据脱敏或去标识化进入分析环境前应根据项目要求处理姓名、身份证号、手机号、住址、病历号等直接或高敏感标识。是否允许保留可回溯键值需要由机构流程确认。最小权限原则原始层、标准化层、分析准备层和报告层应分角色授权。不是所有参与者都需要访问原始明细数据。审计日志记录谁在什么时间访问了哪些数据、执行了哪些任务、导出了哪些结果便于事后追踪。数据出境或跨机构共享如果涉及跨机构、跨区域或出境处理应提前确认数据使用协议、传输方式、脱敏要求和审批流程。结果导出控制质控报告和分析准备表也可能包含敏感信息导出前应检查是否包含不必要的个人标识或小样本可识别信息。AI 工具使用边界不得把未经授权的可识别医疗数据直接提交给外部大模型或第三方服务。若使用本地模型或私有化模型也应纳入机构安全评估和审计范围。在工程实现上可以把合规要求落实为几类可执行控制控制点工程落地方式数据脱敏脱敏任务前置保留脱敏规则版本最小权限按数据层级和角色授权审计日志记录访问、任务运行、导出、失败重跑批次追溯保存批次号、配置版本、代码版本敏感字段控制数据字典标记敏感级别AI 使用限制仅允许摘要、字段匹配等低风险任务禁止上传可识别明细总体架构先稳定数据底座再谈 AI 提效一条实用链路可以按分层设计避免分析代码直接依赖原始文件。---------------- | source files | | csv/xlsx/parquet --------------- | v ---------------- -------------- ------------------ | schema mapping |--| normalize |--| quality validate | | yaml/json | | pandas | | rule configs | ---------------- -------------- ----------------- | | v v ------------- ------------ | DuckDB marts | | QC reports | ------------- ------------- | v -------------- | analysis_ready| ---------------这套结构的核心是把“口径”从脚本中抽出来。字段映射、类型转换、缺失值编码、质控规则、数据批次号都应进入配置或元数据管理而不是散落在多个 Notebook 里。在真实项目中可以把数据分成几层层级作用是否允许修改原始含义raw保留原始导出文件和批次信息不允许normalized统一字段名、类型、编码和缺失值表达只做标准化quality_checked增加质控结果和失败标记不改变原值analysis_ready面向统计或建模准备的分析输入表按方案生成report输出质控报告、批次差异、字段缺失情况不参与专业判断先把这些层级理清后续再讨论 AI 生成摘要、辅助排查异常才有可靠上下文。实现步骤从多源表到分析准备表假设有三类输入研究对象基本信息随访记录实验室指标。下面示例只围绕结构化研究数据处理展开不涉及诊疗推荐或临床决策。1. 定义字段映射配置字段映射建议单独维护例如mapping.yaml。下面是配置示例不是临床规则subject:subject_id:[patient_id,pid,研究编号]birth_date:[birth_date,出生日期]sex:[sex,gender,性别]visit:subject_id:[patient_id,pid,研究编号]visit_date:[visit_date,随访日期]visit_type:[visit_type,随访类型]lab:subject_id:[patient_id,pid,研究编号]test_date:[test_date,检测日期]item_code:[item_code,项目代码]value:[value,结果值]unit:[unit,单位]AI 可以在这里做候选字段匹配提示。例如根据列名、样例值、历史映射记录推荐“研究编号”可能对应subject_id。最终映射仍需由数据管理员或项目负责人确认并记录版本。建议字段映射至少保留这些信息映射版本来源文件名或数据表名原始字段名标准字段名字段说明确认人员生效批次变更原因。字段映射一旦没有版本后面统计结果发生变化时很难判断是数据变了、规则变了还是脚本变了。2. 用 Pandas 做标准化下面代码演示一个最小可运行的标准化函数统一列名、日期、缺失值和性别编码。字段和枚举仅为示例。importpandasaspdfromtypingimportDict,List MISSING_TOKENS{,NA,N/A,null,NULL,未知,未采集}defrename_by_mapping(df:pd.DataFrame,mapping:Dict[str,List[str]])-pd.DataFrame:rename_dict{}lower_cols{c.lower():cforcindf.columns}fortarget,candidatesinmapping.items():forcincandidates:ifc.lower()inlower_cols:rename_dict[lower_cols[c.lower()]]targetbreakreturndf.rename(columnsrename_dict)defnormalize_missing_value(x):ifpd.isna(x):returnNonetextstr(x).strip()returnNoneiftextinMISSING_TOKENSelsexdefnormalize_subject(df:pd.DataFrame)-pd.DataFrame:dfdf.copy()forcolindf.columns:df[col]df[col].map(normalize_missing_value)required{subject_id,birth_date,sex}missing_colsrequired-set(df.columns)ifmissing_cols:raiseValueError(f缺少标准字段:{missing_cols})df[subject_id]df[subject_id].astype(str).str.strip()df[birth_date]pd.to_datetime(df[birth_date],errorscoerce)sex_map{M:male,男:male,1:male,F:female,女:female,2:female}df[sex]df[sex].astype(str).str.strip().map(sex_map).fillna(unknown)returndf[[subject_id,birth_date,sex]].drop_duplicates()if__name____main__:rawpd.DataFrame({研究编号:[001,002,003],出生日期:[1980-01-01,NA,1975/05/20],性别:[男,女,未知]})mapping{subject_id:[patient_id,pid,研究编号],birth_date:[birth_date,出生日期],sex:[sex,gender,性别]}normalizednormalize_subject(rename_by_mapping(raw,mapping))print(normalized)这个阶段建议只做“可解释的标准化”不要做复杂推断。比如出生日期缺失时不应随意用均值或中位数填补。是否允许填补、使用什么方法填补、填补后是否进入敏感分析应写入研究数据处理计划并由专业人员确认。3. 保留缺失值语义真实世界数据中的缺失通常不是一种情况。至少应区分原始表达建议归类说明空字符串、NA、nullmissing_unknown不清楚为何缺失未采集not_collected该字段本应采集但未采集不适用not_applicable对该对象或场景不适用采集失败collection_failed有采集动作但失败拒绝回答refused受试者或研究对象拒绝提供如果直接把这些值都转成NULL后续统计可能会丢失上下文。更稳的做法是同时保留两个字段标准化后的值字段例如birth_date缺失原因字段例如birth_date_missing_reason。这样既能支持分析也能在质控报告中解释缺失来源。用 Great Expectations 思路把质控规则结构化质控规则最好覆盖四类问题完整性唯一性取值范围跨表一致性。考虑到 Great Expectations 不同版本 API 变化较多下面不绑定具体版本代码而是用配置示例表达质控逻辑。实际落地时可以把这些规则转成 Great Expectations Expectation Suite也可以转成自研校验脚本。suite:rws_subject_quality_rulesdataset:subject_normalizedrules:-name:subject_id_not_nulltype:not_nullcolumn:subject_idseverity:error-name:subject_id_uniquetype:uniquecolumn:subject_idseverity:error-name:sex_in_allowed_settype:in_setcolumn:sexallowed_values:[male,female,unknown]severity:error-name:birth_date_not_after_cutofftype:date_betweencolumn:birth_datemin_value:1900-01-01max_value:2026-05-14severity:warning-name:birth_date_missing_ratetype:missing_rate_less_thancolumn:birth_datethreshold:0.20severity:warning这些规则不代表医学判断只是数据工程层面的格式、一致性和可追溯性检查。比如“出生日期不晚于数据截止日期”属于数据逻辑校验而“某对象是否应入组”“某指标异常是否代表风险升高”不能交给这类规则或 AI 模型自动决定。实际项目中建议把规则分成两层硬规则失败后阻断进入analysis_ready例如主键为空、主键重复、跨表主键无法关联。提示规则不阻断流程但进入质控报告例如某字段缺失比例超过示例阈值。阈值应配置化不要写死成所谓“行业标准”。例如某字段缺失率 20% 是否可接受要看研究问题、数据来源、变量重要性和统计分析方案。跨表质控比单表规则更容易暴露返工点单表质控只能发现字段为空、枚举值异常、日期格式错误等问题。真实项目里更难处理的通常是跨表问题随访表里有subject_id主体表里没有检验表存在检测记录但日期早于登记日期或出生日期同一个研究对象在不同系统中性别、出生日期不一致用药、检查、随访记录使用了不同的主键体系历史批次保留旧编码新批次已经切换新编码。跨表规则可以单独维护。下面是配置示例已补齐 YAML 结构并闭合代码围栏suite:rws_cross_table_quality_rulesrules:-name:visit_subject_should_existtype:foreign_keyleft_dataset:visit_normalizedleft_column:subject_idright_dataset:subject_normalizedright_column:subject_idseverity:error-name:lab_subject_should_existtype:foreign_keyleft_dataset:lab_normalizedleft_column:subject_idright_dataset:subject_normalizedright_column:subject_idseverity:error-name:visit_date_not_before_birth_datetype:date_compareleft_dataset:visit_normalizedleft_column:visit_dateright_dataset:subject_normalizedright_column:birth_datejoin_key:subject_idoperator:severity:error-name:lab_date_not_before_birth_datetype:date_compareleft_dataset:lab_normalizedleft_column:test_dateright_dataset:subject_normalizedright_column:birth_datejoin_key:subject_idoperator:severity:warning跨表质控报告最好不要只输出“通过/失败”而要能定位问题样本。报告字段可以包括字段说明batch_id数据批次rule_name规则名称severityerror 或 warningleft_dataset左表right_dataset右表join_key关联键failed_count失败行数failed_rate失败比例sample_keys抽样展示的问题主键rule_version规则版本如果报告中包含样本主键应确认该主键是否属于敏感或可回溯标识。对外沟通时可以使用脱敏后的研究编号或有限抽样结果避免不必要的数据暴露。用 DuckDB 做分析准备表Pandas 适合清洗单表DuckDB 更适合多表连接、聚合和宽表生成。真实世界研究中分析团队通常希望拿到相对稳定的analysis_ready表而不是每次从十几个原始表重新拼接。下面代码是可运行示例用三张内存 DataFrame 注册到 DuckDB 后生成分析准备表。字段均为演示用不代表临床变量定义。importpandasaspdimportduckdb subject_normalizedpd.DataFrame({subject_id:[001,002,003],sex:[male,female,unknown],birth_date:pd.to_datetime([1980-01-01,1990-06-10,1975-05-20])})visit_normalizedpd.DataFrame({subject_id:[001,001,002],visit_date:pd.to_datetime([2024-01-01,2024-03-01,2024-02-15]),visit_type:[baseline,followup,baseline]})lab_normalizedpd.DataFrame({subject_id:[001,001,003],test_date:pd.to_datetime([2024-01-02,2024-03-03,2024-02-20]),item_code:[LAB_A,LAB_B,LAB_A],value:[1.2,3.4,2.1],unit:[u1,u2,u1]})conduckdb.connect()con.register(subject_normalized,subject_normalized)con.register(visit_normalized,visit_normalized)con.register(lab_normalized,lab_normalized)analysis_readycon.execute( SELECT s.subject_id, s.sex, s.birth_date, MIN(v.visit_date) AS first_visit_date, COUNT(DISTINCT v.visit_date) AS visit_count, COUNT(l.item_code) AS lab_record_count FROM subject_normalized s LEFT JOIN visit_normalized v ON s.subject_id v.subject_id LEFT JOIN lab_normalized l ON s.subject_id l.subject_id GROUP BY s.subject_id, s.sex, s.birth_date ORDER BY s.subject_id ).fetchdf()print(analysis_ready)这里生成的是分析准备表不是研究结论表。它可以包含稳定的基础变量、随访次数、检验记录数量等工程加工字段但不应混入未经确认的临床结论字段。如果需要生成分组标签、排除标记或风险相关变量应注意三点规则来源必须可追溯到研究方案或专业人员确认文档规则实现应可审计、可复跑AI 不能自由决定入组、排除、诊疗或风险分层。用 Airflow 编排批处理链路Airflow 的价值在于让流程可重复、可追踪、可失败重跑。一个简化 DAG 可以拆成以下任务extract_sources - normalize_subject - normalize_visit - normalize_lab - run_quality_checks - build_analysis_ready - export_qc_report每个任务都应记录输入路径输出路径数据批次号映射配置版本质控规则版本任务运行日志失败原因重跑记录代码版本或镜像版本执行人或调度账号。不要让任务直接覆盖旧结果。建议按批次保存例如data/ raw/batch_date2026-05-14/ normalized/batch_date2026-05-14/ quality_checked/batch_date2026-05-14/ analysis_ready/batch_date2026-05-14/ reports/batch_date2026-05-14/Airflow 编排时建议把任务输入输出路径和配置版本作为参数传入而不是写死在任务内部。这样后续追溯某一次统计结果时可以明确知道它使用的是哪批原始数据、哪版字段映射、哪版质控规则和哪版分析准备表。一个项目中比较实用的任务拆分方式是任务输入输出extract_sources原始导出文件raw 层归档normalize_subjectraw 主体表、mappingsubject_normalizednormalize_visitraw 随访表、mappingvisit_normalizednormalize_labraw 检验表、mappinglab_normalizedrun_quality_checksnormalized 表、规则配置validation_resultbuild_analysis_ready通过硬规则的数据表analysis_readyexport_qc_reportvalidation_resultHTML/Excel/Markdown 报告失败重跑时不应直接跳过质控。尤其是字段映射、缺失值处理、跨表关联规则发生变更后analysis_ready应重新生成并标记新的规则版本。AI 可以放在哪些低风险环节在这条链路中AI 更适合做辅助工作而不是做专业判断。可考虑的低风险场景包括根据列名和样例值推荐字段映射候选对质控失败结果生成摘要帮助数据管理员归纳异常模式根据历史规则提示可能遗漏的校验项将批次质控结果整理成报告初稿辅助生成数据字典说明草稿根据已有字段字典生成更易读的字段说明。不建议让 AI 执行或决定以下事项研究对象入组研究对象排除诊断或治疗建议用药建议临床风险分层结局事件判定医学因果解释替代统计分析方案审批在未授权情况下处理可识别医疗数据。AI 输出应作为待审核材料而不是最终规则。所有影响研究对象选择、分析口径、医学解释和结论生成的内容都应由研究者、统计人员、数据管理员和相关专业人员按流程确认。如果确实要在数据治理链路中使用 AI建议增加几项控制控制点建议做法输入控制避免输入可识别个人信息输出留痕保存提示词、输入摘要、输出结果和审核记录人工复核字段映射、规则建议、异常解释必须确认后生效权限隔离AI 服务账号不应拥有原始全量数据访问权限模型边界明确 AI 不做诊疗、入组、排除和风险分层判断踩坑记录比算法更影响效率的细节字段映射必须版本化一次字段名调整可能影响后续所有统计结果。如果没有映射版本排查时很难判断结果变化来自数据源、映射关系还是分析逻辑。建议每次变更字段映射时记录变更前字段变更后字段影响的数据表影响的批次变更原因确认人。缺失值要保留语义未采集和不适用在分析时可能含义完全不同。统一转成空值虽然方便计算但会损失上下文。建议至少保留原始值标准化值缺失原因标准化批次处理规则版本。质控失败要定位到行级样本只输出“校验失败”对排查帮助有限。质控报告至少应包含数据集名称字段名规则名严重级别失败行数失败比例样本主键批次号规则版本。如果涉及敏感数据报告中应避免暴露不必要的个人信息并遵循机构的数据安全和隐私保护要求。分析准备表不要混入未确认结论analysis_ready的职责是提供稳定、可追溯、可复用的分析输入。它不应变成“什么都放”的大宽表。建议把字段分成几类基础人口学或登记字段随访过程字段检验检查结构化字段工程派生字段方案确认后的分析变量。未经确认的模型预测、风险评分、诊疗判断不应直接写入正式分析准备表。性能和扩展建议在中小规模真实世界研究数据处理中Pandas 加 DuckDB 通常可以支撑日常批处理。性能优化应先从数据组织和任务拆分入手而不是一开始就上复杂平台。建议优先处理以下问题文件格式原始 CSV 或 Excel 归档后可转换为 Parquet减少重复解析成本。分区管理按batch_date、研究项目、数据域进行分区便于增量处理。大表连接多表 join 和聚合尽量放到 DuckDB降低 Pandas 内存峰值。中间结果落盘关键节点输出中间表避免每次从原始数据全量重算。任务拆分Airflow 任务按数据域拆分失败后只重跑受影响节点。质控分层先跑主键、类型、枚举值等基础规则再跑跨表一致性和批次趋势。质控报告落库把每批次失败数、失败比例、缺失率、规则版本写入表中便于观察趋势。异常样本抽样对失败样本提供有限抽样方便人工复核避免报告过大。规则灰度新增规则可先作为 warning 观察几个批次再决定是否升级为 error。权限控制原始层、标准化层、分析准备层和报告层应按角色授权避免不必要的数据暴露。质控部分尤其要注意扩展性。随着项目推进规则会不断增加如果所有规则都写在一个脚本里后期维护会很困难。更好的做法是把规则配置、规则执行和报告生成拆开rule_config - rule_engine - validation_result - qc_report当数据量继续增长时可以把 DuckDB 层替换成数据仓库、湖仓引擎或分布式计算框架。但数据分层、字段映射版本化、质控规则配置化、批次可追溯这些原则不需要推倒重来。总结先把数据底座做稳真实世界研究的数据治理很少是一条脚本就能解决的问题。多源数据、口径不一、缺失语义、跨表不一致、批次变更和人工补录都会在分析前集中暴露。如果这些问题没有被结构化处理后续再引入 AI、自动化报表或建模流程也只是把不稳定的数据更快地传递到下游。这条链路更适合解决几类具体问题字段映射有版本口径变化能追溯缺失值保留原因后续解释不依赖个人记忆单表和跨表质控前置减少统计阶段返工analysis_ready表稳定输出分析脚本不直接读取原始文件Airflow 记录批次、配置和运行日志方便失败重跑和审计AI 只参与字段候选、报告摘要、异常归纳等可复核环节。它也有明确限制不能替代研究方案设计不能替代医学判断不能自动修复原始数据质量不足不能绕过伦理审批、数据授权和机构规范不能把 AI 输出直接作为入组、排除、诊疗或风险分层依据。如果从一个小批量历史数据开始试跑建议顺序是先固化字段映射再补齐缺失值语义和基础质控规则随后增加跨表一致性检查最后生成稳定的analysis_ready表。数据底座稳定之后再让 AI 辅助质控摘要、异常说明和数据字典维护分析提效才有可靠前提。本文文献检索、文献挖掘以及文献翻译采用的是【超能文献| AI文献检索|AI文档翻译】。