From 75a6238937b09411aa3718b0ad75df9e7327797d Mon Sep 17 00:00:00 2001 From: zc <2649023789@qq.com> Date: Thu, 2 Jul 2026 12:43:11 +0800 Subject: [PATCH] feat: use Neo4j graph retrieval directly --- tools/fault_statistics.py | 78 ++--- tools/graph_tools.py | 708 ++++++++++++++++++++++++++++---------- 2 files changed, 544 insertions(+), 242 deletions(-) diff --git a/tools/fault_statistics.py b/tools/fault_statistics.py index 5b74be3..5e9c7bf 100644 --- a/tools/fault_statistics.py +++ b/tools/fault_statistics.py @@ -49,64 +49,39 @@ async def get_device_system_from_neo4j(device_name: str) -> str: if device_name_clean in neo4j_cache: return neo4j_cache[device_name_clean] - from config import DEVICE_SYSTEM_CONFIG - - endpoint = DEVICE_SYSTEM_CONFIG.get("endpoint") - timeout = DEVICE_SYSTEM_CONFIG.get("timeout", 30.0) - min_hops = DEVICE_SYSTEM_CONFIG.get("default_min_hops", 2) - max_hops = DEVICE_SYSTEM_CONFIG.get("default_max_hops", 8) - top_k = DEVICE_SYSTEM_CONFIG.get("default_top_k", 10) - - if not endpoint: - result = "其他" - neo4j_cache[device_name_clean] = result - return result - try: - payload = { - "device_name": device_name.strip(), - "min_hops": min_hops, - "max_hops": max_hops, - "top_k": top_k - } + from tools.graph_tools import find_system_by_device - async with httpx.AsyncClient(timeout=timeout) as client: - response = await client.post( - endpoint, - json=payload, - headers={"Content-Type": "application/json"} - ) - - if response.status_code != 200: - result = "其他" - neo4j_cache[device_name_clean] = result - return result - - result_json = response.json() - - if not result_json.get("success", False): - result = "其他" - neo4j_cache[device_name_clean] = result - return result - - systems = result_json.get("systems", []) - if not systems or not isinstance(systems, list): - result = "其他" - neo4j_cache[device_name_clean] = result - return result - - first_system = systems[0] - if isinstance(first_system, dict): - system_name = first_system.get("name") or first_system.get("系统名称") or first_system.get("名称") - if system_name: - result = system_name.strip() - neo4j_cache[device_name_clean] = result - return result + result_json = await find_system_by_device( + device_name=device_name.strip(), + min_hops=2, + max_hops=8, + top_k=10, + ) + if not result_json.get("success", False): result = "其他" neo4j_cache[device_name_clean] = result return result + systems = result_json.get("systems", []) + if not systems or not isinstance(systems, list): + result = "其他" + neo4j_cache[device_name_clean] = result + return result + + first_system = systems[0] + if isinstance(first_system, dict): + system_name = first_system.get("name") or first_system.get("系统名称") or first_system.get("名称") + if system_name: + result = system_name.strip() + neo4j_cache[device_name_clean] = result + return result + + result = "其他" + neo4j_cache[device_name_clean] = result + return result + except Exception as e: print(f"查询设备系统失败: {str(e)}") result = "其他" @@ -966,4 +941,3 @@ async def repair_feedback_statistics_tool( ) return result - diff --git a/tools/graph_tools.py b/tools/graph_tools.py index 8d17a1a..f0ed5c8 100644 --- a/tools/graph_tools.py +++ b/tools/graph_tools.py @@ -279,19 +279,17 @@ async def _match_scoped_nodes_with_graph_indexes( vector_index: str, candidate_ids: Optional[List[Any]] = None, fallback_candidates: Optional[List[Dict[str, Any]]] = None, - fulltext_top_k: int = 50, - vector_top_k: int = 50, + fulltext_top_k: int = 500, + vector_top_k: int = 500, text_weight: float = 0.20, fulltext_weight: float = 0.35, vector_weight: float = 0.45, min_text_score: float = 0.34, min_combined_score: float = 0.42, - fallback_text_weight: float = 0.3, - fallback_semantic_weight: float = 0.7, ) -> tuple: """ 使用 Neo4j 全文索引和向量索引在候选范围内匹配节点。 - 索引无结果或得分不足时，退回到原有 OpenaiAPI.hybrid_match_with_embeddings。 + 仅使用索引结果做融合匹配；fallback_candidates 只用于补齐范围查询中的附加字段。 """ query_text = str(query_text or "").strip() if not query_text: @@ -423,10 +421,22 @@ async def _match_scoped_nodes_with_graph_indexes( reverse=True, ) best = scored[0] + second = scored[1] if len(scored) > 1 else None + has_index_signal = best["fulltext_norm"] > 0 or best["vector_score"] > 0 + is_ambiguous = ( + second is not None + and best["text_score"] < 0.82 + and best["combined_score"] < 0.60 + and (best["combined_score"] - second["combined_score"]) < 0.03 + ) if ( best["text_score"] >= 0.82 - or best["combined_score"] >= min_combined_score - or best["text_score"] >= min_text_score + or ( + has_index_signal + and not is_ambiguous + and best["combined_score"] >= min_combined_score + and best["text_score"] >= min_text_score + ) ): return best["candidate"], best["combined_score"], { "match_source": "neo4j_index", @@ -435,17 +445,6 @@ async def _match_scoped_nodes_with_graph_indexes( "vector_score": best["vector_score"], } - if fallback_candidates: - node, score = await OpenaiAPI.hybrid_match_with_embeddings( - query_text=query_text, - candidates=fallback_candidates, - name_key="display_name", - embedding_key="embedding", - text_weight=fallback_text_weight, - semantic_weight=fallback_semantic_weight, - ) - return node, score, {"match_source": "fallback"} - best_score = max([c.get("combined_score", 0.0) for c in indexed_candidates.values()] or [0.0]) return None, best_score, {"match_source": "none"} @@ -519,13 +518,13 @@ async def normalize_device_name_by_ship_graph( device_name: str, fulltext_index: str = "设备_fulltext", vector_index: str = "设备_vector", - fulltext_top_k: int = 50, - vector_top_k: int = 50, + fulltext_top_k: int = 500, + vector_top_k: int = 500, min_text_score: float = 0.34, min_combined_score: float = 0.42, ) -> Dict[str, Any]: """ - 标准化设备名称 + 正在标准化设备名称。。。 标准化失败时返回 success=False，由调用方继续使用原始设备名。 """ original_device_name = str(device_name or "").strip() @@ -800,6 +799,76 @@ def _to_graph_display_node(node_dict: Dict[str, Any]) -> Dict[str, Any]: return {"id": nid, "name": name, "label": label} +def _filter_props_for_graph(props: Dict[str, Any]) -> Dict[str, Any]: + exclude_keys = { + "last_updated", + "created_at", + "knowledge_source", + "fulltext", + "embedding", + } + return {k: v for k, v in (props or {}).items() if k not in exclude_keys} + + +def _record_to_graph_node(data: Dict[str, Any]) -> Dict[str, Any]: + props = data.get("props") or {} + name = data.get("name") or props.get("名称") or props.get("name") or str(data.get("id")) + return { + "id": data.get("id"), + "name": name, + "labels": data.get("labels") or [], + "properties": {"名称": name} if name else _filter_props_for_graph(props), + } + + +async def _get_unique_or_numbered_ship(session, ship_number: str = "") -> Optional[Dict[str, Any]]: + if ship_number and str(ship_number).strip(): + ship_key = str(ship_number).strip() + ship_query = """ + MATCH (s:舰艇) + WHERE s.舷号 = $ship_key + OR s.舷号 CONTAINS $ship_key + OR $ship_key CONTAINS s.舷号 + RETURN id(s) AS id, + elementId(s) AS eid, + labels(s) AS labels, + properties(s) AS props, + coalesce(s.名称, s.name, s.舷号) AS display_name + LIMIT 1 + """ + records = await session.run(ship_query, ship_key=ship_key) + record = await records.single() + return _node_record_to_dict(record) if record else None + + ship_query = """ + MATCH (s:舰艇) + RETURN id(s) AS id, + elementId(s) AS eid, + labels(s) AS labels, + properties(s) AS props, + coalesce(s.名称, s.name, s.舷号) AS display_name + LIMIT 2 + """ + records = await session.run(ship_query) + ships = [_node_record_to_dict(r) async for r in records] + return ships[0] if len(ships) == 1 else None + + +async def _fetch_node_graph_records(session, node_ids: List[Any]) -> List[Dict[str, Any]]: + if not node_ids: + return [] + query = """ + MATCH (n) + WHERE id(n) IN $node_ids + RETURN elementId(n) AS id, + labels(n) AS labels, + coalesce(n.名称, n.name) AS name, + properties(n) AS props + """ + records = await session.run(query, node_ids=node_ids) + return [_record_to_graph_node(r.data()) async for r in records] + + def _build_fault_graph_display( ship_node: Optional[Dict[str, Any]], device_node: Optional[Dict[str, Any]], @@ -884,6 +953,187 @@ def _build_fault_graph_results_text( return "\n".join(parts) +@track_function_calls +async def find_system_by_device( + device_name: str, + min_hops: int = 2, + max_hops: int = 8, + top_k: int = 10, +) -> Dict[str, Any]: + """根据设备名称反向检索所属系统节点。""" + if not device_name or not str(device_name).strip(): + return {"success": False, "error": "设备名称不能为空"} + + uri = NEO4J_CONFIG.get("uri") + user = NEO4J_CONFIG.get("user") + password = NEO4J_CONFIG.get("password") + if not uri or not user or not password: + return {"success": False, "error": "Neo4j 配置不完整，请检查 NEO4J_CONFIG"} + + try: + async with AsyncGraphDatabase.driver(uri, auth=(user, password)) as driver: + async with driver.session() as session: + ship_node = await _get_unique_or_numbered_ship(session, "") + device_candidate_ids = None + if ship_node: + scope_query = """ + MATCH (ship:舰艇) + WHERE id(ship) = $ship_id + MATCH (ship)-[:包含*1..6]->(n:设备) + WITH DISTINCT n, coalesce(n.名称, n.name, n.设备名称) AS nm + WHERE nm IS NOT NULL + RETURN collect(id(n)) AS device_ids + """ + scope_records = await session.run(scope_query, ship_id=ship_node.get("id")) + scope_row = await scope_records.single() + device_candidate_ids = scope_row.get("device_ids") if scope_row else None + if not device_candidate_ids: + return {"success": False, "error": "唯一舰艇节点下未找到设备候选节点"} + + device_node, device_score, _ = await _match_scoped_nodes_with_graph_indexes( + session=session, + query_text=str(device_name).strip(), + fulltext_index="设备_fulltext", + vector_index="设备_vector", + candidate_ids=device_candidate_ids, + fallback_candidates=None, + fulltext_top_k=max(top_k, 500), + vector_top_k=max(top_k, 500), + min_text_score=0.34, + min_combined_score=0.42, + ) + if not device_node: + return {"success": False, "error": f"未找到匹配的设备节点（最高得分: {device_score:.3f}）"} + + device_id = device_node.get("id") + effective_max_hops = max(1, int(max_hops)) + nearest_hops_query = f""" + MATCH (d:设备) + WHERE id(d) = $device_id + MATCH path = (s:系统)-[:包含*1..{effective_max_hops}]->(d) + RETURN min(length(path)) AS nearest_hops + """ + nearest_records = await session.run(nearest_hops_query, device_id=device_id) + nearest_row = await nearest_records.single() + nearest_hops = nearest_row.get("nearest_hops") if nearest_row else None + + if nearest_hops is None: + return { + "success": False, + "error": "未找到设备对应的系统节点", + "device": { + "id": device_id, + "name": _candidate_display_name(device_node), + "labels": device_node.get("labels", []), + "properties": _filter_props_for_graph(device_node.get("props", {})), + }, + } + + system_query = f""" + MATCH (d:设备) + WHERE id(d) = $device_id + MATCH path = (s:系统)-[:包含*1..{effective_max_hops}]->(d) + WHERE length(path) = $nearest_hops + RETURN DISTINCT id(s) AS id, + elementId(s) AS eid, + labels(s) AS labels, + properties(s) AS props, + coalesce(s.名称, s.name) AS display_name, + length(path) AS hops + ORDER BY display_name + """ + system_records = await session.run( + system_query, + device_id=device_id, + nearest_hops=nearest_hops, + ) + systems_raw = [_node_record_to_dict(r) async for r in system_records] + + if not systems_raw: + return { + "success": False, + "error": "未找到设备对应的系统节点", + "device": { + "id": device_id, + "name": _candidate_display_name(device_node), + "labels": device_node.get("labels", []), + "properties": _filter_props_for_graph(device_node.get("props", {})), + }, + } + + systems = [] + for item in systems_raw: + systems.append({ + "id": item.get("id"), + "name": _candidate_display_name(item), + "labels": item.get("labels", []), + "properties": _filter_props_for_graph(item.get("props", {})), + }) + + path_query = f""" + MATCH (d:设备) + WHERE id(d) = $device_id + MATCH path = (s:系统)-[:包含*1..{effective_max_hops}]->(d) + WHERE length(path) = $nearest_hops + WITH path LIMIT 5 + WITH nodes(path) AS ns, relationships(path) AS rs + RETURN ns, rs + """ + path_records = await session.run( + path_query, + device_id=device_id, + nearest_hops=nearest_hops, + ) + nodes_g: List[Dict[str, Any]] = [] + links_g: List[Dict[str, Any]] = [] + seen_nodes = set() + seen_links = set() + async for record in path_records: + for node in record.get("ns") or []: + nid = node.element_id if hasattr(node, "element_id") else node.id + if nid in seen_nodes: + continue + seen_nodes.add(nid) + props = dict(node) if hasattr(node, "__iter__") else {} + nodes_g.append({ + "id": nid, + "name": props.get("名称") or props.get("name") or str(nid), + "labels": list(node.labels) if hasattr(node, "labels") else [], + "properties": _filter_props_for_graph(props), + }) + for rel in record.get("rs") or []: + rid = rel.element_id if hasattr(rel, "element_id") else rel.id + if rid in seen_links: + continue + seen_links.add(rid) + links_g.append({ + "id": rid, + "label": rel.type if hasattr(rel, "type") else str(rel.type), + "source": rel.start_node.element_id if hasattr(rel.start_node, "element_id") else rel.start_node.id, + "target": rel.end_node.element_id if hasattr(rel.end_node, "element_id") else rel.end_node.id, + }) + + return { + "success": True, + "device": { + "id": device_id, + "name": _candidate_display_name(device_node), + "labels": device_node.get("labels", []), + "properties": _filter_props_for_graph(device_node.get("props", {})), + "match_score": device_score, + }, + "systems": systems, + "graph": {"nodes": nodes_g, "links": links_g}, + "meta": { + "system_count": len(systems), + "min_hops": min_hops, + "max_hops": max_hops, + }, + } + except Exception as e: + return {"success": False, "error": f"检索失败: {str(e)}"} + + @track_function_calls async def fault_graph_reason_and_projects( ship_number: Optional[str] = None, @@ -935,7 +1185,7 @@ async def fault_graph_reason_and_projects( device_scope_query = """ MATCH (ship:舰艇) WHERE id(ship) = $ship_id - MATCH (ship)-[:包含*0..8]->(n:设备) + MATCH (ship)-[:包含*0..6]->(n:设备) WITH n, coalesce(n.名称, n.name, n.设备名称) AS nm WHERE nm IS NOT NULL RETURN collect(DISTINCT id(n)) AS device_ids @@ -956,61 +1206,11 @@ async def fault_graph_reason_and_projects( vector_index="设备_vector", candidate_ids=device_candidate_ids if ship_id is not None else None, fallback_candidates=None, - fulltext_top_k=80, - vector_top_k=80, + fulltext_top_k=500, + vector_top_k=500, min_text_score=0.34, min_combined_score=0.42, - fallback_text_weight=0.3, - fallback_semantic_weight=0.7, ) - if not device_node: - if ship_id is not None: - device_fallback_query = """ - MATCH (n:设备) - WHERE id(n) IN $device_ids - WITH n, coalesce(n.名称, n.name, n.设备名称) AS nm - WHERE nm IS NOT NULL - RETURN DISTINCT - id(n) AS id, - labels(n) AS labels, - properties(n) AS props, - nm AS display_name, - n.embedding AS embedding - LIMIT 1000 - """ - dev_records = await session.run( - device_fallback_query, - device_ids=device_candidate_ids, - ) - else: - device_fallback_query = """ - MATCH (n:设备) - WITH n, coalesce(n.名称, n.name, n.设备名称) AS nm - WHERE nm IS NOT NULL - RETURN DISTINCT - id(n) AS id, - labels(n) AS labels, - properties(n) AS props, - nm AS display_name, - n.embedding AS embedding - LIMIT 2000 - """ - dev_records = await session.run(device_fallback_query) - device_candidates = [_node_record_to_dict(r) async for r in dev_records] - device_node, device_score, device_match_meta = await _match_scoped_nodes_with_graph_indexes( - session=session, - query_text=device_key, - fulltext_index="设备_fulltext", - vector_index="设备_vector", - candidate_ids=device_candidate_ids if ship_id is not None else None, - fallback_candidates=device_candidates, - fulltext_top_k=80, - vector_top_k=80, - min_text_score=0.34, - min_combined_score=0.42, - fallback_text_weight=0.3, - fallback_semantic_weight=0.7, - ) if not device_node: return {"success": False, "error": f"未找到匹配的设备节点（最高得分: {device_score:.3f}）"} device_id = device_node["id"] @@ -1043,12 +1243,10 @@ async def fault_graph_reason_and_projects( vector_index="故障模式_vector", candidate_ids=fault_mode_ids, fallback_candidates=None, - fulltext_top_k=80, - vector_top_k=80, + fulltext_top_k=500, + vector_top_k=500, min_text_score=0.30, min_combined_score=0.40, - fallback_text_weight=0.3, - fallback_semantic_weight=0.7, ) if fault_mode_node: scope_by_id = { @@ -1062,37 +1260,6 @@ async def fault_graph_reason_and_projects( fault_mode_node["real_device_id"] = scope_item["real_device_id"] if "real_device_name" in scope_item and "real_device_name" not in fault_mode_node: fault_mode_node["real_device_name"] = scope_item["real_device_name"] - if not fault_mode_node: - fault_mode_fallback_query = """ - MATCH (start_e:设备) - WHERE id(start_e) = $device_id - MATCH (start_e)-[:包含*0..3]-(real_e:设备) - MATCH (real_e)-[:维修]-(w:维修工作)-[:表征]-(fm:故障模式) - RETURN DISTINCT - id(fm) AS id, - labels(fm) AS labels, - properties(fm) AS props, - fm.名称 AS display_name, - fm.embedding AS embedding, - id(real_e) AS real_device_id, - real_e.名称 AS real_device_name - """ - fm_records = await session.run(fault_mode_fallback_query, device_id=device_id) - fault_modes = [_node_record_to_dict(r) async for r in fm_records] - fault_mode_node, fault_score, fault_mode_match_meta = await _match_scoped_nodes_with_graph_indexes( - session=session, - query_text=fault_symptom, - fulltext_index="故障模式_fulltext", - vector_index="故障模式_vector", - candidate_ids=fault_mode_ids, - fallback_candidates=fault_modes, - fulltext_top_k=80, - vector_top_k=80, - min_text_score=0.30, - min_combined_score=0.40, - fallback_text_weight=0.3, - fallback_semantic_weight=0.7, - ) if not fault_mode_node: return {"success": False, "error": f"未能在故障模式节点中找到与故障现象足够匹配的节点（最高得分: {fault_score:.3f}）"} @@ -1158,7 +1325,7 @@ async def fault_graph_reason_and_projects( WHERE id(ship) = $ship_id MATCH (real_e:设备) WHERE id(real_e) = $real_device_id - MATCH path0 = (ship)-[:包含*0..8]->(real_e) + MATCH path0 = (ship)-[:包含*0..6]->(real_e) MATCH path1 = (real_e)-[:维修]-(w:维修工作)-[:表征]-(fm:故障模式) WHERE id(fm) = $fault_mode_id OPTIONAL MATCH (fm)<-[:修理]-(rp:维修项目) @@ -1234,7 +1401,7 @@ async def fault_graph_reason_and_projects( WHERE id(ship) = $ship_id MATCH (real_e:设备) WHERE id(real_e) = $real_device_id - MATCH path0 = (ship)-[:包含*0..8]->(real_e) + MATCH path0 = (ship)-[:包含*0..6]->(real_e) MATCH path1 = (real_e)-[:维修]-(w:维修工作)-[:表征]-(fm:故障模式) WHERE id(fm) = $fault_mode_id WITH relationships(path0) + relationships(path1) AS rs @@ -1467,59 +1634,160 @@ async def operation_graph_search_tool( if not operation_item or not str(operation_item).strip(): return {"success": False, "error": "操作项目不能为空"} - endpoint = GRAPH_SEARCH_CONFIG.get("operation_search_endpoint") - timeout = GRAPH_SEARCH_CONFIG.get("graph_timeout") - - if not endpoint: - return {"success": False, "error": "操作图谱检索服务配置不完整"} - - payload = { - "ship_number": str(ship_number).strip(), - "device_name": str(device_name).strip(), - "operation_item": str(operation_item).strip(), - } + uri = NEO4J_CONFIG.get("uri") + user = NEO4J_CONFIG.get("user") + password = NEO4J_CONFIG.get("password") + if not uri or not user or not password: + return {"success": False, "error": "Neo4j 配置不完整，请检查 NEO4J_CONFIG"} try: - async with httpx.AsyncClient(timeout=timeout) as client: - response = await client.post( - endpoint, - json=payload, - headers={"Content-Type": "application/json"}, - ) - response.raise_for_status() - result = response.json() + async with AsyncGraphDatabase.driver(uri, auth=(user, password)) as driver: + async with driver.session() as session: + # 操作图谱不再依赖舷号；按当前图谱唯一舰艇节点限定范围。 + ship_node = await _get_unique_or_numbered_ship(session, "") + if not ship_node: + return {"success": False, "error": "未找到唯一舰艇节点"} + ship_id = ship_node.get("id") - if not result.get("success", False): - return {"success": False, "error": result.get("error", "检索失败")} + scope_query = """ + MATCH (ship:舰艇) + WHERE id(ship) = $ship_id + MATCH (ship)-[:包含*0..6]->(d:设备) + WITH DISTINCT d, coalesce(d.名称, d.name, d.设备名称) AS nm + WHERE nm IS NOT NULL + RETURN collect(id(d)) AS device_ids + """ + scope_records = await session.run(scope_query, ship_id=ship_id) + scope_row = await scope_records.single() + device_ids = scope_row.get("device_ids") if scope_row else [] + if not device_ids: + return {"success": False, "error": "该舰艇范围内未找到设备节点"} - graph_list = result.get("xxxx.graph") or [] - results_parts = [] - if graph_list: - for i, graph_item in enumerate(graph_list): - if isinstance(graph_item, dict): - _filter_searchable_labels(graph_item.get("nodes", [])) - if "results" in graph_item: - r = graph_item.pop("results") - if r: - results_parts.append(r) - graph_list[i] = apply_styles(graph_item) - result["xxxx.graph"] = graph_list + device_node, device_score, _ = await _match_scoped_nodes_with_graph_indexes( + session=session, + query_text=str(device_name).strip(), + fulltext_index="设备_fulltext", + vector_index="设备_vector", + candidate_ids=device_ids, + fallback_candidates=None, + fulltext_top_k=500, + vector_top_k=500, + min_text_score=0.34, + min_combined_score=0.42, + ) + if not device_node: + return {"success": False, "error": f"在该舰艇范围内未找到匹配的设备节点（最高得分: {device_score:.3f}）"} - if results_parts: - result["results"] = "\n".join(results_parts) - elif "results" not in result: - result["results"] = "" + device_id = device_node.get("id") + op_scope_query = """ + MATCH (start_d:设备) + WHERE id(start_d) = $device_id + MATCH (start_d)-[:包含*0..6]-(real_d:设备) + MATCH (real_d)-[:支持操作]-(op_use:操作使用) + -[:使用]-(op_proc:操作程序) + -[:操作时使用]-(op_item:操作项目) + RETURN DISTINCT + id(op_item) AS id, + elementId(op_item) AS eid, + labels(op_item) AS labels, + properties(op_item) AS props, + coalesce(op_item.名称, op_item.name) AS display_name, + id(real_d) AS real_device_id + """ + op_records = await session.run(op_scope_query, device_id=device_id) + op_scope = [_node_record_to_dict(r) async for r in op_records] + op_ids = [item.get("id") for item in op_scope if item.get("id") is not None] + if not op_ids: + return {"success": False, "error": "该设备下未找到操作项目节点"} - return result + op_node, op_score, _ = await _match_scoped_nodes_with_graph_indexes( + session=session, + query_text=str(operation_item).strip(), + fulltext_index="操作项目_fulltext", + vector_index="操作项目_vector", + candidate_ids=op_ids, + fallback_candidates=op_scope, + fulltext_top_k=500, + vector_top_k=500, + min_text_score=0.30, + min_combined_score=0.40, + ) + if not op_node: + return {"success": False, "error": f"未找到匹配的操作项目节点（最高得分: {op_score:.3f}）"} - except httpx.ReadTimeout: - return {"success": False, "error": "操作图谱检索服务超时"} - except httpx.ConnectError: - return {"success": False, "error": "无法连接操作图谱检索服务"} - except httpx.HTTPStatusError as e: - return {"success": False, "error": f"操作图谱检索服务返回错误: {e.response.status_code}"} + op_item_id = op_node.get("id") + real_device_id = op_node.get("real_device_id") or device_id + + graph_nodes_query = """ + MATCH (ship:舰艇) + WHERE id(ship) = $ship_id + MATCH (real_d:设备) + WHERE id(real_d) = $real_device_id + MATCH (op_item:操作项目) + WHERE id(op_item) = $op_item_id + MATCH path1 = (ship)-[:包含*0..6]->(real_d) + MATCH path2 = (real_d)-[:支持操作]->(op_use:操作使用) + -[:使用]->(op_proc:操作程序) + -[:操作时使用]->(op_item) + WITH nodes(path1) + nodes(path2) AS ns + UNWIND ns AS n + WITH DISTINCT n + RETURN elementId(n) AS id, + labels(n) AS labels, + coalesce(n.名称, n.name) AS name, + properties(n) AS props + """ + node_records = await session.run( + graph_nodes_query, + ship_id=ship_id, + real_device_id=real_device_id, + op_item_id=op_item_id, + ) + nodes_g = [_record_to_graph_node(r.data()) async for r in node_records] + + graph_links_query = """ + MATCH (ship:舰艇) + WHERE id(ship) = $ship_id + MATCH (real_d:设备) + WHERE id(real_d) = $real_device_id + MATCH (op_item:操作项目) + WHERE id(op_item) = $op_item_id + MATCH path1 = (ship)-[:包含*0..6]->(real_d) + MATCH path2 = (real_d)-[:支持操作]->(op_use:操作使用) + -[:使用]->(op_proc:操作程序) + -[:操作时使用]->(op_item) + WITH relationships(path1) + relationships(path2) AS rs + UNWIND rs AS r + WITH DISTINCT r + RETURN elementId(r) AS id, + type(r) AS label, + elementId(startNode(r)) AS source, + elementId(endNode(r)) AS target + """ + link_records = await session.run( + graph_links_query, + ship_id=ship_id, + real_device_id=real_device_id, + op_item_id=op_item_id, + ) + links_g = [] + node_ids = {n.get("id") for n in nodes_g} + async for r in link_records: + data = r.data() + if data.get("source") in node_ids and data.get("target") in node_ids: + links_g.append({ + "id": data.get("id"), + "label": data.get("label", ""), + "source": data.get("source"), + "target": data.get("target"), + }) + + op_props = _filter_props_for_graph(op_node.get("props") or {}) + results_text = json.dumps(op_props, ensure_ascii=False) if op_props else "" + graph_item = apply_styles({"nodes": nodes_g, "links": links_g}) + return {"success": True, "xxxx.graph": [graph_item], "results": results_text} except Exception as e: - return {"success": False, "error": f"操作图谱检索失败: {str(e)}"} + return {"success": False, "error": f"Neo4j 查询失败: {str(e)}"} @tool @@ -1596,8 +1864,6 @@ async def atlas_retrieval( } } """ - api_url = ATLAS_CONFIG["endpoint"] - if not node_names: return { "success": False, @@ -1605,42 +1871,104 @@ async def atlas_retrieval( "data": {} } - payload = { - "node_names": node_names, - "top_k": top_k - } + uri = NEO4J_CONFIG.get("uri") + user = NEO4J_CONFIG.get("user") + password = NEO4J_CONFIG.get("password") + if not uri or not user or not password: + return {"success": False, "error": "Neo4j 配置不完整，请检查 NEO4J_CONFIG", "data": {}} try: - async with httpx.AsyncClient(timeout=timeout) as client: - response = await client.post( - api_url, - json=payload, - headers={ - "accept": "application/json", - "Content-Type": "application/json" + async with AsyncGraphDatabase.driver(uri, auth=(user, password)) as driver: + async with driver.session() as session: + label_indexes = { + "设备": ("设备_fulltext", "设备_vector"), + "系统": ("系统_fulltext", "系统_vector"), + "子系统": ("子系统_fulltext", "子系统_vector"), } - ) + matched_nodes: List[Dict[str, Any]] = [] + seen_node_ids = set() - if response.status_code == 200: - result = response.json() - if result.get("success", False): - return { - "success": True, - "data": result.get("data", {}), - "error": None - } - else: - return { - "success": False, - "error": result.get("message", "API返回失败"), - "data": {} - } - else: - return { - "success": False, - "error": f"HTTP {response.status_code}: {response.text}", - "data": {} - } + for node_name in node_names: + query_name = str(node_name or "").strip() + if not query_name: + continue + label_matches = [] + for label, (fulltext_index, vector_index) in label_indexes.items(): + node, score, meta = await _match_scoped_nodes_with_graph_indexes( + session=session, + query_text=query_name, + fulltext_index=fulltext_index, + vector_index=vector_index, + candidate_ids=None, + fallback_candidates=None, + fulltext_top_k=max(top_k, 500), + vector_top_k=max(top_k, 500), + min_text_score=0.30, + min_combined_score=0.38, + ) + if node: + label_matches.append((node, score, meta, label)) + + if not label_matches: + continue + label_matches.sort(key=lambda x: x[1], reverse=True) + best_node, best_score, best_meta, best_label = label_matches[0] + node_id = best_node.get("id") + if node_id in seen_node_ids: + continue + seen_node_ids.add(node_id) + matched_nodes.append({ + "id": node_id, + "name": _candidate_display_name(best_node), + "labels": best_node.get("labels") or [best_label], + "props": best_node.get("props", {}), + "score": best_score, + }) + + if not matched_nodes: + return {"success": False, "error": "未找到匹配的节点", "data": {}} + + source_ids = [n["id"] for n in matched_nodes if n.get("id") is not None] + atlas_query = """ + MATCH (source) + WHERE id(source) IN $source_ids + MATCH (source)-[*1..1]-(atlas:图册) + RETURN id(source) AS source_id, + coalesce(source.名称, source.name) AS source_name, + id(atlas) AS atlas_id, + coalesce(atlas.名称, atlas.name) AS atlas_name, + properties(atlas) AS atlas_props + """ + records = await session.run(atlas_query, source_ids=source_ids) + result_data: Dict[str, Dict[str, List[str]]] = {} + async for record in records: + source_name = record.get("source_name") + atlas_name = record.get("atlas_name") + atlas_props = record.get("atlas_props") or {} + + pdf_url = None + for key, value in atlas_props.items(): + key_s = str(key) + if "图" in key_s or "pdf" in key_s.lower() or "url" in key_s.lower() or "文件" in key_s: + if isinstance(value, str) and value.startswith(("http", "/")): + pdf_url = value + break + + if not source_name or not atlas_name: + continue + result_data.setdefault(source_name, {}).setdefault(atlas_name, []) + if pdf_url and pdf_url not in result_data[source_name][atlas_name]: + result_data[source_name][atlas_name].append(pdf_url) + + if not result_data: + return { + "success": False, + "error": "未找到关联的图册节点", + "data": {}, + "matched_nodes": matched_nodes, + } + + return {"success": True, "data": result_data, "error": None} except Exception as e: return {