v0.0.5-功能优化-check_item功能增加

- 完整性
- 去除大纲审查
- 融合编制依据

core/construction_review/component/ai_review_engine.py

@@ -647,14 +647,14 @@ class AIReviewEngine(BaseReviewer):
         
         return result
         
-    async def check_completeness(self, trace_id_idx: str, review_content: Dict[str, Any],
+    async def check_completeness(self, trace_id_idx: str, review_content: List[Dict[str, Any]],
                                state: str, stage_name: str) -> Dict[str, Any]:
         """
         完整性检查
 
         Args:
             trace_id_idx: 追踪ID索引
-            review_content: 审查内容，单个文档块（chunk）的字典，格式如文档切分预处理结果.json中的chunks项
+            review_content: 审查内容，文档块（chunks）列表
             state: 状态字典
             stage_name: 阶段名称
 
@@ -670,7 +670,7 @@ class AIReviewEngine(BaseReviewer):
         try:
             # 验证review_content格式
             if not isinstance(review_content, list):
-                raise ValueError(f"review_content必须是字典类型，当前类型: {type(review_content)}")
+                raise ValueError(f"review_content必须是列表类型，当前类型: {type(review_content)}")
             
             # # 获取文档块信息
             # doc = review_content
@@ -756,24 +756,31 @@ class AIReviewEngine(BaseReviewer):
 
             # 生成缺失要点 JSON 列表，便于前端消费
             #missing_issue_json_path = Path(r'temp\document_temp') / 'spec_review_missing_issues.json'
-            issues = await analyzer.build_missing_issue_list(summary_rows,state,start_time)
+            issues = analyzer.build_missing_issue_list(summary_rows)
 
-            return  issues
+            # 包装成外层格式化期望的结构
+            execution_time = time.time() - start_time
+            return {
+                "details": {
+                    "name": "completeness_check",
+                    "response": issues.get("response", []),
+                    "review_location_label": issues.get("review_location_label", ""),
+                    "chapter_code": issues.get("chapter_code", ""),
+                    "original_content": issues.get("original_content", "")
+                },
+                "success": True,
+                "execution_time": execution_time
+            }
         except Exception as e:
             execution_time = time.time() - start_time
             error_msg = f"{name} 审查失败: {str(e)}"
             logger.error(error_msg, exc_info=True)
 
-            from core.construction_review.component.reviewers.base_reviewer import ReviewResult
-            return ReviewResult(
-                success=False,
-                details={
-                    'chunk_id': review_content[0].get('chunk_id', 'unknown') if isinstance(review_content[0], dict) else 'unknown',
-                    'error': str(e)
-                },
-                error_message=error_msg,
-                execution_time=execution_time
-            )
+            # 返回包含错误信息的字典，由外层统一格式化
+            return {
+                'error': error_msg,
+                'exception': str(e)
+            }
 
     async def check_sensitive(self, trace_id_idx: str, review_content: str,
                             state: str, stage_name: str) -> Dict[str, Any]:

core/construction_review/component/reviewers/check_completeness/components/result_analyzer.py

@@ -1,8 +1,6 @@
 """
 结果汇总与规范覆盖分析组件
 """
-import time
-import asyncio
 from typing import Dict, List, Any, Set
 import ast
 import sys
@@ -15,9 +13,6 @@ if str(_root) not in sys.path:
 
 from interfaces import IResultAnalyzer
 from utils.file_utils import read_csv, write_csv
-from ...utils.inter_tool import InterTool
-from ...base_reviewer import ReviewResult
-from core.construction_review.workflows.core_functions.ai_review_core_fun import UnitReviewResult
 
 class ResultAnalyzer(IResultAnalyzer):
     """审查结果汇总分析器"""
@@ -28,7 +23,6 @@ class ResultAnalyzer(IResultAnalyzer):
             spec_csv_path: 规范 CSV 文件路径（Construction_Plan_Content_Specification.csv）
         """
         self.spec_csv_path = spec_csv_path
-        self.inter_tool = InterTool()
 
     def process_results(self, results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
         """
@@ -219,23 +213,22 @@ class ResultAnalyzer(IResultAnalyzer):
         return summary_rows
 
     # 生成缺失要点的 JSON 列表，便于前端或其他系统直接消费
-    async def build_missing_issue_list(
-        self, summary_rows: List[Dict[str, Any]], state: Dict[str, Any] = None, start_time: float = None
-    ) -> List[Dict[str, Any]]:
+    def build_missing_issue_list(
+        self, summary_rows: List[Dict[str, Any]]
+    ) -> Dict[str, Any]:
         """
-        构建缺失要点的问题列表（异步版本）
+        构建缺失要点的问题列表
 
         Args:
             summary_rows: 规范汇总行数据
-            state: 状态字典（可选，用于回调）
-            start_time: 开始时间（可选）
 
         Returns:
-            List[Dict]: 缺失要点的问题列表
+            Dict: 审查结果字典，包含 response 字段的问题列表和必要的元数据
         """
-        issues: List[Dict[str, Any]] = []
+        all_issues = []
+        metadata = {}
 
-        for i, row in enumerate(summary_rows):
+        for row in summary_rows:
             level2 = (row.get("二级目录") or "").strip()
             requirement = (row.get("内容要求") or "").strip()
 
@@ -255,74 +248,30 @@ class ResultAnalyzer(IResultAnalyzer):
             )
             suggestion = f"补充：{requirement_text}" if requirement else "补充缺失要点内容"
             risk_level = self._map_risk_level(len(missing_points))
-            chapter_code = row.get("标签")
-            content = row.get("content", "")
-            section_label = row.get("section_label", "")
 
-            # 构建兼容模型响应格式
-            model_response = [{
+            # 构建问题项并添加到列表
+            issue_item = {
                 "issue_point": issue_point,
                 "location": location,
                 "suggestion": suggestion,
                 "reason": requirement,
                 "risk_level": risk_level,
-            }]
-
-            # 构造审查结果
-            completeness_check_result = ReviewResult(
-                success=True,
-                details={
-                    "name": "completeness_check",
-                    "response": model_response
-                },
-                error_message=None,
-                execution_time=time.time() - start_time if start_time else 0
-            )
-
-            # # 构造 issue 字典并添加到 issues 列表
-            # issue = {
-            #     "issue_point": issue_point,
-            #     "location": location,
-            #     "suggestion": suggestion,
-            #     "requirement": requirement,
-            #     "risk_level": risk_level,
-            #     "chapter_code": chapter_name,
-            #     "section_label": section_label,
-            #     "missing_points": missing_points,
-            #     "check_name": "completeness_check",
-            # }
-            # issues.append(issue)
-
-            # 如果提供了 state，发送进度更新
-            if state is not None:
-                # 构造 UnitReviewResult 对象
-                unit_review_result = UnitReviewResult(
-                    unit_index=i,
-                    unit_content={"section_label": section_label, "chapter_code": chapter_code},
-                    basic_compliance={"completeness_check": completeness_check_result},
-                    technical_compliance={},
-                    rag_enhanced={},
-                    overall_risk=risk_level
-                )
-
-                # 构造 completeness_config
-                completeness_config = {
-                    "section_label": section_label,
-                    "chapter_code": chapter_code,
-                    "content": content
-                    
+            }
+            all_issues.append(issue_item)
+
+            # 收集元数据（从第一行获取）
+            if not metadata:
+                metadata = {
+                    "review_location_label": row.get("section_label", ""),
+                    "chapter_code": row.get("标签", ""),
+                    "original_content": row.get("content", "")
                 }
 
-                # 调用 _complete_format 发送进度更新
-                new_issues = await self.inter_tool._complete_format(
-                    unit_review_result,
-                    state,
-                    i,
-                    len(summary_rows),
-                    completeness_config
-                )
-                issues.append(new_issues)
-        return issues
+        # 返回包含问题和元数据的字典，由外层统一格式化
+        return {
+            "response": all_issues,
+            **metadata
+        }
 
     @staticmethod
     def _parse_list_field(value: Any) -> List[Any]:

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

@@ -11,7 +11,7 @@ from .sensitive_word_checker import (
     check_sensitive_words_async,
     format_check_results
 )
-
+from .directory_extraction import extract_basis_with_langchain_qwen
 __all__ = [
     'PromptLoader',
     'prompt_loader',
@@ -19,5 +19,6 @@ __all__ = [
     'SensitiveWordChecker',
     'check_sensitive_words',
     'check_sensitive_words_async',
-    'format_check_results'
+    'format_check_results',
+    'extract_basis_with_langchain_qwen'
 ]

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

@@ -214,6 +214,11 @@ async def extract_basis_with_langchain_qwen(progress_manager,callback_task_id,te
         return fallback_regex(text)
 
 
+
+
+
+
+
 if __name__ == "__main__":
     demo = """一）编制依据
 （1）相关法律法规

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

@@ -262,6 +262,67 @@ class InterTool:
                 logger.debug(f"跳过分数字段: {check_key}")
                 continue
 
+            # 🔧 特殊处理：timeliness_basis_reviewer 和 reference_basis_reviewer 的返回格式
+            if check_key == 'timeliness_basis_reviewer' and isinstance(check_result, dict):
+                timeliness_data = check_result.get('timeliness_basis_review_results', {})
+                batch_results = timeliness_data.get('review_results', [])
+                logger.debug(f"🔍 [DEBUG] 处理时效性审查结果，批次数: {len(batch_results)}")
+                
+                for batch in batch_results:
+                    if isinstance(batch, list):
+                        # 批次是列表，遍历其中的每个 item
+                        for item in batch:
+                            if isinstance(item, dict):
+                                review_lists.append({
+                                    "check_item": item.get('check_item', 'timeliness_check'),
+                                    "chapter_code": item.get('chapter_code', chapter_code),
+                                    "check_item_code": item.get('check_item_code', f"{chapter_code}_timeliness_check"),
+                                    "check_result": item.get('check_result', item),
+                                    "exist_issue": item.get('exist_issue', False),
+                                    "risk_info": item.get('risk_info', {"risk_level": "low"})
+                                })
+                    elif isinstance(batch, dict):
+                        # 批次本身就是单个 item
+                        review_lists.append({
+                            "check_item": batch.get('check_item', 'timeliness_check'),
+                            "chapter_code": batch.get('chapter_code', chapter_code),
+                            "check_item_code": batch.get('check_item_code', f"{chapter_code}_timeliness_check"),
+                            "check_result": batch.get('check_result', batch),
+                            "exist_issue": batch.get('exist_issue', False),
+                            "risk_info": batch.get('risk_info', {"risk_level": "low"})
+                        })
+                logger.info(f"🔍 时效性审查结果处理完成，添加 {len(review_lists)} 个问题项")
+                continue
+            
+            if check_key == 'reference_basis_reviewer' and isinstance(check_result, dict):
+                reference_data = check_result.get('reference_basis_review_results', {})
+                batch_results = reference_data.get('review_results', [])
+                logger.debug(f"🔍 [DEBUG] 处理规范性审查结果，批次数: {len(batch_results)}")
+                
+                for batch in batch_results:
+                    if isinstance(batch, list):
+                        for item in batch:
+                            if isinstance(item, dict):
+                                review_lists.append({
+                                    "check_item": item.get('check_item', 'reference_check'),
+                                    "chapter_code": item.get('chapter_code', chapter_code),
+                                    "check_item_code": item.get('check_item_code', f"{chapter_code}_reference_check"),
+                                    "check_result": item.get('check_result', item),
+                                    "exist_issue": item.get('exist_issue', False),
+                                    "risk_info": item.get('risk_info', {"risk_level": "low"})
+                                })
+                    elif isinstance(batch, dict):
+                        review_lists.append({
+                            "check_item": batch.get('check_item', 'reference_check'),
+                            "chapter_code": batch.get('chapter_code', chapter_code),
+                            "check_item_code": batch.get('check_item_code', f"{chapter_code}_reference_check"),
+                            "check_result": batch.get('check_result', batch),
+                            "exist_issue": batch.get('exist_issue', False),
+                            "risk_info": batch.get('risk_info', {"risk_level": "low"})
+                        })
+                logger.info(f"🔍 规范性审查结果处理完成，添加 {len(review_lists)} 个问题项")
+                continue
+
             # 🔧 类型安全检查：支持字典和 base_reviewer.ReviewResult 对象
             is_dict = isinstance(check_result, dict)
             is_review_result = hasattr(check_result, 'details') and hasattr(check_result, 'success')

core/construction_review/workflows/ai_review_workflow.py

@@ -346,35 +346,6 @@ class AIReviewWorkflow:
             # 初始化issues列表
             all_issues = []
             completed_chunks = 0
-
-            # 4️ 优先处理大纲审查（在章节循环之前）
-            outline_data = structured_content.get("outline", {})
-            has_outline_check = any(
-                "outline_check" in funcs
-                for funcs in review_item_dict.values()
-            )
-
-            if has_outline_check and outline_data:
-                total_chunks = total_chunks + 1  # 包含大纲审查块
-                await self.core_fun._send_start_review_progress(state, total_chunks, 'check_item_review')
-                logger.info(f"准备执行动态审查任务(含大纲审查)，总章节数: {total_chapters}, 总块数: {total_chunks}")
-                try:
-                    outline_result = await self.ai_review_engine.outline_check(
-                        state["callback_task_id"],
-                        {"outline": outline_data},
-                        state,
-                        "大纲审查"
-                    )
-                    if outline_result:
-                        all_issues.append(outline_result)
-                        logger.info(f"大纲审查完成")
-                except Exception as e:
-                    logger.error(f"大纲审查失败: {str(e)}", exc_info=True)
-            else:
-                await self.core_fun._send_start_review_progress(state, total_chunks, 'check_item_review')
-                logger.info(f"准备执行动态审查任务，总章节数: {total_chapters}, 总块数: {total_chunks}")  
-
-            # 5️ 按章节分组
             chapter_chunks_map = self.core_fun._group_chunks_by_chapter(chunks)
             logger.info(f"内容分组完成，共 {len(chapter_chunks_map)} 个章节")
 
@@ -394,23 +365,12 @@ class AIReviewWorkflow:
                     logger.warning(f"章节 {chapter_code} 没有找到对应内容，跳过")
                     continue
 
-                # 判断章节类型并分支处理
-                if chapter_code == "basis":
-                    # === 编制依据章节：拼接所有chunk后一次性审查 ===
-                    all_issues = await self.core_fun._process_basis_chapter(
-                        chapter_code, chapter_content, func_names, state, all_issues, completed_chunks, total_chunks
-                    )
-                    logger.info(f"🔍 basis章节处理后: all_issues数量={len(all_issues) if all_issues else 0}")  # 调试日志
-                    # 更新已完成块数
-                    completed_chunks += len(chapter_content)
-                else:
-                    # === 普通章节：逐块审查 ===
-                    chunks_completed, all_issues = await self.core_fun._process_normal_chapter(
-                        chapter_code, chapter_content, func_names, state, all_issues
-                    )
-                    logger.info(f"🔍 normal章节处理后: all_issues数量={len(all_issues) if all_issues else 0}, chunks_completed={chunks_completed}")  # 调试日志
-                    # 更新已完成块数
-                    completed_chunks += chunks_completed
+                chunks_completed, all_issues = await self.core_fun._process_chapter_item(
+                    chapter_code, chapter_content, func_names, state, all_issues, completed_chunks, total_chunks
+                )
+                logger.info(f"🔍 章节{chapter_code}处理后: all_issues数量={len(all_issues) if all_issues else 0}, chunks_completed={chunks_completed}")  # 调试日志
+                # 更新已完成块数
+                completed_chunks += chunks_completed
 
                 logger.info(f"章节 {chapter_code} 处理完成")
             logger.info(f"all_issues_结果调试: {all_issues} 个块")

core/construction_review/workflows/core_functions/ai_review_core_fun.py

@@ -39,10 +39,10 @@ from typing import Dict, Union, List, Any, Optional, Tuple
 from dataclasses import dataclass
 from langchain_core.messages import AIMessage
 
+from core.construction_review.component.reviewers.utils import directory_extraction
 from foundation.observability.logger.loggering import server_logger as logger
 from foundation.infrastructure.cache.redis_connection import RedisConnectionFactory
 from core.base.task_models import TaskFileInfo
-# from core.construction_review.component.reviewers.base_reviewer import ReviewResult
 from ...component.reviewers.utils.inter_tool import InterTool
 from ..types import AIReviewState
 
@@ -95,7 +95,7 @@ class AIReviewCoreFun:
         from core.base.workflow_manager import WorkflowManager
         self.workflow_manager = WorkflowManager()
 
-    async def _process_basis_chapter(
+    async def _process_chapter_item(
         self,
         chapter_code: str,
         chapter_content: List[Dict[str, Any]],
@@ -104,151 +104,9 @@ class AIReviewCoreFun:
         all_issues: List[Dict],
         completed_chunks: int,
         total_chunks: int
-    ) -> List[Dict]:
-        """
-        处理编制依据章节（basis）
-
-        特点：
-        - 收集所有chunk的content
-        - 拼接后一次性审查
-        - 整个章节推送一次进度
-
-        Args:
-            chapter_code: 章节代码（应该是"basis"）
-            chapter_content: 章节的所有chunk
-            func_names: 需要执行的审查方法
-            state: AI审查状态
-            all_issues: 累积的issues列表
-            completed_chunks: 已完成的块数
-            total_chunks: 总块数
-
-        Returns:
-            List[Dict]: 更新后的issues列表（包含本次章节的结果）
-        """
-        logger.info(f"🔍 处理编制依据章节: {chapter_code}, 共 {len(chapter_content)} 个chunk")
-
-        # 1. 收集所有chunk的content
-        basis_contents = []
-        for chunk in chapter_content:
-            content = chunk.get("content", "")
-            if content:
-                basis_contents.append(content)
-
-        # 2. 拼接content
-        if not basis_contents:
-            logger.warning(f"编制依据章节 {chapter_code} 没有内容，跳过")
-            return all_issues  # ✅ 返回原始的 all_issues，避免数据丢失
-
-        combined_content = "\n\n".join(basis_contents)
-        logger.info(f"编制依据内容拼接完成，总长度: {len(combined_content)} 字符")
-
-        # 3. 提取编制依据条目（时效性和规范性审查都需要）
-        basis_items = None
-        needs_basis_extraction = any(
-            fn in ["timeliness_basis_reviewer", "reference_basis_reviewer"] 
-            for fn in func_names
-        )
-        
-        if needs_basis_extraction:
-            try:
-                from core.construction_review.component.reviewers.utils.directory_extraction import (
-                    extract_basis_with_langchain_qwen,
-                )
-                basis_items = await extract_basis_with_langchain_qwen(
-                    progress_manager=state.get("progress_manager"),
-                    callback_task_id=state.get("callback_task_id"),
-                    text=combined_content,
-                )
-                logger.info(f"编制依据AI提取完成，条数: {len(getattr(basis_items, 'items', []))}")
-            except Exception as e:
-                logger.error(f"编制依据AI提取失败: {e}", exc_info=True)
-
-        # 4. 执行需要的审查方法
-        chapter_issues = []
-
-        for func_name in func_names:
-            if func_name == "reference_basis_reviewer":
-                # 编制依据审查
-                logger.info(f"执行编制依据审查: {chapter_code}")
-                try:
-                    review_data = {
-                        "content": combined_content,
-                        "basis_items": basis_items,
-                        "max_concurrent": self.max_concurrent
-                    }
-                    result = await self.ai_review_engine.reference_basis_reviewer(
-                        review_data=review_data,
-                        trace_id=f"{state['callback_task_id']}_{chapter_code}",
-                        state=state,
-                        stage_name=f"{chapter_code}_编制依据审查"
-                    )
-
-                    # 提取issues
-                    issues = self._extract_issues_from_result(result)
-                    chapter_issues.extend(issues)
-
-                except Exception as e:
-                    logger.error(f"编制依据审查失败: {str(e)}", exc_info=True)
-
-            elif func_name == "timeliness_basis_reviewer":
-                # 时效性审查
-                logger.info(f"执行时效性审查: {chapter_code}")
-                try:
-                    review_data = {
-                        "content": combined_content,
-                        "basis_items": basis_items,
-                        "max_concurrent": self.max_concurrent
-                    }
-                    result = await self.ai_review_engine.timeliness_basis_reviewer(
-                        review_data=review_data,
-                        trace_id=f"{state['callback_task_id']}_{chapter_code}",
-                        state=state,
-                        stage_name=f"{chapter_code}_时效性审查"
-                    )
-
-                    # 提取issues
-                    issues = self._extract_issues_from_result(result)
-                    chapter_issues.extend(issues)
-
-                except Exception as e:
-                    logger.error(f"时效性审查失败: {str(e)}", exc_info=True)
-
-            else:
-                # 其他方法：不支持对basis章节的操作
-                logger.warning(f"方法 {func_name} 不支持编制依据章节模式，跳过")
-
-        # 4. 累积到总issues
-        all_issues.extend(chapter_issues)
-
-        # 5. 推送章节级别的进度
-        if state.get("progress_manager") and chapter_content:
-            # 计算当前进度
-            current = int(((completed_chunks + len(chapter_content)) / total_chunks) * 100)
-
-            # 推送进度
-            await state["progress_manager"].update_stage_progress(
-                callback_task_id=state["callback_task_id"],
-                stage_name=f"AI审查:{chapter_code}",
-                current=current,
-                status="completed",
-                message=f"编制依据章节审查完成，共 {len(chapter_content)} 个块",
-                event_type="processing"
-            )
-        return all_issues
-    async def _process_normal_chapter(
-        self,
-        chapter_code: str,
-        chapter_content: List[Dict[str, Any]],
-        func_names: List[str],
-        state: AIReviewState,
-        all_issues: List[Dict]
     ) -> Tuple[int, List[Dict]]:
         """
-        处理普通章节（非basis）
-
-        特点：
-        - 逐块审查
-        - 每块完成后立即推送进度
+        统一处理章节审查（所有章节都走逐块审查流程）
 
         Args:
             chapter_code: 章节代码
@@ -256,17 +114,19 @@ class AIReviewCoreFun:
             func_names: 需要执行的审查方法
             state: AI审查状态
             all_issues: 累积的issues列表
+            completed_chunks: 已完成的块数
+            total_chunks: 总块数
 
         Returns:
             Tuple[int, List[Dict]]: (处理的块数量, 更新后的issues列表)
         """
-        logger.info(f"📝 处理普通章节: {chapter_code}, 共 {len(chapter_content)} 个chunk")
-        total_chunks = len(chapter_content)
+        logger.info(f"📝 处理章节: {chapter_code}, 共 {len(chapter_content)} 个chunk")
+        chapter_total_chunks = len(chapter_content)
 
-        # 按块串行遍历
+        # 按块串行遍历（所有章节统一流程）
         for chunk_index, chunk in enumerate(chapter_content):
             chunk_label = chunk.get("section_label", f"chunk_{chunk_index}")
-            logger.info(f"  📄 处理块 {chunk_index+1}/{total_chunks}: {chunk_label}")
+            logger.info(f"  📄 处理块 {chunk_index+1}/{chapter_total_chunks}: {chunk_label}")
 
             # 终止信号检查（块级别）
             if await self.workflow_manager.check_terminate_signal(state["callback_task_id"]):
@@ -277,7 +137,7 @@ class AIReviewCoreFun:
             chunk_results = await self._execute_chunk_methods(
                 chapter_code, chunk, chunk_index, func_names, state
             )
-            logger.info(f"🔍 chunk_results: basic_compliance={chunk_results.get('basic_compliance', {})}, technical_compliance={chunk_results.get('technical_compliance', {})}")
+
             # 格式化当前块的结果为issues
             chunk_page = chunk.get('page', '')
             review_location_label = f"第{chunk_page}页：{chunk_label}"
@@ -293,9 +153,9 @@ class AIReviewCoreFun:
             )
 
             # 推送当前块的进度
-            current = int(((chunk_index + 1) / total_chunks) * 100)
+            current = int(((completed_chunks + chunk_index + 1) / total_chunks) * 100)
             await self._send_unit_review_progress(
-                state, chunk_index, total_chunks, chunk_label, issues, current
+                state, chunk_index, chapter_total_chunks, chunk_label, issues, current
             )
 
             # 累积issues
@@ -305,8 +165,8 @@ class AIReviewCoreFun:
             else:
                 logger.warning(f"⚠️ 块{chunk_index}: issues为空，未添加到all_issues")
 
-        logger.info(f"🔍 章节{chapter_code}完成: 总共处理{total_chunks}个块, all_issues最终数量={len(all_issues)}")
-        return total_chunks,all_issues
+        logger.info(f"🔍 章节{chapter_code}完成: 总共处理{chapter_total_chunks}个块, all_issues最终数量={len(all_issues)}")
+        return chapter_total_chunks, all_issues
 
     def _extract_issues_from_result(self, result: Any) -> List[Dict]:
         """
@@ -417,15 +277,25 @@ class AIReviewCoreFun:
         """
         semaphore = asyncio.Semaphore(5)  # 单个块内限制并发数为5
         rag_enhanced_content = None  # 初始化变量，避免作用域错误
-        if 'check_parameter_compliance' in func_names or 'check_non_parameter_compliance' in func_names:
-                logger.debug("开始执行RAG检索增强")
-                rag_enhanced_content = self.ai_review_engine.rag_enhanced_check(chunk.get('content', ''))
+        basis_content = None  # 初始化变量，避免作用域错误
 
+        if 'check_parameter_compliance' in func_names or 'check_non_parameter_compliance' in func_names:
+            logger.debug("开始执行RAG检索增强")
+            rag_enhanced_content = self.ai_review_engine.rag_enhanced_check(chunk.get('content', ''))
+
+        if 'reference_basis_reviewer' in func_names or 'timeliness_basis_reviewer' in func_names:
+            logger.debug("开始执行编制依据/时效性预处理")
+            # 预处理编制依据/时效性审查所需内容
+            basis_content = await directory_extraction.extract_basis_with_langchain_qwen(
+                state['progress_manager'],
+                state["callback_task_id"],
+                chunk.get('content', '')
+            )
         async def execute_with_semaphore(func_name):
             async with semaphore:
                 try:
                     # 创建并执行单个审查任务
-                    result = await self._execute_single_review(chapter_code, chunk, chunk_index, func_name, state,rag_enhanced_content)
+                    result = await self._execute_single_review(chapter_code, chunk, chunk_index, func_name, state,rag_enhanced_content, basis_content)
                     return func_name, result
                 except Exception as e:
                     logger.error(f"审查任务执行失败 [{chapter_code}.chunk{chunk_index}.{func_name}]: {str(e)}")
@@ -470,7 +340,7 @@ class AIReviewCoreFun:
             'rag_enhanced': merged_rag
         }
 
-    async def _execute_single_review(self, chapter_code: str, chunk: Dict[str, Any], chunk_index: int, func_name: str, state: AIReviewState,rag_enhanced_content :dict = None) -> UnitReviewResult:
+    async def _execute_single_review(self, chapter_code: str, chunk: Dict[str, Any], chunk_index: int, func_name: str, state: AIReviewState,rag_enhanced_content :dict = None, basis_content: dict = None) -> UnitReviewResult:
         """
         执行单个块的审查任务
 
@@ -546,7 +416,8 @@ class AIReviewCoreFun:
             )
 
         elif func_name == "check_completeness":
-            raw_result = await method(trace_id, chunk, state, stage_name)
+            # check_completeness 需要列表类型，将单个 chunk 包装成列表
+            raw_result = await method(trace_id, [chunk], state, stage_name)
             # 基础审查方法，放入 basic_compliance
             return UnitReviewResult(
                 unit_index=chunk_index,
@@ -587,16 +458,62 @@ class AIReviewCoreFun:
                 overall_risk=self._calculate_single_result_risk(raw_result)
             )
 
-        # ⚠️ 以下三个特殊方法不在块级别处理，由主流程统一管理
-        elif func_name in ["outline_check", "timeliness_basis_reviewer", "reference_basis_reviewer"]:
-            logger.warning(f"方法 {func_name} 不应在块级别调用，已在主流程中处理")
+        # # outline_check 仍在章节级别处理
+        # elif func_name == "outline_check":
+        #     logger.warning(f"方法 {func_name} 不应在块级别调用，已在主流程中处理")
+        #     return UnitReviewResult(
+        #         unit_index=chunk_index,
+        #         unit_content=chunk,
+        #         basic_compliance={},
+        #         technical_compliance={},
+        #         rag_enhanced={},
+        #         overall_risk="low"
+        #     )
+
+        # reference_basis_reviewer：编制依据审查（逐块处理）
+        elif func_name == "reference_basis_reviewer":
+            review_data = {
+                "content": review_content,  # 原始文本内容
+                "basis_items": basis_content,  # 提取的 BasisItems 对象
+                "max_concurrent": self.max_concurrent
+            }
+            raw_result = await method(
+                review_data=review_data,
+                trace_id=trace_id,
+                state=state,
+                stage_name=stage_name
+            )
+            # 基础审查方法，放入 basic_compliance
             return UnitReviewResult(
                 unit_index=chunk_index,
                 unit_content=chunk,
-                basic_compliance={},
+                basic_compliance={func_name: raw_result},
+                technical_compliance={},
+                rag_enhanced={},
+                overall_risk=self._calculate_single_result_risk(raw_result)
+            )
+
+        # timeliness_basis_reviewer：时效性审查（逐块处理）
+        elif func_name == "timeliness_basis_reviewer":
+            review_data = {
+                "content": review_content,  # 原始文本内容
+                "basis_items": basis_content,  # 提取的 BasisItems 对象
+                "max_concurrent": self.max_concurrent
+            }
+            raw_result = await method(
+                review_data=review_data,
+                trace_id=trace_id,
+                state=state,
+                stage_name=stage_name
+            )
+            # 基础审查方法，放入 basic_compliance
+            return UnitReviewResult(
+                unit_index=chunk_index,
+                unit_content=chunk,
+                basic_compliance={func_name: raw_result},
                 technical_compliance={},
                 rag_enhanced={},
-                overall_risk="low"
+                overall_risk=self._calculate_single_result_risk(raw_result)
             )
 
         else: