import inspect from src.rag.qa_v1_1_73.graph_path_chain import GraphHop, GraphPath, NativeGraphPathRetriever def hop(index: int, source: str, target: str, assertion: str) -> GraphHop: return GraphHop( hop_index=index, from_id=source, from_name=source, to_id=target, to_name=target, source=source, relation="related_to", target=target, assertion_id=assertion, edge_score=0.8, ) def path(*hops: GraphHop, score: float) -> GraphPath: return GraphPath( hops=list(hops), node_ids=[hops[0].from_id] + [item.to_id for item in hops], path_score=score, score_breakdown={"path_coherence": 1.0, "provenance_confidence": 0.8}, ) def test_direct_path_is_not_suppressed_by_longer_path_with_different_terminal() -> None: direct = path(hop(1, "A", "B", "a1"), score=0.95) longer = path(hop(1, "A", "B", "a1"), hop(2, "B", "C", "a2"), score=0.99) selected = NativeGraphPathRetriever._select_multihop_paths([direct, longer], top_k=10) assert direct in selected def test_top10_keeps_two_distinct_paths_with_same_endpoints() -> None: first = path(hop(1, "A", "B", "a1"), hop(2, "B", "Z", "a2"), score=0.90) second = path(hop(1, "A", "C", "b1"), hop(2, "C", "Z", "b2"), score=0.85) selected = NativeGraphPathRetriever._select_multihop_paths([first, second], top_k=10) assert selected == [first, second] def test_quoted_relation_phrases_become_search_clues() -> None: clues = NativeGraphPathRetriever._relation_clues( '相关研究中，与 ash content 存在“determined by”关系的对象是什么？' ) assert "determined by" in clues assert "determined_by" in clues def test_blocked_method_endpoint_is_retained_but_not_expanded_when_unmentioned() -> None: method_hop = hop(3, "ash content", "AOAC", "a3") method_hop.target_type = "method" method_hop.relation_category = "method_tool" terminal_path = path(method_hop, score=0.8) assert NativeGraphPathRetriever._should_expand_path(terminal_path, "ash content determined by what") is False assert NativeGraphPathRetriever._should_expand_path(terminal_path, "aoac used for analysis") is True def test_start_node_selection_preserves_each_seed_before_ambiguous_fill() -> None: rows = [ {"seed_index": 0, "canonical_id": f"na_{i}", "name": "NA", "entity_type": "unknown"} for i in range(20) ] + [ {"seed_index": 1, "canonical_id": "gene_1", "name": "PgbZIP44", "entity_type": "gene"}, {"seed_index": 2, "canonical_id": "root_1", "name": "roots", "entity_type": "tissue"}, ] selected = NativeGraphPathRetriever._fair_seed_nodes(rows, limit=3) assert [row["canonical_id"] for row in selected] == ["na_0", "gene_1", "root_1"] def test_path_score_rewards_coverage_of_question_quoted_relations() -> None: matching_hop = hop(1, "Na+", "safflower", "gold") matching_hop.relation = "content_apparent_in" matching = path(matching_hop, score=0.0) distractor_hop = hop(1, "Na+", "root", "other") distractor_hop.relation = "associated_with" distractor = path(distractor_hop, score=0.0) retriever = NativeGraphPathRetriever.__new__(NativeGraphPathRetriever) matching_score, _ = retriever._score_path( matching, "na content apparent in what", {"na"}, set(), ["content apparent in"] ) distractor_score, _ = retriever._score_path( distractor, "na content apparent in what", {"na"}, set(), ["content apparent in"] ) assert matching_score > distractor_score def test_only_seed_names_visible_in_question_become_endpoint_clues() -> None: clues = NativeGraphPathRetriever._visible_entity_clues( "l canum collected from gt using acetone", ["l canum", "gt", "acetone", "unrelated provisional guess"], ) assert clues == ["l canum", "gt", "acetone"] def test_beam_selection_preserves_a_candidate_from_each_start_anchor() -> None: crowded = [path(hop(1, "A", f"X{i}", f"a{i}"), score=0.9 - i * 0.01) for i in range(5)] second_anchor = path(hop(1, "B", "Y", "b1"), score=0.2) selected = NativeGraphPathRetriever._select_beam_paths(crowded + [second_anchor], beam_width=2) assert selected == [crowded[0], second_anchor] def test_beam_prefers_paths_reaching_question_visible_bridge() -> None: distractor = path(hop(1, "A", "motif", "a1"), score=0.95) visible_bridge = path(hop(1, "A", "Senna", "a2"), score=0.60) selected = NativeGraphPathRetriever._select_beam_paths( [distractor, visible_bridge], beam_width=1, entity_clues=["senna"] ) assert selected == [visible_bridge] def test_neighbor_query_limits_directed_outgoing_edges_not_undirected_noise() -> None: source = inspect.getsource(NativeGraphPathRetriever._fetch_neighbors) assert "a.source_canonical_id = e.canonical_id" in source assert "n.canonical_id = a.target_canonical_id" in source assert "a.target_canonical_id <> e.canonical_id" in source def test_start_node_query_supports_punctuation_normalized_abbreviations() -> None: source = inspect.getsource(NativeGraphPathRetriever._resolve_start_nodes) assert "replace(e.name_norm, '.', '') = seed_name" in source def test_stitch_paths_joins_exact_canonical_endpoint_without_inventing_edges() -> None: prefix = path(hop(1, "A", "B", "a1"), hop(2, "B", "C", "a2"), score=0.8) suffix = path(hop(1, "C", "D", "a3"), score=0.9) stitched = NativeGraphPathRetriever._stitch_paths([prefix, suffix], max_hops=4) assert len(stitched) == 1 assert stitched[0].node_ids == ["A", "B", "C", "D"] assert stitched[0].assertion_ids() == ["a1", "a2", "a3"] def test_stitch_paths_rejects_cycles_and_overlong_paths() -> None: prefix = path(hop(1, "A", "B", "a1"), hop(2, "B", "C", "a2"), score=0.8) cycle = path(hop(1, "C", "A", "a3"), score=0.9) long_suffix = path( hop(1, "C", "D", "b1"), hop(2, "D", "E", "b2"), hop(3, "E", "F", "b3"), score=0.9, ) assert NativeGraphPathRetriever._stitch_paths([prefix, cycle, long_suffix], max_hops=4) == [] def test_terminal_type_scoring_prefers_unknown_classification_answer_over_known_bridge() -> None: known_bridge_hop = hop(2, "cage", "pilose beggarticks", "a2") known_bridge_hop.target_type = "species" known_bridge = path(known_bridge_hop, score=0.0) answer_hop = hop(3, "pilose beggarticks", "Bidens pilosa", "a3") answer_hop.target_type = "species" answer = path(answer_hop, score=0.0) question = "pilose beggarticks belongs to what family category or subclass" assert NativeGraphPathRetriever._target_type_match( answer, question, ["classification"] ) > NativeGraphPathRetriever._target_type_match( known_bridge, question, ["classification"] ) def test_path_score_rewards_coverage_of_visible_bridge_entities() -> None: via_bridge = path( hop(1, "Salvia", "ITS2", "a1"), hop(2, "ITS2", "Senna", "a2"), hop(3, "Senna", "mucilage", "a3"), score=0.0, ) distractor = path( hop(1, "Salvia", "ITS2", "a1"), hop(2, "ITS2", "helix", "b2"), score=0.0, ) retriever = NativeGraphPathRetriever.__new__(NativeGraphPathRetriever) args = ("salvia its2 senna contains what", {"salvia"}, set(), ["contains"], [], ["salvia", "its2", "senna"]) via_score, _ = retriever._score_path(via_bridge, *args) distractor_score, _ = retriever._score_path(distractor, *args) assert via_score > distractor_score def test_top10_treats_visible_bridge_coverage_as_hard_ranking_constraint() -> None: complete = path(hop(1, "A", "B", "a1"), hop(2, "B", "C", "a2"), score=0.60) complete.score_breakdown["visible_entity_coverage"] = 1.0 distractor = path(hop(1, "A", "X", "b1"), score=0.99) distractor.score_breakdown["visible_entity_coverage"] = 0.33 selected = NativeGraphPathRetriever._select_multihop_paths([distractor, complete], top_k=1) assert selected == [complete] def test_top10_prefers_complete_chain_over_its_equally_visible_suffix() -> None: complete = path( hop(1, "A", "B", "a1"), hop(2, "B", "C", "a2"), hop(3, "C", "D", "a3"), hop(4, "D", "answer", "a4"), score=0.80, ) suffix = path( hop(1, "B", "C", "a2"), hop(2, "C", "D", "a3"), hop(3, "D", "other", "b4"), score=0.99, ) complete.score_breakdown["visible_entity_coverage"] = 1.0 suffix.score_breakdown["visible_entity_coverage"] = 1.0 selected = NativeGraphPathRetriever._select_multihop_paths([suffix, complete], top_k=1) assert selected == [complete] def test_question_chain_spec_compiles_explicit_pairs_and_terminal_relation() -> None: question = ( "AhbHLH109.4 在 Silihong中表达；Silihong 通过 A. duranensis进行研究；" "研究报道 A. duranensis 与 Hsf genes 存在“has gene family”关系。" "相关研究中，与 Hsf genes 存在“analyzed for”关系的对象是什么？" ) pairs, terminal = NativeGraphPathRetriever._question_chain_spec(question) assert pairs == [ ("ahbhlh109 4", "silihong"), ("silihong", "a duranensis"), ("a duranensis", "hsf genes"), ] assert terminal == ("hsf genes", "analyzed for") def test_question_chain_spec_handles_interaction_and_non_object_terminal_forms() -> None: pairs, terminal = NativeGraphPathRetriever._question_chain_spec( "PcNF-YA5 与 miR169相互作用；研究报道 miR169 与 NFYA5 存在“targets”关系；" "NFYA5 调控 drought-responsive genes。drought-responsive genes 属于什么家族、类别或亚类？" ) assert pairs == [ ("pcnf ya5", "mir169"), ("mir169", "nfya5"), ("nfya5", "drought responsive genes"), ] assert terminal == ("drought responsive genes", "") def test_question_chain_spec_preserves_chemical_charge_in_entity_names() -> None: pairs, terminal = NativeGraphPathRetriever._question_chain_spec( "PgbZIP44 在 roots中表达；研究报道 roots 与 Na+ 存在“site of accumulation”关系。" "相关研究中，与 Na+ 存在“content apparent in”关系的对象是什么？" ) assert pairs[-1] == ("roots", "na+") assert terminal == ("na+", "content apparent in") def test_question_chain_spec_handles_treatment_or_environment_terminal() -> None: pairs, terminal = NativeGraphPathRetriever._question_chain_spec( "研究报道 parents 与 PH 存在“showed phenotypic variation for”关系。" "PH 与什么处理或环境条件有关？" ) assert pairs == [("parents", "ph")] assert terminal == ("ph", "")