如何把PV数据录入从“人肉战场“变成了全自动流水线

去年Q2,我们的药物警戒(Pharmacovigilance,PV)团队在做年度复盘,有人做了一张饼图。整个部门的工时分布:62%在录入数据,18%在核查录入的数据,只有20%在做真正的信号检测和风险分析。这张图在会议室里沉默了很久。我们公司同时跑着三十几个临床项目光是SUSAR(Suspected Unexpected Serious Adverse Reaction)的上报每周就有几十条。每条从PDF/传真/邮件里扒出来再填进安全数据库——MedDRA编码要对时间线要对项目编号要对E2B格式要合规——一个有经验的专员吗认认真真做也要2-3小时。出错了重做。监管要求更新了重做。我坐在那个会议室里看着这张饼图脑子里只有一个念头:这件事不应该由人来做。问题到底在哪里在开始写代码之前我们做了一件事:画了一张现实流程图。不是PPT里那种漂亮的示意图是真实观察PV专员工作一周之后一步一步还原出来的。大概是这样:这个流程每一步都依赖人的眼睛和判断但90%的步骤其实是规则明确的信息提取和格式转换。这就是自动化的切入点。技术选型:我们为什么没有选最新的内部开始讨论方案时有人提议直接上大模型让LLM自己解析报告、自己填字段。我拒绝了。原因很实际:PV数据是监管数据,有严格的audit trail要求。每一个字段的来源必须可追溯,每一步操作必须有日志。大模型的黑盒特性在这个场景下是监管风险。另外,PV系统的表单操作非常具体——点哪个按钮、怎么触发MedDRA查询——这些系统级交互LLM根本不理解。最终的技术栈:层次技术负责什么文档解析层UiPath Document Understanding 自定义ML模型从PDF/邮件中提取结构化字段规则引擎层自研Python规则库严重性判断、预期性匹配、项目级个性化规则RPA执行层UiPath Studio操控Argus界面完成录入质控层Python 正则字段完整性校验、逻辑一致性检查日志层ELK Stack全流程audit trail,满足21 CFR Part 11不是最酷的方案但是合规的、可落地的、三个月能上线的。核心代码:文档解析部分(脱敏版)这是整个系统最难的部分。AE报告格式千奇百怪:有些是结构化表单,有些是医生用Word写的叙述,有些是传真扫描件OCR之后满是噪声。我们用UiPath Document Understanding做了一个两阶段提取:第一阶段:分类器判断报告类型# ae_classifier.py (脱敏版)# 根据文档特征判断报告类型,走不同的提取模板importrefromenumimportEnumclassReportType(Enum):STRUCTURED_FORMstructured# 标准化表单(如MedWatch)NARRATIVE_TEXTnarrative# 叙述性文本LITERATUREliterature# 文献来源UNKNOWNunknownFORM_SIGNALS[rMedWatch,rCIOMS,rPatient Initials,rDate of Birth,rSuspect Drug,rIndication]NARRATIVE_SIGNALS[rDear.*Safety,radverse event,rside effect,rexperienced,rreported]defclassify_report(text:str)-ReportType: 基于关键词密度判断报告类型 生产环境中建议替换为训练好的分类模型 text_lowertext.lower()form_scoresum(1forpatterninFORM_SIGNALSifre.search(pattern,text,re.IGNORECASE))narrative_scoresum(1forpatterninNARRATIVE_SIGNALSifre.search(pattern,text,re.IGNORECASE))ifform_score3:returnReportType.STRUCTURED_FORMelifnarrative_score2:returnReportType.NARRATIVE_TEXTelifet alintext_lowerorjournalintext_lower:returnReportType.LITERATUREelse:returnReportType.UNKNOWN第二阶段:字段提取(针对叙述性文本)# ae_extractor.py (脱敏版)# 从叙述性AE报告中提取关键字段importrefromdataclassesimportdataclass,fieldfromtypingimportOptional,ListfromdatetimeimportdatetimedataclassclassAECase:# 患者信息(均为匿名化处理)patient_initials:Optional[str]Noneage:Optional[int]Noneage_unit:stryearssex:Optional[str]Noneweight_kg:Optional[float]None# 药品信息suspect_drugs:List[str]field(default_factorylist)dose:Optional[str]Noneroute:Optional[str]Noneindication:Optional[str]Nonestart_date:Optional[str]None# 不良反应ae_terms:List[str]field(default_factorylist)ae_onset_date:Optional[str]Noneae_outcome:Optional[str]Noneserious:Optional[bool]None# 上报信息reporter_type:Optional[str]Nonecountry:Optional[str]None# 提取置信度(用于QC标记)confidence_scores:dictfield(default_factorydict)classNarrativeExtractor: 从自由文本叙述中提取AE字段 注意:本版本为规则引擎,生产中建议结合NLP模型提升召回率 # 年龄匹配:支持多种表达方式AGE_PATTERNS[r(\d{1,3})[-\s]?year[-\s]?old,raged?\s(\d{1,3}),r(\d{1,3})\s*yo\b,rage[:\s](\d{1,3}),]# 性别匹配SEX_MAP{r\b(male|man|boy|him|his)\b:M,r\b(female|woman|girl|her|she)\b:F,}# 转归匹配OUTCOME_PATTERNS{recovered:rrecover|resolv|better|improved,recovering:rrecovering|improving,not recovered:rnot recover|ongoing|persist|continu,fatal:rdied|death|fatal|deceased|passed away,sequelae:rsequela|permanent|disability,unknown:runknown|unclear,}defextract(self,text:str)-AECase:caseAECase()text_normalizedself._normalize_text(text)case.age,case.age_unitself._extract_age(text_normalized)case.sexself._extract_sex(text_normalized)case.suspect_drugsself._extract_drugs(text_normalized)case.ae_termsself._extract_ae_terms(text_normalized)case.ae_outcomeself._extract_outcome(text_normalized)case.seriousself._assess_seriousness(text_normalized)case.countryself._extract_country(text_normalized)# 计算整体置信度case.confidence_scoresself._calc_confidence(case)returncasedef_normalize_text(self,text:str)-str:统一换行、去除多余空白textre.sub(r\r\n|\r,\n,text)textre.sub(r[ \t], ,text)returntext.strip()def_extract_age(self,text:str):forpatterninself.AGE_PATTERNS:mre.search(pattern,text,re.IGNORECASE)ifm:ageint(m.group(1))if0age120:# 合理性校验unitmonthsifage2andmonthintext.lower()elseyearsreturnage,unitreturnNone,yearsdef_extract_sex(self,text:str)-Optional[str]:forpattern,valueinself.SEX_MAP.items():ifre.search(pattern,text,re.IGNORECASE):returnvaluereturnNonedef_extract_drugs(self,text:str)-List[str]: 真实环境建议对接公司药品主数据库做字典匹配 此处为简化版本 drugs[]# 提取大写药品名(通常为INN或商品名)candidatesre.findall(r\b[A-Z][A-Za-z]{4,}\b,text)STOPWORDS{Patient,Subject,Report,Event,Study,Serious,Treatment,Hospital,Doctor}forwordincandidates:ifwordnotinSTOPWORDSandnotword.isupper():drugs.append(word)returnlist(dict.fromkeys(drugs))[:5]# 去重,最多返回5个def_extract_ae_terms(self,text:str)-List[str]: 生产环境应对接MedDRA字典API做精准匹配 此处为占位示例 ae_keywordsre.findall(r\b(nausea|vomiting|headache|rash|fever|fatigue|rdizziness|dyspnea|pain|swelling|pruritus|erythema)\b,text,re.IGNORECASE)returnlist(set(t.lower()fortinae_keywords))def_assess_seriousness(self,text:str)-bool: ICH E2A定义的严重性标准:死亡/危及生命/住院/致残/先天异常/重要医学事件 serious_indicators[rhospitali[sz],rlife.threatening,rdied|death|fatal,rdisabl,rcongenital,rimportant medical,rICU|intensive care,remergency,]text_lowertext.lower()returnany(re.search(p,text_lower)forpinserious_indicators)def_extract_outcome(self,text:str)-Optional[str]:text_lowertext.lower()foroutcome,patterninself.OUTCOME_PATTERNS.items():ifre.search(pattern,text_lower):returnoutcomereturnunknowndef_extract_country(self,text:str)-Optional[str]:# 真实版本维护一个国家名称词典,此处简化common_countries[United States,USA,Germany,France,Japan,China,United Kingdom,UK,Canada,Australia,Italy,Spain]forcountryincommon_countries:ifcountryintext:returncountryreturnNonedef_calc_confidence(self,case:AECase)-dict: 字段级置信度评估,用于QC决策: - HIGH: 自动提交 - MEDIUM: 人工复核后提交 - LOW: 退回人工处理 scores{}scores[age]HIGHifcase.ageelseLOWscores[sex]HIGHifcase.sexelseLOWscores[drugs]HIGHifcase.suspect_drugselseLOWscores[ae_terms]HIGHifcase.ae_termselseLOWscores[seriousness]HIGH# 规则引擎输出视为高置信high_countsum(1forvinscores.values()ifvHIGH)scores[overall](HIGHifhigh_count4elseMEDIUMifhigh_count2elseLOW)returnscores第三阶段:RPA写入(UiPath侧的配置逻辑,用Python伪代码说明)# argus_writer.py (脱敏版,对应UiPath Invoke Code活动)# 这部分在UiPath中实现,这里用Python说明逻辑classArgusWriter: 将提取的AECase写入Argus Safety数据库界面 实际通过UiPath RPA操控浏览器/客户端完成 def__init__(self,project_config:dict):# project_config包含该项目的特定规则:# 比如哪些AE需要走快速通道,report timeline要求等self.configproject_configdefwrite_case(self,case:AECase,session)-str: 返回生成的Case ID session为UiPath Browser Session对象 # 1. 只有高置信度的case才允许全自动提交ifcase.confidence_scores.get(overall)LOW:raiseLowConfidenceError(fCase置信度不足,转人工处理:{case.confidence_scores})case_idself._create_new_case(session)self._fill_patient_tab(session,case)self._fill_drug_tab(session,case)self._fill_event_tab(session,case)self._fill_narrative(session,case)# 项目特定逻辑:是否需要勾选额外的监管flagifself.config.get(eu_reporting_required):self._set_eu_flag(session)# MEDIUM置信度:保存草稿,推送QC队列;不直接提交ifcase.confidence_scores.get(overall)MEDIUM:self._save_draft(session)self._push_to_qc_queue(case_id)else:self._submit_for_review(session)self._write_audit_log(case_id,case)returncase_iddef_fill_patient_tab(self,session,case:AECase):点击Patient标签页,填写患者信息# UiPath: Click(Patient Tab)# UiPath: TypeInto(Age Field, str(case.age))# ... 以此类推passdef_write_audit_log(self,case_id:str,case:AECase):写入操作日志,满足21 CFR Part 11要求log_entry{timestamp:datetime.utcnow().isoformat(),case_id:case_id,action:AUTO_ENTRY,operator:RPA_BOT_v2,confidence:case.confidence_scores,fields_auto_filled:self._count_filled_fields(case),fields_manual_required:[],}# 写入ELK或合规日志系统self._send_to_audit_system(log_entry)上线之后发生了什么第一个月:机器人上线只敢处理MEDIUM/HIGH置信度的structured form。大概覆盖了35%的案例量。剩下65%还是人工。团队不太信任机器人每个case都要double check。这很正常。我们不急。第二个月:QC数据出来了。自动录入的case字段错误率2.1%人工录入历史基线是4.7%。机器人错得比人少。团队开始放心。第三个月:扩展到叙述性文本,覆盖率升到78%。剩下22%是质量极差的扫描件或者多语言混合文档,继续人工处理。到现在的数据:指标自动化前自动化后单案例平均处理时长2.1小时28分钟字段错误率4.7%2.1%专员加班频率每周3-4次基本消除人工覆盖率(需人工介入)100%22%那张曾经让人沉默的饼图现在长这样数据录入降到了21%信号检测和风险分析升到了58%。几个踩过的坑,送给同行坑一:MedDRA版本对齐问题MedDRA每年更新两次。自动化系统用的词典版本必须和数据库里的版本严格一致否则PT编码会对不上。我们上线第三天就发现了这个问题紧急加了版本锁定逻辑。坑二:过度追求自动化率项目初期KPI设的是自动化率90%“结果为了凑数字把一些本不该自动提交的低质量报告也推进去了被QC打回来了一批。后来把KPI改成自动提交后零退回率”反而质量和速度都上去了。坑三:变更控制(Change Control)没做好RPA机器人改了一个字段映射逻辑没有走正式的变更控制流程。被内审发现了。在制药行业系统变更必须有完整的文档和审批这一点不能偷懒。最后有人问我:以后PV专员会被取代吗不会。药物警戒的核心价值从来不是录入数据而是在海量的信号噪声里识别出真正的风险信号是在监管政策变动的时候做出正确的判断是在一个真实的病人可能因此受害的时候负起那份责任。这些事机器人做不了。我们做自动化是为了让那些聪明的人从重复劳动里解放出来去做只有人才能做的事。代码已脱敏处理生产环境中请根据实际系统进行适配。如有问题欢迎评论区交流。