246 lines
10 KiB
Python
246 lines
10 KiB
Python
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", "")
|