Spaces:
Paused
Paused
File size: 2,379 Bytes
8c1b9fe | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 | from __future__ import annotations
import pytest
from auralynq.agent.cache import SemanticCache
from auralynq.agent.runner import answer_question, stream_answer_question
from auralynq.pipeline import build_index
@pytest.fixture
def indexed(corpus_dir):
build_index(corpus_dir)
# caches are process-global; clear between tests
from auralynq.agent import runner
runner._CACHE.clear()
return corpus_dir
def test_answer_simple_question_is_grounded(indexed):
res = answer_question("What is the capital of France?")
assert res.answer
assert res.citations, "grounded answer must have citations"
assert res.route in ("fast", "hybrid", "graph")
# every citation marker maps to a real source/locator
for c in res.citations:
assert c["source"] and "marker" in c
def test_relational_question_routes_to_graph_or_hybrid(indexed):
res = answer_question("How is Paris connected to France and Europe through their relationship?")
assert res.route in ("graph", "hybrid")
# path evidence should be populated when PathRAG runs
if res.route in ("graph", "hybrid"):
assert isinstance(res.path_evidence, list)
def test_trace_has_all_core_nodes(indexed):
res = answer_question("What is PathRAG?")
names = {span["name"] for span in res.trace}
assert {
"planner",
"router",
"context_fusion",
"synthesizer",
"self_check",
"citation_validator",
} <= names
def test_citation_validator_strips_dangling_markers(indexed):
res = answer_question("What is the capital of France?")
import re
markers = {int(m) for m in re.findall(r"\[(\d+)\]", res.answer)}
assert all(1 <= m <= max(len(res.citations), 1) for m in markers) or not markers
def test_semantic_cache_hit():
cache = SemanticCache(threshold=0.9)
cache.store("what is pathrag", "PathRAG is graph retrieval. [1]", [{"marker": 1}])
hit = cache.lookup("what is pathrag")
assert hit is not None
assert hit[0].startswith("PathRAG")
def test_streaming_yields_tokens_and_final(indexed):
events = list(stream_answer_question("What is the capital of France?"))
types = [e["type"] for e in events]
assert types[0] == "meta"
assert "token" in types
assert types[-1] == "final"
final = events[-1]
assert "answer" in final and "citations" in final
|