地址去括号、统称谓，MGeo前处理这样做-程序员充电站

地址去括号、统称谓，MGeo前处理这样做

在地址数据清洗与标准化任务中，同一地理位置常因表述差异导致匹配失败。例如，“北京市海淀区中关村大街27号”与“中关村大街27号（海淀区）”本应指向同一地点，却因括号内容和行政区划命名方式不同而被误判。阿里达摩院联合高德推出的MGeo模型为中文地址相似度匹配提供了强大支持，但要充分发挥其能力，合理的前处理策略至关重要。

本文将围绕“去括号”与“统称谓”两大核心预处理步骤，结合MGeo地址相似度匹配镜像的实际使用流程，系统讲解如何提升地址对齐的准确率与效率。

1. MGeo地址匹配的核心能力与挑战

1.1 模型背景与技术优势

MGeo是专为中文地理文本设计的多模态预训练模型，融合了地图结构信息与自然语言语义，在以下任务中表现优异：

地址要素识别：自动提取省、市、区、街道、门牌等结构化信息
地址相似度判断：区分完全匹配、部分匹配与不匹配三类关系
模糊地址理解：解析“静安寺附近”、“五道口商圈”等非精确描述

相比传统正则规则或编辑距离方法，MGeo能理解“朝阳区”与“朝阳”属于同一行政层级，具备更强的语义泛化能力。

1.2 实际应用中的主要干扰因素

尽管MGeo具备较强的鲁棒性，但在真实业务场景中仍面临以下挑战：

干扰类型	示例	影响
括号标注	“杭州市西湖区文三路159号(东部软件园)”	模型可能误认为括号内为地址主体
行政区简称	“浙江杭州” vs “浙江省杭州市”	增加语义歧义风险
标点差异	使用顿号、逗号或空格分隔	可能影响token切分一致性
字符冗余	包含电话、联系人等附加信息	引入噪声降低匹配精度

因此，在输入模型前进行标准化预处理，是提升整体效果的关键环节。

2. 预处理关键技术：去括号与统称谓

2.1 去除括号及其内容

括号常用于补充说明，如园区名、楼宇别名、联系方式等，这些信息虽有助于人工识别，但会干扰模型对主地址的判断。

正则表达式实现方案

import re def remove_parentheses(address: str) -> str: """ 移除字符串中的各类括号及其内容 支持：()、[]、【】、<>、{} 等常见括号类型 """ # 定义多种括号模式 patterns = [ r'\([^)]*\)', # 圆括号 r'\[[^\]]*\]', # 方括号 r'【[^】]*】', # 中文方括号 r'<[^>]*>', # 尖括号 r'{[^}]*}' # 花括号 ] cleaned = address for pattern in patterns: cleaned = re.sub(pattern, '', cleaned) return cleaned.strip() # 测试示例 print(remove_parentheses("杭州市西湖区文三路159号(东部软件园)")) # 输出：杭州市西湖区文三路159号

注意事项

若括号内包含关键地址信息（如“虹桥火车站（高铁站）”），建议保留并替换为通用表述
可设置白名单机制，对特定关键词（如“医院”、“学校”）做例外处理

2.2 统一行政区划称谓

不同来源的数据中，省市区常以全称、简称混合出现，影响模型一致性判断。

标准化映射表构建

# 构建行政区划归一化字典 ADMIN_LEVEL_NORMALIZATION = { # 省级单位 '省': '', '自治区': '', '直辖市': '', # 市级单位 '市': '', '地区': '', '自治州': '', '盟': '', # 区县级单位 '区': '', '县': '', '旗': '', '自治县': '', '市辖区': '', '县级市': '' } def normalize_administrative_terms(address: str) -> str: """ 统一行政区划术语，去除冗余后缀 """ result = address for term, replacement in ADMIN_LEVEL_NORMALIZATION.items(): result = result.replace(term, replacement) # 多个空格合并为一个 result = re.sub(r'\s+', ' ', result) return result.strip() # 测试示例 print(normalize_administrative_terms("浙江省杭州市西湖区")) # 输出：浙江杭州西湖

进阶优化建议

结合NLP实体识别结果，仅对已识别为“行政区”的token进行处理
对于“北京”、“上海”等直辖市，无需去除“市”字也可接受，可选择性保留

3. 基于MGeo镜像的完整实践流程

3.1 环境部署与初始化

CSDN算力平台提供的预置镜像已集成MGeo所需全部依赖，部署步骤如下：

创建实例时选择镜像：“MGeo地址相似度匹配实体对齐-中文-地址领域”
启动后通过JupyterLab进入终端环境
激活Python环境：
```
conda activate py37testmaas
```

验证环境可用性：

from modelscope.pipelines import pipeline tagger = pipeline('token-classification', 'damo/mgeo_geographic_elements_tagging_chinese_base') print(tagger('北京市海淀区中关村'))

预期输出包含prov、city、district等地域标签。

3.2 完整地址比对脚本示例

import pandas as pd import re from modelscope.pipelines import pipeline from modelscope.utils.constant import Tasks # 初始化地址相似度管道 similarity_pipeline = pipeline( task=Tasks.sentence_similarity, model='damo/mgeo_address_similarity_chinese_base', batch_size=16 # 根据显存调整 ) def preprocess_address(addr: str) -> str: """综合预处理函数""" if not isinstance(addr, str): return "" # 1. 去除各类括号内容 addr = re.sub(r'\([^)]*\)', '', addr) addr = re.sub(r'\[[^\]]*\]', '', addr) addr = re.sub(r'【[^】]*】', '', addr) # 2. 统一行政区划称谓 for term in ['省', '市', '区', '县', '镇', '乡', '街道']: addr = addr.replace(term, '') # 3. 清理多余空白 addr = re.sub(r'\s+', '', addr) return addr.strip() # 加载待匹配数据 df = pd.read_excel('/root/workspace/addresses.xlsx') # 包含address1, address2列 # 预处理两列地址 df['addr1_clean'] = df['address1'].apply(preprocess_address) df['addr2_clean'] = df['address2'].apply(preprocess_address) # 批量执行相似度匹配 results = [] for _, row in df.iterrows(): try: result = similarity_pipeline(input=(row['addr1_clean'], row['addr2_clean'])) results.append(result['output']['label']) except Exception as e: results.append('error') df['match_label'] = results # 保存结果 df.to_excel('/root/workspace/matched_results.xlsx', index=False)

3.3 性能优化与错误处理

显存不足应对策略

当batch_size过大导致OOM时，可通过以下方式优化：

# 动态调整batch_size def get_optimal_batch_size(model_name): import torch total_memory = torch.cuda.get_device_properties(0).total_memory / (1024**3) if total_memory < 8: return 8 elif total_memory < 16: return 16 else: return 32 similarity_pipeline = pipeline( task=Tasks.sentence_similarity, model='damo/mgeo_address_similarity_chinese_base', batch_size=get_optimal_batch_size('mgeo') )

添加重试机制保障稳定性

from tenacity import retry, stop_after_attempt, wait_exponential @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, max=10)) def robust_match(pipe, addr1, addr2): return pipe(input=(addr1, addr2))['output']['label']