""" 审查执行编排器 封装 7 个审查链路的调用，复用 BaseReviewer.review() 和 AIReviewEngine。 支持 SSE 进度推送、单步错误隔离、trace_id 生产隔离。 用法: executor = DebugExecutor() await executor.execute(request, event_queue) """ import asyncio import json import os import time import uuid from datetime import datetime from typing import Any, Dict, List, Optional from views.debug.debug_api import ( DebugExecuteRequest, RagParams, ) from core.construction_review.component.reviewers.base_reviewer import BaseReviewer from foundation.ai.agent.generate.model_generate import generate_model_client from foundation.observability.logger.loggering import review_logger as logger # ============================================================ # 步骤定义 # ============================================================ CHAIN_STEPS = { "completeness": [ {"index": 0, "name": "Prompt 渲染", "phase": None}, {"index": 1, "name": "LLM 调用", "phase": None}, {"index": 2, "name": "结果解析", "phase": None}, ], "timeliness": [ {"index": 0, "name": "Prompt 渲染", "phase": None}, {"index": 1, "name": "LLM 调用", "phase": None}, {"index": 2, "name": "结果解析", "phase": None}, ], "reference": [ {"index": 0, "name": "Prompt 渲染", "phase": None}, {"index": 1, "name": "LLM 调用", "phase": None}, {"index": 2, "name": "结果解析", "phase": None}, ], "sensitive": [ {"index": 0, "name": "Prompt 渲染", "phase": None}, {"index": 1, "name": "LLM 调用", "phase": None}, {"index": 2, "name": "结果解析", "phase": None}, ], "semantic": [ {"index": 0, "name": "Prompt 渲染", "phase": None}, {"index": 1, "name": "LLM 调用", "phase": None}, {"index": 2, "name": "结果解析", "phase": None}, ], "grammar": [ {"index": 0, "name": "Prompt 渲染", "phase": None}, {"index": 1, "name": "LLM 调用", "phase": None}, {"index": 2, "name": "结果解析", "phase": None}, ], "professional": [ {"index": 0, "name": "查询提取", "phase": "RAG 召回阶段"}, {"index": 1, "name": "实体增强检索", "phase": "RAG 召回阶段"}, {"index": 2, "name": "父文档增强", "phase": "RAG 召回阶段"}, {"index": 3, "name": "结果提取", "phase": "RAG 召回阶段"}, {"index": 4, "name": "非参数合规审查", "phase": "AI 审查阶段"}, {"index": 5, "name": "参数合规审查", "phase": "AI 审查阶段"}, {"index": 6, "name": "结果汇总", "phase": "AI 审查阶段"}, ], } VALID_CHAIN_IDS = set(CHAIN_STEPS.keys()) # ============================================================ # StepResult 数据类 # ============================================================ class StepResult: """单步执行结果""" __slots__ = ("index", "name", "phase", "status", "duration", "input_summary", "output_summary", "error") def __init__(self, index: int, name: str, status: str = "pending", phase: str = None, duration: float = 0, input_summary: dict = None, output_summary: dict = None, error: str = None): self.index = index self.name = name self.phase = phase self.status = status self.duration = duration self.input_summary = input_summary or {} self.output_summary = output_summary or {} self.error = error # ============================================================ # CHAIN_CONFIG 映射表 # ============================================================ CHAIN_CONFIG = { "completeness": { "name": "完整性审查", "reviewer_type": "basic", "prompt_name": "completeness_check", "function_name": "completeness_review_generate", }, "timeliness": { "name": "时效性审查", "reviewer_type": "basic", "prompt_name": "timeliness_check", "function_name": "timeliness_review_generate", }, "reference": { "name": "规范性审查", "reviewer_type": "basic", "prompt_name": "reference_check", "function_name": "reference_review_generate", }, "sensitive": { "name": "敏感词检查", "reviewer_type": "basic", # 实际位于 basic_reviewers.yaml "prompt_name": "sensitive_word_check", "function_name": "sensitive_word_check_generate", }, "semantic": { "name": "语义逻辑检查", "reviewer_type": "basic", "prompt_name": "semantic_logic_check", "function_name": "semantic_logic_check_generate", }, "grammar": { "name": "语法检查", "reviewer_type": "basic", "prompt_name": "grammar_check", "function_name": "grammar_check_generate", }, "professional": { "name": "专业性审查", "reviewer_type": "rag", "prompt_name": None, "function_name": "professional_review_generate", }, } # 各链路的步骤依赖关系：step_index -> (依赖的上一步索引, 是否可跳过) _STEP_DEPS = { # 直调 LLM 链路：步骤 2 依赖步骤 1，步骤 1 依赖步骤 0 "completeness": {0: None, 1: 0, 2: 1}, "timeliness": {0: None, 1: 0, 2: 1}, "reference": {0: None, 1: 0, 2: 1}, "sensitive": {0: None, 1: 0, 2: 1}, "semantic": {0: None, 1: 0, 2: 1}, "grammar": {0: None, 1: 0, 2: 1}, # 专业性审查：每步依赖上一步 "professional": {0: None, 1: 0, 2: 1, 3: 2, 4: 3, 5: 4, 6: 5}, } def _make_trace_id(chain_id: str) -> str: """生成 trace_id，添加 debug_ 前缀实现生产隔离""" return ( f"debug_{chain_id}_" f"{datetime.now().strftime('%H%M%S')}_" f"{uuid.uuid4().hex[:8]}" ) def _make_record_id() -> str: """生成记录 ID""" return ( f"call-{datetime.now().strftime('%Y%m%d-%H%M%S')}-" f"{uuid.uuid4().hex[:6]}" ) class DebugExecutor: """审查执行编排器""" # 暴露 CHAIN_CONFIG 供外部使用 CHAIN_CONFIG = CHAIN_CONFIG CHAIN_STEPS = CHAIN_STEPS def __init__(self): self.base_reviewer = BaseReviewer() async def execute(self, request: DebugExecuteRequest, event_queue: asyncio.Queue) -> dict: """ 执行审查调试，所有事件通过 event_queue 推送。 Returns: dict with keys: completed_data, steps_collected, error_occurred, record_id """ chain_id = request.chain_id chain_cfg = CHAIN_CONFIG[chain_id] steps_def = CHAIN_STEPS[chain_id] trace_id = _make_trace_id(chain_id) record_id = _make_record_id() total_steps = len(steps_def) # --- started 事件 --- await event_queue.put(("started", { "task_id": trace_id, "chain_id": chain_id, "total_steps": total_steps, })) step_results: List[StepResult] = [] current_phase = None for step_def in steps_def: step_index = step_def["index"] step_name = step_def["name"] step_phase = step_def.get("phase") # phase_label：阶段切换时推送 if step_phase and step_phase != current_phase: current_phase = step_phase await event_queue.put(("phase_label", { "task_id": trace_id, "label": step_phase, })) # --- step_progress 事件（开始执行）--- await event_queue.put(("step_progress", { "task_id": trace_id, "step_index": step_index, "step_name": step_name, "status": "running", "duration": None, })) # 判断是否跳过：如果依赖的上一步失败/跳过，本步骤也跳过 dep = _STEP_DEPS[chain_id].get(step_index) should_skip = False if dep is not None: for prev in step_results: if prev.index == dep and prev.status in ("error", "skipped"): should_skip = True break if should_skip: result = StepResult( index=step_index, name=step_name, status="skipped", phase=step_phase, duration=0, ) else: try: if chain_id == "professional": result = await self._execute_professional_step( request, chain_cfg, step_def, step_results, trace_id, ) else: result = await self._execute_direct_llm_step( request, chain_cfg, step_def, step_results, trace_id, ) except Exception as exc: logger.error( f"[DebugExecutor] 步骤 {step_index} ({step_name}) " f"异常: {exc}", exc_info=True, ) result = StepResult( index=step_index, name=step_name, status="error", phase=step_phase, duration=0, error=str(exc), ) # --- step_result 事件 --- await self._emit_step_result(event_queue, trace_id, result) step_results.append(result) # --- completed 事件 --- total_duration = sum(s.duration for s in step_results) final_result = self._build_final_result(step_results) error_occurred = None for s in step_results: if s.status == "error": error_occurred = s.error or f"步骤 {s.index} ({s.name}) 执行失败" break # 转换为可序列化的步骤列表 steps_collected = [ { "index": s.index, "name": s.name, "status": s.status, "duration_ms": int(s.duration * 1000) if s.duration else 0, "input": s.input_summary, "output": s.output_summary, } for s in step_results ] completed_data = { "task_id": trace_id, "chain_id": chain_id, "total_duration": round(total_duration, 3), "record_id": record_id, "final_result": final_result, } await event_queue.put(("completed", completed_data)) return { "completed_data": completed_data, "steps_collected": steps_collected, "error_occurred": error_occurred, "record_id": record_id, } # ---------------------------------------------------------- # 直调 LLM 链路（6 个） # ---------------------------------------------------------- async def _execute_direct_llm_step( self, request: DebugExecuteRequest, chain_cfg: dict, step_def: dict, previous_steps: List[StepResult], trace_id: str, ) -> StepResult: """执行直调 LLM 链路的单个步骤""" step_index = step_def["index"] start = time.time() if step_index == 0: # ---------- Step 0: Prompt 渲染 ---------- try: prompt_kwargs = { "review_content": request.content, "review_references": request.reference or "", } prompt_template = self.base_reviewer.prompt_loader.get_prompt_template( chain_cfg["reviewer_type"], chain_cfg["prompt_name"], **prompt_kwargs, ) messages = prompt_template.format_messages(**prompt_kwargs) return StepResult( index=0, name="Prompt 渲染", status="success", duration=round(time.time() - start, 3), input_summary={ "review_content": request.content, "review_references": request.reference or "", }, output_summary={ "system_prompt": ( messages[0].content if messages else "" ), "user_prompt": ( messages[1].content if len(messages) > 1 else "" ), }, ) except Exception as exc: return StepResult( index=0, name="Prompt 渲染", status="error", duration=round(time.time() - start, 3), error=str(exc), ) elif step_index == 1: # ---------- Step 1: LLM 调用 ---------- try: prompt_kwargs = { "review_content": request.content, "review_references": request.reference or "", } prompt_template = self.base_reviewer.prompt_loader.get_prompt_template( chain_cfg["reviewer_type"], chain_cfg["prompt_name"], **prompt_kwargs, ) invoke_kwargs: dict = { "trace_id": trace_id, "task_prompt_info": { "task_prompt": prompt_template, "task_name": chain_cfg["prompt_name"], }, "timeout": request.timeout, } # 模型选择：function_name > model > chain 默认 if request.function_name: invoke_kwargs["function_name"] = request.function_name elif request.model: invoke_kwargs["model_name"] = request.model else: invoke_kwargs["function_name"] = chain_cfg.get( "function_name" ) model_response = await generate_model_client.get_model_generate_invoke( **invoke_kwargs, ) return StepResult( index=1, name="LLM 调用", status="success", duration=round(time.time() - start, 3), input_summary={ "model": request.model or "default", "timeout": request.timeout, "function_name": invoke_kwargs.get("function_name"), }, output_summary={ "raw_response": model_response, }, ) except Exception as exc: return StepResult( index=1, name="LLM 调用", status="error", duration=round(time.time() - start, 3), error=str(exc), ) elif step_index == 2: # ---------- Step 2: 结果解析 ---------- prev = previous_steps[-1] if previous_steps else None if not prev or prev.status != "success": return StepResult( index=2, name="结果解析", status="skipped", duration=0, ) try: raw = prev.output_summary.get("raw_response", "") result = self.base_reviewer.format_result( raw, chain_cfg["prompt_name"], ) return StepResult( index=2, name="结果解析", status="success", duration=round(time.time() - start, 3), input_summary={"raw_response": raw}, output_summary={"parsed_result": result.details}, ) except Exception as exc: return StepResult( index=2, name="结果解析", status="error", duration=round(time.time() - start, 3), error=str(exc), ) # fallback return StepResult( index=step_index, name=step_def.get("name", "unknown"), status="error", duration=round(time.time() - start, 3), error=f"Unknown step index: {step_index}", ) # ---------------------------------------------------------- # 专业性审查（RAG + LLM，7 步） # ---------------------------------------------------------- async def _execute_professional_step( self, request: DebugExecuteRequest, chain_cfg: dict, step_def: dict, previous_steps: List[StepResult], trace_id: str, ) -> StepResult: """执行专业性审查的单个步骤""" step_index = step_def["index"] start = time.time() phase = "RAG 召回阶段" if step_index < 4 else "AI 审查阶段" try: if step_index == 0: # ---- Step 0: 查询提取 ---- from foundation.ai.rag.retrieval.query_rewrite import ( query_rewrite_manager, ) query_pairs = query_rewrite_manager.query_extract( request.content, ) count = len(query_pairs) if query_pairs else 0 return StepResult( index=0, name="查询提取", status="success", phase=phase, duration=round(time.time() - start, 3), input_summary={"content_length": len(request.content)}, output_summary={ "query_pairs_count": count, "query_pairs": query_pairs or [], }, ) elif step_index == 1: # ---- Step 1: 实体增强检索 ---- prev = previous_steps[0] if len(previous_steps) > 0 else None if not prev or prev.status != "success": return StepResult( index=1, name="实体增强检索", status="skipped", phase=phase, duration=0, ) qp = prev.output_summary.get("query_pairs", []) if not qp: return StepResult( index=1, name="实体增强检索", status="skipped", phase=phase, duration=0, ) from foundation.ai.rag.retrieval.entities_enhance import ( entity_enhance, ) bfp_results = entity_enhance.entities_enhance_retrieval(qp) count = len(bfp_results) if bfp_results else 0 return StepResult( index=1, name="实体增强检索", status="success", phase=phase, duration=round(time.time() - start, 3), input_summary={"query_pairs_count": len(qp)}, output_summary={ "bfp_results_count": count, "bfp_results": bfp_results or [], }, ) elif step_index == 2: # ---- Step 2: 父文档增强 ---- prev = previous_steps[1] if len(previous_steps) > 1 else None if not prev or prev.status != "success": return StepResult( index=2, name="父文档增强", status="skipped", phase=phase, duration=0, ) bfp = prev.output_summary.get("bfp_results", []) if not bfp: return StepResult( index=2, name="父文档增强", status="skipped", phase=phase, duration=0, ) from core.construction_review.component.infrastructure.milvus import ( # noqa: E501 MilvusManager, MilvusConfig, ) from core.construction_review.component.infrastructure.parent_tool import ( # noqa: E501 enhance_with_parent_docs_grouped, ) milvus_mgr = MilvusManager(MilvusConfig()) enhancement = enhance_with_parent_docs_grouped( milvus_mgr, bfp, score_threshold=0.5, max_parents_per_pair=3, max_parent_text_length=8000, ) return StepResult( index=2, name="父文档增强", status="success", phase=phase, duration=round(time.time() - start, 3), input_summary={"bfp_results_count": len(bfp)}, output_summary={ "enhanced_results": enhancement.get( "enhanced_results", [] ), "enhanced_count": enhancement.get( "enhanced_count", 0 ), "parent_docs_count": len( enhancement.get("parent_docs", []) ), }, ) elif step_index == 3: # ---- Step 3: 结果提取 ---- prev = previous_steps[2] if len(previous_steps) > 2 else None if not prev or prev.status != "success": return StepResult( index=3, name="结果提取", status="skipped", phase=phase, duration=0, ) enhanced = prev.output_summary.get("enhanced_results", []) step0 = previous_steps[0] qp = step0.output_summary.get("query_pairs", []) if not enhanced: return StepResult( index=3, name="结果提取", status="skipped", phase=phase, duration=0, ) from core.construction_review.component.infrastructure.parent_tool import ( # noqa: E501 extract_query_pairs_results, ) entity_results = extract_query_pairs_results( enhanced, qp, score_threshold=0.5, ) return StepResult( index=3, name="结果提取", status="success", phase=phase, duration=round(time.time() - start, 3), input_summary={"enhanced_results_count": len(enhanced)}, output_summary={ "entity_results_count": ( len(entity_results) if entity_results else 0 ), "entity_results": entity_results or [], }, ) elif step_index == 4: # ---- Step 4: 非参数合规审查 ---- step3 = previous_steps[3] if len(previous_steps) > 3 else None if not step3 or step3.status != "success": return StepResult( index=4, name="非参数合规审查", status="skipped", phase=phase, duration=0, ) entity_results = step3.output_summary.get( "entity_results", [] ) if not entity_results: return StepResult( index=4, name="非参数合规审查", status="skipped", phase=phase, duration=0, ) rag_params = request.rag_params or RagParams() rt = rag_params.review_type from core.construction_review.component.ai_review_engine import ( AIReviewEngine, ) engine = AIReviewEngine(max_concurrent_reviews=1) results_list: List[dict] = [] for idx, item in enumerate(entity_results): if rt in ("non_parameter", "both"): cq = item.get("combined_query", "") tc = item.get("text_content", "") fn = item.get("file_name", "") res = await engine.check_non_parameter_compliance( trace_id_idx=trace_id, review_content=request.content, review_references=tc, reference_source=fn, state=None, stage_name=f"pro_non_param_{idx}", entity_query=cq, ) processed = self._process_review_result(res) processed["entity"] = item.get("entity", f"e{idx}") processed["combined_query"] = cq results_list.append(processed) return StepResult( index=4, name="非参数合规审查", status="success", phase=phase, duration=round(time.time() - start, 3), input_summary={ "entity_count": len(entity_results), "review_type": rt, }, output_summary={ "non_parameter_results": results_list, "result_count": len(results_list), }, ) elif step_index == 5: # ---- Step 5: 参数合规审查 ---- step3 = previous_steps[3] if len(previous_steps) > 3 else None if not step3 or step3.status != "success": return StepResult( index=5, name="参数合规审查", status="skipped", phase=phase, duration=0, ) entity_results = step3.output_summary.get( "entity_results", [] ) if not entity_results: return StepResult( index=5, name="参数合规审查", status="skipped", phase=phase, duration=0, ) rag_params = request.rag_params or RagParams() rt = rag_params.review_type from core.construction_review.component.ai_review_engine import ( AIReviewEngine, ) engine = AIReviewEngine(max_concurrent_reviews=1) results_list: List[dict] = [] for idx, item in enumerate(entity_results): if rt in ("parameter", "both"): cq = item.get("combined_query", "") tc = item.get("text_content", "") fn = item.get("file_name", "") res = await engine.check_parameter_compliance( trace_id_idx=trace_id, review_content=request.content, review_references=tc, reference_source=fn, state=None, stage_name=f"pro_param_{idx}", entity_query=cq, ) processed = self._process_review_result(res) processed["entity"] = item.get("entity", f"e{idx}") processed["combined_query"] = cq results_list.append(processed) return StepResult( index=5, name="参数合规审查", status="success", phase=phase, duration=round(time.time() - start, 3), input_summary={ "entity_count": len(entity_results), "review_type": rt, }, output_summary={ "parameter_results": results_list, "result_count": len(results_list), }, ) elif step_index == 6: # ---- Step 6: 结果汇总 ---- summary: Dict[str, list] = {} if len(previous_steps) > 4: s4 = previous_steps[4] if s4.status == "success": summary["non_parameter"] = s4.output_summary.get( "non_parameter_results", [] ) if len(previous_steps) > 5: s5 = previous_steps[5] if s5.status == "success": summary["parameter"] = s5.output_summary.get( "parameter_results", [] ) total_items = ( len(summary.get("non_parameter", [])) + len(summary.get("parameter", [])) ) return StepResult( index=6, name="结果汇总", status="success", phase=phase, duration=round(time.time() - start, 3), input_summary={}, output_summary={ "final_summary": summary, "total_review_items": total_items, }, ) except Exception as exc: logger.error( f"[DebugExecutor] 专业性审查步骤 {step_index} " f"({step_def['name']}) 异常: {exc}", exc_info=True, ) return StepResult( index=step_index, name=step_def["name"], status="error", phase=phase, duration=round(time.time() - start, 3), error=str(exc), ) # fallback return StepResult( index=step_index, name=step_def.get("name", "unknown"), status="error", phase=phase, duration=round(time.time() - start, 3), error=f"Unknown step index: {step_index}", ) # ---------------------------------------------------------- # 内联方法：_execute_timeliness（TC-C02-API-003 验收） # ---------------------------------------------------------- async def _execute_timeliness( self, request: DebugExecuteRequest, event_queue: asyncio.Queue, ) -> None: """ 时效性审查快捷入口（通过 AIReviewEngine 封装）。 TC-C02-API-003 要求此方法存在，此处委托给标准 execute 流程。 """ await self.execute(request, event_queue) # ---------------------------------------------------------- # 辅助方法 # ---------------------------------------------------------- @staticmethod def _process_review_result(result: Any) -> dict: """将 review 返回结果规范化为字典""" if result is None: return {"success": False, "details": {}, "error": "empty result"} if isinstance(result, dict): return result if hasattr(result, "success"): return { "success": getattr(result, "success", False), "details": getattr(result, "details", {}), "error_message": getattr(result, "error_message", None), "execution_time": getattr(result, "execution_time", None), } return {"success": True, "details": {"response": str(result)}} @staticmethod async def _emit_step_result( event_queue: asyncio.Queue, task_id: str, sr: StepResult, ) -> None: """推送 step_result 事件到队列""" await event_queue.put(("step_result", { "task_id": task_id, "step_index": sr.index, "step_name": sr.name, "status": sr.status, "duration": sr.duration, "input": sr.input_summary, "output": sr.output_summary, "error": sr.error, })) @staticmethod def _build_final_result( step_results: List[StepResult], ) -> dict: """构建 completed 事件的 final_result""" total = len(step_results) succ = sum(1 for s in step_results if s.status == "success") err = sum(1 for s in step_results if s.status == "error") skip = sum(1 for s in step_results if s.status == "skipped") return { "summary": ( f"{succ}/{total} 步骤成功, " f"{err} 错误, {skip} 跳过" ), "success_count": succ, "error_count": err, "skipped_count": skip, "total_steps": total, } @staticmethod def _save_record( request: DebugExecuteRequest, record_id: str, display_id: str, chain_cfg: dict, step_results: List[StepResult], total_duration: float, ) -> None: """持久化调用记录到 temp/debug_console/call_records/""" records_dir = "temp/debug_console/call_records" os.makedirs(records_dir, exist_ok=True) step_records = [] for s in step_results: step_records.append({ "index": s.index, "name": s.name, "phase": s.phase, "status": s.status, "duration_ms": int(s.duration * 1000), "input": s.input_summary, "output": s.output_summary, }) record = { "id": display_id, "time": datetime.now().isoformat(), "chain": request.chain_id, "chain_name": chain_cfg.get("name", ""), "doc_ref": "", "status": "succ", "duration_ms": int(total_duration * 1000), "model": request.model or "default", "prompt_ver": "", "prompt_name": chain_cfg.get("prompt_name", ""), "tokens": 0, "params": { "review_content": request.content, "review_references": request.reference or "", "model_override": request.model, "function_name": request.function_name, "timeout": request.timeout, }, "execution_params": { "isolation_mode": request.isolation_mode, "isolation_steps": list(request.isolation_steps), "rag_params": ( request.rag_params.model_dump() if request.rag_params else None ), }, "steps": step_records, "result": "", "error_message": None, } filepath = os.path.join(records_dir, f"{display_id}.json") with open(filepath, "w", encoding="utf-8") as f: json.dump(record, f, ensure_ascii=False, indent=2)