1020 lines
33 KiB
Python
1020 lines
33 KiB
Python
from src.rag.qa_v1_1_73.agentic_controller import AgenticRetrievalController
|
||
from src.rag.qa_v1_1_73.schemas import EvidenceItem, HypothesisEntity, RetrievalHypothesis
|
||
|
||
|
||
QUESTION = "荔枝中有哪些基因家族已经被研究"
|
||
|
||
|
||
class FakeLLM:
|
||
available = True
|
||
|
||
def __init__(self, *responses):
|
||
self.responses = list(responses)
|
||
self.calls = []
|
||
|
||
def complete_json(self, system, user, *, max_tokens=2500):
|
||
self.calls.append((system, user, max_tokens))
|
||
response = self.responses.pop(0)
|
||
if isinstance(response, Exception):
|
||
raise response
|
||
return response
|
||
|
||
|
||
def test_controller_caps_replans_at_one():
|
||
controller = AgenticRetrievalController(FakeLLM(), max_replans=99)
|
||
|
||
assert controller.max_replans == 1
|
||
|
||
|
||
def hypothesis():
|
||
return RetrievalHypothesis(
|
||
question_type="survey",
|
||
direct_question_entities=[
|
||
HypothesisEntity(
|
||
mention="荔枝",
|
||
canonical_candidate="Litchi chinensis",
|
||
entity_type="species",
|
||
origin="question",
|
||
confidence=1.0,
|
||
verification_status="verified_alias",
|
||
matched_kb_entity="Litchi chinensis",
|
||
use_as_graph_seed=True,
|
||
),
|
||
HypothesisEntity(
|
||
mention="基因家族",
|
||
canonical_candidate="gene family",
|
||
entity_type="gene_family",
|
||
origin="question",
|
||
confidence=1.0,
|
||
verification_status="question_direct_unverified",
|
||
use_as_graph_seed=True,
|
||
),
|
||
],
|
||
relation_intents=["studied_gene_families", "identified"],
|
||
answer_slots=["list of gene family names"],
|
||
)
|
||
|
||
|
||
def agent_plan(**overrides):
|
||
value = {
|
||
"mode": "PLAN",
|
||
"question_analysis": {
|
||
"language": "zh",
|
||
"intent": "列出荔枝中研究过的基因家族",
|
||
"operation": "enumerate",
|
||
"expected_answer_shape": "entity_list",
|
||
"requires_graph": True,
|
||
},
|
||
"anchors": [
|
||
{
|
||
"mention": "荔枝",
|
||
"canonical_name": "Litchi chinensis",
|
||
"canonical_id": "",
|
||
"entity_type": "species",
|
||
"role": "start_anchor",
|
||
"confidence": 1.0,
|
||
"required_in_every_answer_path": True,
|
||
}
|
||
],
|
||
"answer_targets": [
|
||
{
|
||
"description": "基因家族",
|
||
"allowed_entity_types": ["gene_family"],
|
||
"role": "answer_endpoint",
|
||
}
|
||
],
|
||
"relation_plan": {
|
||
"relation_intents": ["identified", "studied_gene_families"],
|
||
"allowed_relation_categories": [],
|
||
"relation_name_hints": ["identified_in", "reported_in"],
|
||
"allow_incoming": True,
|
||
"allow_outgoing": True,
|
||
"semantic_direction": "anchor_to_answer",
|
||
"preserve_assertion_direction": True,
|
||
},
|
||
"graph_plan": {
|
||
"enabled": True,
|
||
"max_hops": 1,
|
||
"min_hops": 1,
|
||
"anchor_only_start": True,
|
||
"required_anchor_coverage": 1.0,
|
||
"required_terminal_types": ["gene_family"],
|
||
"permit_reverse_navigation": True,
|
||
"exclude_unrelated_vector_start_seeds": True,
|
||
},
|
||
"vector_plan": {
|
||
"enabled": True,
|
||
"purpose": "supporting_evidence",
|
||
"queries": ["Litchi chinensis gene family"],
|
||
"may_create_graph_start_seeds": False,
|
||
},
|
||
"acceptance_gate": {
|
||
"minimum_supported_candidates": 1,
|
||
"require_anchor_in_each_path": True,
|
||
"require_terminal_type_match": True,
|
||
"require_traceable_provenance": True,
|
||
"reject_cross_species_substitution": True,
|
||
},
|
||
"reasoning_summary": "以荔枝为锚点枚举相连基因家族",
|
||
}
|
||
value.update(overrides)
|
||
return value
|
||
|
||
|
||
def test_plan_gives_agent_direction_seed_and_endpoint_control():
|
||
controller = AgenticRetrievalController(FakeLLM(agent_plan()))
|
||
|
||
plan = controller.plan(QUESTION, hypothesis())
|
||
|
||
assert plan["anchors"] == [
|
||
{
|
||
"mention": "荔枝",
|
||
"canonical_name": "Litchi chinensis",
|
||
"canonical_id": "",
|
||
"entity_type": "species",
|
||
"role": "start_anchor",
|
||
"confidence": 1.0,
|
||
"required_in_every_answer_path": True,
|
||
}
|
||
]
|
||
assert plan["answer_targets"][0]["allowed_entity_types"] == ["gene_family"]
|
||
assert plan["relation_plan"]["allow_incoming"] is True
|
||
assert plan["graph_plan"]["anchor_only_start"] is True
|
||
assert plan["graph_plan"]["max_hops"] == 1
|
||
|
||
|
||
def test_agent_prompt_allows_structured_assertions_without_graph_for_definition_questions():
|
||
llm = FakeLLM(agent_plan())
|
||
controller = AgenticRetrievalController(llm)
|
||
|
||
controller.plan(QUESTION, hypothesis())
|
||
|
||
system_prompt = llm.calls[0][0]
|
||
assert "definition, naming, or structure" in system_prompt
|
||
assert "source, relation, target" in system_prompt
|
||
assert "graph path is optional" in system_prompt
|
||
|
||
|
||
def test_plan_removes_invented_anchor_and_never_uses_generic_target_as_start():
|
||
unsafe = agent_plan(
|
||
anchors=[
|
||
{
|
||
"mention": "TCP",
|
||
"canonical_name": "TCP gene family",
|
||
"canonical_id": "cent_invented",
|
||
"entity_type": "gene_family",
|
||
"role": "start_anchor",
|
||
"confidence": 1.0,
|
||
"required_in_every_answer_path": True,
|
||
},
|
||
{
|
||
"mention": "基因家族",
|
||
"canonical_name": "gene family",
|
||
"canonical_id": "",
|
||
"entity_type": "gene_family",
|
||
"role": "start_anchor",
|
||
"confidence": 1.0,
|
||
"required_in_every_answer_path": True,
|
||
},
|
||
],
|
||
graph_plan={"max_hops": 99},
|
||
)
|
||
controller = AgenticRetrievalController(FakeLLM(unsafe))
|
||
|
||
plan = controller.plan(QUESTION, hypothesis())
|
||
|
||
assert [item["canonical_name"] for item in plan["anchors"]] == ["Litchi chinensis"]
|
||
assert plan["graph_plan"]["max_hops"] == 3
|
||
|
||
|
||
def test_plan_falls_back_deterministically_when_agent_json_fails():
|
||
controller = AgenticRetrievalController(FakeLLM(ValueError("bad json")))
|
||
|
||
plan = controller.plan(QUESTION, hypothesis())
|
||
|
||
assert plan["plan_source"] == "deterministic_fallback"
|
||
assert plan["anchors"][0]["canonical_name"] == "Litchi chinensis"
|
||
assert plan["answer_targets"][0]["allowed_entity_types"] == ["gene_family"]
|
||
assert plan["relation_plan"]["allow_incoming"] is True
|
||
assert plan["graph_plan"]["max_hops"] == 1
|
||
|
||
|
||
def litchi_evidence():
|
||
return EvidenceItem(
|
||
evidence_id="E1",
|
||
kb="V16/V15",
|
||
source="SPL gene family",
|
||
relation="18_homologues_identified",
|
||
target="Litchi chinensis",
|
||
evidence="Eighteen SPL homologues were identified.",
|
||
paper_id="10.1080_14620316.2018.1543557",
|
||
doi="10.1080/14620316.2018.1543557",
|
||
raw={
|
||
"assertion_id": "cass_spl_litchi",
|
||
"source_type": "gene_family",
|
||
"target_type": "species",
|
||
"anchor_name": "Litchi chinensis",
|
||
"answer_entity": "SPL gene family",
|
||
"answer_entity_type": "gene_family",
|
||
},
|
||
)
|
||
|
||
|
||
def test_review_accepts_only_real_anchor_connected_ids():
|
||
llm = FakeLLM(
|
||
{
|
||
"mode": "REVIEW",
|
||
"decision": "ACCEPT",
|
||
"attempt": 0,
|
||
"accepted_evidence_ids": ["E1", "E_INVENTED"],
|
||
"accepted_chain_ids": ["AG1", "G_INVENTED"],
|
||
"problems": [],
|
||
"replan": {"required": False},
|
||
"final_handoff": {"allowed": True},
|
||
}
|
||
)
|
||
controller = AgenticRetrievalController(llm)
|
||
chains = [
|
||
{
|
||
"chain_id": "AG1",
|
||
"start": "Litchi chinensis",
|
||
"end": "SPL gene family",
|
||
"answer_entity": "SPL gene family",
|
||
"answer_entity_type": "gene_family",
|
||
"anchor_name": "Litchi chinensis",
|
||
"doi": "10.1080/14620316.2018.1543557",
|
||
"hops": [],
|
||
}
|
||
]
|
||
|
||
review = controller.review(
|
||
QUESTION, agent_plan(), [litchi_evidence()], chains, attempt=0
|
||
)
|
||
|
||
assert review["decision"] == "ACCEPT"
|
||
assert review["accepted_evidence_ids"] == ["E1"]
|
||
assert review["accepted_chain_ids"] == ["AG1"]
|
||
assert review["final_handoff"]["allowed"] is True
|
||
|
||
|
||
def test_review_overrides_false_accept_when_anchor_is_absent():
|
||
llm = FakeLLM(
|
||
{
|
||
"mode": "REVIEW",
|
||
"decision": "ACCEPT",
|
||
"attempt": 0,
|
||
"accepted_evidence_ids": ["E2"],
|
||
"accepted_chain_ids": ["G2"],
|
||
"problems": [],
|
||
"replan": {"required": False},
|
||
"final_handoff": {"allowed": True},
|
||
}
|
||
)
|
||
controller = AgenticRetrievalController(llm)
|
||
unrelated = EvidenceItem(
|
||
evidence_id="E2",
|
||
kb="V16/V15",
|
||
source="TCP gene family",
|
||
relation="identified_in",
|
||
target="strawberry",
|
||
evidence="TCP evidence",
|
||
doi="10.1/tcp",
|
||
raw={"source_type": "gene_family", "target_type": "species"},
|
||
)
|
||
|
||
review = controller.review(
|
||
QUESTION,
|
||
agent_plan(),
|
||
[unrelated],
|
||
[{"chain_id": "G2", "start": "TCP gene family", "end": "MYB family"}],
|
||
attempt=0,
|
||
)
|
||
|
||
assert review["decision"] == "REPLAN"
|
||
assert review["final_handoff"]["allowed"] is False
|
||
|
||
|
||
def test_review_rejects_cross_species_count_comparison_even_when_llm_accepts_it():
|
||
llm = FakeLLM(
|
||
{
|
||
"mode": "REVIEW",
|
||
"decision": "ACCEPT",
|
||
"attempt": 0,
|
||
"accepted_evidence_ids": ["E_COMPARE"],
|
||
"accepted_chain_ids": [],
|
||
"final_handoff": {"allowed": True},
|
||
}
|
||
)
|
||
controller = AgenticRetrievalController(llm)
|
||
comparison = EvidenceItem(
|
||
evidence_id="E_COMPARE",
|
||
kb="V16/V15",
|
||
source="HcARF genes",
|
||
relation="number_similar_to",
|
||
target="Litchi chinensis",
|
||
evidence="The number of HcARF genes was similar to that in Litchi chinensis.",
|
||
pmid="34052884",
|
||
raw={
|
||
"source_type": "gene_family",
|
||
"target_type": "species",
|
||
"anchor_name": "Litchi chinensis",
|
||
"answer_entity_type": "gene_family",
|
||
},
|
||
)
|
||
|
||
review = controller.review(QUESTION, agent_plan(), [comparison], [], attempt=0)
|
||
|
||
assert review["decision"] == "REPLAN"
|
||
assert review["accepted_evidence_ids"] == []
|
||
|
||
|
||
def test_review_abstains_after_one_failed_replan():
|
||
controller = AgenticRetrievalController(FakeLLM({"decision": "REPLAN"}))
|
||
|
||
review = controller.review(QUESTION, agent_plan(), [], [], attempt=1)
|
||
|
||
assert review["decision"] == "ABSTAIN"
|
||
assert review["final_handoff"]["allowed"] is True
|
||
assert review["accepted_evidence_ids"] == []
|
||
assert review["accepted_chain_ids"] == []
|
||
|
||
|
||
def test_review_can_accept_the_result_of_the_first_replan():
|
||
controller = AgenticRetrievalController(
|
||
FakeLLM(
|
||
{
|
||
"decision": "ACCEPT",
|
||
"accepted_evidence_ids": ["E1"],
|
||
"accepted_chain_ids": [],
|
||
"final_handoff": {"allowed": True},
|
||
}
|
||
)
|
||
)
|
||
|
||
review = controller.review(
|
||
QUESTION, agent_plan(), [litchi_evidence()], [], attempt=1
|
||
)
|
||
|
||
assert review["decision"] == "ACCEPT"
|
||
assert review["accepted_evidence_ids"] == ["E1"]
|
||
|
||
|
||
def test_review_treats_unknown_terminal_type_as_unconstrained():
|
||
controller = AgenticRetrievalController(
|
||
FakeLLM(
|
||
{
|
||
"decision": "ACCEPT",
|
||
"accepted_evidence_ids": ["E_R2R3"],
|
||
"accepted_chain_ids": [],
|
||
"final_handoff": {"allowed": True},
|
||
}
|
||
)
|
||
)
|
||
evidence = EvidenceItem(
|
||
evidence_id="E_R2R3",
|
||
kb="V16/V15",
|
||
source="R2R3-MYB proteins",
|
||
relation="contain",
|
||
target="R2 and R3 MYB repeats",
|
||
evidence="R2R3-MYB proteins contain the R2 and R3 MYB repeats.",
|
||
doi="10.1000/r2r3",
|
||
raw={
|
||
"source_type": "protein",
|
||
"target_type": "domain",
|
||
"answer_entity_type": "domain",
|
||
},
|
||
)
|
||
plan = {
|
||
"anchors": [],
|
||
"question_analysis": {
|
||
"operation": "lookup",
|
||
"expected_answer_shape": "fact",
|
||
},
|
||
"graph_plan": {"required_terminal_types": ["unknown"]},
|
||
"acceptance_gate": {
|
||
"require_anchor_in_each_path": False,
|
||
"require_traceable_provenance": True,
|
||
},
|
||
}
|
||
|
||
review = controller.review(
|
||
"R2R3-MYB蛋白为什么被称为R2R3-MYB？",
|
||
plan,
|
||
[evidence],
|
||
[],
|
||
attempt=0,
|
||
)
|
||
|
||
assert review["decision"] == "ACCEPT"
|
||
assert review["accepted_evidence_ids"] == ["E_R2R3"]
|
||
|
||
|
||
def test_deterministic_handoff_keeps_structured_direct_evidence_but_enforces_anchor():
|
||
controller = AgenticRetrievalController(FakeLLM())
|
||
direct = EvidenceItem(
|
||
evidence_id="E_R2R3",
|
||
kb="V16/V15",
|
||
source="R2R3-MYB proteins",
|
||
relation="contain",
|
||
target="R2 and R3 MYB repeats",
|
||
evidence="R2R3-MYB proteins contain the R2 and R3 MYB repeats.",
|
||
doi="10.1000/r2r3",
|
||
raw={"answer_entity_type": "domain"},
|
||
)
|
||
unrelated = EvidenceItem(
|
||
evidence_id="E_TCP",
|
||
kb="V16/V15",
|
||
source="TCP gene family",
|
||
relation="identified_in",
|
||
target="Arabidopsis thaliana",
|
||
evidence="TCP genes were identified in Arabidopsis thaliana.",
|
||
doi="10.1000/tcp",
|
||
raw={"answer_entity_type": "gene_family"},
|
||
)
|
||
|
||
direct_items, _ = controller.deterministic_handoff(
|
||
{
|
||
"anchors": [],
|
||
"graph_plan": {"required_terminal_types": ["unknown"]},
|
||
},
|
||
[direct, unrelated],
|
||
[],
|
||
)
|
||
strict_items, _ = controller.deterministic_handoff(
|
||
{
|
||
"anchors": [{"canonical_name": "Litchi chinensis"}],
|
||
"graph_plan": {"required_terminal_types": ["gene_family"]},
|
||
},
|
||
[unrelated],
|
||
[],
|
||
)
|
||
|
||
assert [item.evidence_id for item in direct_items] == ["E_R2R3", "E_TCP"]
|
||
assert strict_items == []
|
||
|
||
|
||
def hsp_hypothesis():
|
||
return RetrievalHypothesis(
|
||
question_type="mechanism",
|
||
direct_question_entities=[
|
||
HypothesisEntity(
|
||
mention="HSP基因家族",
|
||
canonical_candidate="HSP gene family",
|
||
entity_type="gene_family",
|
||
origin="question",
|
||
confidence=0.9,
|
||
verification_status="question_direct_unverified",
|
||
),
|
||
HypothesisEntity(
|
||
mention="植物",
|
||
canonical_candidate="plants",
|
||
entity_type="species",
|
||
origin="question",
|
||
confidence=0.9,
|
||
verification_status="question_direct_unverified",
|
||
),
|
||
HypothesisEntity(
|
||
mention="高温",
|
||
canonical_candidate="high temperature",
|
||
entity_type="condition",
|
||
origin="question",
|
||
confidence=0.9,
|
||
verification_status="question_direct_unverified",
|
||
),
|
||
],
|
||
relation_intents=["heat stress response", "chaperone function"],
|
||
answer_slots=["molecular mechanism", "protected target", "biological outcome"],
|
||
)
|
||
|
||
|
||
def test_mechanism_plan_keeps_question_literal_anchors_and_rejects_wrong_gene_terminal():
|
||
unsafe = agent_plan(
|
||
anchors=[
|
||
{
|
||
"mention": "HSP基因家族",
|
||
"canonical_name": "HSP gene family",
|
||
"entity_type": "gene_family",
|
||
},
|
||
{
|
||
"mention": "植物",
|
||
"canonical_name": "plants",
|
||
"entity_type": "species",
|
||
},
|
||
{
|
||
"mention": "高温",
|
||
"canonical_name": "high temperature",
|
||
"entity_type": "condition",
|
||
},
|
||
],
|
||
answer_targets=[
|
||
{
|
||
"description": "wrong endpoint",
|
||
"allowed_entity_types": ["gene_family"],
|
||
}
|
||
],
|
||
graph_plan={
|
||
"enabled": True,
|
||
"max_hops": 3,
|
||
"required_terminal_types": ["gene"],
|
||
},
|
||
)
|
||
controller = AgenticRetrievalController(FakeLLM(unsafe))
|
||
|
||
plan = controller.plan("HSP基因家族如何帮助植物应对高温？", hsp_hypothesis())
|
||
|
||
assert [item["canonical_name"] for item in plan["anchors"]] == [
|
||
"HSP gene family",
|
||
"high temperature",
|
||
]
|
||
allowed = plan["answer_targets"][0]["allowed_entity_types"]
|
||
assert "process" in allowed
|
||
assert "protein" in allowed
|
||
assert "gene" not in allowed
|
||
assert "gene_family" not in allowed
|
||
assert set(plan["graph_plan"]["required_terminal_types"]).issubset(set(allowed))
|
||
assert plan["graph_plan"]["max_hops"] == 2
|
||
assert plan["question_analysis"]["operation"] == "mechanism"
|
||
|
||
|
||
def _hsp_evidence(evidence_id, relation, target, evidence):
|
||
return EvidenceItem(
|
||
evidence_id=evidence_id,
|
||
kb="V16/V15",
|
||
source="HSPs",
|
||
relation=relation,
|
||
target=target,
|
||
evidence=evidence,
|
||
paper_id="10.1000_hsp",
|
||
doi="10.1000/hsp",
|
||
raw={"source_type": "protein", "target_type": "process"},
|
||
)
|
||
|
||
|
||
def test_group_review_does_not_apply_hsp_specific_mechanism_vocabulary_gate():
|
||
llm = FakeLLM(
|
||
{
|
||
"groups": [
|
||
{
|
||
"subquestion_id": "Q0",
|
||
"status": "SATISFIED",
|
||
"accepted_evidence_ids": [
|
||
"E_UP",
|
||
"E_TOL",
|
||
"E_PREVENT",
|
||
"E_REFOLD",
|
||
"E_INVENTED",
|
||
],
|
||
"accepted_chain_ids": ["G1", "G_INVENTED"],
|
||
"missing_answer_slots": [],
|
||
"refinement": {},
|
||
}
|
||
]
|
||
}
|
||
)
|
||
controller = AgenticRetrievalController(llm)
|
||
evidence = [
|
||
_hsp_evidence("E_UP", "upregulated_in", "heat stress", "HSPs were upregulated under heat stress."),
|
||
_hsp_evidence("E_TOL", "enhance", "heat tolerance", "HSPs enhance heat tolerance."),
|
||
_hsp_evidence(
|
||
"E_PREVENT",
|
||
"prevents",
|
||
"protein aggregation",
|
||
"HSPs act as molecular chaperones and prevent protein aggregation.",
|
||
),
|
||
_hsp_evidence(
|
||
"E_REFOLD",
|
||
"assist_refolding",
|
||
"heat-damaged proteins",
|
||
"HSPs promote refolding of heat-damaged proteins.",
|
||
),
|
||
]
|
||
chains = [
|
||
{
|
||
"chain_id": "G1",
|
||
"hops": [
|
||
{
|
||
"source": "HSPs",
|
||
"relation": "prevents",
|
||
"target": "protein aggregation",
|
||
"evidence": "HSPs prevent protein aggregation.",
|
||
"doi": "10.1000/hsp",
|
||
}
|
||
],
|
||
}
|
||
]
|
||
|
||
review = controller.review_subquestion_groups(
|
||
"HSP基因家族如何帮助植物应对高温？",
|
||
[
|
||
{
|
||
"subquestion_id": "Q0",
|
||
"question": "HSP基因家族如何帮助植物应对高温？",
|
||
"answer_slot": "molecular mechanism",
|
||
"target_entity_types": ["process", "protein", "trait"],
|
||
"acceptance_requirements": {
|
||
"requires_structured_assertion": True,
|
||
"requires_paper_provenance": True,
|
||
},
|
||
"evidence": evidence,
|
||
"chains": chains,
|
||
}
|
||
],
|
||
attempt=1,
|
||
)
|
||
|
||
group = review["groups"][0]
|
||
assert group["status"] == "SATISFIED"
|
||
assert group["accepted_evidence_ids"] == [
|
||
"E_UP",
|
||
"E_TOL",
|
||
"E_PREVENT",
|
||
"E_REFOLD",
|
||
]
|
||
assert group["accepted_chain_ids"] == ["G1"]
|
||
assert "protein-homeostasis mechanisms" not in llm.calls[0][0]
|
||
|
||
|
||
def test_group_review_recovers_valid_mechanism_ids_when_llm_describes_support_but_omits_ids():
|
||
llm = FakeLLM(
|
||
{
|
||
"groups": [
|
||
{
|
||
"subquestion_id": "Q1",
|
||
"status": "PARTIAL",
|
||
"accepted_evidence_ids": [],
|
||
"accepted_chain_ids": [],
|
||
"missing_answer_slots": [],
|
||
"review_reason": "Evidence directly states that HSPs prevent aggregation and refold damaged proteins, satisfying the answer slot.",
|
||
"refinement": {},
|
||
}
|
||
]
|
||
}
|
||
)
|
||
controller = AgenticRetrievalController(llm)
|
||
evidence = [
|
||
_hsp_evidence("E_UP", "upregulated_in", "heat stress", "HSPs were upregulated under heat stress."),
|
||
_hsp_evidence(
|
||
"E_MECH",
|
||
"act_as",
|
||
"molecular chaperones",
|
||
"HSPs prevent protein aggregation and promote refolding of heat-damaged proteins.",
|
||
),
|
||
]
|
||
|
||
review = controller.review_subquestion_groups(
|
||
"HSP基因家族如何帮助植物应对高温？",
|
||
[
|
||
{
|
||
"subquestion_id": "Q1",
|
||
"question": "HSP如何通过分子伴侣作用维持蛋白质稳态？",
|
||
"answer_slot": "molecular mechanism",
|
||
"acceptance_requirements": {
|
||
"requires_structured_assertion": True,
|
||
"requires_paper_provenance": True,
|
||
},
|
||
"evidence": evidence,
|
||
"chains": [],
|
||
}
|
||
],
|
||
attempt=1,
|
||
)
|
||
|
||
group = review["groups"][0]
|
||
assert group["accepted_evidence_ids"] == ["E_UP", "E_MECH"]
|
||
assert group["status"] == "PARTIAL"
|
||
assert group["selection_source"] == "deterministic_valid_evidence_recovery"
|
||
|
||
|
||
def test_group_review_accepts_gst_mechanism_evidence_without_hsp_terms():
|
||
llm = FakeLLM(
|
||
{
|
||
"groups": [
|
||
{
|
||
"subquestion_id": "Q1",
|
||
"status": "PARTIAL",
|
||
"accepted_evidence_ids": [],
|
||
"accepted_chain_ids": [],
|
||
"missing_answer_slots": [],
|
||
"review_reason": (
|
||
"Evidence provides direct molecular actions including "
|
||
"detoxification, peroxide scavenging, and redox regulation."
|
||
),
|
||
"refinement": {},
|
||
}
|
||
]
|
||
}
|
||
)
|
||
controller = AgenticRetrievalController(llm)
|
||
evidence = EvidenceItem(
|
||
evidence_id="Q1.A1.S1",
|
||
kb="V16/V15",
|
||
source="GST gene family",
|
||
relation="detoxifies",
|
||
target="reactive electrophiles and peroxides",
|
||
evidence=(
|
||
"Plant GSTs catalyze glutathione conjugation and contribute to "
|
||
"peroxide scavenging and cellular redox homeostasis under stress."
|
||
),
|
||
pmid="28848585",
|
||
raw={"source_type": "gene_family", "target_type": "biological_process"},
|
||
)
|
||
|
||
review = controller.review_subquestion_groups(
|
||
"GST基因家族在植物胁迫响应中有什么作用？",
|
||
[
|
||
{
|
||
"subquestion_id": "Q1",
|
||
"question": "GST通过哪些分子作用参与植物胁迫响应？",
|
||
"answer_slot": "molecular mechanism",
|
||
"anchors": [
|
||
{"mention": "GST基因家族", "canonical_name": "GST gene family"}
|
||
],
|
||
"target_entity_types": ["biological_process"],
|
||
"acceptance_requirements": {
|
||
"requires_anchor_connection": True,
|
||
"requires_structured_assertion": True,
|
||
"requires_paper_provenance": True,
|
||
},
|
||
"evidence": [evidence],
|
||
"chains": [],
|
||
}
|
||
],
|
||
attempt=1,
|
||
)
|
||
|
||
group = review["groups"][0]
|
||
assert group["accepted_evidence_ids"] == ["Q1.A1.S1"]
|
||
assert group["status"] == "PARTIAL"
|
||
assert group["selection_source"] == "deterministic_valid_evidence_recovery"
|
||
|
||
|
||
def test_group_review_recovers_valid_ids_without_matching_review_phrase():
|
||
llm = FakeLLM(
|
||
{
|
||
"groups": [
|
||
{
|
||
"subquestion_id": "Q0",
|
||
"status": "PARTIAL",
|
||
"accepted_evidence_ids": [],
|
||
"accepted_chain_ids": [],
|
||
"missing_answer_slots": [],
|
||
"review_reason": (
|
||
"Both direct and chain evidence support the overall role."
|
||
),
|
||
"refinement": {},
|
||
}
|
||
]
|
||
}
|
||
)
|
||
controller = AgenticRetrievalController(llm)
|
||
evidence = EvidenceItem(
|
||
evidence_id="Q0.A1.S1",
|
||
kb="V16/V15",
|
||
source="GST gene family",
|
||
relation="involved_in",
|
||
target="plant stress response",
|
||
evidence="GST genes participate in plant responses to abiotic stress.",
|
||
pmid="28848585",
|
||
raw={"source_type": "gene_family", "target_type": "biological_process"},
|
||
)
|
||
|
||
review = controller.review_subquestion_groups(
|
||
"GST基因家族在植物胁迫响应中有什么作用？",
|
||
[
|
||
{
|
||
"subquestion_id": "Q0",
|
||
"question": "GST基因家族在植物胁迫响应中有什么作用？",
|
||
"answer_slot": "biological role",
|
||
"anchors": [
|
||
{"mention": "GST基因家族", "canonical_name": "GST gene family"}
|
||
],
|
||
"acceptance_requirements": {
|
||
"requires_anchor_connection": True,
|
||
"requires_structured_assertion": True,
|
||
"requires_paper_provenance": True,
|
||
},
|
||
"evidence": [evidence],
|
||
"chains": [],
|
||
}
|
||
],
|
||
attempt=1,
|
||
)
|
||
|
||
group = review["groups"][0]
|
||
assert group["accepted_evidence_ids"] == ["Q0.A1.S1"]
|
||
assert group["status"] == "PARTIAL"
|
||
assert group["selection_source"] == "deterministic_valid_evidence_recovery"
|
||
|
||
|
||
def test_group_review_creates_one_deterministic_refinement_when_no_evidence_or_ids():
|
||
llm = FakeLLM(
|
||
{
|
||
"groups": [
|
||
{
|
||
"subquestion_id": "Q1",
|
||
"status": "EMPTY",
|
||
"accepted_evidence_ids": [],
|
||
"accepted_chain_ids": [],
|
||
"missing_answer_slots": [],
|
||
"review_reason": "No usable evidence was selected.",
|
||
"refinement": {},
|
||
}
|
||
]
|
||
}
|
||
)
|
||
controller = AgenticRetrievalController(llm)
|
||
group = {
|
||
"subquestion_id": "Q1",
|
||
"question": "GST通过哪些分子作用参与植物胁迫响应？",
|
||
"answer_slot": "molecular mechanism",
|
||
"anchors": [
|
||
{"mention": "GST基因家族", "canonical_name": "GST gene family"}
|
||
],
|
||
"relation_intents": ["detoxifies", "scavenges", "regulates"],
|
||
"target_entity_types": ["biological_process"],
|
||
"evidence": [],
|
||
"chains": [],
|
||
}
|
||
|
||
first = controller.review_subquestion_groups(
|
||
"GST基因家族在植物胁迫响应中有什么作用？", [group], attempt=1
|
||
)["groups"][0]
|
||
second = controller.review_subquestion_groups(
|
||
"GST基因家族在植物胁迫响应中有什么作用？", [group], attempt=2
|
||
)["groups"][0]
|
||
|
||
assert first["refinement"]["vector_queries"]
|
||
assert first["refinement"]["relation_name_hints"] == [
|
||
"detoxifies",
|
||
"scavenges",
|
||
"regulates",
|
||
]
|
||
assert second["refinement"] == {}
|
||
|
||
|
||
def test_group_review_accepts_hsp_subfamily_evidence_for_hsp_family_mechanism_anchor():
|
||
controller = AgenticRetrievalController(FakeLLM({"groups": []}))
|
||
evidence = EvidenceItem(
|
||
evidence_id="E_SHSP",
|
||
kb="V16/V15",
|
||
source="sHSP",
|
||
relation="helps",
|
||
target="damaged protein refolding",
|
||
evidence="sHSPs prevent damaged protein aggregation and help damaged proteins refold.",
|
||
pmcid="PMC9201338",
|
||
raw={"source_type": "protein_family", "target_type": "biological_process"},
|
||
)
|
||
|
||
review = controller.review_subquestion_groups(
|
||
"HSP基因家族如何帮助植物应对高温？",
|
||
[
|
||
{
|
||
"subquestion_id": "Q1",
|
||
"question": "HSP通过哪些分子伴侣机制维持蛋白质稳态？",
|
||
"answer_slot": "molecular mechanism",
|
||
"anchors": [
|
||
{
|
||
"mention": "HSP基因家族",
|
||
"canonical_name": "HSP gene family",
|
||
}
|
||
],
|
||
"target_entity_types": ["biological_process"],
|
||
"acceptance_requirements": {
|
||
"requires_anchor_connection": True,
|
||
"requires_structured_assertion": True,
|
||
"requires_paper_provenance": True,
|
||
},
|
||
"evidence": [evidence],
|
||
"chains": [],
|
||
}
|
||
],
|
||
attempt=1,
|
||
)
|
||
|
||
assert review["groups"][0]["accepted_evidence_ids"] == ["E_SHSP"]
|
||
|
||
|
||
def test_group_review_rejects_cold_stress_mechanism_for_heat_stress_question():
|
||
controller = AgenticRetrievalController(FakeLLM({"groups": []}))
|
||
cold = _hsp_evidence(
|
||
"E_COLD",
|
||
"upregulated_in",
|
||
"cold environments",
|
||
"HSPs protect plants from freezing-induced protein denaturation under cold stress.",
|
||
)
|
||
heat = _hsp_evidence(
|
||
"E_HEAT",
|
||
"prevents",
|
||
"protein denaturation",
|
||
"Under heat stress, HSPs prevent protein denaturation and aggregation.",
|
||
)
|
||
|
||
review = controller.review_subquestion_groups(
|
||
"HSP基因家族如何帮助植物应对高温？",
|
||
[
|
||
{
|
||
"subquestion_id": "Q1",
|
||
"question": "HSP应对高温的分子机制是什么？",
|
||
"answer_slot": "molecular mechanism",
|
||
"evidence": [cold, heat],
|
||
"chains": [],
|
||
}
|
||
],
|
||
attempt=1,
|
||
)
|
||
|
||
assert review["groups"][0]["accepted_evidence_ids"] == ["E_HEAT"]
|
||
|
||
|
||
def test_group_review_sends_compact_evidence_not_full_raw_payload():
|
||
llm = FakeLLM({"groups": []})
|
||
controller = AgenticRetrievalController(llm)
|
||
evidence = _hsp_evidence(
|
||
"E_COMPACT",
|
||
"prevents",
|
||
"protein aggregation",
|
||
"HSPs prevent protein aggregation.",
|
||
)
|
||
evidence.raw = {
|
||
"source_type": "protein",
|
||
"target_type": "process",
|
||
"large_internal_blob": "DO_NOT_SEND" * 1000,
|
||
}
|
||
|
||
controller.review_subquestion_groups(
|
||
"HSP基因家族如何帮助植物应对高温？",
|
||
[
|
||
{
|
||
"subquestion_id": "Q0",
|
||
"question": "HSP基因家族如何帮助植物应对高温？",
|
||
"answer_slot": "mechanism",
|
||
"evidence": [evidence],
|
||
"chains": [],
|
||
}
|
||
],
|
||
attempt=1,
|
||
)
|
||
|
||
user_payload = llm.calls[0][1]
|
||
assert "E_COMPACT" in user_payload
|
||
assert '"source_type": "protein"' in user_payload
|
||
assert "DO_NOT_SEND" not in user_payload
|
||
|
||
|
||
def test_group_review_rejects_comparison_only_evidence_for_species_enumeration():
|
||
llm = FakeLLM(
|
||
{
|
||
"groups": [
|
||
{
|
||
"subquestion_id": "Q0",
|
||
"status": "SATISFIED",
|
||
"accepted_evidence_ids": ["E_COMPARE"],
|
||
"accepted_chain_ids": [],
|
||
"refinement": {},
|
||
}
|
||
]
|
||
}
|
||
)
|
||
controller = AgenticRetrievalController(llm)
|
||
comparison = EvidenceItem(
|
||
evidence_id="E_COMPARE",
|
||
kb="V16/V15",
|
||
source="HcARF genes",
|
||
relation="number_similar_to",
|
||
target="Litchi chinensis",
|
||
evidence="The number of HcARF genes was similar to Litchi chinensis.",
|
||
pmid="34052884",
|
||
raw={"source_type": "gene_family", "target_type": "species"},
|
||
)
|
||
|
||
review = controller.review_subquestion_groups(
|
||
"荔枝中有哪些基因家族已经被研究？",
|
||
[
|
||
{
|
||
"subquestion_id": "Q0",
|
||
"question": "荔枝中有哪些基因家族已经被研究？",
|
||
"answer_slot": "gene family list",
|
||
"anchors": [
|
||
{
|
||
"mention": "荔枝",
|
||
"canonical_name": "Litchi chinensis",
|
||
"entity_type": "species",
|
||
}
|
||
],
|
||
"target_entity_types": ["gene_family"],
|
||
"acceptance_requirements": {
|
||
"requires_anchor_connection": True,
|
||
"requires_terminal_type_match": True,
|
||
"requires_structured_assertion": True,
|
||
"requires_paper_provenance": True,
|
||
},
|
||
"evidence": [comparison],
|
||
"chains": [],
|
||
}
|
||
],
|
||
attempt=1,
|
||
)
|
||
|
||
assert review["groups"][0]["status"] == "PARTIAL"
|
||
assert review["groups"][0]["accepted_evidence_ids"] == []
|