LQConstPlanWriterAgent/core/document_chat/component/retrieval_service.py

# -*- coding: utf-8 -*-
"""质量优先的多路向量检索服务。

四路召回架构：
    1. parent_vector：父表向量检索（主体内容向量）
    2. child_locator：子表向量定位 → 反查父行（精确定位片段）
    3. tag_keyword：标签关键词匹配（设备型号、标准号等）
    4. chapter_similarity：章节相似度检索（同类型章节参考）

合并策略：
    - RRF（Reciprocal Rank Fusion）融合多路排名
    - 按路径加权：parent_vector 1.0, child_locator 0.8, tag 1.2, chapter 0.5
    - 多源加分：同一条候选在多个路径中被召回时额外加分
    - 标签匹配加分：关键词出现在 tag_list 或文本中时额外加分
    - Scope 匹配加分：与当前项目/章节范围一致时额外加分

去重策略：
    - candidate_key 去重（基于 document_id + parent_id + chunk_id）
    - 内容哈希去重（同一文件同一文本内容仅保留一条）
"""

from __future__ import annotations

from dataclasses import dataclass
from hashlib import md5
from pathlib import Path
import re
from typing import Any, Callable, Dict, List, Optional

import yaml

from core.document_chat.component.document_chat_logger import document_chat_logger as logger


PROJECT_ROOT = Path(__file__).resolve().parents[3]
RETRIEVAL_CONFIG_PATH = PROJECT_ROOT / "config" / "document_chat_retrieval.yaml"


@dataclass(frozen=True)
class RetrievalConfig:
    """检索配置（不可变）。各参数含义见下方字段注释。"""
    enabled: bool = True
    parent_collection: str = "t_kngs_construction_plan_parent"
    child_collection: str = "t_kngs_construction_plan_child"
    # 各路径召回上限
    parent_recall_top_k: int = 30
    child_recall_top_k: int = 40
    tag_recall_top_k: int = 30
    chapter_recall_top_k: int = 15
    recall_top_k: int = 30
    rerank_top_k: int = 8
    submit_top_k: int = 3  # 最终送入 LLM prompt 的参考条数上限
    # 质量阈值
    min_vector_similarity: float = 0.45
    min_rerank_score: float = 0.65  # 重排质量门，低于此值被过滤
    min_qualified_count: int = 1
    # 参考内容长度限制
    max_reference_chars: int = 4000  # 所有参考总字符上限
    max_single_reference_chars: int = 1500  # 单条参考字符上限
    # 降级策略
    allow_vector_fallback: bool = False
    allow_unscoped_search: bool = False
    # 混合搜索权重（dense=sparse 向量融合）
    dense_weight: float = 0.7
    sparse_weight: float = 0.3
    child_dense_weight: float = 0.6
    child_sparse_weight: float = 0.4
    ranker_type: str = "weighted"
    # 标签召回
    tag_recall_enabled: bool = True
    tag_terms_limit: int = 8
    # RRF 参数
    rrf_k: int = 60
    # 路径权重
    parent_vector_weight: float = 1.0
    child_locator_weight: float = 0.8
    tag_weight: float = 1.2
    chapter_similarity_weight: float = 0.5
    # 加分项
    tag_exact_bonus: float = 0.08
    tag_partial_bonus: float = 0.03
    multi_source_bonus: float = 0.02
    scope_bonus: float = 0.03
    warnings: Optional[Dict[str, str]] = None


def load_retrieval_config() -> RetrievalConfig:
    """从 YAML 配置文件加载检索参数，文件不存在时使用默认值。"""
    if not RETRIEVAL_CONFIG_PATH.exists():
        return RetrievalConfig(warnings=_default_warnings())

    with open(RETRIEVAL_CONFIG_PATH, "r", encoding="utf-8") as handle:
        raw = yaml.safe_load(handle) or {}

    retrieval = raw.get("retrieval") or {}
    warnings = raw.get("warnings") or _default_warnings()
    return RetrievalConfig(
        enabled=bool(retrieval.get("enabled", True)),
        parent_collection=str(retrieval.get("parent_collection") or "t_kngs_construction_plan_parent"),
        child_collection=str(retrieval.get("child_collection") or "t_kngs_construction_plan_child"),
        parent_recall_top_k=_to_int(retrieval.get("parent_recall_top_k"), 30),
        child_recall_top_k=_to_int(retrieval.get("child_recall_top_k"), 40),
        tag_recall_top_k=_to_int(retrieval.get("tag_recall_top_k"), 30),
        chapter_recall_top_k=_to_int(retrieval.get("chapter_recall_top_k"), 15),
        recall_top_k=_to_int(retrieval.get("recall_top_k"), 30),
        rerank_top_k=_to_int(retrieval.get("rerank_top_k"), 8),
        submit_top_k=_to_int(retrieval.get("submit_top_k"), 3),
        min_vector_similarity=_to_float(retrieval.get("min_vector_similarity"), 0.45),
        min_rerank_score=_to_float(retrieval.get("min_rerank_score"), 0.65),
        min_qualified_count=_to_int(retrieval.get("min_qualified_count"), 1),
        max_reference_chars=_to_int(retrieval.get("max_reference_chars"), 4000),
        max_single_reference_chars=_to_int(retrieval.get("max_single_reference_chars"), 1500),
        allow_vector_fallback=bool(retrieval.get("allow_vector_fallback", False)),
        allow_unscoped_search=bool(retrieval.get("allow_unscoped_search", False)),
        dense_weight=_to_float(retrieval.get("dense_weight"), 0.7),
        sparse_weight=_to_float(retrieval.get("sparse_weight"), 0.3),
        child_dense_weight=_to_float(retrieval.get("child_dense_weight"), 0.6),
        child_sparse_weight=_to_float(retrieval.get("child_sparse_weight"), 0.4),
        ranker_type=str(retrieval.get("ranker_type") or "weighted"),
        tag_recall_enabled=bool(retrieval.get("tag_recall_enabled", True)),
        tag_terms_limit=_to_int(retrieval.get("tag_terms_limit"), 8),
        rrf_k=_to_int(retrieval.get("rrf_k"), 60),
        parent_vector_weight=_to_float(retrieval.get("parent_vector_weight"), 1.0),
        child_locator_weight=_to_float(retrieval.get("child_locator_weight"), 0.8),
        tag_weight=_to_float(retrieval.get("tag_weight"), 1.2),
        chapter_similarity_weight=_to_float(retrieval.get("chapter_similarity_weight"), 0.5),
        tag_exact_bonus=_to_float(retrieval.get("tag_exact_bonus"), 0.08),
        tag_partial_bonus=_to_float(retrieval.get("tag_partial_bonus"), 0.03),
        multi_source_bonus=_to_float(retrieval.get("multi_source_bonus"), 0.02),
        scope_bonus=_to_float(retrieval.get("scope_bonus"), 0.03),
        warnings=warnings,
    )


class DocumentChatRetrievalService:
    """构建检索查询，从向量库召回高质量候选。

    核心流程：
    1. build_query：将用户输入、章节信息、意图拼接为检索 query
    2. recall：执行多路召回 → RRF 合并 → 去重
    """

    # 父表查询输出字段
    PARENT_OUTPUT_FIELDS = [
        "pk", "text", "document_id", "parent_id", "index", "tag_list",
        "metadata", "file_name", "chapter_title",
        "chapter_level_1", "chapter_level_2", "chapter_level_3",
    ]
    # 子表查询输出字段
    CHILD_OUTPUT_FIELDS = [
        "pk", "text", "document_id", "parent_id", "index", "tag_list",
        "metadata", "file_name", "chapter_title",
        "chapter_level_1", "chapter_level_2", "chapter_level_3",
    ]

    def __init__(self, config: Optional[RetrievalConfig] = None):
        self.config = config or load_retrieval_config()

    # ============================================================
    # Query 构建
    # ============================================================
    def build_query(self, state: Dict[str, Any]) -> str:
        """构建精炼检索 query，避免冗余的项目摘要。

        拼接内容：
        - 用户原始输入
        - 意图识别后的规范化指令
        - 当前选中章节编号 + 标题
        - 提取的关键词（最多 8 个）
        去重后截取 120 字符。
        """
        selected_section = state.get("selected_section") or {}
        intent_result = state.get("intent_result") or {}
        keywords = self.build_query_keywords(state)

        parts = [
            state.get("user_message") or "",
            intent_result.get("normalized_instruction") or "",
            f"{selected_section.get('index', '')} {selected_section.get('title', '')}".strip(),
            " ".join(keywords[:8]),
        ]
        return _dedupe_join(parts, max_chars=120)

    def build_query_keywords(self, state: Dict[str, Any], query: Optional[str] = None) -> List[str]:
        """从多来源提取检索关键词。

        来源优先级：
        1. 用户输入
        2. 意图规范化指令
        3. 章节编号 + 标题
        4. 章节正文内容（前 500 字）
        5. 已拼接的 query
        6. 历史对话中用户消息（排除 AI 回复，防止助手建议污染检索）

        关键词提取规则见 _extract_retrieval_keywords。
        """
        selected_section = state.get("selected_section") or {}
        intent_result = state.get("intent_result") or {}
        history = state.get("conversation_history") or []

        sources = [
            state.get("user_message") or "",
            intent_result.get("normalized_instruction") or "",
            f"{selected_section.get('index', '')} {selected_section.get('title', '')}",
            str(selected_section.get("content") or "")[:500],
            query or "",
        ]
        if history:
            for turn in history[-6:]:
                if not isinstance(turn, dict):
                    continue
                role = str(turn.get("role") or turn.get("sender") or "").lower()
                # 仅取用户消息，跳过 AI 助手回复
                if role in ("assistant", "ai", "bot", "model"):
                    continue
                content = str(turn.get("content") or turn.get("message") or "")
                if content:
                    sources.append(content)

        keywords: List[str] = []
        seen = set()
        for text in sources:
            for keyword in _extract_retrieval_keywords(str(text or "")):
                normalized = keyword.strip()
                if not normalized or normalized in seen:
                    continue
                seen.add(normalized)
                keywords.append(normalized)
                if len(keywords) >= 20:
                    return keywords
        return keywords

    # ============================================================
    # 主召回入口
    # ============================================================
    def recall(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """执行多路向量召回，RRF 合并，去重过滤。

        返回：
        - retrieval_candidates：去重后的候选列表
        - retrieval_status：recalled / no_scope / no_recall / disabled
        - retrieval_metrics：各路径召回统计
        - retrieval_steps：每步详细日志
        """
        if not self.config.enabled:
            return self._empty_result("disabled", [], retrieval_method="disabled")

        query = str(state.get("retrieval_query") or "").strip()
        if not query:
            return self._empty_result("no_recall", [self._warning("no_recall")], retrieval_method="empty_query")

        # 提取检索范围（项目ID、工程类型、章节分类等）
        scope = self._extract_scope(state)
        if not self._has_reliable_scope(scope) and not self.config.allow_unscoped_search:
            return self._empty_result(
                "no_scope",
                [self._warning("no_scope")],
                retrieval_method="no_scope",
                retrieval_scope=scope,
            )

        keywords = list(state.get("retrieval_keywords") or self.build_query_keywords(state, query))
        retrieval_steps: List[Dict[str, Any]] = []
        source_results: Dict[str, List[Dict[str, Any]]] = {}

        # ===== 四路召回 =====
        source_results["parent_vector"] = self._run_recall_path(
            "parent_vector",
            lambda: self._recall_by_parent_vector(scope, query),
            retrieval_steps,
            query=query,
            scope=scope,
        )
        source_results["child_locator"] = self._run_recall_path(
            "child_locator",
            lambda: self._recall_by_child_locator(scope, query),
            retrieval_steps,
            query=query,
            scope=scope,
        )
        if self.config.tag_recall_enabled:
            source_results["tag"] = self._run_recall_path(
                "tag",
                lambda: self._recall_by_tag(scope, keywords),
                retrieval_steps,
                query=" ".join(keywords[: self.config.tag_terms_limit]),
                scope=scope,
            )
        if scope.get("chapter_level_1") and scope.get("chapter_level_2"):
            source_results["chapter_similarity"] = self._run_recall_path(
                "chapter_similarity",
                lambda: self._recall_by_chapter(scope, query),
                retrieval_steps,
                query=query,
                scope=scope,
            )

        # ===== RRF 合并 + 去重 =====
        merged_candidates = self._merge_recall_results(source_results, scope, keywords)
        cleaned = self._clean_candidates(merged_candidates)

        retrieval_steps.append(
            {
                "step": "rrf_merge",
                "query": query,
                "scope": {key: value for key, value in scope.items() if value},
                "count": len(merged_candidates),
                "items": _pack_log_items(merged_candidates),
            }
        )
        retrieval_steps.append(
            {
                "step": "clean_candidates",
                "count": len(cleaned),
                "items": _pack_log_items(cleaned),
            }
        )

        if not cleaned:
            return self._empty_result(
                "no_recall",
                [self._warning("no_recall")],
                retrieval_method="multi_path_rrf",
                retrieval_scope=scope,
                retrieval_steps=retrieval_steps,
            )

        source_counts = {source: len(items or []) for source, items in source_results.items()}

        # 日志：区分请求的 scope、实际应用的过滤、实际召回的文件
        applied_expr = self._build_filter_expr(scope)
        actual_files = list(dict.fromkeys(
            str(item.get("source", ""))[:40]
            for item in cleaned
            if item.get("source")
        ))[:5]
        logger.info(
            f"[DocumentChat] recall completed: method=multi_path_rrf "
            f"requested_scope={dict((k, v) for k, v in scope.items() if v)} "
            f"applied_filter='{applied_expr}' "
            f"actual_sources={actual_files} "
            f"source_counts={source_counts} "
            f"total={len(cleaned)} max_sim={max((item.get('vector_similarity', 0.0) for item in cleaned), default=0.0):.4f}"
        )

        metrics = {
            "recall_count": len(cleaned),
            "merged_count": len(merged_candidates),
            "source_counts": source_counts,
            "max_vector_similarity": max((item.get("vector_similarity", 0.0) for item in cleaned), default=0.0),
            "max_fusion_score": max((item.get("fusion_score", 0.0) for item in cleaned), default=0.0),
            "scope": {key: value for key, value in scope.items() if value},
            "retrieval_method": "multi_path_rrf",
        }
        return {
            "retrieval_candidates": cleaned,
            "retrieval_steps": retrieval_steps,
            "retrieval_status": "recalled",
            "retrieval_method": "multi_path_rrf",
            "retrieval_metrics": metrics,
            "warnings": [],
        }

    def _run_recall_path(
        self,
        step: str,
        func: Callable[[], List[Dict[str, Any]]],
        retrieval_steps: List[Dict[str, Any]],
        query: str,
        scope: Dict[str, Any],
    ) -> List[Dict[str, Any]]:
        """执行单路召回，异常时不阻断其他路径。"""
        try:
            candidates = func() or []
            retrieval_steps.append(
                {
                    "step": step,
                    "query": query,
                    "scope": {key: value for key, value in scope.items() if value},
                    "count": len(candidates),
                    "items": _pack_log_items(candidates),
                }
            )
            return candidates
        except Exception as exc:
            logger.warning(f"[DocumentChat] {step} recall failed: {exc}", exc_info=True)
            retrieval_steps.append(
                {
                    "step": step,
                    "query": query,
                    "scope": {key: value for key, value in scope.items() if value},
                    "count": 0,
                    "error": str(exc),
                    "items": [],
                }
            )
            return []

    # ============================================================
    # 四路召回具体实现
    # ============================================================
    def _recall_by_parent_vector(self, scope: Dict[str, Any], query: str) -> List[Dict[str, Any]]:
        """父表向量检索：Milvus 混合搜索（dense + sparse），直接返回父表文档。"""
        from foundation.database.base.vector.milvus_vector import MilvusVectorManager

        expr = self._build_filter_expr(scope)
        results = MilvusVectorManager().hybrid_search(
            param={"collection_name": self.config.parent_collection, "expr": expr},
            query_text=query,
            top_k=self.config.parent_recall_top_k,
            ranker_type=self.config.ranker_type,
            dense_weight=self.config.dense_weight,
            sparse_weight=self.config.sparse_weight,
        )
        return [
            self._candidate_from_vector_row(row, "parent_vector", scope)
            for row in results
            if str(row.get("text_content") or "").strip()
        ]

    def _recall_by_child_locator(self, scope: Dict[str, Any], query: str) -> List[Dict[str, Any]]:
        """子表向量定位 + 父表反查：先用 query 在子表中找到匹配片段，
        再通过 parent_id 反查父表行，获取完整的父文档内容。

        优势：子表粒度更细，能精确定位到段落级别，然后拉取对应父文档的完整内容。
        """
        from foundation.database.base.vector.milvus_vector import MilvusVectorManager

        expr = self._build_filter_expr(scope)
        child_rows = MilvusVectorManager().hybrid_search(
            param={"collection_name": self.config.child_collection, "expr": expr},
            query_text=query,
            top_k=self.config.child_recall_top_k,
            ranker_type=self.config.ranker_type,
            dense_weight=self.config.child_dense_weight,
            sparse_weight=self.config.child_sparse_weight,
        )

        # 按 parent_id 分组子表命中结果
        child_groups: Dict[str, List[Dict[str, Any]]] = {}
        for row in child_rows:
            metadata = self._normalize_row_metadata(row.get("metadata") or {})
            parent_id = str(self._metadata_value(metadata, "parent_id") or "").strip()
            if not parent_id:
                continue
            child_groups.setdefault(parent_id, []).append(row)

        # 通过 parent_id 反查父表
        parent_rows = self._fetch_parent_rows_by_parent_ids(list(child_groups.keys()), scope)
        candidates = []
        for parent_row in parent_rows:
            parent_id = str(parent_row.get("parent_id") or "").strip()
            matches = child_groups.get(parent_id) or []
            max_similarity = max((_to_float(item.get("similarity"), 0.0) for item in matches), default=0.0)
            candidate = self._candidate_from_parent_row(parent_row, "child_locator", scope, max_similarity)
            metadata = candidate.setdefault("metadata", {})
            metadata["child_hit_count"] = len(matches)  # 子表命中次数
            metadata["matched_child_texts"] = [
                str(item.get("text_content") or "").strip()
                for item in matches[:5]
                if str(item.get("text_content") or "").strip()
            ]
            candidates.append(candidate)
        return candidates

    def _recall_by_tag(self, scope: Dict[str, Any], keywords: List[str]) -> List[Dict[str, Any]]:
        """标签关键词召回：从关键词中筛选标准号、设备名等专业术语，
        在 tag_list 字段上做 LIKE 匹配。

        注意：标签召回容易过度匹配，因此结果相似度乘以 0.7 打折。
        """
        tag_terms = self._select_tag_terms(keywords)
        if not tag_terms:
            return []

        tag_expr = self._build_tag_expr(tag_terms)
        scope_expr = self._build_filter_expr(scope)
        expr = _combine_expr(scope_expr, tag_expr)

        # 父表标签匹配
        parent_rows = self._condition_query(
            collection_name=self.config.parent_collection,
            filter_expr=expr,
            output_fields=self.PARENT_OUTPUT_FIELDS,
            limit=self.config.tag_recall_top_k,
        )
        candidates = [
            self._candidate_from_parent_row(row, "tag", scope, self.config.min_vector_similarity)
            for row in parent_rows
        ]

        # 子表标签匹配，再反查父行
        child_rows = self._condition_query(
            collection_name=self.config.child_collection,
            filter_expr=expr,
            output_fields=self.CHILD_OUTPUT_FIELDS,
            limit=self.config.tag_recall_top_k,
        )
        child_parent_ids = []
        child_tag_map: Dict[str, List[str]] = {}
        for row in child_rows:
            parent_id = str(row.get("parent_id") or self._metadata_value(row, "parent_id") or "").strip()
            if not parent_id:
                continue
            child_parent_ids.append(parent_id)
            text = str(row.get("text") or "").strip()
            if text:
                child_tag_map.setdefault(parent_id, []).append(text)

        for row in self._fetch_parent_rows_by_parent_ids(child_parent_ids, scope):
            parent_id = str(row.get("parent_id") or "").strip()
            candidate = self._candidate_from_parent_row(row, "tag", scope, self.config.min_vector_similarity)
            metadata = candidate.setdefault("metadata", {})
            metadata["matched_child_texts"] = child_tag_map.get(parent_id, [])[:5]
            candidates.append(candidate)

        # 标签结果打折，防止过度匹配
        for candidate in candidates:
            candidate["vector_similarity"] *= 0.7

        # 记录匹配的标签术语
        for candidate in candidates:
            metadata = candidate.setdefault("metadata", {})
            tag_text = " ".join(
                str(value or "")
                for value in (
                    metadata.get("tag_list"),
                    candidate.get("text"),
                    " ".join(metadata.get("matched_child_texts") or []),
                )
            )
            metadata["tag_match_terms"] = [term for term in tag_terms if term and term in tag_text]
        return candidates

    def _recall_by_chapter(self, scope: Dict[str, Any], query: str) -> List[Dict[str, Any]]:
        """章节相似度检索：调用现有 similar_fragment_service，
        按 chapter_level_1 + chapter_level_2 限定范围搜索相似片段。
        """
        from core.construction_write.component.similar_fragment_service import search_similar_fragments

        rows = search_similar_fragments(
            level1=str(scope.get("chapter_level_1") or ""),
            level2=str(scope.get("chapter_level_2") or ""),
            search_text=query,
            top_k=self.config.chapter_recall_top_k,
        )
        candidates = []
        for row in rows:
            metadata = {
                "tenant_id": scope.get("tenant_id") or "",
                "project_id": scope.get("project_id") or "",
                "knowledge_base_id": scope.get("knowledge_base_id") or "",
                "file_name": row.get("file_name") or "",
                "chapter_level_1": row.get("chapter_level_1") or scope.get("chapter_level_1") or "",
                "chapter_level_2": row.get("chapter_level_2") or scope.get("chapter_level_2") or "",
                "parent_count": row.get("parent_count", 0),
                "source_scope_valid": True,  # 通过章节分类限定，天然 scope 匹配
            }
            text = str(row.get("text") or "").strip()
            candidates.append(
                {
                    "candidate_key": self._build_candidate_key({**row, "metadata": metadata}, text),
                    "text": text,
                    "source": metadata.get("file_name") or "向量知识库",
                    "vector_similarity": _to_float(row.get("similarity"), 0.0),
                    "fusion_score": 0.0,
                    "metadata": metadata,
                    "source_hits": {},
                    "retrieval_source": "chapter_similarity",
                }
            )
        return candidates

    # ============================================================
    # RRF 合并
    # ============================================================
    def _merge_recall_results(
        self,
        source_results: Dict[str, List[Dict[str, Any]]],
        scope: Dict[str, Any],
        keywords: List[str],
    ) -> List[Dict[str, Any]]:
        """多路召回结果 RRF 融合合并。

        融合分数计算：
        - 基础分：weight / (rrf_k + rank)，按路径权重和排名计算
        - 多源加分：同一条候选在多个路径中被召回时额外加分
        - Scope 加分：与当前项目范围一致时额外加分
        - 标签加分：关键词出现在候选文本中时额外加分
        """
        weights = {
            "parent_vector": self.config.parent_vector_weight,
            "child_locator": self.config.child_locator_weight,
            "tag": self.config.tag_weight,
            "chapter_similarity": self.config.chapter_similarity_weight,
        }
        merged: Dict[str, Dict[str, Any]] = {}

        for source, candidates in source_results.items():
            weight = weights.get(source, 0.0)
            for rank, item in enumerate(candidates or [], start=1):
                key = str(item.get("candidate_key") or self._build_candidate_key(item, item.get("text")))
                if not key:
                    continue
                if key not in merged:
                    candidate = dict(item)
                    candidate["candidate_key"] = key
                    candidate["source_hits"] = {}
                    candidate["fusion_score"] = 0.0
                    merged[key] = candidate

                current = merged[key]
                # RRF 公式：累加 weight / (rrf_k + rank)
                current["fusion_score"] = _to_float(current.get("fusion_score"), 0.0) + weight / (self.config.rrf_k + rank)
                current["vector_similarity"] = max(
                    _to_float(current.get("vector_similarity"), 0.0),
                    _to_float(item.get("vector_similarity"), 0.0),
                )
                current.setdefault("source_hits", {})[source] = {
                    "rank": rank,
                    "vector_similarity": _to_float(item.get("vector_similarity"), 0.0),
                }
                self._merge_metadata(current, item)

        # 额外加分
        for candidate in merged.values():
            source_hits = candidate.get("source_hits") if isinstance(candidate.get("source_hits"), dict) else {}
            metadata = candidate.get("metadata") if isinstance(candidate.get("metadata"), dict) else {}
            if len(source_hits) > 1:
                candidate["fusion_score"] += self.config.multi_source_bonus
            if self._metadata_matches_scope(metadata, scope):
                candidate["fusion_score"] += self.config.scope_bonus
            candidate["fusion_score"] += self._calc_tag_bonus(candidate, keywords)

        return sorted(merged.values(), key=lambda item: item.get("fusion_score", 0.0), reverse=True)[: self.config.recall_top_k]

    # ============================================================
    # Milvus 查询辅助
    # ============================================================
    def _fetch_parent_rows_by_parent_ids(self, parent_ids: List[str], scope: Dict[str, Any]) -> List[Dict[str, Any]]:
        """根据 parent_id 列表反查父表行，去重后逐条查询。"""
        unique_ids = []
        seen = set()
        for parent_id in parent_ids:
            value = str(parent_id or "").strip()
            if value and value not in seen:
                seen.add(value)
                unique_ids.append(value)

        rows: List[Dict[str, Any]] = []
        scope_expr = self._build_filter_expr(scope)
        for parent_id in unique_ids[: self.config.recall_top_k]:
            parent_expr = f"parent_id == '{_escape_milvus_string(parent_id)}'"
            expr = _combine_expr(parent_expr, scope_expr)
            rows.extend(
                self._condition_query(
                    collection_name=self.config.parent_collection,
                    filter_expr=expr,
                    output_fields=self.PARENT_OUTPUT_FIELDS,
                    limit=100,
                )
            )
        return rows

    def _condition_query(
        self,
        collection_name: str,
        filter_expr: str,
        output_fields: List[str],
        limit: int,
    ) -> List[Dict[str, Any]]:
        """Milvus 条件查询（非向量），按 filter 表达式筛选文档。"""
        from core.construction_write.component.similar_fragment_service import get_milvus_manager

        if not filter_expr:
            return []
        return get_milvus_manager().condition_query(
            collection_name=collection_name,
            filter=filter_expr,
            output_fields=output_fields,
            limit=limit,
        )

    # ============================================================
    # 候选构建
    # ============================================================
    def _candidate_from_vector_row(self, row: Dict[str, Any], source: str, scope: Dict[str, Any]) -> Dict[str, Any]:
        """从 Milvus 混合搜索结果行构建标准候选。"""
        metadata = self._normalize_row_metadata(row.get("metadata") or {})
        text = str(row.get("text_content") or row.get("text") or "").strip()
        metadata["source_scope_valid"] = self._metadata_matches_scope(metadata, scope)
        return {
            "candidate_key": self._build_candidate_key(metadata, text),
            "text": text,
            "source": metadata.get("file_name") or metadata.get("title") or "向量知识库",
            "vector_similarity": _to_float(row.get("similarity"), 0.0),
            "fusion_score": 0.0,
            "metadata": metadata,
            "source_hits": {},
            "retrieval_source": source,
        }

    def _candidate_from_parent_row(
        self,
        row: Dict[str, Any],
        source: str,
        scope: Dict[str, Any],
        vector_similarity: float,
    ) -> Dict[str, Any]:
        """从父表行构建标准候选。"""
        metadata = self._normalize_row_metadata(row)
        text = str(row.get("text") or "").strip()
        metadata["source_scope_valid"] = self._metadata_matches_scope(metadata, scope)
        return {
            "candidate_key": self._build_candidate_key(metadata, text),
            "text": text,
            "source": metadata.get("file_name") or "向量知识库",
            "vector_similarity": _to_float(vector_similarity, 0.0),
            "fusion_score": 0.0,
            "metadata": metadata,
            "source_hits": {},
            "retrieval_source": source,
        }

    # ============================================================
    # Scope 提取与过滤
    # ============================================================
    def _extract_scope(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """从工作流状态中提取检索范围信息。

        按优先级从 selected_section、document_context、project_info、retrieval_filters
        中查找字段值，兼容多种字段命名。
        """
        selected = state.get("selected_section") or {}
        context = state.get("document_context") or {}
        project = state.get("project_info") or {}
        filters = context.get("retrieval_filters") if isinstance(context.get("retrieval_filters"), dict) else {}
        filters = filters or project.get("retrieval_filters") if isinstance(project.get("retrieval_filters"), dict) else filters

        def pick(*keys: str) -> str:
            for source in (selected, context, project, filters or {}):
                for key in keys:
                    value = source.get(key) if isinstance(source, dict) else None
                    if value not in (None, ""):
                        return str(value).strip()
            return ""

        return {
            "tenant_id": pick("tenant_id"),
            "project_id": pick("project_id"),
            "knowledge_base_id": pick("knowledge_base_id", "kb_id"),
            "engineering_type": pick("engineering_type", "project_type"),
            "plan_type": pick("plan_type"),
            "chapter_level_1": pick("chapter_level_1", "level1"),
            "chapter_level_2": pick("chapter_level_2", "level2"),
            "chapter_level_3": pick("chapter_level_3", "level3"),
        }

    @staticmethod
    def _has_reliable_scope(scope: Dict[str, Any]) -> bool:
        """判断是否有足够可靠的 scope 用于限定检索范围。"""
        if scope.get("chapter_level_1") and scope.get("chapter_level_2"):
            return True
        return bool(scope.get("plan_type"))

    def _build_filter_expr(self, scope: Dict[str, Any]) -> str:
        """构建 Milvus 过滤表达式，按章节层级限定检索范围。"""
        conditions = []
        for key in ("plan_type", "chapter_level_1", "chapter_level_2", "chapter_level_3"):
            value = str(scope.get(key) or "").strip()
            if value:
                conditions.append(f"{key} == '{_escape_milvus_string(value)}'")
        return " and ".join(conditions)

    def _build_tag_expr(self, tag_terms: List[str]) -> str:
        """构建标签 LIKE 查询表达式。"""
        conditions = []
        for term in tag_terms[: self.config.tag_terms_limit]:
            conditions.append(f'tag_list like "%{_escape_milvus_string(term)}%"')
        return " or ".join(conditions)

    def _select_tag_terms(self, keywords: List[str]) -> List[str]:
        """从关键词中筛选高价值标签术语。

        排除：验收、标准、规范等通用词（几乎匹配所有文档）
        优先：标准号（如 TB10212-2012）、设备名（架桥机、龙门吊等）
        """
        generic_terms = {
            "验收", "标准", "规范", "检查", "检测", "试验", "安装", "拆除",
            "要求", "安全", "环保", "质量", "进度", "交底",
        }
        device_terms = {"架桥机", "龙门吊", "吊车", "塔吊", "施工电梯", "挂篮", "支架", "台车"}
        selected = []
        priority = []  # 标准号和设备名优先
        seen = set()
        for keyword in keywords:
            value = str(keyword or "").strip()
            if len(value) < 2 or value in seen:
                continue
            seen.add(value)
            if value in generic_terms:
                continue
            if re.match(r"[A-Z]{1,3}\d{4,}", value) or value in device_terms:
                priority.append(value)
            elif len(selected) < self.config.tag_terms_limit:
                selected.append(value)
        return priority + selected

    @staticmethod
    def _metadata_matches_scope(metadata: Dict[str, Any], scope: Dict[str, Any]) -> bool:
        """检查候选 metadata 是否与当前检索 scope 兼容。

        不要求所有字段都匹配，仅校验 scope 和 metadata 同时存在且不一致的字段。
        """
        required_keys = ["tenant_id", "project_id", "knowledge_base_id", "chapter_level_1", "chapter_level_2", "chapter_level_3"]
        for key in required_keys:
            expected = str(scope.get(key) or "").strip()
            if not expected:
                continue
            actual = str(metadata.get(key) or "").strip()
            if actual and actual != expected:
                return False
        return True

    # ============================================================
    # Metadata 处理
    # ============================================================
    def _normalize_row_metadata(self, row_or_metadata: Any) -> Dict[str, Any]:
        """规范化行数据为统一的 metadata 字典。处理嵌套 metadata 和 YAML 字符串。"""
        metadata = self._normalize_metadata(row_or_metadata)
        inner = self._normalize_metadata(metadata.get("metadata")) if metadata.get("metadata") else {}
        for key, value in inner.items():
            metadata.setdefault(key, value)
        for key in self.PARENT_OUTPUT_FIELDS:
            if isinstance(row_or_metadata, dict) and row_or_metadata.get(key) not in (None, ""):
                metadata[key] = row_or_metadata.get(key)
        return metadata

    @staticmethod
    def _normalize_metadata(metadata: Any) -> Dict[str, Any]:
        """将 metadata 转为字典，支持 YAML 字符串解析。"""
        if isinstance(metadata, dict):
            return dict(metadata)
        if isinstance(metadata, str) and metadata.strip():
            try:
                loaded = yaml.safe_load(metadata)
                return dict(loaded) if isinstance(loaded, dict) else {}
            except Exception:
                return {}
        return {}

    @staticmethod
    def _metadata_value(metadata: Dict[str, Any], key: str) -> Any:
        """安全获取 metadata 值，支持嵌套 metadata.metadata 和 YAML 字符串。"""
        if key in metadata:
            return metadata.get(key)
        nested = metadata.get("metadata")
        if isinstance(nested, dict):
            return nested.get(key)
        if isinstance(nested, str) and nested.strip():
            try:
                parsed = yaml.safe_load(nested)
                if isinstance(parsed, dict):
                    return parsed.get(key)
            except Exception:
                return None
        return None

    def _build_candidate_key(self, metadata: Dict[str, Any], text: Any = "") -> str:
        """构建候选唯一标识键，按优先级尝试不同字段组合。"""
        metadata = self._normalize_row_metadata(metadata)
        document_id = str(self._metadata_value(metadata, "document_id") or "").strip()
        parent_id = str(self._metadata_value(metadata, "parent_id") or "").strip()
        chunk_id = str(self._metadata_value(metadata, "chunk_id") or "").strip()
        chapter_title = str(self._metadata_value(metadata, "chapter_title") or "").strip()
        index = self._metadata_value(metadata, "index")
        pk = str(self._metadata_value(metadata, "pk") or "").strip()

        if document_id and parent_id and chunk_id:
            return f"{document_id}::{parent_id}::{chunk_id}"
        if document_id and parent_id and chapter_title and index not in (None, ""):
            return f"{document_id}::{parent_id}::{chapter_title}::{index}"
        if pk:
            return pk
        if parent_id and chapter_title and index not in (None, ""):
            return f"{parent_id}::{chapter_title}::{index}"
        return str(text or "")[:300]

    def _merge_metadata(self, current: Dict[str, Any], incoming: Dict[str, Any]) -> None:
        """合并两条候选的 metadata，不覆盖已有非空值。"""
        current_meta = current.setdefault("metadata", {})
        incoming_meta = incoming.get("metadata") if isinstance(incoming.get("metadata"), dict) else {}
        for key, value in incoming_meta.items():
            if key not in current_meta or current_meta.get(key) in (None, "", []):
                current_meta[key] = value
        if incoming.get("source") and not current.get("source"):
            current["source"] = incoming.get("source")

    # ============================================================
    # 加分计算
    # ============================================================
    def _calc_tag_bonus(self, candidate: Dict[str, Any], keywords: List[str]) -> float:
        """计算标签匹配加分：关键词精确匹配 tag_list 加分更多，仅出现在文本中加分较少。"""
        metadata = candidate.get("metadata") if isinstance(candidate.get("metadata"), dict) else {}
        text = " ".join(
            str(value or "")
            for value in (
                candidate.get("text"),
                metadata.get("tag_list"),
                " ".join(metadata.get("matched_child_texts") or []),
            )
        )
        bonus = 0.0
        for keyword in self._select_tag_terms(keywords):
            if not keyword:
                continue
            if keyword in str(metadata.get("tag_list") or ""):
                bonus += self.config.tag_exact_bonus
            elif keyword in text:
                bonus += self.config.tag_partial_bonus
        return bonus

    # ============================================================
    # 候选清理
    # ============================================================
    def _clean_candidates(self, candidates: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """清理候选：过滤过短文本、双重去重（candidate_key + 内容哈希）。

        去重策略：
        1. candidate_key 去重：相同 document+parent+chunk 视为同一条
        2. 内容哈希去重：同一文件同一文本内容（即使路径不同）只保留一条
        """
        cleaned = []
        seen_keys = set()
        seen_hashes = set()
        for item in candidates:
            text = str(item.get("text") or "").strip()
            if len(text) < 20:
                continue
            metadata = item.get("metadata") if isinstance(item.get("metadata"), dict) else {}
            dedupe_key = str(item.get("candidate_key") or text[:300])
            # 内容哈希去重
            content_hash = _content_hash(text[:300])
            file_name = str(metadata.get("file_name") or "")
            hash_key = f"{file_name}::{content_hash}"
            if dedupe_key in seen_keys or hash_key in seen_hashes:
                continue
            seen_keys.add(dedupe_key)
            seen_hashes.add(hash_key)
            metadata["candidate_key"] = dedupe_key
            cleaned.append(
                {
                    "candidate_key": dedupe_key,
                    "text": text[: self.config.max_single_reference_chars],
                    "source": str(item.get("source") or metadata.get("file_name") or "向量知识库"),
                    "vector_similarity": _to_float(item.get("vector_similarity"), 0.0),
                    "fusion_score": _to_float(item.get("fusion_score"), 0.0),
                    "source_hits": item.get("source_hits") if isinstance(item.get("source_hits"), dict) else {},
                    "metadata": metadata,
                }
            )
        cleaned.sort(key=lambda item: (item.get("fusion_score", 0.0), item.get("vector_similarity", 0.0)), reverse=True)
        return cleaned[: self.config.recall_top_k]

    # ============================================================
    # 空结果/告警
    # ============================================================
    def _empty_result(
        self,
        status: str,
        warnings: List[str],
        retrieval_method: str = "",
        retrieval_scope: Optional[Dict[str, Any]] = None,
        retrieval_steps: Optional[List[Dict[str, Any]]] = None,
    ) -> Dict[str, Any]:
        """构建空召回结果。"""
        return {
            "retrieval_candidates": [],
            "retrieval_steps": retrieval_steps or [],
            "retrieval_status": status,
            "retrieval_method": retrieval_method,
            "retrieval_metrics": {
                "recall_count": 0,
                "retrieval_method": retrieval_method,
                "scope": {key: value for key, value in (retrieval_scope or {}).items() if value},
            },
            "warnings": warnings,
        }

    def _warning(self, key: str) -> str:
        """获取指定键的告警文案。"""
        warnings = self.config.warnings or _default_warnings()
        return warnings.get(key) or ""


def _default_warnings() -> Dict[str, str]:
    return {
        "no_scope": "缺少可靠的知识库检索范围，本次未引用向量库内容。",
        "no_recall": "未召回可信知识库内容，本次回答不引用向量库。",
        "low_confidence": "未找到可信度足够的知识库片段，本次未引用向量库内容。",
        "rerank_failed": "知识库片段重排不可用，本次未引用向量库内容。",
    }


def _escape_milvus_string(value: str) -> str:
    """转义 Milvus 字符串中的特殊字符（反斜杠、单引号、双引号）。"""
    return str(value).replace("\\", "\\\\").replace("'", "\\'").replace('"', '\\"')


def _combine_expr(*exprs: str) -> str:
    """用 AND 连接多个过滤表达式，每个子表达式加括号。"""
    parts = [f"({expr})" for expr in exprs if str(expr or "").strip()]
    return " and ".join(parts)


def _dedupe_join(parts: List[str], max_chars: int) -> str:
    """去重后拼接文本片段，限制总长度。"""
    values = []
    seen = set()
    for part in parts:
        value = re.sub(r"\s+", " ", str(part or "")).strip()
        if not value or value in seen:
            continue
        seen.add(value)
        values.append(value)
    return " ".join(values)[:max_chars]


def _extract_retrieval_keywords(text: str) -> List[str]:
    """从文本中提取检索关键词，支持多种模式：

    1. 标准号/型号：如 TB10212-2012、φ48.3×3.6
    2. 规范名称：《XXX规范》
    3. 领域专业术语：验收、架桥机、箱梁等
    4. 术语+动作组合：XX验收、XX安装
    5. 长词中的领域术语片段
    """
    if not text:
        return []

    keywords: List[str] = []
    # 模式1：标准号/型号（字母+数字，可选连字符）
    for match in re.findall(r"[A-Za-z]{1,8}\s*\d{2,8}(?:[-—]\d{2,4})?", text):
        keywords.append(re.sub(r"\s+", "", match).upper())
    # 模式2：《XXX》规范名称
    for match in re.findall(r"《([^》]{2,40})》", text):
        keywords.append(match.strip())

    # 模式3：领域专业术语
    domain_terms = (
        "验收", "标准", "规范", "检查", "检测", "试验", "安装", "拆除", "吊装",
        "架桥机", "龙门吊", "吊车", "箱梁", "T梁", "梁板", "钢丝绳", "支座",
        "地基", "安全装置", "操作证", "合格证", "静载", "动载", "空载",
    )
    for term in domain_terms:
        if term in text:
            keywords.append(term)

    # 模式4：术语+动作组合
    for match in re.findall(r"[一-鿿A-Za-z0-9.-]{0,12}(?:验收|标准|规范|检查|检测|试验|安装|拆除|吊装|要求)", text):
        if 2 <= len(match) <= 20:
            keywords.append(match)

    # 模式5：分词后含领域术语的片段
    normalized = re.sub(r"[\s,，。；;:：、/\\|()\[\]{}<>《》\"'""？?]+", " ", text)
    for token in normalized.split():
        token = token.strip()
        if len(token) < 2 or len(token) > 12:
            continue
        if any(term in token for term in domain_terms):
            keywords.append(token)

    seen = set()
    unique = []
    for keyword in keywords:
        keyword = keyword.strip()
        if keyword and keyword not in seen:
            seen.add(keyword)
            unique.append(keyword)
    return unique


def _pack_log_items(items: List[Dict[str, Any]], limit: int = 20, text_limit: int = 1500) -> List[Dict[str, Any]]:
    """打包候选条目为日志格式，限制条数和文本长度。"""
    packed = []
    for item in (items or [])[:limit]:
        if not isinstance(item, dict):
            continue
        metadata = item.get("metadata") if isinstance(item.get("metadata"), dict) else {}
        text = str(item.get("text") or item.get("text_content") or item.get("content") or "").strip()
        packed.append(
            {
                "candidate_key": item.get("candidate_key"),
                "source": item.get("source") or metadata.get("file_name") or "",
                "text": text[:text_limit],
                "vector_similarity": _to_float(item.get("vector_similarity", item.get("similarity")), 0.0),
                "fusion_score": _to_float(item.get("fusion_score"), 0.0),
                "rerank_score": _to_float(item.get("rerank_score"), 0.0) if "rerank_score" in item else None,
                "source_hits": item.get("source_hits") if isinstance(item.get("source_hits"), dict) else {},
                "metadata": {
                    key: metadata.get(key)
                    for key in (
                        "document_id", "parent_id", "file_name", "chapter_title",
                        "chapter_level_1", "chapter_level_2", "chapter_level_3",
                        "parent_count", "child_hit_count", "matched_child_texts",
                        "tag_match_terms", "source_scope_valid",
                    )
                    if metadata.get(key) not in (None, "")
                },
            }
        )
    return packed


def _to_int(value: Any, default: int) -> int:
    """安全整数转换。"""
    try:
        return int(value)
    except (TypeError, ValueError):
        return default


def _to_float(value: Any, default: float = 0.0) -> float:
    """安全浮点数转换。"""
    try:
        return float(value)
    except (TypeError, ValueError):
        return default


def _content_hash(text: str) -> str:
    """基于归一化文本的短 MD5 哈希，用于内容去重。"""
    normalized = re.sub(r"\s+", " ", text.strip().lower())
    return md5(normalized.encode("utf-8")).hexdigest()[:12]