feat(kb): 新增检索配置、图种子查询、实体/三元组搜索与融合检索

- 新增 MilvusRetrievalConfig 数据类，支持向量/关键词/混合/图检索参数化配置
- 新增 query_seed_subgraph 和图检索融合逻辑
- MilvusGraphVectorStore 新增 search_entities/search_triples
- MilvusGraphService.get_status 支持构建任务状态与进度查询
- cypher 语句增加 entity_id/triple_id 属性
- 前端 KnowledgeGraphSection 和 SearchConfigModal 适配新配置
- 新增 igraph 依赖

backend/package/pyproject.toml

@@ -23,6 +23,7 @@ dependencies = [
     "docling>=2.68.0",
     "docx2txt>=0.9",
     "httpx>=0.27.0",
+    "igraph>=0.11.8",
     "json-repair>=0.54.0",
     "langchain>=1.2.0",
     "langchain-community>=0.4",

backend/package/yuxi/knowledge/graphs/graph_utils.py

@@ -116,7 +116,8 @@ def cypher_merge_entity_mention(db_label: str) -> str:
         normalized_name: $normalized_name,
         label: $entity_label
     }})
-    SET e.name = $name,
+    SET e.entity_id = $entity_id,
+        e.name = $name,
         e.attributes = $attributes
     MERGE (c)-[m:MENTIONS {{chunk_id: $chunk_id, file_id: $file_id, db_id: $db_id}}]->(e)
     """
@@ -142,7 +143,8 @@ def cypher_merge_relation(db_label: str) -> str:
         target_name: $target_name,
         type: $relation_type
     }}]->(target)
-    SET r.text = $text,
+    SET r.triple_id = $triple_id,
+        r.text = $text,
         r.file_id = $file_id,
         r.extractor_type = $extractor_type
     """

backend/package/yuxi/knowledge/graphs/milvus_graph_service.py

@@ -66,7 +66,7 @@ class MilvusGraphService:
     def driver(self):
         return self.connection.driver
 
-    async def get_status(self, db_id: str) -> dict[str, Any]:
+    async def get_status(self, db_id: str, *, tasker: Any = None) -> dict[str, Any]:
         kb = await self._get_milvus_kb(db_id)
         params = dict(kb.additional_params or {})
         config = params.get(GRAPH_CONFIG_KEY) or {}
@@ -75,6 +75,28 @@ class MilvusGraphService:
             self.chunk_repo.count_graph_pending_by_db_id(db_id),
             self.chunk_repo.count_graph_indexed_by_db_id(db_id),
         )
+
+        build_task_status = None
+        build_task_progress = 0
+        if tasker is not None:
+            active_task = await tasker.find_task_by_payload(
+                task_type=GRAPH_TASK_TYPE,
+                payload_match={"db_id": db_id},
+                statuses={"pending", "running"},
+            )
+            if active_task:
+                build_task_status = active_task.status
+                build_task_progress = round(active_task.progress)
+            else:
+                failed_task = await tasker.find_task_by_payload(
+                    task_type=GRAPH_TASK_TYPE,
+                    payload_match={"db_id": db_id},
+                    statuses={"failed", "cancelled"},
+                )
+                if failed_task:
+                    build_task_status = "failed"
+                    build_task_progress = 0
+
         return {
             "db_id": db_id,
             "kb_type": kb.kb_type,
@@ -84,6 +106,8 @@ class MilvusGraphService:
             "total_chunks": total_chunks,
             "pending_chunks": pending_chunks,
             "indexed_chunks": indexed_chunks,
+            "build_task_status": build_task_status,
+            "build_task_progress": build_task_progress,
         }
 
     async def configure(
@@ -228,11 +252,13 @@ class MilvusGraphService:
 
             # 2. MERGE Entity 节点 + Chunk→Entity (MENTIONS)
             for entity in entities:
+                entity_record = entity_record_by_local_id[entity["id"]]
                 tx.run(
                     merge_entity_cypher,
                     chunk_id=chunk.chunk_id,
                     file_id=chunk.file_id,
                     db_id=db_id,
+                    entity_id=entity_record["entity_id"],
                     normalized_name=normalize_entity_name(entity["text"]),
                     entity_label=entity.get("label") or "Entity",
                     name=entity["text"],
@@ -243,6 +269,17 @@ class MilvusGraphService:
             for relation in relations:
                 source = entity_by_id[relation["source"]]
                 target = entity_by_id[relation["target"]]
+                source_record = entity_record_by_local_id[relation["source"]]
+                target_record = entity_record_by_local_id[relation["target"]]
+                relation_type = relation.get("label") or "RELATED_TO"
+                triple_id = compute_triple_id(
+                    db_id,
+                    source_record["normalized_name"],
+                    source_record["label"],
+                    relation_type,
+                    target_record["normalized_name"],
+                    target_record["label"],
+                )
                 tx.run(
                     merge_relation_cypher,
                     db_id=db_id,
@@ -252,7 +289,8 @@ class MilvusGraphService:
                     source_label=source.get("label") or "Entity",
                     target_name=normalize_entity_name(target["text"]),
                     target_label=target.get("label") or "Entity",
-                    relation_type=relation.get("label") or "RELATED_TO",
+                    relation_type=relation_type,
+                    triple_id=triple_id,
                     text=relation["text"],
                     extractor_type=relation_extractor_type,
                 )
@@ -416,6 +454,43 @@ class MilvusGraphService:
             logger.error(f"Milvus graph query failed: {e}")
             return {"nodes": [], "edges": []}
 
+    async def query_seed_subgraph(
+        self,
+        db_id: str,
+        *,
+        entity_ids: list[str],
+        max_nodes: int,
+    ) -> dict[str, Any]:
+        if not entity_ids:
+            return {"nodes": [], "edges": []}
+        label = safe_neo4j_label(db_id)
+        cypher = f"""
+        MATCH (seed:Entity:MilvusKB:`{label}`)
+        WHERE seed.entity_id IN $entity_ids
+        MATCH p = (seed)-[*1..2]-(n:MilvusKB:`{label}`)
+        WITH p LIMIT $path_limit
+        WITH collect(p) AS paths
+        UNWIND paths AS node_path
+        UNWIND nodes(node_path) AS node
+        WITH paths, collect(DISTINCT node) AS graph_nodes
+        UNWIND paths AS rel_path
+        UNWIND relationships(rel_path) AS rel
+        RETURN graph_nodes AS nodes, collect(DISTINCT rel) AS edges
+        """
+        try:
+            with self.driver.session() as session:
+                record = session.run(
+                    cypher,
+                    entity_ids=list(dict.fromkeys(entity_ids)),
+                    path_limit=max(max_nodes, 1) * 4,
+                ).single()
+                if not record:
+                    return {"nodes": [], "edges": []}
+                return self._process_subgraph_record(record, max_nodes, db_id)
+        except Exception as e:
+            logger.error(f"Milvus seed subgraph query failed: {e}")
+            return {"nodes": [], "edges": []}
+
     async def get_labels(self, db_id: str | None = None) -> list[str]:
         effective_db_id = db_id or self.db_id
         if not effective_db_id:
@@ -558,6 +633,32 @@ class MilvusGraphService:
 
         return {"nodes": nodes[:limit], "edges": edges[: limit * 2]}
 
+    def _process_subgraph_record(self, record: Any, limit: int, db_id: str) -> dict[str, Any]:
+        nodes = []
+        edges = []
+        node_ids = set()
+        edge_ids = set()
+
+        for raw_node in record.get("nodes") or []:
+            node = self._normalize_node(raw_node, db_id)
+            if not node or node["id"] in node_ids:
+                continue
+            nodes.append(node)
+            node_ids.add(node["id"])
+            if len(nodes) >= limit:
+                break
+
+        for raw_edge in record.get("edges") or []:
+            edge = self._normalize_edge(raw_edge)
+            if not edge or edge["id"] in edge_ids:
+                continue
+            if edge["source_id"] not in node_ids or edge["target_id"] not in node_ids:
+                continue
+            edges.append(edge)
+            edge_ids.add(edge["id"])
+
+        return {"nodes": nodes, "edges": edges}
+
     def _normalize_node(self, raw_node: Any, db_id: str | None = None) -> dict[str, Any]:
         if hasattr(raw_node, "element_id"):
             node_id = raw_node.element_id

backend/package/yuxi/knowledge/graphs/milvus_graph_vector_store.py

@@ -92,6 +92,44 @@ class MilvusGraphVectorStore:
         if tasks:
             await asyncio.gather(*tasks)
 
+    async def search_entities(
+        self,
+        *,
+        db_id: str,
+        query_text: str,
+        embedding_model_spec: str,
+        top_k: int,
+    ) -> list[dict[str, Any]]:
+        collection_name = graph_entity_collection_name(db_id)
+        if not utility.has_collection(collection_name, using=self.connection_alias):
+            return []
+        return await self._search_graph_collection(
+            collection_name=collection_name,
+            query_text=query_text,
+            embedding_model_spec=embedding_model_spec,
+            top_k=top_k,
+            output_fields=["id", "content"],
+        )
+
+    async def search_triples(
+        self,
+        *,
+        db_id: str,
+        query_text: str,
+        embedding_model_spec: str,
+        top_k: int,
+    ) -> list[dict[str, Any]]:
+        collection_name = graph_triple_collection_name(db_id)
+        if not utility.has_collection(collection_name, using=self.connection_alias):
+            return []
+        return await self._search_graph_collection(
+            collection_name=collection_name,
+            query_text=query_text,
+            embedding_model_spec=embedding_model_spec,
+            top_k=top_k,
+            output_fields=["id", "content", "source_id", "target_id"],
+        )
+
     def drop_graph_collections(self, db_id: str) -> None:
         for collection_name in [graph_entity_collection_name(db_id), graph_triple_collection_name(db_id)]:
             try:
@@ -109,6 +147,54 @@ class MilvusGraphVectorStore:
         batch_size = int(getattr(model, "batch_size", 40) or 40)
         return partial(model.abatch_encode, batch_size=batch_size)
 
+    async def _search_graph_collection(
+        self,
+        *,
+        collection_name: str,
+        query_text: str,
+        embedding_model_spec: str,
+        top_k: int,
+        output_fields: list[str],
+    ) -> list[dict[str, Any]]:
+        if top_k <= 0:
+            return []
+        collection = Collection(name=collection_name, using=self.connection_alias)
+        collection.load()
+        embed = self._get_embedding_function(embedding_model_spec)
+        query_embedding = await embed([query_text])
+        return await asyncio.to_thread(
+            self._search_loaded_collection,
+            collection,
+            query_embedding,
+            max(top_k, 1),
+            output_fields,
+        )
+
+    def _search_loaded_collection(
+        self,
+        collection: Collection,
+        query_embedding: list,
+        top_k: int,
+        output_fields: list[str],
+    ) -> list[dict[str, Any]]:
+        results = collection.search(
+            data=query_embedding,
+            anns_field="embedding",
+            param={"metric_type": VECTOR_METRIC_TYPE, "params": {"nprobe": 10}},
+            limit=top_k,
+            output_fields=output_fields,
+        )
+        if not results or not results[0]:
+            return []
+
+        records = []
+        for hit in results[0]:
+            entity = hit.entity
+            record = {field: entity.get(field) for field in output_fields}
+            record["score"] = float(hit.distance or 0.0)
+            records.append(record)
+        return records
+
     def _get_or_create_entity_collection(self, db_id: str, embedding_info: Any) -> Collection:
         collection_name = graph_entity_collection_name(db_id)
         fields = [

backend/package/yuxi/knowledge/implementations/milvus.py

@@ -2,6 +2,7 @@ import asyncio
 import os
 import time
 import traceback
+from dataclasses import MISSING, dataclass, field, fields
 from functools import partial
 from typing import Any
 
@@ -34,6 +35,207 @@ CONTENT_ANALYZER_PARAMS = {"type": "chinese"}
 VECTOR_METRIC_TYPE = "COSINE"
 
 
+@dataclass(kw_only=True)
+class MilvusRetrievalConfig:
+    search_mode: str = field(
+        default="vector",
+        metadata={
+            "label": "检索模式",
+            "type": "select",
+            "options": [
+                {"value": "vector", "label": "向量检索", "description": "仅使用向量相似度检索"},
+                {"value": "keyword", "label": "BM25 全文检索", "description": "仅使用 Milvus BM25 检索"},
+                {"value": "hybrid", "label": "混合检索", "description": "Milvus 向量检索与 BM25 融合检索"},
+            ],
+            "description": "选择检索模式",
+        },
+    )
+    final_top_k: int = field(
+        default=10,
+        metadata={
+            "label": "最终返回 Chunk 数",
+            "type": "number",
+            "min": 1,
+            "max": 100,
+            "description": "重排序后返回给前端的文档数量",
+        },
+    )
+    similarity_threshold: float = field(
+        default=0.0,
+        metadata={
+            "label": "相似度阈值（0-1）",
+            "type": "number",
+            "min": 0.0,
+            "max": 1.0,
+            "step": 0.1,
+            "description": "过滤相似度低于此值的结果",
+        },
+    )
+    bm25_top_k: int = field(
+        default=50,
+        metadata={
+            "label": "BM25 召回数量",
+            "type": "number",
+            "min": 1,
+            "max": 200,
+            "description": "BM25 全文检索和混合检索中的 BM25 候选数量",
+        },
+    )
+    vector_weight: float = field(
+        default=0.7,
+        metadata={
+            "label": "向量检索权重",
+            "type": "number",
+            "min": 0.0,
+            "max": 1.0,
+            "step": 0.1,
+            "description": "混合检索中向量召回结果的融合权重",
+        },
+    )
+    bm25_weight: float = field(
+        default=0.3,
+        metadata={
+            "label": "BM25 权重",
+            "type": "number",
+            "min": 0.0,
+            "max": 1.0,
+            "step": 0.1,
+            "description": "混合检索中 BM25 召回结果的融合权重",
+        },
+    )
+    bm25_drop_ratio_search: float = field(
+        default=0.0,
+        metadata={
+            "label": "BM25 稀疏项丢弃比例",
+            "type": "number",
+            "min": 0.0,
+            "max": 1.0,
+            "step": 0.1,
+            "description": "BM25 检索时丢弃低分稀疏项的比例，数值越大检索越快但可能降低召回",
+        },
+    )
+    include_distances: bool = field(
+        default=True,
+        metadata={"label": "显示相似度", "type": "boolean", "description": "在结果中显示相似度分数"},
+    )
+    use_graph_retrieval: bool = field(
+        default=False,
+        metadata={"label": "启用图检索", "type": "boolean", "description": "是否启用实体和三元组扩散检索"},
+    )
+    graph_entity_top_k: int = field(
+        default=10,
+        metadata={
+            "label": "图实体召回数量",
+            "type": "number",
+            "min": 1,
+            "max": 100,
+            "depend_on": ("use_graph_retrieval", True),
+            "description": "通过 Query 召回的实体数量",
+        },
+    )
+    graph_triple_top_k: int = field(
+        default=10,
+        metadata={
+            "label": "图三元组召回数量",
+            "type": "number",
+            "min": 1,
+            "max": 100,
+            "depend_on": ("use_graph_retrieval", True),
+            "description": "通过 Query 召回的三元组数量",
+        },
+    )
+    graph_max_nodes: int = field(
+        default=10000,
+        metadata={
+            "label": "图检索最大节点数",
+            "type": "number",
+            "min": 100,
+            "max": 50000,
+            "depend_on": ("use_graph_retrieval", True),
+            "description": "2-hop 扩散子图最多读取的节点数量",
+        },
+    )
+    graph_top_k: int = field(
+        default=20,
+        metadata={
+            "label": "图召回 Chunk 数",
+            "type": "number",
+            "min": 1,
+            "max": 200,
+            "depend_on": ("use_graph_retrieval", True),
+            "description": "PPR 后从图谱路径召回的 Chunk 数量",
+        },
+    )
+    graph_weight: float = field(
+        default=1.0,
+        metadata={
+            "label": "图检索融合权重",
+            "type": "number",
+            "min": 0.0,
+            "max": 5.0,
+            "step": 0.1,
+            "depend_on": ("use_graph_retrieval", True),
+            "description": "排名融合时图检索结果的权重",
+        },
+    )
+    ppr_damping: float = field(
+        default=0.85,
+        metadata={
+            "label": "PPR 阻尼系数",
+            "type": "number",
+            "min": 0.1,
+            "max": 0.99,
+            "step": 0.01,
+            "depend_on": ("use_graph_retrieval", True),
+            "description": "Personalized PageRank 的阻尼系数",
+        },
+    )
+    use_reranker: bool = field(
+        default=False,
+        metadata={"label": "启用重排序", "type": "boolean", "description": "是否使用精排模型对检索结果进行重排序"},
+    )
+    reranker_model: str = field(
+        default="",
+        metadata={
+            "label": "重排序模型",
+            "type": "select",
+            "depend_on": ("use_reranker", True),
+            "description": "选择用于本次查询的重排序模型",
+            "options_provider": "rerank_models",
+        },
+    )
+    recall_top_k: int = field(
+        default=50,
+        metadata={
+            "label": "召回数量",
+            "type": "number",
+            "min": 10,
+            "max": 200,
+            "depend_on": ("use_reranker", True),
+            "description": "向量检索或混合检索保留的候选数量（启用重排序时有效）",
+        },
+    )
+
+
+def _retrieval_config_options() -> list[dict[str, Any]]:
+    options = []
+    for config_field in fields(MilvusRetrievalConfig):
+        metadata = dict(config_field.metadata)
+        options_provider = metadata.pop("options_provider", None)
+        default = None if config_field.default is MISSING else config_field.default
+        option = {
+            "key": config_field.name,
+            "default": default,
+            **metadata,
+        }
+        if options_provider == "rerank_models":
+            option["options"] = [
+                {"label": info.display_name, "value": info.spec} for info in model_cache.get_all_specs("rerank")
+            ]
+        options.append(option)
+    return options
+
+
 class MilvusKB(KnowledgeBase):
     """基于 Milvus 的生产级向量库"""
 
@@ -627,7 +829,8 @@ class MilvusKB(KnowledgeBase):
                 search_mode = "vector"
 
             use_reranker = bool(merged_kwargs.get("use_reranker", False))
-            if use_reranker:
+            use_graph_retrieval = bool(merged_kwargs.get("use_graph_retrieval", False))
+            if use_reranker or use_graph_retrieval:
                 recall_top_k = int(merged_kwargs.get("recall_top_k", 50))
                 recall_top_k = max(recall_top_k, final_top_k)
             else:
@@ -736,6 +939,12 @@ class MilvusKB(KnowledgeBase):
 
                 logger.debug(f"Milvus hybrid query response: {len(retrieved_chunks)} chunks found")
 
+            if use_graph_retrieval:
+                graph_chunks = await self._retrieve_graph_chunks(query_text, db_id, retrieved_chunks, merged_kwargs)
+                if graph_chunks:
+                    graph_weight = float(merged_kwargs.get("graph_weight", 1.0))
+                    retrieved_chunks = self._fuse_chunk_rankings(retrieved_chunks, graph_chunks, graph_weight)
+
             if not retrieved_chunks:
                 return []
 
@@ -780,6 +989,204 @@ class MilvusKB(KnowledgeBase):
             logger.error(f"Milvus query error: {e}, {traceback.format_exc()}")
             return []
 
+    async def _retrieve_graph_chunks(
+        self,
+        query_text: str,
+        db_id: str,
+        base_chunks: list[dict],
+        query_params: dict[str, Any],
+    ) -> list[dict]:
+        try:
+            from yuxi.knowledge.graphs.milvus_graph_service import MilvusGraphService
+            from yuxi.knowledge.graphs.milvus_graph_vector_store import MilvusGraphVectorStore
+
+            embedding_model_spec = self.databases_meta[db_id].get("embedding_model_spec")
+            if not embedding_model_spec:
+                return []
+
+            entity_top_k = max(int(query_params.get("graph_entity_top_k", 10)), 1)
+            triple_top_k = max(int(query_params.get("graph_triple_top_k", 10)), 1)
+            graph_top_k = max(int(query_params.get("graph_top_k", 20)), 1)
+            graph_max_nodes = max(int(query_params.get("graph_max_nodes", 10000)), 1)
+
+            vector_store = MilvusGraphVectorStore()
+            entity_hits, triple_hits = await asyncio.gather(
+                vector_store.search_entities(
+                    db_id=db_id,
+                    query_text=query_text,
+                    embedding_model_spec=embedding_model_spec,
+                    top_k=entity_top_k,
+                ),
+                vector_store.search_triples(
+                    db_id=db_id,
+                    query_text=query_text,
+                    embedding_model_spec=embedding_model_spec,
+                    top_k=triple_top_k,
+                ),
+            )
+            seed_weights = await self._build_graph_seed_weights(db_id, base_chunks, entity_hits, triple_hits)
+            if not seed_weights:
+                return []
+
+            graph_service = MilvusGraphService()
+            subgraph = await graph_service.query_seed_subgraph(
+                db_id,
+                entity_ids=list(seed_weights.keys()),
+                max_nodes=graph_max_nodes,
+            )
+            graph_scores = self._rank_graph_chunks_by_ppr(
+                subgraph,
+                seed_weights,
+                top_k=graph_top_k,
+                damping=float(query_params.get("ppr_damping", 0.85)),
+            )
+            if not graph_scores:
+                return []
+
+            chunks = await KnowledgeChunkRepository().list_by_chunk_ids([chunk_id for chunk_id, _ in graph_scores])
+            score_by_chunk_id = dict(graph_scores)
+            return [
+                self._build_chunk_from_record(chunk, score_by_chunk_id[chunk.chunk_id], score_field="graph_score")
+                for chunk in chunks
+            ]
+        except Exception as exc:  # noqa: BLE001
+            logger.error(f"Graph retrieval failed for {db_id}: {exc}")
+            return []
+
+    async def _build_graph_seed_weights(
+        self,
+        db_id: str,
+        base_chunks: list[dict],
+        entity_hits: list[dict[str, Any]],
+        triple_hits: list[dict[str, Any]],
+    ) -> dict[str, float]:
+        seed_weights: dict[str, float] = {}
+
+        def add_seed(entity_id: str | None, score: float, weight: float) -> None:
+            if not entity_id:
+                return
+            seed_weights[entity_id] = seed_weights.get(entity_id, 0.0) + max(float(score or 0.0), 0.0) * weight
+
+        for hit in entity_hits:
+            add_seed(hit.get("id"), hit.get("score", 0.0), 1.0)
+
+        for hit in triple_hits:
+            score = float(hit.get("score") or 0.0)
+            add_seed(hit.get("source_id"), score, 0.8)
+            add_seed(hit.get("target_id"), score, 0.8)
+
+        chunk_scores = {
+            chunk.get("metadata", {}).get("chunk_id"): float(chunk.get("score") or 0.0)
+            for chunk in base_chunks
+            if chunk.get("metadata", {}).get("chunk_id")
+        }
+        if chunk_scores:
+            chunks = await KnowledgeChunkRepository().list_by_chunk_ids(list(chunk_scores))
+            for chunk in chunks:
+                for entity_id in chunk.ent_ids or []:
+                    add_seed(entity_id, chunk_scores.get(chunk.chunk_id, 0.0), 0.3)
+
+        total = sum(seed_weights.values())
+        if total <= 0:
+            return {}
+        return {entity_id: weight / total for entity_id, weight in seed_weights.items()}
+
+    def _rank_graph_chunks_by_ppr(
+        self,
+        subgraph: dict[str, Any],
+        seed_weights: dict[str, float],
+        *,
+        top_k: int,
+        damping: float,
+    ) -> list[tuple[str, float]]:
+        nodes = subgraph.get("nodes") or []
+        edges = subgraph.get("edges") or []
+        if not nodes:
+            return []
+
+        try:
+            import igraph as ig
+        except ImportError:
+            logger.error("Graph retrieval requires python-igraph. Please install igraph.")
+            return []
+
+        node_ids = [node["id"] for node in nodes]
+        index_by_id = {node_id: index for index, node_id in enumerate(node_ids)}
+        edge_indices = [
+            (index_by_id[edge["source_id"]], index_by_id[edge["target_id"]])
+            for edge in edges
+            if edge.get("source_id") in index_by_id and edge.get("target_id") in index_by_id
+        ]
+        if not edge_indices:
+            return []
+
+        graph = ig.Graph(n=len(nodes), edges=edge_indices, directed=False)
+        reset = [0.0] * len(nodes)
+        chunk_node_indexes: list[tuple[int, str]] = []
+        for index, node in enumerate(nodes):
+            properties = node.get("properties") or {}
+            if node.get("type") == "Chunk" and properties.get("chunk_id"):
+                chunk_node_indexes.append((index, properties["chunk_id"]))
+                continue
+            entity_id = properties.get("entity_id")
+            if entity_id in seed_weights:
+                reset[index] = seed_weights[entity_id]
+
+        reset_total = sum(reset)
+        if reset_total <= 0 or not chunk_node_indexes:
+            return []
+        reset = [value / reset_total for value in reset]
+        scores = graph.personalized_pagerank(damping=min(max(damping, 0.1), 0.99), reset=reset)
+        ranked = sorted(
+            ((chunk_id, float(scores[index])) for index, chunk_id in chunk_node_indexes),
+            key=lambda item: item[1],
+            reverse=True,
+        )
+        return ranked[:top_k]
+
+    def _build_chunk_from_record(self, chunk: Any, score: float, score_field: str | None = None) -> dict:
+        metadata = {
+            "source": self._get_filename_for_file_id(chunk.file_id),
+            "chunk_id": chunk.chunk_id,
+            "file_id": chunk.file_id,
+            "chunk_index": chunk.chunk_index,
+        }
+        result = {"content": chunk.content, "metadata": metadata, "score": float(score or 0.0)}
+        if score_field:
+            result[score_field] = float(score or 0.0)
+        return result
+
+    def _fuse_chunk_rankings(
+        self,
+        base_chunks: list[dict],
+        graph_chunks: list[dict],
+        graph_weight: float,
+    ) -> list[dict]:
+        fused: dict[str, dict[str, Any]] = {}
+        rrf_k = 60.0
+
+        def merge_chunk(chunk: dict, rank: int, weight: float, source: str) -> None:
+            chunk_id = chunk.get("metadata", {}).get("chunk_id")
+            if not chunk_id:
+                return
+            score = weight / (rrf_k + rank)
+            existing = fused.get(chunk_id)
+            if existing is None:
+                existing = {**chunk, "fusion_score": 0.0, "fusion_sources": []}
+                fused[chunk_id] = existing
+            existing["fusion_score"] += score
+            existing["score"] = existing["fusion_score"]
+            existing["fusion_sources"].append(source)
+            if source == "graph" and "graph_score" in chunk:
+                existing["graph_score"] = chunk["graph_score"]
+
+        for rank, chunk in enumerate(base_chunks, start=1):
+            merge_chunk(chunk, rank, 1.0, "chunk")
+        for rank, chunk in enumerate(graph_chunks, start=1):
+            merge_chunk(chunk, rank, max(graph_weight, 0.0), "graph")
+
+        return sorted(fused.values(), key=lambda item: item.get("fusion_score", 0.0), reverse=True)
+
     async def delete_file_chunks_only(self, db_id: str, file_id: str) -> None:
         """仅删除文件的chunks数据，保留元数据（用于更新操作）"""
         graph_config = (self.databases_meta.get(db_id, {}).get("metadata") or {}).get("graph_build_config")
@@ -894,115 +1301,7 @@ class MilvusKB(KnowledgeBase):
 
     def get_query_params_config(self, db_id: str, **kwargs) -> dict:
         """获取 Milvus 知识库的查询参数配置"""
-        # 构建 Milvus 特定查询参数
-        # TODO 使用 dataclass 重新配置并解析
-        options = [
-            {
-                "key": "search_mode",
-                "label": "检索模式",
-                "type": "select",
-                "default": "vector",
-                "options": [
-                    {"value": "vector", "label": "向量检索", "description": "仅使用向量相似度检索"},
-                    {"value": "keyword", "label": "BM25 全文检索", "description": "仅使用 Milvus BM25 检索"},
-                    {"value": "hybrid", "label": "混合检索", "description": "Milvus 向量检索与 BM25 融合检索"},
-                ],
-                "description": "选择检索模式",
-            },
-            {
-                "key": "final_top_k",
-                "label": "最终返回 Chunk 数",
-                "type": "number",
-                "default": 10,
-                "min": 1,
-                "max": 100,
-                "description": "重排序后返回给前端的文档数量",
-            },
-            {
-                "key": "similarity_threshold",
-                "label": "相似度阈值（0-1）",
-                "type": "number",
-                "default": 0.0,
-                "min": 0.0,
-                "max": 1.0,
-                "step": 0.1,
-                "description": "过滤相似度低于此值的结果",
-            },
-            {
-                "key": "bm25_top_k",
-                "label": "BM25 召回数量",
-                "type": "number",
-                "default": 50,
-                "min": 1,
-                "max": 200,
-                "description": "BM25 全文检索和混合检索中的 BM25 候选数量",
-            },
-            {
-                "key": "vector_weight",
-                "label": "向量检索权重",
-                "type": "number",
-                "default": 0.7,
-                "min": 0.0,
-                "max": 1.0,
-                "step": 0.1,
-                "description": "混合检索中向量召回结果的融合权重",
-            },
-            {
-                "key": "bm25_weight",
-                "label": "BM25 权重",
-                "type": "number",
-                "default": 0.3,
-                "min": 0.0,
-                "max": 1.0,
-                "step": 0.1,
-                "description": "混合检索中 BM25 召回结果的融合权重",
-            },
-            {
-                "key": "bm25_drop_ratio_search",
-                "label": "BM25 稀疏项丢弃比例",
-                "type": "number",
-                "default": 0.0,
-                "min": 0.0,
-                "max": 1.0,
-                "step": 0.1,
-                "description": "BM25 检索时丢弃低分稀疏项的比例，数值越大检索越快但可能降低召回",
-            },
-            {
-                "key": "include_distances",
-                "label": "显示相似度",
-                "type": "boolean",
-                "default": True,
-                "description": "在结果中显示相似度分数",
-            },
-            {
-                "key": "use_reranker",
-                "label": "启用重排序",
-                "type": "boolean",
-                "default": False,
-                "description": "是否使用精排模型对检索结果进行重排序",
-            },
-            {
-                "key": "reranker_model",
-                "label": "重排序模型",
-                "type": "select",
-                "default": "",
-                "options": [
-                    {"label": info.display_name, "value": info.spec} for info in model_cache.get_all_specs("rerank")
-                ],
-                "description": "选择用于本次查询的重排序模型",
-            },
-            {
-                "key": "recall_top_k",
-                "label": "召回数量",
-                "type": "number",
-                "default": 50,
-                "min": 10,
-                "max": 200,
-                "description": "向量检索或混合检索保留的候选数量（启用重排序时有效）",
-            },
-        ]
-
-        return {"type": "milvus", "options": options}
+        return {"type": "milvus", "options": _retrieval_config_options()}
 
     def __del__(self):
         """清理连接"""

backend/package/yuxi/repositories/knowledge_chunk_repository.py

@@ -46,6 +46,14 @@ class KnowledgeChunkRepository:
             )
             return list(result.scalars().all())
 
+    async def list_by_chunk_ids(self, chunk_ids: list[str]) -> list[KnowledgeChunk]:
+        if not chunk_ids:
+            return []
+        async with pg_manager.get_async_session_context() as session:
+            result = await session.execute(select(KnowledgeChunk).where(KnowledgeChunk.chunk_id.in_(chunk_ids)))
+            chunks_by_id = {chunk.chunk_id: chunk for chunk in result.scalars().all()}
+            return [chunks_by_id[chunk_id] for chunk_id in chunk_ids if chunk_id in chunks_by_id]
+
     async def batch_upsert(self, chunks: list[dict[str, Any]]) -> list[KnowledgeChunk]:
         if not chunks:
             return []

backend/server/routers/knowledge_router.py

@@ -298,7 +298,7 @@ async def delete_database(db_id: str, current_user: User = Depends(get_admin_use
 @knowledge.get("/databases/{db_id}/graph-build/status")
 async def get_graph_build_status(db_id: str, current_user: User = Depends(get_admin_user)):
     try:
-        return await MilvusGraphService().get_status(db_id)
+        return await MilvusGraphService().get_status(db_id, tasker=tasker)
     except ValueError as e:
         raise HTTPException(status_code=400, detail=str(e))
     except Exception as e:

backend/test/unit/knowledge/test_milvus_retrieval_config.py

@@ -0,0 +1,51 @@
+from yuxi.knowledge.implementations.milvus import MilvusKB, _retrieval_config_options
+
+
+def test_milvus_retrieval_config_exposes_graph_and_dependencies():
+    options = _retrieval_config_options()
+    by_key = {option["key"]: option for option in options}
+
+    assert by_key["use_graph_retrieval"]["default"] is False
+    assert by_key["graph_max_nodes"]["default"] == 10000
+    assert by_key["graph_max_nodes"]["depend_on"] == ("use_graph_retrieval", True)
+    assert by_key["graph_top_k"]["depend_on"] == ("use_graph_retrieval", True)
+    assert by_key["reranker_model"]["depend_on"] == ("use_reranker", True)
+
+
+def test_graph_ppr_ranks_chunk_nodes_from_seed_entities():
+    kb = object.__new__(MilvusKB)
+    subgraph = {
+        "nodes": [
+            {"id": "e1", "type": "Entity", "properties": {"entity_id": "seed"}},
+            {"id": "c1", "type": "Chunk", "properties": {"chunk_id": "chunk_a"}},
+            {"id": "e2", "type": "Entity", "properties": {"entity_id": "other"}},
+            {"id": "c2", "type": "Chunk", "properties": {"chunk_id": "chunk_b"}},
+        ],
+        "edges": [
+            {"source_id": "e1", "target_id": "c1"},
+            {"source_id": "e1", "target_id": "e2"},
+            {"source_id": "e2", "target_id": "c2"},
+        ],
+    }
+
+    ranked = kb._rank_graph_chunks_by_ppr(subgraph, {"seed": 1.0}, top_k=2, damping=0.85)
+
+    assert [chunk_id for chunk_id, _ in ranked] == ["chunk_a", "chunk_b"]
+
+
+def test_rrf_fusion_merges_chunk_and_graph_rankings():
+    kb = object.__new__(MilvusKB)
+    base_chunks = [
+        {"content": "base a", "metadata": {"chunk_id": "a"}, "score": 0.9},
+        {"content": "base b", "metadata": {"chunk_id": "b"}, "score": 0.8},
+    ]
+    graph_chunks = [
+        {"content": "graph b", "metadata": {"chunk_id": "b"}, "score": 0.7, "graph_score": 0.7},
+        {"content": "graph c", "metadata": {"chunk_id": "c"}, "score": 0.6, "graph_score": 0.6},
+    ]
+
+    fused = kb._fuse_chunk_rankings(base_chunks, graph_chunks, graph_weight=1.0)
+
+    assert [chunk["metadata"]["chunk_id"] for chunk in fused] == ["b", "a", "c"]
+    assert fused[0]["graph_score"] == 0.7
+    assert fused[0]["fusion_sources"] == ["chunk", "graph"]

backend/uv.lock

@@ -1511,6 +1511,26 @@ wheels = [
     { url = "https://pypi.tuna.tsinghua.edu.cn/packages/0e/61/66938bbb5fc52dbdf84594873d5b51fb1f7c7794e9c0f5bd885f30bc507b/idna-3.11-py3-none-any.whl", hash = "sha256:771a87f49d9defaf64091e6e6fe9c18d4833f140bd19464795bc32d966ca37ea", size = 71008, upload-time = "2025-10-12T14:55:18.883Z" },
 ]
 
+[[package]]
+name = "igraph"
+version = "1.0.0"
+source = { registry = "https://pypi.tuna.tsinghua.edu.cn/simple" }
+dependencies = [
+    { name = "texttable" },
+]
+sdist = { url = "https://pypi.tuna.tsinghua.edu.cn/packages/23/be/56bef1919005b4caf1f71522b300d359f7faeb7ae93a3b0baa9b4f146a87/igraph-1.0.0.tar.gz", hash = "sha256:2414d0be2e4d77ee5357807d100974b40f6082bb1bb71988ec46cfb6728651ee", size = 5077105, upload-time = "2025-10-23T12:22:50.127Z" }
+wheels = [
+    { url = "https://pypi.tuna.tsinghua.edu.cn/packages/a5/03/3278ad0ceb3ea0e84d8ae3a85bdded4d0e57853aeb802a200feb43847b93/igraph-1.0.0-cp39-abi3-macosx_10_15_x86_64.whl", hash = "sha256:c2cbc415e02523e5a241eecee82319080bf928a70b1ba299f3b3e25bf029b6d4", size = 2257415, upload-time = "2025-10-23T12:22:27.246Z" },
+    { url = "https://pypi.tuna.tsinghua.edu.cn/packages/0d/bc/6281ec7f9baaf71ee57c3b1748da2d3148d15d253e1a03006f204aa68ca5/igraph-1.0.0-cp39-abi3-macosx_11_0_arm64.whl", hash = "sha256:1a27753cd80680a8f676c2d5a467aaa4a95e510b30748398ec4e4aeb982130e8", size = 2048555, upload-time = "2025-10-23T12:22:29.49Z" },
+    { url = "https://pypi.tuna.tsinghua.edu.cn/packages/2a/38/3cd6428a4ed4c09a56df05998438e7774fd1d799ee4fb8fc481674f5f7fc/igraph-1.0.0-cp39-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:a55dc3a2a4e3fc3eba42479910c1511bfc3ecb33cdf5f0406891fd85f14b5aee", size = 5314141, upload-time = "2025-10-23T12:22:31.023Z" },
+    { url = "https://pypi.tuna.tsinghua.edu.cn/packages/7d/da/dd2867c25adbb41563720f14b5fc895c98bf88be682a3faff4f7b3118d2a/igraph-1.0.0-cp39-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:2d04c2c76f686fb1f554ee35dfd3085f5e73b7965ba6b4cf06d53e66b1955522", size = 5683134, upload-time = "2025-10-23T12:22:32.423Z" },
+    { url = "https://pypi.tuna.tsinghua.edu.cn/packages/e5/40/243c118d34ab80382d7009c4dcb99b887384c3d2ce84d29eeac19e2a007a/igraph-1.0.0-cp39-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:f2b52dc1757fff0fed29a9f7a276d971a11db4211569ed78b9eab36288dfcc9d", size = 6211583, upload-time = "2025-10-23T12:22:34.238Z" },
+    { url = "https://pypi.tuna.tsinghua.edu.cn/packages/1d/b7/88f433819c54b496cb0315fce28e658970cb20ff5dbd52a5a605ce2888de/igraph-1.0.0-cp39-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:05c79a2a8fca695b2f217a6fa7f2549f896f757d4db41be32a055400cb19cc30", size = 6594509, upload-time = "2025-10-23T12:22:35.831Z" },
+    { url = "https://pypi.tuna.tsinghua.edu.cn/packages/7b/5d/8f7f6f619d374e959aa3664ebc4b24c10abc90c2e8efbed97f2623fadaf5/igraph-1.0.0-cp39-abi3-win32.whl", hash = "sha256:c2bce3cd472fec3dd9c4d8a3ea5b6b9be65fb30edf760beb4850760dd4f2d479", size = 2725406, upload-time = "2025-10-23T12:22:37.588Z" },
+    { url = "https://pypi.tuna.tsinghua.edu.cn/packages/af/77/a85b3745cf40a0572bae2de8cd9c2a2a8af78e5cf3e880fc0a249114e609/igraph-1.0.0-cp39-abi3-win_amd64.whl", hash = "sha256:faeff8ede0cf15eb4ded44b0fcea6e1886740146e60504c24ad2da14e0939563", size = 3221663, upload-time = "2025-10-23T12:22:39.404Z" },
+    { url = "https://pypi.tuna.tsinghua.edu.cn/packages/ef/7e/5df541c37bdf6493035e89c22bd53f30d99b291bcda6c78e9a8afeecec2b/igraph-1.0.0-cp39-abi3-win_arm64.whl", hash = "sha256:b607cafc24b10a615e713ee96e58208ef27e0764af80140c7cc45d4724a3f2df", size = 2785701, upload-time = "2025-10-23T12:22:41.03Z" },
+]
+
 [[package]]
 name = "importlib-metadata"
 version = "8.7.1"
@@ -4639,6 +4659,15 @@ wheels = [
     { url = "https://pypi.tuna.tsinghua.edu.cn/packages/d7/c1/eb8f9debc45d3b7918a32ab756658a0904732f75e555402972246b0b8e71/tenacity-9.1.4-py3-none-any.whl", hash = "sha256:6095a360c919085f28c6527de529e76a06ad89b23659fa881ae0649b867a9d55", size = 28926, upload-time = "2026-02-07T10:45:32.24Z" },
 ]
 
+[[package]]
+name = "texttable"
+version = "1.7.0"
+source = { registry = "https://pypi.tuna.tsinghua.edu.cn/simple" }
+sdist = { url = "https://pypi.tuna.tsinghua.edu.cn/packages/1c/dc/0aff23d6036a4d3bf4f1d8c8204c5c79c4437e25e0ae94ffe4bbb55ee3c2/texttable-1.7.0.tar.gz", hash = "sha256:2d2068fb55115807d3ac77a4ca68fa48803e84ebb0ee2340f858107a36522638" }
+wheels = [
+    { url = "https://pypi.tuna.tsinghua.edu.cn/packages/24/99/4772b8e00a136f3e01236de33b0efda31ee7077203ba5967fcc76da94d65/texttable-1.7.0-py2.py3-none-any.whl", hash = "sha256:72227d592c82b3d7f672731ae73e4d1f88cd8e2ef5b075a7a7f01a23a3743917" },
+]
+
 [[package]]
 name = "thinc"
 version = "8.3.13"
@@ -5545,6 +5574,7 @@ dependencies = [
     { name = "docling" },
     { name = "docx2txt" },
     { name = "httpx" },
+    { name = "igraph" },
     { name = "json-repair" },
     { name = "langchain" },
     { name = "langchain-community" },
@@ -5623,6 +5653,7 @@ requires-dist = [
     { name = "docling", specifier = ">=2.68.0" },
     { name = "docx2txt", specifier = ">=0.9" },
     { name = "httpx", specifier = ">=0.27.0" },
+    { name = "igraph", specifier = ">=0.11.8" },
     { name = "json-repair", specifier = ">=0.54.0" },
     { name = "langchain", specifier = ">=1.2.0" },
     { name = "langchain-community", specifier = ">=0.4" },

web/src/components/KnowledgeGraphSection.vue

@@ -118,10 +118,18 @@
             <div class="panel-body">
               <div class="status-row">
                 <span class="status-label">状态</span>
-                <a-tag :color="graphBuildStatus?.locked ? 'green' : 'orange'" size="small">
-                  {{ graphBuildStatus?.locked ? '已配置' : '未配置' }}
-                </a-tag>
+                <a-tag v-if="isBuildActive" color="blue" size="small">构建中</a-tag>
+                <a-tag v-else-if="isBuildFailed" color="red" size="small">构建失败</a-tag>
+                <a-tag v-else-if="graphBuildStatus?.locked" color="green" size="small">已配置</a-tag>
+                <a-tag v-else color="orange" size="small">未配置</a-tag>
               </div>
+              <a-progress
+                v-if="isBuildActive"
+                :percent="graphBuildStatus?.build_task_progress ?? 0"
+                :stroke-color="{ '0%': '#108ee9', '100%': '#87d068' }"
+                size="small"
+                style="margin-bottom: 10px"
+              />
               <div class="stats-grid">
                 <div class="stat-item">
                   <span class="stat-value">{{ graphBuildStatus?.total_chunks ?? '-' }}</span>
@@ -145,6 +153,23 @@
                 >
                   配置抽取器
                 </a-button>
+                <a-button
+                  v-else-if="isBuildActive"
+                  type="primary"
+                  block
+                  disabled
+                >
+                  构建中 {{ graphBuildStatus?.build_task_progress ?? 0 }}%
+                </a-button>
+                <a-button
+                  v-else-if="isBuildFailed"
+                  type="primary"
+                  block
+                  :disabled="!graphBuildStatus?.pending_chunks"
+                  @click="startGraphBuild"
+                >
+                  重试索引
+                </a-button>
                 <a-button
                   v-else
                   type="primary"
@@ -255,6 +280,38 @@ const searchInput = ref('')
 const graphBuildStatus = ref(null)
 const graphBuildLoading = ref(false)
 const showGraphConfig = ref(false)
+let buildStatusPollTimer = null
+
+const isBuildActive = computed(() => {
+  const s = graphBuildStatus.value?.build_task_status
+  return s === 'pending' || s === 'running'
+})
+
+const isBuildFailed = computed(() => {
+  return graphBuildStatus.value?.build_task_status === 'failed'
+})
+
+const stopBuildStatusPoll = () => {
+  if (buildStatusPollTimer) {
+    clearInterval(buildStatusPollTimer)
+    buildStatusPollTimer = null
+  }
+}
+
+const startBuildStatusPoll = () => {
+  stopBuildStatusPoll()
+  buildStatusPollTimer = setInterval(() => {
+    loadGraphBuildStatus()
+  }, 5000)
+}
+
+watch(isBuildActive, (active) => {
+  if (active) {
+    startBuildStatusPoll()
+  } else {
+    stopBuildStatusPoll()
+  }
+}, { immediate: true })
 const graphConfigForm = reactive({
   extractor_type: 'llm',
   model_spec: '',
@@ -477,6 +534,7 @@ onUnmounted(() => {
     clearTimeout(pendingLoadTimer)
     pendingLoadTimer = null
   }
+  stopBuildStatusPoll()
 })
 </script>
 

web/src/components/SearchConfigModal.vue

@@ -26,7 +26,7 @@
       <div v-else class="config-forms">
         <a-form layout="vertical">
           <a-row :gutter="16">
-            <a-col :span="12" v-for="param in queryParams" :key="param.key">
+            <a-col :span="12" v-for="param in visibleQueryParams" :key="param.key">
               <a-form-item :label="param.label">
                 <template #extra v-if="param.description">
                   <div class="param-description">{{ param.description }}</div>
@@ -96,6 +96,15 @@ const error = ref('')
 const queryParams = ref([])
 const meta = reactive({})
 
+const isDependencySatisfied = (param) => {
+  const dependency = param.depend_on
+  if (!dependency || dependency.length < 2) return true
+  const [key, expectedValue] = dependency
+  return meta[key] === expectedValue
+}
+
+const visibleQueryParams = computed(() => queryParams.value.filter(isDependencySatisfied))
+
 // 计算属性：处理布尔类型的双向绑定
 const computedMeta = computed(() => {
   const result = {}