feat: add finding preprocessor

core/finding_preprocessor.py

+from __future__ import annotations
+
+import json
+import logging
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.memory import Finding
+    from core.tools.segment_llm import LLMTool
+
+logger = logging.getLogger(__name__)
+
+
+FINDING_PREPROCESSOR_SYSTEM_PROMPT = """你是合同审查结果文本处理助手。
+你的任务是严格按照给定的处理要求修改建议文本。
+只允许执行明确要求的文字调整，不得补充、删减、解释或改写其他内容。
+返回 JSON 对象，格式为：{"suggestion": "处理后的建议文本"}。"""
+
+FINDING_PREPROCESSOR_USER_PROMPT = """请处理以下合同审查建议：
+
+审查项：{rule_title}
+处理要求：{processing_requirement}
+原始建议：{suggestion}
+
+请仅返回 JSON 对象。"""
+
+
+class FindingPreprocessor:
+    """在 finding 入库前按审查项执行文本处理。"""
+
+    _LLM_PROCESSING_REQUIREMENTS = {
+        "安装调试审查": "将建议文本中的“指导”替换为“支持”，其他内容保持不变。",
+    }
+
+    _SUGGESTION_REPLACEMENTS = {
+        "安装调试审查": {
+            "指导": "支持",
+        },
+    }
+
+    def __init__(self, llm_tool: LLMTool | None = None) -> None:
+        self._llm_tool = llm_tool
+
+    def process(self, finding: Finding,use_rule=True) -> Finding:
+        """优先使用 LLM 处理，调用失败时回退到确定性规则。"""
+        if use_rule:
+            return self.process_with_rules(finding)
+        return self.process_with_llm(finding)
+
+    def process_with_llm(self, finding: Finding) -> Finding:
+        rule_title = (finding.rule_title or "").strip()
+        requirement = self._LLM_PROCESSING_REQUIREMENTS.get(rule_title)
+        if not requirement or not finding.suggestion:
+            return finding
+
+        user_content = FINDING_PREPROCESSOR_USER_PROMPT.format(
+            rule_title=rule_title,
+            processing_requirement=requirement,
+            suggestion=json.dumps(finding.suggestion, ensure_ascii=False),
+        )
+
+        try:
+            llm_tool = self._get_llm_tool()
+            messages = llm_tool.build_messages(
+                user_content,
+                system_content=FINDING_PREPROCESSOR_SYSTEM_PROMPT,
+            )
+            response = llm_tool.run_with_loop(llm_tool.chat_async(messages))
+            data = llm_tool.parse_first_json(response)
+            suggestion = str(data.get("suggestion", "") or "").strip()
+            if not suggestion:
+                raise ValueError("LLM returned an empty suggestion")
+            finding.suggestion = suggestion
+            return finding
+        except Exception as exc:
+            logger.warning(
+                "LLM finding preprocessing failed, fallback to rule processing: %s",
+                exc,
+            )
+            return self.process_with_rules(finding)
+
+    def process_with_rules(self, finding: Finding) -> Finding:
+        replacements = self._SUGGESTION_REPLACEMENTS.get(
+            (finding.rule_title or "").strip(), {}
+        )
+        for source, target in replacements.items():
+            finding.suggestion = (finding.suggestion or "").replace(source, target)
+        return finding
+
+    def _get_llm_tool(self) -> LLMTool:
+        if self._llm_tool is None:
+            from core.tools.segment_llm import LLMTool
+
+            self._llm_tool = LLMTool(llm_key="base_tool_llm")
+        return self._llm_tool

core/memory.py

@@ -12,6 +12,7 @@ from uuid import uuid4
 from utils.http_util import upload_file
 from utils.doc_util import DocBase
 from core.config import META_KEY, FILE_SUFFIX, use_lufa
+from core.finding_preprocessor import FindingPreprocessor
 
 logger = logging.getLogger(__name__)
 
@@ -65,7 +66,7 @@ class Finding:
     def __repr__(self):
         return (
             f"Finding(id={self.id!r}, rule_title={self.rule_title!r}, segment_id={self.segment_id}, "
-            f"issue={self.issue!r}, risk_level={self.risk_level!r}, result={self.result!r})"
+            f"issue={self.issue!r}, risk_level={self.risk_level!r}, result={self.result!r}),suggestion={self.suggestion!r}"
         )
 
 
@@ -82,6 +83,7 @@ class MemoryStore:
         self.facts: List[Dict[str, Any]] = []
         self.merge_facts: List[Dict[str, Any]] = []
         self.findings: Dict[str, List[Finding]] = {}
+        self._finding_preprocessor = FindingPreprocessor()
         self._load()
 
     # ---------------------- facts ----------------------
@@ -179,6 +181,7 @@ class MemoryStore:
 
     def _add_finding(self, key: str, finding: Finding) -> Finding:
         with self._lock:
+            finding = self._finding_preprocessor.process(finding)
             finding_key = self._normalize_finding_key(key)
             if not finding.id:
                 finding.id = uuid4().hex
@@ -635,12 +638,12 @@ def test_memory_and_export_excel():
     )
     # print( store.search_facts(['支付']))
     finding1 = Finding(
-        rule_title="违约责任",
+        rule_title="安装调试审查",
         segment_id=1,
-        original_text="违约方应赔偿全部损失",
-        issue="未约定违约金上限，可能导致赔偿范围过大",
+        original_text="安装调试审查测试原文",
+        issue="安装调试审查中不能有指导",
         risk_level="H",
-        suggestion="建议增加‘赔偿金额不超过合同总额的30%’",
+        suggestion="修改为安装调试指导。",
     )
     finding2 = Finding(
         rule_title="违约责任",
@@ -659,11 +662,12 @@ def test_memory_and_export_excel():
     # print("Findings search:")
     # for f in hits:
     #   print(json.dumps(asdict(f), ensure_ascii=False, indent=2))
-    print(store.export_to_excel("测试"))
+    # print(store.export_to_excel("测试"))
 
 
 if __name__ == "__main__":
-    test_export_findings_to_doc_comments(
-        "/home/ccran/lufa-contract/tmp/1_金盘箱变采购合同.docx"
-    )
-    # test_memory_and_export_excel()
+    # test_export_findings_to_doc_comments(
+    #     "/home/ccran/lufa-contract/tmp/1_金盘箱变采购合同.docx"
+    # )
+    test_memory_and_export_excel()
+    pass