Merge pull request 'feat: 用llm生成脱敏地址' (#3) from dev into main

Reviewed-on: #3

backend/app/core/document_handlers/ner_processor.py

@@ -616,12 +616,45 @@ class NerProcessor:
 
     def _mask_address(self, address: str) -> str:
         """
-        对地址进行脱敏处理：
+        对地址进行脱敏处理，使用LLM直接生成脱敏地址：
         保留区级以上地址，路名以大写首字母替代，门牌数字以****代替，大厦名、小区名以大写首字母替代
         """
         if not address:
             return address
         
+        try:
+            # 使用LLM生成脱敏地址
+            from ..prompts.masking_prompts import get_address_masking_prompt
+            
+            prompt = get_address_masking_prompt(address)
+            logger.info(f"Calling ollama to mask address: {address}")
+            
+            # 使用ollama客户端生成脱敏地址，带验证
+            response = self.ollama_client.generate_with_validation(
+                prompt=prompt,
+                response_type='address_masking',
+                return_parsed=True
+            )
+            
+            if response and isinstance(response, dict) and "masked_address" in response:
+                masked_address = response["masked_address"]
+                logger.info(f"Successfully masked address: {address} -> {masked_address}")
+                return masked_address
+            else:
+                logger.warning(f"Invalid response format for address masking: {response}")
+                return self._mask_address_fallback(address)
+                
+        except Exception as e:
+            logger.error(f"Error masking address with LLM: {e}")
+            return self._mask_address_fallback(address)
+    
+    def _mask_address_fallback(self, address: str) -> str:
+        """
+        地址脱敏的回退方法，使用原有的正则表达式和拼音转换逻辑
+        """
+        if not address:
+            return address
+        
         # 提取地址组件
         components = self._extract_address_components(address)
         

backend/app/core/prompts/masking_prompts.py

@@ -112,6 +112,46 @@ def get_ner_address_prompt(text: str) -> str:
     return prompt.format(text=text)
 
 
+def get_address_masking_prompt(address: str) -> str:
+    """
+    Returns a prompt that generates a masked version of an address following specific rules.
+    
+    Args:
+        address (str): The original address to be masked
+        
+    Returns:
+        str: The formatted prompt that will generate a masked address
+    """
+    prompt = textwrap.dedent("""
+你是一个专业的地址脱敏助手。请对给定的地址进行脱敏处理，遵循以下规则：
+
+脱敏规则：
+1. 保留区级以上地址（省、市、区、县）
+2. 路名以大写首字母替代，例如：恒丰路 -> HF路
+3. 门牌数字以**代替，例如：66号 -> **号
+4. 大厦名、小区名以大写首字母替代，例如：白云大厦 -> BY大厦
+5. 房间号以****代替，例如：1607室 -> ****室
+
+示例：
+- 输入：上海市静安区恒丰路66号白云大厦1607室
+- 输出：上海市静安区HF路**号BY大厦****室
+
+- 输入：北京市海淀区北小马厂6号1号楼华天大厦1306室
+- 输出：北京市海淀区北小马厂**号**号楼HT大厦****室
+
+请严格按照JSON格式输出结果：
+
+{{
+"masked_address": "脱敏后的地址"
+}}
+
+原始地址：{address}
+
+请严格按照JSON格式输出结果。
+    """)
+    return prompt.format(address=address)
+
+
 def get_ner_project_prompt(text: str) -> str:
     """
     Returns a prompt that generates a mapping of original project names to their masked versions.

backend/app/core/utils/llm_validator.py

@@ -125,6 +125,18 @@ class LLMResponseValidator:
         "required": ["road_name", "house_number", "building_name", "community_name"]
     }
     
+    # Schema for address masking responses
+    ADDRESS_MASKING_SCHEMA = {
+        "type": "object",
+        "properties": {
+            "masked_address": {
+                "type": "string",
+                "description": "The masked address following the specified rules"
+            }
+        },
+        "required": ["masked_address"]
+    }
+    
     @classmethod
     def validate_entity_extraction(cls, response: Dict[str, Any]) -> bool:
         """
@@ -230,6 +242,26 @@ class LLMResponseValidator:
             logger.warning(f"Response that failed validation: {response}")
             return False
     
+    @classmethod
+    def validate_address_masking(cls, response: Dict[str, Any]) -> bool:
+        """
+        Validate address masking response from LLM.
+        
+        Args:
+            response: The parsed JSON response from LLM
+            
+        Returns:
+            bool: True if valid, False otherwise
+        """
+        try:
+            validate(instance=response, schema=cls.ADDRESS_MASKING_SCHEMA)
+            logger.debug(f"Address masking validation passed for response: {response}")
+            return True
+        except ValidationError as e:
+            logger.warning(f"Address masking validation failed: {e}")
+            logger.warning(f"Response that failed validation: {response}")
+            return False
+    
     @classmethod
     def _validate_linkage_content(cls, response: Dict[str, Any]) -> bool:
         """
@@ -291,7 +323,8 @@ class LLMResponseValidator:
             'entity_linkage': cls.validate_entity_linkage,
             'regex_entity': cls.validate_regex_entity,
             'business_name_extraction': cls.validate_business_name_extraction,
-            'address_extraction': cls.validate_address_extraction
+            'address_extraction': cls.validate_address_extraction,
+            'address_masking': cls.validate_address_masking
         }
         
         validator = validators.get(response_type)
@@ -326,6 +359,8 @@ class LLMResponseValidator:
                 validate(instance=response, schema=cls.BUSINESS_NAME_EXTRACTION_SCHEMA)
             elif response_type == 'address_extraction':
                 validate(instance=response, schema=cls.ADDRESS_EXTRACTION_SCHEMA)
+            elif response_type == 'address_masking':
+                validate(instance=response, schema=cls.ADDRESS_MASKING_SCHEMA)
             else:
                 return f"Unknown response type: {response_type}"
             

backend/docs/ADDRESS_MASKING_IMPROVEMENT.md

@@ -0,0 +1,239 @@
+# 地址脱敏改进文档
+
+## 问题描述
+
+原始的地址脱敏方法使用正则表达式和拼音转换来手动处理地址组件，存在以下问题：
+- 需要手动维护复杂的正则表达式模式
+- 拼音转换可能失败，需要回退处理
+- 难以处理复杂的地址格式
+- 代码维护成本高
+
+## 解决方案
+
+### 1. LLM 直接生成脱敏地址
+
+使用 LLM 直接生成脱敏后的地址，遵循指定的脱敏规则：
+
+- **保留区级以上地址**：省、市、区、县
+- **路名缩写**：以大写首字母替代，如：恒丰路 -> HF路
+- **门牌号脱敏**：数字以**代替，如：66号 -> **号
+- **大厦名缩写**：以大写首字母替代，如：白云大厦 -> BY大厦
+- **房间号脱敏**：以****代替，如：1607室 -> ****室
+
+### 2. 实现架构
+
+#### 核心组件
+
+1. **`get_address_masking_prompt()`** - 生成地址脱敏 prompt
+2. **`_mask_address()`** - 主要的脱敏方法，使用 LLM
+3. **`_mask_address_fallback()`** - 回退方法，使用原有逻辑
+
+#### 调用流程
+
+```
+输入地址
+    ↓
+生成脱敏 prompt
+    ↓
+调用 Ollama LLM
+    ↓
+解析 JSON 响应
+    ↓
+返回脱敏地址
+    ↓
+失败时使用回退方法
+```
+
+### 3. Prompt 设计
+
+#### 脱敏规则说明
+```
+脱敏规则：
+1. 保留区级以上地址（省、市、区、县）
+2. 路名以大写首字母替代，例如：恒丰路 -> HF路
+3. 门牌数字以**代替，例如：66号 -> **号
+4. 大厦名、小区名以大写首字母替代，例如：白云大厦 -> BY大厦
+5. 房间号以****代替，例如：1607室 -> ****室
+```
+
+#### 示例展示
+```
+示例：
+- 输入：上海市静安区恒丰路66号白云大厦1607室
+- 输出：上海市静安区HF路**号BY大厦****室
+
+- 输入：北京市海淀区北小马厂6号1号楼华天大厦1306室
+- 输出：北京市海淀区北小马厂**号**号楼HT大厦****室
+```
+
+#### JSON 输出格式
+```json
+{
+"masked_address": "脱敏后的地址"
+}
+```
+
+## 实现细节
+
+### 1. 主要方法
+
+#### `_mask_address(address: str) -> str`
+```python
+def _mask_address(self, address: str) -> str:
+    """
+    对地址进行脱敏处理，使用LLM直接生成脱敏地址
+    """
+    if not address:
+        return address
+    
+    try:
+        # 使用LLM生成脱敏地址
+        prompt = get_address_masking_prompt(address)
+        response = self.ollama_client.generate_with_validation(
+            prompt=prompt,
+            response_type='address_masking',
+            return_parsed=True
+        )
+        
+        if response and isinstance(response, dict) and "masked_address" in response:
+            return response["masked_address"]
+        else:
+            return self._mask_address_fallback(address)
+            
+    except Exception as e:
+        logger.error(f"Error masking address with LLM: {e}")
+        return self._mask_address_fallback(address)
+```
+
+#### `_mask_address_fallback(address: str) -> str`
+```python
+def _mask_address_fallback(self, address: str) -> str:
+    """
+    地址脱敏的回退方法，使用原有的正则表达式和拼音转换逻辑
+    """
+    # 原有的脱敏逻辑作为回退
+```
+
+### 2. Ollama 调用模式
+
+遵循现有的 Ollama 客户端调用模式，使用验证：
+
+```python
+response = self.ollama_client.generate_with_validation(
+    prompt=prompt,
+    response_type='address_masking',
+    return_parsed=True
+)
+```
+
+- `response_type='address_masking'`：指定响应类型进行验证
+- `return_parsed=True`：返回解析后的 JSON
+- 自动验证响应格式是否符合 schema
+
+## 测试结果
+
+### 测试案例
+
+| 原始地址 | 期望脱敏结果 |
+|----------|-------------|
+| 上海市静安区恒丰路66号白云大厦1607室 | 上海市静安区HF路**号BY大厦****室 |
+| 北京市海淀区北小马厂6号1号楼华天大厦1306室 | 北京市海淀区北小马厂**号**号楼HT大厦****室 |
+| 天津市津南区双港镇工业园区优谷产业园5号楼-1505 | 天津市津南区双港镇工业园区优谷产业园**号楼-**** |
+
+### Prompt 验证
+
+- ✓ 包含脱敏规则说明
+- ✓ 提供具体示例
+- ✓ 指定 JSON 输出格式
+- ✓ 包含原始地址
+- ✓ 指定输出字段名
+
+## 优势
+
+### 1. 智能化处理
+- LLM 能够理解复杂的地址格式
+- 自动处理各种地址变体
+- 减少手动维护成本
+
+### 2. 可靠性
+- 回退机制确保服务可用性
+- 错误处理和日志记录
+- 保持向后兼容性
+
+### 3. 可扩展性
+- 易于添加新的脱敏规则
+- 支持多语言地址处理
+- 可配置的脱敏策略
+
+### 4. 一致性
+- 统一的脱敏标准
+- 可预测的输出格式
+- 便于测试和验证
+
+## 性能影响
+
+### 1. 延迟
+- LLM 调用增加处理时间
+- 网络延迟影响响应速度
+- 回退机制提供快速响应
+
+### 2. 成本
+- LLM API 调用成本
+- 需要稳定的网络连接
+- 回退机制降低依赖风险
+
+### 3. 准确性
+- 显著提高脱敏准确性
+- 减少人工错误
+- 更好的地址理解能力
+
+## 配置参数
+
+- `response_type`: 响应类型，用于验证 (默认: 'address_masking')
+- `return_parsed`: 是否返回解析后的 JSON (默认: True)
+- `max_retries`: 最大重试次数 (默认: 3)
+
+## 验证 Schema
+
+地址脱敏响应必须符合以下 JSON schema：
+
+```json
+{
+  "type": "object",
+  "properties": {
+    "masked_address": {
+      "type": "string",
+      "description": "The masked address following the specified rules"
+    }
+  },
+  "required": ["masked_address"]
+}
+```
+
+## 使用示例
+
+```python
+from app.core.document_handlers.ner_processor import NerProcessor
+
+processor = NerProcessor()
+original_address = "上海市静安区恒丰路66号白云大厦1607室"
+masked_address = processor._mask_address(original_address)
+print(f"Original: {original_address}")
+print(f"Masked: {masked_address}")
+```
+
+## 未来改进方向
+
+1. **缓存机制**：缓存常见地址的脱敏结果
+2. **批量处理**：支持批量地址脱敏
+3. **自定义规则**：支持用户自定义脱敏规则
+4. **多语言支持**：扩展到其他语言的地址处理
+5. **性能优化**：异步处理和并发调用
+
+## 相关文件
+
+- `backend/app/core/document_handlers/ner_processor.py` - 主要实现
+- `backend/app/core/prompts/masking_prompts.py` - Prompt 函数
+- `backend/app/core/services/ollama_client.py` - Ollama 客户端
+- `backend/app/core/utils/llm_validator.py` - 验证 schema 和验证方法
+- `backend/test_validation_schema.py` - 验证 schema 测试