dev：终极版规范性审查

core/construction_review/component/reviewers/reference_basis_reviewer.py

@@ -8,6 +8,8 @@ from typing import Any, Dict, List, Optional
 from core.construction_review.component.reviewers.utils.directory_extraction import BasisItem, BasisItems
 from core.construction_review.component.reviewers.utils.inter_tool import InterTool
 from core.construction_review.component.reviewers.utils.prompt_loader import PromptLoader
+from core.construction_review.component.reviewers.utils.punctuation_checker import check_punctuation
+from core.construction_review.component.reviewers.utils.punctuation_result_processor import process_punctuation_results
 from foundation.observability.logger.loggering import server_logger as logger
 from langchain_core.prompts import ChatPromptTemplate
 from langchain_openai import ChatOpenAI
@@ -145,16 +147,23 @@ class BasisReviewService:
 
         async with self._semaphore:
             try:
-                # 构建提示词模板和用户内容
-                prompt_template = self.message_builder.get_prompt_template()
-                message = prompt_template.partial(check_content=basis_items)
-                trace_id = f"prep_basis_batch_{int(time.time())}"
-                llm_out = await self.llm_client.review_basis(message, trace_id)
-                print("LLM输出：\n")
-                print(llm_out)
+                # 第一步：调用标点符号检查器
                 
-                # 使用标准化处理器处理响应
-                standardized_result = self.response_processor.process_llm_response(llm_out, "reference_check", "basis","basis_reference_check")
+                checker_result = await check_punctuation(basis_items)
+                print(checker_result)
+                
+                # 第二步：调用结果处理器，生成详细的问题分析报告
+                processor_result = await process_punctuation_results(checker_result)
+                print("\n【第二步】问题分析报告输出：")
+                print(processor_result)
+                
+                # 第三步：转换为标准格式
+                standardized_result = self.response_processor.process_llm_response(
+                    processor_result, 
+                    "reference_check", 
+                    "basis",
+                    "basis_reference_check"
+                )
 
                 # 统计问题数量
                 issue_count = sum(1 for item in standardized_result if item.get('exist_issue', False))

core/construction_review/component/reviewers/utils/punctuation_checker.py

@@ -0,0 +1,272 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+import json
+from typing import List, Optional
+
+from pydantic import BaseModel, Field, ValidationError
+from langchain_core.prompts import ChatPromptTemplate
+from langchain_core.output_parsers import PydanticOutputParser, StrOutputParser
+from langchain_openai import ChatOpenAI
+
+
+# ===== 1) 定义结构 =====
+class PunctuationResult(BaseModel):
+    original_text: str = Field(..., description="审查的规范原文，与输入完全一致")
+    title_mark_status: bool = Field(..., description="书名号使用是否正确，true表示正确，false表示错误")
+    bracket_status: Optional[bool] = Field(..., description="括号使用是否正确，true表示正确，false表示错误，null表示没有编号")
+
+
+class PunctuationResults(BaseModel):
+    items: List[PunctuationResult]
+
+
+# ===== 2) SYSTEM Prompt =====
+SYSTEM = """
+你是【标点符号规范性检查助手】。
+
+【任务】
+仅对已通过“成对出现”预检的文本，检查书名号和括号是否**包裹完整且位置正确**。
+（预检已经保证：
+    - 书名号《》至少各出现一次且数量相等
+    - 括号（/（ ）/）至少各出现一次且数量相等（中文、英文括号视为同类）
+    因此你只需判断包裹范围是否正确、是否遗漏内容。）
+
+【判断标准】
+- title_mark_status：书名号需完全包裹规范名称，且不多包/漏包
+- bracket_status：括号需完全包裹规范编号，且不多包/漏包；编号可能是各种形式，如果文本中没有编号，设置为null
+
+【输出要求】
+- 为每个输入文本输出一个检查结果
+- 确保输出数量与输入一致
+- original_text 必须与输入完全一致
+- title_mark_status 必须是布尔值：true表示正确，false表示错误
+- bracket_status 必须是布尔值或null：true表示正确，false表示错误，null表示没有编号
+"""
+
+HUMAN = """
+请检查以下文本中书名号和括号的**内容是否全部被包裹**，以及是否有编号。
+（所有文本已通过成对出现的预检，至少各有一对《》且数量相等。）
+
+【判断原则】
+- 仅检查包裹的**完整性**：书名号是否包裹了规范名称的全部内容；括号是否包裹了编号的全部内容
+- 中文括号（）和英文括号()混用视为正常，不区分
+- 若内容在符号外遗漏，或符号包裹了多余内容，则判定为false
+- **重要**：如果文本中没有编号（完全没有任何()或（）符号），则bracket_status设置为null
+
+【简单示例】
+示例1：《建筑抗震设》计规范 （GB 50011-2001）
+- 规范名称是"建筑抗震设计规范"，但只有"建筑抗震设"被包裹，"计规范"在外 → title_mark_status=false
+- 编号被完整包裹 → bracket_status=true
+
+示例2：《建筑抗震设计规范》
+- 书名号包裹了完整的规范名称 → title_mark_status=true
+- 没有编号 → bracket_status=null
+
+示例3：《建筑抗震设计规范》(GB 50011-2001)
+- 书名号包裹了完整的规范名称 → title_mark_status=true
+- 英文括号包裹了完整的编号 → bracket_status=true（混用不算错）
+
+【待检查文本】
+{items}
+
+【输出格式要求】
+{format_instructions}
+/no_think
+"""
+
+# ===== 3) Output Parser =====
+parser = PydanticOutputParser(pydantic_object=PunctuationResults)
+
+# ===== 4) Prompt =====
+prompt = ChatPromptTemplate.from_messages([
+    ("system", SYSTEM),
+    ("human", HUMAN)
+])
+
+# ===== 5) LLM =====
+llm = ChatOpenAI(
+    model="qwen3-30b",
+    base_url="http://192.168.91.253:8003/v1",
+    api_key="sk-123456",
+    temperature=0.7,
+)
+
+
+# ===== 6) 提取第一个 JSON =====
+def extract_first_json(text: str) -> dict:
+    """从任意模型输出中提取第一个完整 JSON 对象 { ... }"""
+    start = text.find("{")
+    if start == -1:
+        raise ValueError("未找到 JSON 起始 '{'")
+
+    depth = 0
+    for i in range(start, len(text)):
+        ch = text[i]
+        if ch == "{":
+            depth += 1
+        elif ch == "}":
+            depth -= 1
+            if depth == 0:
+                return json.loads(text[start:i + 1])
+
+    raise ValueError("JSON 花括号未闭合")
+
+
+# ===== 7) 核心方法 =====
+async def check_punctuation(items: List[str]) -> str:
+    """
+    检查规范文本中的书名号和括号使用是否正确，先进行成对预检，再用LLM判断包裹完整性
+    
+    Args:
+        items: 待检查的规范文本列表
+        
+    Returns:
+        检查结果的JSON字符串，包含三个字段：
+        - original_text: 原文
+        - title_mark_status: 书名号使用是否正确（true/false）
+        - bracket_status: 括号使用是否正确（true/false/null，null表示没有编号）
+    """
+    # 1) 预检：是否存在且成对出现
+    pre_results = []  # 预填结果，若需LLM再补充
+    llm_inputs = []   # 需要LLM判定包裹完整性的文本
+
+    for text in items:
+        # 书名号成对判定
+        left_title = text.count("《")
+        right_title = text.count("》")
+        title_pair_ok = left_title == right_title and left_title > 0
+
+        # 括号成对判定（中英文括号混用视为同类）
+        left_br = text.count("(") + text.count("（")
+        right_br = text.count(")") + text.count("）")
+        bracket_pair_ok = left_br == right_br and left_br > 0
+        
+        # 只有书名号和括号都存在时，才判断一一对应
+        # 情况1：都不存在 → 都为False
+        if left_title == 0 and left_br == 0:
+            pre_results.append({
+                "original_text": text,
+                "title_mark_status": False,
+                "bracket_status": None
+            })
+            continue
+        
+        # 情况2：只有书名号，没有括号 → bracket_status为None
+        if left_title > 0 and left_br == 0:
+            if title_pair_ok:
+                llm_inputs.append(text)
+            else:
+                pre_results.append({
+                    "original_text": text,
+                    "title_mark_status": False,
+                    "bracket_status": None
+                })
+            continue
+        
+        # 情况3：只有括号，没有书名号 → title_mark_status为False
+        if left_title == 0 and left_br > 0:
+            pre_results.append({
+                "original_text": text,
+                "title_mark_status": False,
+                "bracket_status": bool(bracket_pair_ok)
+            })
+            continue
+        
+        # 情况4：两者都存在，判断一一对应
+        if left_title != left_br:
+            # 数量不对应，两个都为False
+            pre_results.append({
+                "original_text": text,
+                "title_mark_status": False,
+                "bracket_status": False
+            })
+            continue
+        
+        # 检查括号是否在书名号之后
+        bracket_after_title = True
+        if bracket_pair_ok and title_pair_ok:
+            # 找最后一个书名号的位置
+            last_title_pos = max(text.rfind("《"), text.rfind("》"))
+            # 找第一个括号的位置
+            first_bracket_pos = min(
+                text.find("(") if "(" in text else float('inf'),
+                text.find("（") if "（" in text else float('inf')
+            )
+            bracket_after_title = last_title_pos < first_bracket_pos
+
+        if not title_pair_ok or not bracket_pair_ok or not bracket_after_title:
+            # 预检失败或位置不正确，直接判定对应项为False，无需LLM
+            pre_results.append({
+                "original_text": text,
+                "title_mark_status": bool(title_pair_ok),
+                "bracket_status": bool(bracket_pair_ok and bracket_after_title)
+            })
+        else:
+            # 成对且位置正确通过，交给LLM判断包裹是否完整和是否有编号
+            llm_inputs.append(text)
+
+    # 若无需要LLM的，直接返回预检结果
+    if not llm_inputs:
+        return json.dumps(pre_results, ensure_ascii=False, indent=2)
+
+    chain = prompt | llm | StrOutputParser()
+    format_instructions = parser.get_format_instructions()
+
+    payload = {
+        "items": json.dumps(llm_inputs, ensure_ascii=False, indent=2),
+        "format_instructions": format_instructions
+    }
+
+    last_err = None
+
+    llm_result: List[dict] = []
+    for _ in range(2):
+        try:
+            raw = await chain.ainvoke(payload)
+            data = extract_first_json(raw)
+            findings = PunctuationResults.model_validate(data)
+            llm_result = [x.model_dump() for x in findings.items]
+            break
+        except (Exception, ValidationError, json.JSONDecodeError) as e:
+            last_err = e
+            print(f"[标点符号检查] 解析失败，重试中: {e}")
+
+    if last_err and not llm_result:
+        raise RuntimeError(f"标点符号检查失败：{last_err}") from last_err
+
+    # 合并预检与LLM结果，按原输入顺序输出
+    merged = []
+    llm_map = {item["original_text"]: item for item in llm_result}
+    for text in items:
+        # 先看预检是否已有
+        found = next((r for r in pre_results if r["original_text"] == text), None)
+        if found:
+            merged.append(found)
+        else:
+            merged.append(llm_map.get(text, {
+                "original_text": text,
+                "title_mark_status": False,
+                "bracket_status": None
+            }))
+
+    return json.dumps(merged, ensure_ascii=False, indent=2)
+
+
+# ===== 8) 示例 =====
+if __name__ == "__main__":
+    import asyncio
+
+    # 测试用例
+    test_items = [
+        "(4)《中华人民共和国突发事件应对法》【主席令〔2007〕第 69 号】；",  # 正确
+        "《混》凝土结构设计规范（GB 50010-2010）",      # 缺少书名号
+        "建筑施工组织设计规范GB/T 50502-2015",  # 缺少括号
+        "《建筑抗震设计规范》（GB 50011）-2001",       # 括号不成对
+        "《城市道路工程设计规范（CJJ 37-2012）",    # 书名号不成对
+        "《公路工程技术标准》(JTG B01-2014）",     # 正确
+    ]
+
+    result = asyncio.run(check_punctuation(test_items))
+    print("\n标点符号检查结果：")
+    print(result)

core/construction_review/component/reviewers/utils/punctuation_result_processor.py

@@ -0,0 +1,202 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+import json
+from typing import List, Literal
+
+from pydantic import BaseModel, Field, ValidationError
+from langchain_core.prompts import ChatPromptTemplate
+from langchain_core.output_parsers import PydanticOutputParser, StrOutputParser
+from langchain_openai import ChatOpenAI
+
+
+# ===== 1) 定义结构 =====
+RiskLevel = Literal["无风险", "中风险"]
+
+
+class PunctuationIssueResult(BaseModel):
+    issue_point: str = Field(..., description="问题点描述")
+    location: str = Field(..., description="审查内容，与输入完全一致")
+    suggestion: str = Field(..., description="修改建议，可执行动作")
+    reason: str = Field(..., description="问题的原因分析")
+    risk_level: RiskLevel = Field(..., description='风险水平，只能是 "无风险" / "中风险"')
+
+
+class PunctuationIssueResults(BaseModel):
+    items: List[PunctuationIssueResult]
+
+
+# ===== 2) SYSTEM Prompt =====
+SYSTEM = """
+你是【编制依据格式问题分析专家】。
+
+【任务】
+根据标点符号检查结果，生成详细的问题分析报告。
+
+【重要说明（必须严格遵守）】
+- location 字段必须与输入的 original_text 完全一致（一字不差）
+- 根据 title_mark_status 和 bracket_status 的值判断问题类型
+- 提供具体的修改建议和原因分析
+
+【输出要求】
+- 为每个检查结果输出一个详细的问题分析
+- 确保输出数量与输入一致
+- location 必须与 original_text 完全一致
+- 严格按照判定规则生成内容
+"""
+
+HUMAN = """
+请根据以下标点符号检查结果，生成详细的问题分析报告：
+
+【判定规则】
+
+当 title_mark_status = true 且 bracket_status = true：
+- issue_point：编制依据格式正确
+- reason：规范名称和编号的标点符号使用规范
+- suggestion：无
+- risk_level：无风险
+
+当 title_mark_status != true 时：
+- issue_point：编制依据格式错误
+- reason：从以下三种情况中选择最符合实际的问题描述：
+    1. 规范名称未被书名号包裹
+    2. 书名号不成对
+    3. 规范名称未完全被书名号包裹
+- suggestion：将规范名称用书名号《》包裹，正确格式：《规范名称》（编号）
+- risk_level：中风险
+
+当 bracket_status ！= true 时：
+- issue_point：编制依据格式错误
+- reason：如果bracket_status = null，则问题原因是"编号缺失"；
+    否则从以下三种情况中从上到下选择符合的问题描述：
+    1. 规范编号未被括号包裹
+    2. 规范编号未完全被括号包裹
+    3. 括号不成对
+- suggestion：
+  * 如果是"编号缺失"：补充编号，格式为（编号）
+  * 否则：将编号用括号（）包裹，正确格式：《规范名称》（编号）
+- risk_level：中风险
+
+当 title_mark_status != true 且 bracket_status != true：
+- issue_point：编制依据格式错误
+- reason：引用不符合正确格式：《规范名称》（编号）
+- suggestion：请将引用调正为正确格式：《规范名称》（编号）并保证名称与编号一一对应
+- risk_level：中风险
+
+【标点符号检查结果】
+{check_results}
+
+【输出格式要求】
+{format_instructions}
+/no_think
+"""
+
+# ===== 3) Output Parser =====
+parser = PydanticOutputParser(pydantic_object=PunctuationIssueResults)
+
+# ===== 4) Prompt =====
+prompt = ChatPromptTemplate.from_messages([
+    ("system", SYSTEM),
+    ("human", HUMAN)
+])
+
+# ===== 5) LLM =====
+llm = ChatOpenAI(
+    model="qwen3-30b",
+    base_url="http://192.168.91.253:8003/v1",
+    api_key="sk-123456",
+    temperature=0,
+)
+
+
+# ===== 6) 提取第一个 JSON =====
+def extract_first_json(text: str) -> dict:
+    """从任意模型输出中提取第一个完整 JSON 对象 { ... }"""
+    start = text.find("{")
+    if start == -1:
+        raise ValueError("未找到 JSON 起始 '{'")
+
+    depth = 0
+    for i in range(start, len(text)):
+        ch = text[i]
+        if ch == "{":
+            depth += 1
+        elif ch == "}":
+            depth -= 1
+            if depth == 0:
+                return json.loads(text[start:i + 1])
+
+    raise ValueError("JSON 花括号未闭合")
+
+
+# ===== 7) 核心方法 =====
+async def process_punctuation_results(check_results: str) -> str:
+    """
+    根据标点符号检查结果生成详细的问题分析报告
+    
+    Args:
+        check_results: punctuation_checker 的返回结果（JSON字符串）
+        
+    Returns:
+        问题分析报告的JSON字符串，包含五个字段：
+        - issue_point: 问题点描述
+        - location: 审查内容（与原文一致）
+        - suggestion: 修改建议
+        - reason: 问题原因分析
+        - risk_level: 风险水平
+    """
+    chain = prompt | llm | StrOutputParser()
+    format_instructions = parser.get_format_instructions()
+
+    payload = {
+        "check_results": check_results,
+        "format_instructions": format_instructions
+    }
+
+    last_err = None
+
+    for _ in range(2):
+        try:
+            raw = await chain.ainvoke(payload)
+            #print(f"[标点符号问题分析] 模型输出: {raw}...")
+            data = extract_first_json(raw)
+            findings = PunctuationIssueResults.model_validate(data)
+            result = [x.model_dump() for x in findings.items]
+            return json.dumps(result, ensure_ascii=False, indent=2)
+        except (Exception, ValidationError, json.JSONDecodeError) as e:
+            last_err = e
+
+    raise RuntimeError(f"标点符号问题分析失败：{last_err}") from last_err
+
+
+# ===== 8) 示例 =====
+if __name__ == "__main__":
+    import asyncio
+
+    # 模拟 punctuation_checker 的返回结果
+    check_results = json.dumps([
+        {
+            "original_text": "《混凝土结构设计规范》",
+            "title_mark_status": True,
+            "bracket_status": "null"
+        },
+        {
+            "original_text": "《混凝土结构设计规范》【GB 50010-2010】",
+            "title_mark_status": True,
+            "bracket_status": False
+        },
+        {
+            "original_text": "《建筑施工组织设计规范》(GB/T 50502-2015）",
+            "title_mark_status": True,
+            "bracket_status": True
+        },
+        {
+            "original_text": "建筑抗震设计规范 GB 50011-2010",
+            "title_mark_status": False,
+            "bracket_status": False
+        }
+    ], ensure_ascii=False)
+
+    result = asyncio.run(process_punctuation_results(check_results))
+    print("\n标点符号问题分析结果：")
+    print(result)

core/construction_review/component/reviewers/utils/reference_matcher.py

@@ -55,7 +55,7 @@ HUMAN = """
    - 完全找不到任何相关文件，返回 false
 
 3. **has_exact_match**（是否有名称编号都相同的文件）
-   - 找到名称且编号相同的文件，返回 true
+   - 忽略书写格式不同，找到名称且编号相同的文件，返回 true
    - 否则返回 false
 
 4. **exact_match_info**（名称编号相同的文件及状态）
@@ -96,7 +96,7 @@ llm = ChatOpenAI(
     model="qwen3-30b",
     base_url="http://192.168.91.253:8003/v1",
     api_key="sk-123456",
-    temperature=0,
+    temperature=0.7,
 )
 
 # ===== 6) 提取第一个 JSON =====