feat(rag): honest deflection gate with amber UI state and alternative-question chips
This commit is contained in:
@@ -0,0 +1,322 @@
|
||||
"""Unit: the honesty gate (A8) — boundary, prompts, and suggestion chips.
|
||||
|
||||
Pure gate logic runs against fake retriever output (``RetrievedChunk``
|
||||
rows from a fake retriever) with no Postgres and no network. The
|
||||
endpoint-level tests drive ``POST /api/chat`` with the retriever, the DB
|
||||
session, and the LLM all faked, so the whole deflection contract
|
||||
(prompt → deltas → done event → query_log) is verified without a stack.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import uuid
|
||||
from collections.abc import Iterator
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.api import chat as chat_api
|
||||
from app.config import Settings
|
||||
from app.main import app as fastapi_app
|
||||
from app.models import Document, QueryLog
|
||||
from app.rag.retriever import RetrievedChunk, weak_hit_titles
|
||||
from app.rag.suggestions import MAX_SUGGESTIONS, derive_suggestions
|
||||
|
||||
ANSWER = "I haven't done anything like that — try one of these instead!"
|
||||
|
||||
|
||||
def _settings(threshold: float = 0.30) -> Settings:
|
||||
return Settings(
|
||||
_env_file=None, # pyright: ignore[reportCallIssue]
|
||||
relevance_threshold=threshold,
|
||||
)
|
||||
|
||||
|
||||
def _doc(title: str, content: str) -> Document:
|
||||
return Document(
|
||||
id=uuid.uuid4(),
|
||||
source="Homelab",
|
||||
path=f"{title.lower().replace(' ', '-')}.md",
|
||||
full_path="/tmp/doc.md",
|
||||
title=title,
|
||||
content=content,
|
||||
content_hash="0" * 64,
|
||||
)
|
||||
|
||||
|
||||
def _chunk(doc: Document, score: float) -> RetrievedChunk:
|
||||
return RetrievedChunk(
|
||||
chunk_id=uuid.uuid4(),
|
||||
position=0,
|
||||
content=doc.content[:32],
|
||||
score=score,
|
||||
document=doc,
|
||||
)
|
||||
|
||||
|
||||
# ---------- gate boundary (fake retriever rows, no LLM) ----------
|
||||
|
||||
|
||||
def test_gate_boundary_score_at_threshold_answers() -> None:
|
||||
"""Score exactly at the threshold ⇒ HIGH (the gate is strict <)."""
|
||||
doc = _doc("Kubernetes Homelab Cluster", "TALOS_DOC_CONTENT")
|
||||
plan = chat_api.plan_turn([_chunk(doc, 0.30)], _settings(threshold=0.30))
|
||||
assert plan.deflected is False
|
||||
assert plan.top_score == pytest.approx(0.30)
|
||||
assert "<relevance>HIGH</relevance>" in plan.system_prompt
|
||||
assert "DEFLECT_MODE" not in plan.system_prompt
|
||||
assert "TALOS_DOC_CONTENT" in plan.system_prompt
|
||||
assert plan.suggestions == []
|
||||
|
||||
|
||||
def test_gate_boundary_just_below_threshold_deflects() -> None:
|
||||
doc = _doc("Kubernetes Homelab Cluster", "TALOS_DOC_CONTENT")
|
||||
plan = chat_api.plan_turn([_chunk(doc, 0.2999)], _settings(threshold=0.30))
|
||||
assert plan.deflected is True
|
||||
assert plan.top_score == pytest.approx(0.2999)
|
||||
assert "<relevance>LOW</relevance>" in plan.system_prompt
|
||||
assert "DEFLECT_MODE" in plan.system_prompt
|
||||
# Titles only: the full document content must never reach the LLM.
|
||||
assert "TALOS_DOC_CONTENT" not in plan.system_prompt
|
||||
assert "Kubernetes Homelab Cluster" in plan.system_prompt
|
||||
|
||||
|
||||
def test_gate_is_env_tunable_via_settings() -> None:
|
||||
doc = _doc("Backup Strategy", "BACKUP_DOC_CONTENT")
|
||||
hits = [_chunk(doc, 0.30)]
|
||||
assert chat_api.plan_turn(hits, _settings(threshold=0.35)).deflected is True
|
||||
assert chat_api.plan_turn(hits, _settings(threshold=0.25)).deflected is False
|
||||
|
||||
|
||||
def test_gate_zero_chunks_deflects_with_fallback_chips() -> None:
|
||||
plan = chat_api.plan_turn([], _settings())
|
||||
assert plan.deflected is True
|
||||
assert plan.top_score == 0.0
|
||||
assert "nothing close at all" in plan.system_prompt
|
||||
# No weak hits ⇒ onboarding fallback fills the chips.
|
||||
assert 2 <= len(plan.suggestions) <= MAX_SUGGESTIONS
|
||||
|
||||
|
||||
# ---------- prompt content (LOW vs HIGH) ----------
|
||||
|
||||
|
||||
def test_low_prompt_has_titles_only_no_content() -> None:
|
||||
a = _doc("Kubernetes Homelab Cluster", "ALPHA_DOC_CONTENT")
|
||||
b = _doc("Backup Strategy", "BETA_DOC_CONTENT")
|
||||
plan = chat_api.plan_turn([_chunk(b, 0.10), _chunk(a, 0.20)], _settings())
|
||||
prompt = plan.system_prompt
|
||||
assert "<relevance>LOW</relevance>" in prompt
|
||||
assert "DEFLECT_MODE" in prompt
|
||||
assert "HONESTY GATE" in prompt # the LOW rule is what the model follows
|
||||
assert "- Kubernetes Homelab Cluster" in prompt
|
||||
assert "- Backup Strategy" in prompt
|
||||
assert "ALPHA_DOC_CONTENT" not in prompt
|
||||
assert "BETA_DOC_CONTENT" not in prompt
|
||||
assert "<documents>" not in prompt
|
||||
|
||||
|
||||
def test_high_path_unaffected() -> None:
|
||||
a = _doc("Kubernetes Homelab Cluster", "ALPHA_DOC_CONTENT")
|
||||
b = _doc("Backup Strategy", "BETA_DOC_CONTENT")
|
||||
plan = chat_api.plan_turn([_chunk(a, 0.90), _chunk(b, 0.40)], _settings())
|
||||
assert plan.deflected is False
|
||||
assert plan.suggestions == []
|
||||
assert "<relevance>HIGH</relevance>" in plan.system_prompt
|
||||
assert "DEFLECT_MODE" not in plan.system_prompt
|
||||
assert "ALPHA_DOC_CONTENT" in plan.system_prompt
|
||||
assert "BETA_DOC_CONTENT" in plan.system_prompt
|
||||
assert [d.title for d in plan.docs] == ["Kubernetes Homelab Cluster", "Backup Strategy"]
|
||||
|
||||
|
||||
# ---------- weak_hit_titles (fake retriever mapping) ----------
|
||||
|
||||
|
||||
def test_weak_hit_titles_dedupe_and_rank_by_best_score() -> None:
|
||||
a = _doc("Kubernetes Homelab Cluster", "AAA")
|
||||
b = _doc("Backup Strategy", "BBB")
|
||||
chunks = [_chunk(b, 0.5), _chunk(a, 0.2), _chunk(a, 0.9)]
|
||||
assert weak_hit_titles(chunks) == ["Kubernetes Homelab Cluster", "Backup Strategy"]
|
||||
|
||||
|
||||
# ---------- suggestions derivation ----------
|
||||
|
||||
|
||||
def test_suggestions_derived_from_titles_in_order() -> None:
|
||||
got = derive_suggestions(
|
||||
["Kubernetes Homelab Cluster", "Backup Strategy", "Deploying a New Service"],
|
||||
fallback=["should not appear"],
|
||||
)
|
||||
assert len(got) == 3
|
||||
assert all(s.strip() for s in got)
|
||||
assert "Kubernetes Homelab Cluster" in got[0]
|
||||
assert "Backup Strategy" in got[1]
|
||||
assert "Deploying a New Service" in got[2]
|
||||
|
||||
|
||||
def test_suggestions_capped_at_three() -> None:
|
||||
got = derive_suggestions([f"Title {i}" for i in range(6)], fallback=["F"])
|
||||
assert len(got) == MAX_SUGGESTIONS == 3
|
||||
|
||||
|
||||
def test_suggestions_top_up_from_fallback_when_titles_thin() -> None:
|
||||
got = derive_suggestions(
|
||||
["Backup Strategy"],
|
||||
fallback=["How is my Kubernetes cluster set up?", "What's my backup strategy?"],
|
||||
)
|
||||
assert len(got) == 3
|
||||
assert got[0] == "What's in your notes about Backup Strategy?"
|
||||
assert got[1] == "How is my Kubernetes cluster set up?"
|
||||
|
||||
|
||||
def test_suggestions_dedupes_and_ignores_blank() -> None:
|
||||
got = derive_suggestions(
|
||||
["Backup Strategy", "backup strategy", " "],
|
||||
fallback=["What's my backup strategy?", " "],
|
||||
)
|
||||
# "backup strategy" is a case-insensitive dup; blank title/fallback are
|
||||
# skipped — including ones that only look blank after formatting. Only
|
||||
# two valid items remain, and the list never pads with junk.
|
||||
assert got == [
|
||||
"What's in your notes about Backup Strategy?",
|
||||
"What's my backup strategy?",
|
||||
]
|
||||
assert all("about ?" not in s and s == s.strip() for s in got)
|
||||
|
||||
|
||||
def test_suggestions_empty_input_yields_fallback_only() -> None:
|
||||
assert derive_suggestions([], fallback=[]) == []
|
||||
got = derive_suggestions([], fallback=["One?", "Two?"])
|
||||
assert got == ["One?", "Two?"]
|
||||
|
||||
|
||||
# ---------- endpoint-level gate (fake retriever + fake LLM + fake session) ----------
|
||||
|
||||
|
||||
class _CannedLLM:
|
||||
"""Records the messages it is given; streams a canned answer."""
|
||||
|
||||
def __init__(self, answer: str = ANSWER) -> None:
|
||||
self.settings = Settings(_env_file=None) # pyright: ignore[reportCallIssue]
|
||||
self.embed_batches = 0
|
||||
self.answer = answer
|
||||
self.seen: list[list[dict[str, str]]] = []
|
||||
|
||||
async def embed_one(self, _text: str) -> list[float]:
|
||||
return [0.0] * 768
|
||||
|
||||
async def chat_stream(self, messages: list[dict[str, str]]):
|
||||
self.seen.append(messages)
|
||||
for i in range(0, len(self.answer), 12):
|
||||
yield self.answer[i : i + 12]
|
||||
|
||||
|
||||
class _FakeSession:
|
||||
"""Stands in for the DB session: records the QueryLog row it is given."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.added: list[Any] = []
|
||||
self.commits = 0
|
||||
|
||||
def add(self, obj: Any) -> None:
|
||||
self.added.append(obj)
|
||||
|
||||
def commit(self) -> None:
|
||||
self.commits += 1
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def gate_env(monkeypatch: pytest.MonkeyPatch) -> Iterator[tuple[_FakeSession, _CannedLLM]]:
|
||||
"""``POST /api/chat`` with retriever, session, and LLM all faked."""
|
||||
monkeypatch.setattr(chat_api, "db_available", lambda: True)
|
||||
session = _FakeSession()
|
||||
llm = _CannedLLM()
|
||||
monkeypatch.setitem(fastapi_app.dependency_overrides, chat_api.get_db, lambda: session)
|
||||
monkeypatch.setitem(fastapi_app.dependency_overrides, chat_api.get_llm, lambda: llm)
|
||||
yield session, llm
|
||||
fastapi_app.dependency_overrides.clear()
|
||||
|
||||
|
||||
def _ask(client: TestClient, message: str) -> list[dict[str, Any]]:
|
||||
with client.stream("POST", "/api/chat", json={"message": message}) as r:
|
||||
assert r.status_code == 200
|
||||
frames: list[dict[str, Any]] = []
|
||||
buf = ""
|
||||
for part in r.iter_text():
|
||||
buf += part
|
||||
while "\n\n" in buf:
|
||||
frame, buf = buf.split("\n\n", 1)
|
||||
frame = frame.strip()
|
||||
if frame.startswith("data:"):
|
||||
frames.append(json.loads(frame.removeprefix("data:").strip()))
|
||||
assert buf.strip() == ""
|
||||
return frames
|
||||
|
||||
|
||||
def _fake_retriever(chunks: list[RetrievedChunk]) -> Any:
|
||||
def retrieve(_db: Any, _vec: list[float]) -> list[RetrievedChunk]:
|
||||
return chunks
|
||||
|
||||
return retrieve
|
||||
|
||||
|
||||
def test_endpoint_just_below_threshold_deflects(
|
||||
client: TestClient,
|
||||
gate_env: tuple[_FakeSession, _CannedLLM],
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
session, llm = gate_env
|
||||
doc = _doc("Deploying a New Service", "DOC_CONTENT_NEVER_SENT")
|
||||
monkeypatch.setattr(chat_api, "retrieve", _fake_retriever([_chunk(doc, 0.2999)]))
|
||||
|
||||
frames = _ask(client, "How do I bake sourdough bread?")
|
||||
|
||||
deltas = [f for f in frames if f["type"] == "delta"]
|
||||
assert "".join(d["text"] for d in deltas) == ANSWER # the LLM was still called
|
||||
done = frames[-1]
|
||||
assert done["type"] == "done"
|
||||
assert done["deflected"] is True
|
||||
assert 2 <= len(done["suggestions"]) <= MAX_SUGGESTIONS # title chip + fallback
|
||||
assert all(s.strip() for s in done["suggestions"])
|
||||
assert any("Deploying a New Service" in s for s in done["suggestions"])
|
||||
|
||||
# The LLM saw the LOW prompt: DEFLECT_MODE + titles, never doc content.
|
||||
(system, user) = llm.seen[0][0], llm.seen[0][1]
|
||||
assert user["content"] == "How do I bake sourdough bread?"
|
||||
assert "DEFLECT_MODE" in system["content"]
|
||||
assert "DOC_CONTENT_NEVER_SENT" not in system["content"]
|
||||
|
||||
# Durable record: deflected + the weak score.
|
||||
(row,) = session.added
|
||||
assert isinstance(row, QueryLog)
|
||||
assert row.deflected is True
|
||||
assert row.top_score == pytest.approx(0.2999)
|
||||
assert session.commits == 1
|
||||
|
||||
|
||||
def test_endpoint_score_at_threshold_answers(
|
||||
client: TestClient,
|
||||
gate_env: tuple[_FakeSession, _CannedLLM],
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
session, llm = gate_env
|
||||
doc = _doc("Kubernetes Homelab Cluster", "TALOS_DOC_SENT")
|
||||
monkeypatch.setattr(chat_api, "retrieve", _fake_retriever([_chunk(doc, 0.30)]))
|
||||
|
||||
frames = _ask(client, "How is my Kubernetes cluster set up?")
|
||||
|
||||
done = frames[-1]
|
||||
assert done["type"] == "done"
|
||||
assert done["deflected"] is False
|
||||
assert done["suggestions"] == []
|
||||
assert done["sources"] and done["sources"][0]["title"] == "Kubernetes Homelab Cluster"
|
||||
|
||||
(system, _user) = llm.seen[0][0], llm.seen[0][1]
|
||||
assert "<relevance>HIGH</relevance>" in system["content"]
|
||||
assert "DEFLECT_MODE" not in system["content"]
|
||||
assert "TALOS_DOC_SENT" in system["content"]
|
||||
|
||||
(row,) = session.added
|
||||
assert isinstance(row, QueryLog)
|
||||
assert row.deflected is False
|
||||
assert row.top_score == pytest.approx(0.30)
|
||||
Reference in New Issue
Block a user