feat(qa): soft gate 兼容策略(fail-closed to legacy)
This commit is contained in:
potaser
@@ -0,0 +1,109 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
from collections.abc import Iterable, Mapping
|
||||
from typing import Any
|
||||
|
||||
|
||||
LEGACY = "legacy"
|
||||
SOFT_COMPAT_V1 = "soft_compat_v1"
|
||||
_POLICIES = {LEGACY, SOFT_COMPAT_V1}
|
||||
|
||||
_ANSWER_SLOT_TYPES = {
|
||||
"biochemical_activity",
|
||||
"chromatin_modification",
|
||||
"enzyme_activity",
|
||||
"expression_level",
|
||||
"expression_pattern",
|
||||
"expression_relationship",
|
||||
"function",
|
||||
"gene_expression",
|
||||
"regulation",
|
||||
"regulatory_relationship",
|
||||
"transporter_activity",
|
||||
}
|
||||
|
||||
_SYMMETRIC_COMPATIBILITY = (
|
||||
{"pathogen", "organism_group"},
|
||||
{"compartment", "organelle"},
|
||||
{"compartment", "cellular_component"},
|
||||
{"gene_group", "biosynthetic_gene_set"},
|
||||
{"enzyme", "enzyme_activity"},
|
||||
)
|
||||
|
||||
_DOI_INTERNAL_RE = re.compile(
|
||||
r"^(10\.\d{4,9})_([A-Za-z0-9][A-Za-z0-9.;()/:+\-]*)$",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_OXFORD_DOI_INTERNAL_RE = re.compile(
|
||||
r"^(10\.1093)_([A-Za-z0-9][A-Za-z0-9.;()+\-]*)_([A-Za-z0-9][A-Za-z0-9.;()+\-]*)$",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_DOI_RE = re.compile(
|
||||
r"^10\.\d{4,9}/[A-Za-z0-9][A-Za-z0-9._;()/:+\-]*$",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
|
||||
def _type_token(value: Any) -> str:
|
||||
text = re.sub(r"[^a-z0-9]+", "_", str(value or "").strip().lower())
|
||||
return text.strip("_")
|
||||
|
||||
|
||||
def gate_policy(env: Mapping[str, str] | None = None) -> str:
|
||||
values = os.environ if env is None else env
|
||||
value = str(values.get("QA_AGENTIC_GATE_POLICY", LEGACY) or LEGACY).strip().lower()
|
||||
return value if value in _POLICIES else LEGACY
|
||||
|
||||
|
||||
def type_compatibility(
|
||||
actual_type: Any,
|
||||
requested_types: Iterable[Any],
|
||||
*,
|
||||
relationship_context: bool = False,
|
||||
) -> tuple[bool, str]:
|
||||
actual = _type_token(actual_type)
|
||||
requested = {_type_token(value) for value in requested_types}
|
||||
requested.discard("")
|
||||
requested.difference_update({"any", "unknown", "unspecified"})
|
||||
|
||||
if not requested:
|
||||
return True, "no_terminal_type_constraint"
|
||||
if actual in requested:
|
||||
return True, "exact_type_match"
|
||||
if requested.intersection(_ANSWER_SLOT_TYPES):
|
||||
return True, "answer_slot_not_endpoint_type"
|
||||
for family in _SYMMETRIC_COMPATIBILITY:
|
||||
if actual in family and family.intersection(requested):
|
||||
return True, "controlled_type_family"
|
||||
if actual == "transcription_factor" and requested.intersection({"gene", "protein"}):
|
||||
return True, "transcription_factor_endpoint"
|
||||
if (
|
||||
relationship_context
|
||||
and actual == "genetic_variant"
|
||||
and requested.intersection({"gene", "condition"})
|
||||
):
|
||||
return True, "genetic_variant_relationship_endpoint"
|
||||
return False, "terminal_type_mismatch"
|
||||
|
||||
|
||||
def doi_from_paper_id(paper_id: Any) -> str:
|
||||
value = str(paper_id or "").strip()
|
||||
oxford_match = _OXFORD_DOI_INTERNAL_RE.fullmatch(value)
|
||||
if oxford_match:
|
||||
normalized = (
|
||||
f"{oxford_match.group(1)}/{oxford_match.group(2)}/{oxford_match.group(3)}"
|
||||
)
|
||||
return normalized if _DOI_RE.fullmatch(normalized) else ""
|
||||
match = _DOI_INTERNAL_RE.fullmatch(value)
|
||||
if not match:
|
||||
return ""
|
||||
normalized = f"{match.group(1)}/{match.group(2)}"
|
||||
return normalized if _DOI_RE.fullmatch(normalized) else ""
|
||||
|
||||
|
||||
def pmcid_from_paper_id(paper_id: Any) -> str:
|
||||
value = str(paper_id or "").strip()
|
||||
match = re.fullmatch(r"(PMC\d+)(?:_[A-Za-z0-9][A-Za-z0-9.\-]*)?", value, re.IGNORECASE)
|
||||
return match.group(1).upper() if match else ""
|
||||
@@ -0,0 +1,73 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from src.rag.qa_v1_1_73 import soft_gate_policy as module
|
||||
|
||||
|
||||
def test_gate_policy_defaults_and_fails_closed_to_legacy():
|
||||
assert module.gate_policy({}) == "legacy"
|
||||
assert (
|
||||
module.gate_policy({"QA_AGENTIC_GATE_POLICY": "soft_compat_v1"})
|
||||
== "soft_compat_v1"
|
||||
)
|
||||
assert module.gate_policy({"QA_AGENTIC_GATE_POLICY": "unexpected"}) == "legacy"
|
||||
|
||||
|
||||
def test_closed_type_compatibility_map_and_answer_slots():
|
||||
compatible = [
|
||||
("pathogen", ["organism_group"], False),
|
||||
("organism_group", ["pathogen"], False),
|
||||
("compartment", ["organelle"], False),
|
||||
("compartment", ["cellular_component"], False),
|
||||
("organelle", ["compartment"], False),
|
||||
("transcription_factor", ["gene"], False),
|
||||
("transcription_factor", ["protein"], False),
|
||||
("gene_group", ["biosynthetic_gene_set"], False),
|
||||
("biosynthetic_gene_set", ["gene_group"], False),
|
||||
("enzyme", ["enzyme_activity"], False),
|
||||
("enzyme_activity", ["enzyme"], False),
|
||||
("genetic_variant", ["gene", "condition"], True),
|
||||
("gene", ["expression_pattern"], False),
|
||||
("tissue", ["regulatory_relationship"], False),
|
||||
]
|
||||
|
||||
for actual, requested, relationship_context in compatible:
|
||||
accepted, reason = module.type_compatibility(
|
||||
actual,
|
||||
requested,
|
||||
relationship_context=relationship_context,
|
||||
)
|
||||
assert accepted, (actual, requested, reason)
|
||||
|
||||
assert not module.type_compatibility("pathogen", ["phenotype"])[0]
|
||||
assert not module.type_compatibility("genetic_variant", ["gene"])[0]
|
||||
assert not module.type_compatibility("unknown_new_type", ["gene"])[0]
|
||||
|
||||
|
||||
def test_doi_from_paper_id_accepts_only_unambiguous_internal_form():
|
||||
|
||||
assert (
|
||||
module.doi_from_paper_id("10.3389_fpls.2016.00954")
|
||||
== "10.3389/fpls.2016.00954"
|
||||
)
|
||||
assert (
|
||||
module.doi_from_paper_id("10.1016_j.cj.2021.04.004")
|
||||
== "10.1016/j.cj.2021.04.004"
|
||||
)
|
||||
assert module.doi_from_paper_id("10.1093_jxb_ery350") == "10.1093/jxb/ery350"
|
||||
assert (
|
||||
module.doi_from_paper_id("10.1093_plphys_kiac335")
|
||||
== "10.1093/plphys/kiac335"
|
||||
)
|
||||
assert module.doi_from_paper_id("PMC123") == ""
|
||||
assert module.doi_from_paper_id("12345678") == ""
|
||||
assert module.doi_from_paper_id("10.3389_fpls_2016_00954") == ""
|
||||
assert module.doi_from_paper_id("10.3389/") == ""
|
||||
assert module.doi_from_paper_id("not_a_doi") == ""
|
||||
|
||||
|
||||
def test_pmcid_from_paper_id_accepts_only_leading_pmc_accession():
|
||||
|
||||
assert module.pmcid_from_paper_id("PMC10299364_ijms-24-10165") == "PMC10299364"
|
||||
assert module.pmcid_from_paper_id("pmc8508941") == "PMC8508941"
|
||||
assert module.pmcid_from_paper_id("prefix_PMC10299364") == ""
|
||||
assert module.pmcid_from_paper_id("PMC_10299364") == ""
|
||||
@@ -0,0 +1,145 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from src.rag.qa_v1_1_73.agentic_controller import AgenticRetrievalController
|
||||
from src.rag.qa_v1_1_73.schemas import EvidenceItem
|
||||
|
||||
|
||||
def _controller():
|
||||
return AgenticRetrievalController.__new__(AgenticRetrievalController)
|
||||
|
||||
|
||||
def _item(
|
||||
actual_type: str,
|
||||
*,
|
||||
source: str = "ANCHOR",
|
||||
relation: str = "regulates",
|
||||
target: str = "ANSWER",
|
||||
evidence: str = "ANCHOR regulates ANSWER under the reported condition.",
|
||||
paper_id: str = "paper-1",
|
||||
) -> EvidenceItem:
|
||||
return EvidenceItem(
|
||||
evidence_id="Q0.A1.E1",
|
||||
kb="V16/V15",
|
||||
source=source,
|
||||
relation=relation,
|
||||
target=target,
|
||||
evidence=evidence,
|
||||
paper_id=paper_id,
|
||||
raw={
|
||||
"answer_entity_type": actual_type,
|
||||
"source_type": "gene",
|
||||
"target_type": actual_type,
|
||||
"anchor_name": "ANCHOR",
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def _group(requested_types: list[str], *, question: str = "How does ANCHOR regulate ANSWER?"):
|
||||
return {
|
||||
"subquestion_id": "Q0",
|
||||
"question": question,
|
||||
"role": "required",
|
||||
"anchors": [{"mention": "ANCHOR", "canonical_name": "ANCHOR"}],
|
||||
"target_entity_types": requested_types,
|
||||
"acceptance_requirements": {
|
||||
"requires_structured_assertion": True,
|
||||
"requires_paper_provenance": True,
|
||||
"requires_anchor_connection": True,
|
||||
"requires_terminal_type_match": True,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("actual_type", "requested_types"),
|
||||
[
|
||||
("pathogen", ["organism_group"]),
|
||||
("compartment", ["organelle", "cellular_component"]),
|
||||
("transcription_factor", ["gene", "regulatory_relationship"]),
|
||||
("gene_group", ["biosynthetic_gene_set"]),
|
||||
("enzyme", ["enzyme_activity"]),
|
||||
("genetic_variant", ["gene", "condition", "phenotype"]),
|
||||
("tissue", ["gene", "expression_pattern"]),
|
||||
("gene", ["process", "pathway", "function"]),
|
||||
("genomic_region", ["gene", "chromatin_modification"]),
|
||||
],
|
||||
)
|
||||
def test_group_type_gate_is_legacy_exact_and_soft_compatible(
|
||||
monkeypatch, actual_type, requested_types
|
||||
):
|
||||
controller = _controller()
|
||||
item = _item(actual_type)
|
||||
group = _group(requested_types)
|
||||
|
||||
monkeypatch.setenv("QA_AGENTIC_GATE_POLICY", "legacy")
|
||||
assert not controller._group_evidence_satisfies(group, item)
|
||||
|
||||
monkeypatch.setenv("QA_AGENTIC_GATE_POLICY", "soft_compat_v1")
|
||||
assert controller._group_evidence_satisfies(group, item)
|
||||
|
||||
|
||||
def test_plan_evidence_and_chain_type_gates_use_same_policy(monkeypatch):
|
||||
controller = _controller()
|
||||
item = _item("transcription_factor")
|
||||
plan = {
|
||||
"anchors": [{"mention": "ANCHOR"}],
|
||||
"graph_plan": {"required_terminal_types": ["gene"]},
|
||||
"acceptance_gate": {},
|
||||
}
|
||||
chain = {
|
||||
"chain_id": "Q0.A1.G1",
|
||||
"start": "ANCHOR",
|
||||
"end": "ANSWER",
|
||||
"anchor_name": "ANCHOR",
|
||||
"answer_entity_type": "transcription_factor",
|
||||
"hops": [
|
||||
{
|
||||
"source": "ANCHOR",
|
||||
"relation": "regulates",
|
||||
"target": "ANSWER",
|
||||
"paper_id": "paper-1",
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
monkeypatch.setenv("QA_AGENTIC_GATE_POLICY", "legacy")
|
||||
assert not controller._evidence_satisfies(plan, item)
|
||||
assert not controller._chain_satisfies(plan, chain)
|
||||
|
||||
monkeypatch.setenv("QA_AGENTIC_GATE_POLICY", "soft_compat_v1")
|
||||
assert controller._evidence_satisfies(plan, item)
|
||||
assert controller._chain_satisfies(plan, chain)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("group", "item"),
|
||||
[
|
||||
(_group(["gene"]), _item("gene", paper_id="")),
|
||||
(
|
||||
_group(["gene"], question="Which genes are associated with ANCHOR?"),
|
||||
_item(
|
||||
"gene",
|
||||
source="OTHER",
|
||||
evidence="OTHER regulates ANSWER under the reported condition.",
|
||||
),
|
||||
),
|
||||
(
|
||||
_group(["gene"], question="How does ANCHOR respond to heat stress?"),
|
||||
_item(
|
||||
"gene",
|
||||
evidence="ANCHOR responds only to cold stress and freezing.",
|
||||
),
|
||||
),
|
||||
(
|
||||
_group(["gene"], question="Which genes does ANCHOR regulate?"),
|
||||
_item("gene", relation="compared_to"),
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_non_type_safety_gates_remain_active_in_both_policies(monkeypatch, group, item):
|
||||
controller = _controller()
|
||||
for policy in ("legacy", "soft_compat_v1"):
|
||||
monkeypatch.setenv("QA_AGENTIC_GATE_POLICY", policy)
|
||||
assert not controller._group_evidence_satisfies(group, item)
|
||||
Reference in New Issue
Block a user