95 lines
3.4 KiB
Python
95 lines
3.4 KiB
Python
import pytest
|
||
from pydantic import ValidationError
|
||
|
||
from src.rag.qa_v1_1_73 import pipeline as pipeline_module
|
||
from src.rag.qa_v1_1_73.api import QARequest
|
||
from src.rag.qa_v1_1_73.schemas import RetrievalHypothesis
|
||
|
||
|
||
def test_native_multihop_backend_clamps_requested_four_hops_to_three(monkeypatch) -> None:
|
||
captured = {}
|
||
|
||
class FakeRetriever:
|
||
def retrieve_chains(self, *args, **kwargs):
|
||
captured.update(kwargs)
|
||
return [{"chain_id": "G1", "hops": []}], {"status": "ok"}
|
||
|
||
def close(self):
|
||
captured["closed"] = True
|
||
|
||
monkeypatch.setattr(pipeline_module, "NativeGraphPathRetriever", FakeRetriever, raising=False)
|
||
instance = pipeline_module.QAV1Pipeline.__new__(pipeline_module.QAV1Pipeline)
|
||
chains, audit = instance._retrieve_native_chains(
|
||
backend="native_multihop",
|
||
question="What terminal object is reached?",
|
||
hypothesis=RetrievalHypothesis(question_type="relation"),
|
||
seed_ids=["cent_anchor"],
|
||
seed_names=["Anchor"],
|
||
relation_groups=["expression_condition"],
|
||
answer_type_groups=["treatment"],
|
||
max_hops=4,
|
||
min_hops=1,
|
||
beam_width=90,
|
||
per_node_expand=24,
|
||
top_k=10,
|
||
)
|
||
assert len(chains) == 1
|
||
assert audit["status"] == "ok"
|
||
assert captured == {
|
||
"max_hops": 3,
|
||
"min_hops": 1,
|
||
"beam_width": 90,
|
||
"per_node_expand": 24,
|
||
"top_k_chains": 10,
|
||
"relation_categories": ["expression_condition"],
|
||
"answer_type_groups": ["treatment"],
|
||
"closed": True,
|
||
}
|
||
|
||
|
||
def test_api_native_chain_contract_is_one_to_three_hops() -> None:
|
||
assert QARequest(question="test").native_chain_max_hops == 3
|
||
with pytest.raises(ValidationError):
|
||
QARequest(question="test", native_chain_max_hops=4)
|
||
|
||
|
||
def test_deterministic_question_seed_extraction_recovers_explicit_chinese_relations() -> None:
|
||
question = (
|
||
"研究报道 HORVU4Hr1G054060 与 parents 存在“unchanged in”关系；"
|
||
"研究报道 parents 与 PH 存在“showed phenotypic variation for”关系。"
|
||
)
|
||
seeds = pipeline_module.QAV1Pipeline._deterministic_question_seed_names(question)
|
||
assert "horvu4hr1g054060" in seeds
|
||
assert "parents" in seeds
|
||
assert "ph" in seeds
|
||
|
||
|
||
def test_deterministic_question_seed_extraction_handles_collected_from_clauses() -> None:
|
||
seeds = pipeline_module.QAV1Pipeline._deterministic_question_seed_names(
|
||
"L. canum 采集于 GT；GT 采集于 acetone。"
|
||
)
|
||
assert seeds == ["l canum", "gt", "acetone"]
|
||
|
||
|
||
def test_deterministic_question_seed_extraction_handles_common_chain_clauses() -> None:
|
||
seeds = pipeline_module.QAV1Pipeline._deterministic_question_seed_names(
|
||
"AhbHLH109.4 在 Silihong中表达；Silihong 通过 A. duranensis进行研究；"
|
||
"NFYA5 调控 drought-responsive genes；Pfam database 包含 SPL gene family。"
|
||
)
|
||
assert seeds == [
|
||
"ahbhlh109 4",
|
||
"silihong",
|
||
"a duranensis",
|
||
"nfya5",
|
||
"drought responsive genes",
|
||
"pfam database",
|
||
"spl gene family",
|
||
]
|
||
|
||
|
||
def test_chinese_terminal_question_produces_answer_type_groups() -> None:
|
||
groups = pipeline_module.QAV1Pipeline._native_answer_type_groups_from_hypothesis(
|
||
RetrievalHypothesis(), "PH 与什么处理或环境条件有关？"
|
||
)
|
||
assert {"treatment", "condition", "stress"}.issubset(groups)
|