From 409df5b0738d24facd3814b499e46a8f4f00ab50 Mon Sep 17 00:00:00 2001 From: Wenjie Zhang Date: Sun, 17 May 2026 18:06:39 +0800 Subject: [PATCH] =?UTF-8?q?feat(kb):=20=E6=96=B0=E5=A2=9E=E6=A3=80?= =?UTF-8?q?=E7=B4=A2=E9=85=8D=E7=BD=AE=E3=80=81=E5=9B=BE=E7=A7=8D=E5=AD=90?= =?UTF-8?q?=E6=9F=A5=E8=AF=A2=E3=80=81=E5=AE=9E=E4=BD=93/=E4=B8=89?= =?UTF-8?q?=E5=85=83=E7=BB=84=E6=90=9C=E7=B4=A2=E4=B8=8E=E8=9E=8D=E5=90=88?= =?UTF-8?q?=E6=A3=80=E7=B4=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 MilvusRetrievalConfig 数据类，支持向量/关键词/混合/图检索参数化配置 - 新增 query_seed_subgraph 和图检索融合逻辑 - MilvusGraphVectorStore 新增 search_entities/search_triples - MilvusGraphService.get_status 支持构建任务状态与进度查询 - cypher 语句增加 entity_id/triple_id 属性 - 前端 KnowledgeGraphSection 和 SearchConfigModal 适配新配置 - 新增 igraph 依赖 --- backend/package/pyproject.toml | 1 + .../yuxi/knowledge/graphs/graph_utils.py | 6 +- .../knowledge/graphs/milvus_graph_service.py | 105 +++- .../graphs/milvus_graph_vector_store.py | 86 +++ .../yuxi/knowledge/implementations/milvus.py | 519 ++++++++++++++---- .../knowledge_chunk_repository.py | 8 + backend/server/routers/knowledge_router.py | 2 +- .../knowledge/test_milvus_retrieval_config.py | 51 ++ backend/uv.lock | 31 ++ web/src/components/KnowledgeGraphSection.vue | 64 ++- web/src/components/SearchConfigModal.vue | 11 +- 11 files changed, 765 insertions(+), 119 deletions(-) create mode 100644 backend/test/unit/knowledge/test_milvus_retrieval_config.py diff --git a/backend/package/pyproject.toml b/backend/package/pyproject.toml index 135e383b..ebfe3dcb 100644 --- a/backend/package/pyproject.toml +++ b/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", diff --git a/backend/package/yuxi/knowledge/graphs/graph_utils.py b/backend/package/yuxi/knowledge/graphs/graph_utils.py index 9b9dd62a..51bfd08c 100644 --- a/backend/package/yuxi/knowledge/graphs/graph_utils.py +++ b/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 """ diff --git a/backend/package/yuxi/knowledge/graphs/milvus_graph_service.py b/backend/package/yuxi/knowledge/graphs/milvus_graph_service.py index 8a56547c..45dbc464 100644 --- a/backend/package/yuxi/knowledge/graphs/milvus_graph_service.py +++ b/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 diff --git a/backend/package/yuxi/knowledge/graphs/milvus_graph_vector_store.py b/backend/package/yuxi/knowledge/graphs/milvus_graph_vector_store.py index 967a6624..6230efe8 100644 --- a/backend/package/yuxi/knowledge/graphs/milvus_graph_vector_store.py +++ b/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 = [ diff --git a/backend/package/yuxi/knowledge/implementations/milvus.py b/backend/package/yuxi/knowledge/implementations/milvus.py index 339c602e..024b4a33 100644 --- a/backend/package/yuxi/knowledge/implementations/milvus.py +++ b/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): """清理连接""" diff --git a/backend/package/yuxi/repositories/knowledge_chunk_repository.py b/backend/package/yuxi/repositories/knowledge_chunk_repository.py index bbb13250..e0d12d1d 100644 --- a/backend/package/yuxi/repositories/knowledge_chunk_repository.py +++ b/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 [] diff --git a/backend/server/routers/knowledge_router.py b/backend/server/routers/knowledge_router.py index 3be8e40b..04c9a2f9 100644 --- a/backend/server/routers/knowledge_router.py +++ b/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: diff --git a/backend/test/unit/knowledge/test_milvus_retrieval_config.py b/backend/test/unit/knowledge/test_milvus_retrieval_config.py new file mode 100644 index 00000000..6cce0ae3 --- /dev/null +++ b/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"] diff --git a/backend/uv.lock b/backend/uv.lock index 3d97f9dc..1a93f9b1 100644 --- a/backend/uv.lock +++ b/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" }, diff --git a/web/src/components/KnowledgeGraphSection.vue b/web/src/components/KnowledgeGraphSection.vue index c3f75bbe..7a6c5abb 100644 --- a/web/src/components/KnowledgeGraphSection.vue +++ b/web/src/components/KnowledgeGraphSection.vue @@ -118,10 +118,18 @@
状态 - - {{ graphBuildStatus?.locked ? '已配置' : '未配置' }} - + 构建中 + 构建失败 + 已配置 + 未配置
+
{{ graphBuildStatus?.total_chunks ?? '-' }} @@ -145,6 +153,23 @@ > 配置抽取器 + + 构建中 {{ graphBuildStatus?.build_task_progress ?? 0 }}% + + + 重试索引 + { + 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() }) diff --git a/web/src/components/SearchConfigModal.vue b/web/src/components/SearchConfigModal.vue index 5bed1096..9d8f6859 100644 --- a/web/src/components/SearchConfigModal.vue +++ b/web/src/components/SearchConfigModal.vue @@ -26,7 +26,7 @@
- +