背景数据分类分级是数据治理的核心环节也是合规建设的必答题。随着《数据安全法》实施政企单位纷纷引入大模型自动化处理这项工作。但实际项目中我们发现一个普遍问题AI分类系统上线初期准确率90%三个月后退化到60%以下。这不是大模型能力问题而是工程化架构缺失导致的系统性缺陷。本文从真实项目经验出发剖析问题根因给出可落地的架构设计方案。一、问题剖析简单调用大模型的三大技术缺陷1.1 信息输入不完整大模型的判断能力来源于输入信息。典型的简单调用模式只输入字段基本信息# 简单调用示例input_text字段名phone类型varchar(20)promptf请判断以下字段的数据级别{input_text}resultllm_api.call(prompt)问题数据分级取决于三个维度——字段属性、业务系统归属、数据使用场景。简单调用只输入了第一维大模型只能靠通用知识猜测。1.2 学习机制缺失大模型在推理阶段是静态的——每次调用独立之前的结果不影响后续判断。# 人工修正无法影响模型wrong_result一般数据# AI把身份证号标成一般corrected核心数据# 人工修正# 下次调用AI仍会犯相同错误next_resultllm_api.call(字段名id_card)# 还是一般数据结论没有反馈机制AI永远停留在上线那天的能力水平。1.3 风险控制缺失大模型输出是概率性的——同一输入不同调用可能给出不同结果。在合规场景大概率正确是危险的。# 同一输入不同调用结果可能不同result_1llm_api.call(prompt)# 核心数据result_2llm_api.call(prompt)# 重要数据result_3llm_api.call(prompt)# 一般数据 # ← 灾难性错误简单调用模式没有风险拦截机制。二、工程化架构设计2.1 核心公式工程化 上下文注入反馈闭环规则兜底2.2 五层架构┌─────────────────────────────────────────────┐ │ Layer1: 数据资产层 │ │ ├── AssetCatalog: 表/字段元信息 │ │ ├── SystemRegistry: 业务系统台账 │ │ └── ComplianceRuleLib: 合规规则库 │ └─────────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────────┐ │ Layer2: 上下文引擎层 │ │ ├── ContextExtractor: 自动提取关联信息 │ │ ├── PromptAssembler: 组装完整输入 │ │ └── PromptTemplateMgr: 模板管理 │ └─────────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────────┐ │ Layer3: AI判断层 │ │ ├── LLMClient: 大模型API封装 │ │ ├── ModelComparison: 多模型对比可选 │ │ └── ResultCache: 结果缓存 │ └─────────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────────┐ │ Layer4: 规则引擎层 │ │ ├── BlacklistRule: 黑名单强制提升 │ │ ├── WhitelistRule: 白名单强制降低 │ │ └── BusinessRule: 系统级强制覆盖 │ └─────────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────────┐ │ Layer5: 反馈闭环层 │ │ ├── ReviewInterface: 人工审核API │ │ ├── AnnotationStore: 标注库存储 │ │ └── LearningUpdater: 定期学习更新 │ └─────────────────────────────────────────────┘2.3 核心代码实现思路上下文引擎实现classContextEngine:上下文引擎自动提取并组装判断输入def__init__(self,asset_catalog,system_registry,rule_lib):self.asset_catalogasset_catalog self.system_registrysystem_registry self.rule_librule_lib self.prompt_templateself._load_template()defbuild_context(self,field_id:str)-dict:构建完整上下文信息# 1. 获取字段基本信息field_infoself.asset_catalog.get_field(field_id)# 2. 获取所属表/库信息table_infoself.asset_catalog.get_table(field_info[table_id])db_infoself.asset_catalog.get_database(table_info[db_id])# 3. 获取业务系统信息system_infoself.system_registry.get_by_database(db_info[id])# 4. 获取合规规则compliance_rulesself.rule_lib.get_rules(system_info[type])return{field_name:field_info[name],field_type:field_info[type],table_name:table_info[name],database_name:db_info[name],system_name:system_info[name],system_type:system_info[type],compliance_level:system_info[compliance_level],data_usage:field_info.get(usage,unknown),rules:compliance_rules}defassemble_prompt(self,context:dict)-str:组装完整promptreturnself.prompt_template.format(**context)规则引擎实现classRuleEngine:规则引擎高风险场景强制干预BLACKLIST{id_card:核心数据,bank_card:核心数据,passport:核心数据,phone:重要数据,email:重要数据,}WHITELIST{public_notice:一般数据,announcement:一般数据,}def__init__(self,business_rules:dict):self.business_rulesbusiness_rules# 系统级规则defapply_rules(self,field_name:str,system_type:str,ai_result:str)-str:应用规则返回最终分级# 1. 黑名单优先iffield_nameinself.BLACKLIST:returnself.BLACKLIST[field_name]# 2. 白名单次之iffield_nameinself.WHITELIST:returnself.WHITELIST[field_name]# 3. 业务规则覆盖ifsystem_typeinself.business_rules:returnself.business_rules[system_type]# 4. 无匹配规则返回AI结果returnai_result反馈闭环实现classFeedbackLoop:反馈闭环人工标注→学习更新def__init__(self,annotation_store,learning_updater):self.storeannotation_store self.updaterlearning_updaterdefrecord_correction(self,field_id:str,original:str,corrected:str,reason:str):记录人工修正annotation{field_id:field_id,original_level:original,corrected_level:corrected,reason:reason,timestamp:datetime.now(),reviewer:get_current_user()}self.store.save(annotation)defweekly_review(self):每周学习更新new_annotationsself.store.get_since_last_update()iflen(new_annotations)0:self.updater.update_with_examples(new_annotations)self.store.mark_updated()主流程整合classAIClassificationSystem:AI分类分级主系统def__init__(self):self.context_engineContextEngine(...)self.llm_clientLLMClient(...)self.rule_engineRuleEngine(...)self.feedback_loopFeedbackLoop(...)defclassify(self,field_id:str)-str:完整分类流程# Step 1: 构建上下文contextself.context_engine.build_context(field_id)promptself.context_engine.assemble_prompt(context)# Step 2: AI判断ai_resultself.llm_client.call(prompt)# Step 3: 规则兜底final_resultself.rule_engine.apply_rules(context[field_name],context[system_type],ai_result)# Step 4: 记录结果用于后续反馈self.feedback_loop.record_classification(field_id,final_result)returnfinal_result三、避坑指南3.1 上下文引擎是核心不可省略错误做法为快速上线跳过上下文引擎只做字段级判断。# ❌ 错误只输入字段名resultllm.call(字段名phone)# ✅ 正确输入完整上下文contextcontext_engine.build_context(field_id)resultllm.call(assemble_prompt(context))后果上线快退化也快。一个月开发投入 → 系统稳定运行三年。3.2 规则引擎是防线不嫌笨错误观点规则引擎不够智能想完全依赖大模型。正确理解智能系统必须有保守防线。# ❌ 错误纯AI判断resultllm.call(prompt)# 无兜底高风险字段漏标# ✅ 正确规则优先ai_resultllm.call(prompt)final_resultrule_engine.apply_rules(field_name,system_type,ai_result)3.3 反馈闭环需持续运营错误做法上线后做一次标注就结束。# ❌ 错误一次性标注annotationscollect_initial_annotations()updater.update(annotations)# 上线后无更新# ✅ 正确持续运营defweekly_update_job():feedback_loop.weekly_review()# 每周自动运行建议节奏每周review一次每月更新标注库。四、系统成熟度自检五、技术选型建议5.1 大模型选择5.2 标注库实现# 推荐方案关系数据库存储CREATE TABLE annotations(idSERIAL PRIMARY KEY,field_id VARCHAR(64),original_level VARCHAR(32),corrected_level VARCHAR(32),reason TEXT,reviewer VARCHAR(64),created_at TIMESTAMP,synced BOOLEAN DEFAULT FALSE);5.3 定时任务# 推荐方案使用定时任务框架fromapschedulerimportScheduler schedulerScheduler()scheduler.add_job(feedback_loop.weekly_review,cron,day_of_weekmon,hour9)scheduler.start()六、总结AI驱动的数据分类分级不是把大模型当万能工具而是构建三层架构上下文注入让AI看见业务有依据判断反馈闭环让AI持续进化不原地踏步规则兜底让AI有安全网高风险可控核心投入优先级上下文引擎 规则引擎 反馈闭环。缺少这三层架构任何AI分类系统都注定不稳定。
