Coverage for src/lilbee/wiki/quality.py: 100%
81 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-14 11:46 +0000
1"""Faithfulness scoring and drift heuristics for wiki page bodies.
3Holds the deterministic body-vs-source cosine score, the title/body
4coherence pre-check that gates a page on a structurally valid heading,
5and the unified-diff helpers used when an existing page's content
6changed by more than the configured threshold (drift detection).
7"""
9from __future__ import annotations
11import difflib
12import logging
14import numpy as np
16from lilbee.app.services import get_services
17from lilbee.core.config import Config
18from lilbee.core.text import clean_label_for_display, is_valid_label
19from lilbee.core.vectors import Vector
20from lilbee.data.store import SearchChunk
21from lilbee.wiki.citations import strip_citation_block
23log = logging.getLogger(__name__)
25_MAX_DIFF_PREVIEW_LINES = 20 # lines of unified diff shown in drift warnings
28def content_change_ratio(old_text: str, new_text: str) -> float:
29 """Fraction of lines that changed between two texts (0.0 = identical, 1.0 = total rewrite)."""
30 old_lines = old_text.splitlines()
31 new_lines = new_text.splitlines()
32 if not old_lines and not new_lines:
33 return 0.0
34 total = max(len(old_lines), len(new_lines))
35 matcher = difflib.SequenceMatcher(None, old_lines, new_lines)
36 changed = total - sum(block.size for block in matcher.get_matching_blocks())
37 return changed / total
40def diff_summary(old_text: str, new_text: str) -> str:
41 """Human-readable unified diff summary (first 20 diff lines)."""
42 diff = difflib.unified_diff(
43 old_text.splitlines(),
44 new_text.splitlines(),
45 lineterm="",
46 fromfile="old",
47 tofile="new",
48 )
49 lines = list(diff)
50 if len(lines) > _MAX_DIFF_PREVIEW_LINES:
51 extra = len(lines) - _MAX_DIFF_PREVIEW_LINES
52 return "\n".join(lines[:_MAX_DIFF_PREVIEW_LINES]) + f"\n... ({extra} more lines)"
53 return "\n".join(lines)
56def _title_content_coherence(wiki_text: str, label: str) -> bool:
57 """Deterministic pre-check: title and body must reference the concept.
59 The LLM faithfulness score evaluates whether the prose reflects
60 the source chunks but does not penalize structural noise in the
61 title (bb-8b7s: ``| | designer`` passed at 0.90 because the body
62 was coherent). This pre-check asserts three invariants:
64 1. The first ``# `` heading must be a sanity-valid label per
65 :func:`is_valid_label`. A heading like ``| | designer`` fails
66 the structural-char gate even though it contains the cleaned
67 display name as a substring.
68 2. The cleaned display name must appear in the heading as a
69 case-insensitive substring. Covers LLM drift where the
70 heading names a different concept than requested.
71 3. The body must mention the display name at least once outside
72 the heading. Covers the "LLM talked about something adjacent
73 but never named the concept" regression.
75 Returns True when all three hold, False otherwise.
76 """
77 display = clean_label_for_display(label).lower()
78 if not display:
79 return False
80 heading: str | None = None
81 body_parts: list[str] = []
82 for line in wiki_text.splitlines():
83 if heading is None and line.startswith("# "):
84 heading = line[2:].strip()
85 continue
86 body_parts.append(line)
87 if heading is None:
88 return False
89 if not is_valid_label(heading):
90 return False
91 if display not in heading.lower():
92 return False
93 body = "\n".join(body_parts).lower()
94 return display in body
97def _mean_vector(vectors: list[list[float]]) -> list[float]:
98 """Compute the element-wise mean of a non-empty vector list.
100 Empty input returns an empty list; callers must check before any
101 downstream dot-product so we do not leak a shape mismatch.
103 Routes through numpy so the inner loop runs in C: for the typical
104 ``D=768``, ``N=10`` case this cuts per-call cost from ~8k Python
105 ops to a single SIMD-backed reduction.
106 """
107 if not vectors:
108 return []
109 result: list[float] = np.asarray(vectors, dtype=np.float32).mean(axis=0).tolist()
110 return result
113def _embedding_faithfulness_score(
114 body_vec: Vector,
115 source_vectors: list[list[float]],
116) -> float:
117 """Cosine-similarity score between the body and the mean source vector.
119 Assumes L2-normalized vectors (both the embedder and the store
120 return normalized vectors); cosine reduces to a dot product.
121 Falls through to :func:`cosine_sim` so a non-normalized vector
122 does not silently produce an out-of-range value. Result is
123 clamped at zero because a negative cosine means the body vector
124 points the other way from the mean of the sources: treat that
125 the same as uncorrelated for threshold purposes.
127 Returns 0.0 on a dimension mismatch between the body vector and
128 the source-vector mean. That is not expected in production (the
129 embedder and the chunk vectors come from the same model), but a
130 stub-driven test may hand in off-shape vectors and crashing the
131 whole pipeline on the shape-check hides the real assertion.
132 """
133 from lilbee.data.store import cosine_sim
135 mean_vec = _mean_vector(source_vectors)
136 if len(mean_vec) == 0 or len(body_vec) == 0:
137 return 0.0
138 if len(mean_vec) != len(body_vec):
139 log.warning(
140 "Body vector dim %d does not match source vector dim %d; scoring 0.0",
141 len(body_vec),
142 len(mean_vec),
143 )
144 return 0.0
145 return max(0.0, cosine_sim(body_vec, mean_vec))
148def check_faithfulness(
149 chunks: list[SearchChunk],
150 wiki_text: str,
151 label: str,
152 config: Config | None = None,
153) -> float:
154 """Score the wiki body's similarity to its source chunks, 0.0 on failure.
156 Faithfulness is a deterministic cosine-similarity score between
157 the page body and the mean of its source chunk vectors. The B3
158 title/body coherence pre-check still runs first as a hard gate: a
159 garbage H1 returns 0.0 regardless of embedding similarity, so
160 structurally broken pages route to drafts even when the prose
161 happens to be coherent.
163 ``chunks`` carries ``.vector`` populated by LanceDB (see
164 ``SearchChunk`` in ``lilbee.data.store``), so no extra embedder call is
165 needed for the source side. The body is embedded once via the
166 shared services embedder. Any exception in the embedder (model
167 missing, network issue, invalid config) is caught and reported as
168 0.0 so a single faulty page drops to drafts instead of aborting
169 the whole build.
170 """
171 if not _title_content_coherence(wiki_text, label):
172 log.info(
173 "Faithfulness title/body coherence failed for %r; scoring 0.0",
174 label,
175 )
176 return 0.0
177 source_vectors = [c.vector for c in chunks if c.vector]
178 if not source_vectors:
179 log.warning("No source vectors for %s; scoring 0.0", label)
180 return 0.0
182 # Strip the citation block so we embed only the body prose.
183 # Frontmatter is not attached yet at this point: build_frontmatter
184 # runs after scoring. strip_citation_block is idempotent on a body
185 # whose trailer has not been rendered yet.
186 body_text = strip_citation_block(wiki_text).strip()
187 if not body_text:
188 log.warning("Empty body for %s; scoring 0.0", label)
189 return 0.0
191 try:
192 body_vectors = get_services().embedder.embed_batch([body_text])
193 except Exception as exc:
194 log.warning("Body embedding failed for %s: %s", label, exc)
195 return 0.0
196 if not body_vectors:
197 return 0.0
198 return _embedding_faithfulness_score(body_vectors[0], source_vectors)