119 lines
4.3 KiB
Python
119 lines
4.3 KiB
Python
import re
|
|
from pathlib import Path
|
|
|
|
from src.rag.qa_v1_1_73.hypothesis_extractor import HypothesisExtractor
|
|
from src.rag.qa_v1_1_73.provisional_answer import ProvisionalAnswerGenerator
|
|
from src.rag.qa_v1_1_73.query_builder import QueryBuilder
|
|
from src.rag.qa_v1_1_73.schemas import RetrievalHypothesis
|
|
|
|
|
|
class CapturingJSONLLM:
|
|
def __init__(self) -> None:
|
|
self.calls = []
|
|
|
|
def complete_json(self, system: str, user: str, **kwargs):
|
|
self.calls.append((system, user, kwargs))
|
|
return {
|
|
"question_type": "function",
|
|
"direct_question_entities": [
|
|
{
|
|
"mention": "WRKY基因家族",
|
|
"canonical_candidate": "WRKY gene family",
|
|
"type": "gene_family",
|
|
"confidence": 0.99,
|
|
}
|
|
],
|
|
"concept_queries": [
|
|
"WRKY基因家族 功能 调控机制",
|
|
"WRKY gene family functions regulatory mechanisms",
|
|
],
|
|
}
|
|
|
|
|
|
class CapturingTextLLM:
|
|
def __init__(self) -> None:
|
|
self.calls = []
|
|
|
|
def complete(self, system: str, user: str, **kwargs) -> str:
|
|
self.calls.append((system, user, kwargs))
|
|
return "临时答案"
|
|
|
|
|
|
class StubDomain:
|
|
def expanded_relations(self, intents):
|
|
return ["involved_in"] if "function" in intents else []
|
|
|
|
def evidence_query_terms(self, entities):
|
|
return []
|
|
|
|
|
|
def test_hypothesis_planner_builds_bilingual_queries_in_one_llm_call() -> None:
|
|
llm = CapturingJSONLLM()
|
|
hypothesis = HypothesisExtractor(llm).extract(
|
|
"WRKY基因家族的功能有哪些？",
|
|
"WRKY proteins participate in stress responses.",
|
|
)
|
|
|
|
prompt = "\n".join(llm.calls[0][:2]).lower()
|
|
assert len(llm.calls) == 1
|
|
assert "preserve each entity `mention` exactly" in prompt
|
|
assert "same planner call" in prompt
|
|
assert "at least one concept query in the original question language" in prompt
|
|
assert "english scientific retrieval variants" in prompt
|
|
assert "retrieval hypotheses only" in prompt
|
|
assert hypothesis.direct_question_entities[0].mention == "WRKY基因家族"
|
|
assert hypothesis.direct_question_entities[0].canonical_candidate == "WRKY gene family"
|
|
|
|
|
|
def test_provisional_answer_uses_question_language_without_translating_identifiers() -> None:
|
|
llm = CapturingTextLLM()
|
|
ProvisionalAnswerGenerator(llm).generate("WRKY基因家族的功能有哪些？")
|
|
|
|
system = llm.calls[0][0].lower()
|
|
assert "same language as the question" in system
|
|
assert "gene symbols and latin species names unchanged" in system
|
|
|
|
|
|
def test_query_builder_preserves_original_question_and_bilingual_variants() -> None:
|
|
hypothesis = RetrievalHypothesis(
|
|
relation_intents=["function"],
|
|
concept_queries=[
|
|
"WRKY基因家族 功能 调控机制",
|
|
"WRKY gene family functions regulatory mechanisms",
|
|
],
|
|
)
|
|
|
|
queries = QueryBuilder(StubDomain()).build(
|
|
"WRKY基因家族的功能有哪些？",
|
|
"WRKY家族参与多种植物过程。",
|
|
hypothesis,
|
|
)
|
|
|
|
assert queries["question_lane"][0] == "WRKY基因家族的功能有哪些？"
|
|
assert "WRKY基因家族 功能 调控机制" in queries["hypothesis_lane"]
|
|
assert "WRKY gene family functions regulatory mechanisms" in queries["hypothesis_lane"]
|
|
|
|
|
|
def test_7925_startup_script_is_local_and_does_not_print_secrets() -> None:
|
|
script_path = Path("scripts/start_qa_v1_1_73_7925.sh")
|
|
assert script_path.is_file()
|
|
script = script_path.read_text(encoding="utf-8")
|
|
|
|
assert "source .env" in script
|
|
assert script.index("source .env") < script.index("python -m uvicorn")
|
|
assert "--host 127.0.0.1" in script
|
|
assert "--port 7925" in script
|
|
assert "QA_PAPER_METADATA_REGISTRY" in script
|
|
assert "V15_QDRANT_COLLECTION" in script
|
|
assert "bolt://127.0.0.1:7687" in script
|
|
assert 'QA_V1_GRAPH_SCHEMA="${QA_V1_GRAPH_SCHEMA:-legacy}"' in script
|
|
assert (
|
|
'QA_V1_ENABLE_PROVISIONAL_DRAFT="${QA_V1_ENABLE_PROVISIONAL_DRAFT:-0}"'
|
|
in script
|
|
)
|
|
assert 'QA_V1_MAX_ROUNDS="${QA_V1_MAX_ROUNDS:-1}"' in script
|
|
assert "plant_srt_direct_password" not in script
|
|
assert "QA_V1_LEGACY_NEO4J_PASSWORD:-password" in script
|
|
assert "set -x" not in script
|
|
assert not re.search(r"echo[^\n]*(?:key|password|secret)", script, re.IGNORECASE)
|