AI驱动的数据分类分级：工程化架构设计与落地实践详解-平芜编程栈

背景

数据分类分级是数据治理的核心环节，也是合规建设的必答题。随着《数据安全法》实施，政企单位纷纷引入大模型自动化处理这项工作。
但实际项目中，我们发现一个普遍问题：AI分类系统上线初期准确率90%+，三个月后退化到60%以下。
这不是大模型能力问题，而是工程化架构缺失导致的系统性缺陷。本文从真实项目经验出发，剖析问题根因，给出可落地的架构设计方案。

一、问题剖析：简单调用大模型的三大技术缺陷

1.1 信息输入不完整

大模型的判断能力来源于输入信息。典型的"简单调用"模式只输入字段基本信息：

# 简单调用示例input_text="字段名：phone，类型：varchar(20)"prompt=f"请判断以下字段的数据级别：{input_text}"result=llm_api.call(prompt)

问题：数据分级取决于三个维度——字段属性、业务系统归属、数据使用场景。简单调用只输入了第一维，大模型只能靠"通用知识"猜测。

1.2 学习机制缺失

大模型在推理阶段是"静态"的——每次调用独立，之前的结果不影响后续判断。

# 人工修正无法影响模型wrong_result="一般数据"# AI把身份证号标成一般corrected="核心数据"# 人工修正# 下次调用，AI仍会犯相同错误next_result=llm_api.call("字段名：id_card")# 还是"一般数据"

结论：没有反馈机制，AI永远停留在上线那天的能力水平。

1.3 风险控制缺失

大模型输出是概率性的——同一输入不同调用可能给出不同结果。在合规场景，"大概率正确"是危险的。

# 同一输入，不同调用结果可能不同result_1=llm_api.call(prompt)# "核心数据"result_2=llm_api.call(prompt)# "重要数据"result_3=llm_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_catalog=asset_catalog self.system_registry=system_registry self.rule_lib=rule_lib self.prompt_template=self._load_template()defbuild_context(self,field_id:str)->dict:"""构建完整上下文信息"""# 1. 获取字段基本信息field_info=self.asset_catalog.get_field(field_id)# 2. 获取所属表/库信息table_info=self.asset_catalog.get_table(field_info['table_id'])db_info=self.asset_catalog.get_database(table_info['db_id'])# 3. 获取业务系统信息system_info=self.system_registry.get_by_database(db_info['id'])# 4. 获取合规规则compliance_rules=self.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:"""组装完整prompt"""returnself.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_rules=business_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.store=annotation_store self.updater=learning_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_annotations=self.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_engine=ContextEngine(...)self.llm_client=LLMClient(...)self.rule_engine=RuleEngine(...)self.feedback_loop=FeedbackLoop(...)defclassify(self,field_id:str)->str:"""完整分类流程"""# Step 1: 构建上下文context=self.context_engine.build_context(field_id)prompt=self.context_engine.assemble_prompt(context)# Step 2: AI判断ai_result=self.llm_client.call(prompt)# Step 3: 规则兜底final_result=self.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 上下文引擎是核心，不可省略

错误做法：为快速上线，跳过上下文引擎，只做字段级判断。

# ❌ 错误：只输入字段名result=llm.call("字段名：phone")# ✅ 正确：输入完整上下文context=context_engine.build_context(field_id)result=llm.call(assemble_prompt(context))

后果：上线快，退化也快。一个月开发投入 → 系统稳定运行三年。

3.2 规则引擎是防线，不嫌"笨"

错误观点：规则引擎不够智能，想完全依赖大模型。
正确理解：智能系统必须有保守防线。

# ❌ 错误：纯AI判断result=llm.call(prompt)# 无兜底，高风险字段漏标# ✅ 正确：规则优先ai_result=llm.call(prompt)final_result=rule_engine.apply_rules(field_name,system_type,ai_result)

3.3 反馈闭环需持续运营

错误做法：上线后做一次标注就结束。

# ❌ 错误：一次性标注annotations=collect_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 scheduler=Scheduler()scheduler.add_job(feedback_loop.weekly_review,'cron',day_of_week='mon',hour=9)scheduler.start()