Closes #114, #115, #116, #117. - Deterministic, offline, dependency-free char-ngram embedding backend + config section (semantic.enabled default false, min_similarity, backend). - Committed embedding index (data/kb_embedding_index.json) mapping vectors to real KB entry ids only; built by scripts/build_embedding_index.py, loaded once/cached. - Resolver falls back to the nearest entry on a deterministic miss when enabled, accepting >= min_similarity and emitting SC0020; NSFW-gated neighbours are skipped. Disabled => byte-identical to before. - Guarantee tests: deterministic lookup always wins (exact hit never queries the index), fallback emits only KB tags (no invention), identical input => identical output. SC0020 registered. Co-authored-by: kunail0804 <kunail0804@gmail.com> Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Crespin Alexandre
co-authored by
kunail0804
Claude Opus 4.8
kunail0804
Claude Opus 4.8
parent
50e7b0a8a4
commit
87d9525176
@@ -7,6 +7,16 @@ All notable changes to Scene Compiler are documented here. The project follows
|
||||
|
||||
### Added
|
||||
|
||||
- **Semantic & embedding fallback (epic #34).** An **opt-in** nearest-neighbour
|
||||
fallback for the Resolver, off by default (`semantic.enabled`). When enabled, a
|
||||
concept that misses both deterministic lookup and the head-noun reduction falls
|
||||
back to its nearest Knowledge Base entry in a committed embedding index
|
||||
(`data/kb_embedding_index.json`), accepted only above `semantic.min_similarity`
|
||||
and emitting `SC0020`. The default embedding backend is a deterministic, offline,
|
||||
dependency-free character-n-gram model. Guarantees: deterministic lookup always
|
||||
wins, the fallback can only return an existing KB entry (never invents a tag),
|
||||
and identical input yields identical output. Config surfaced on the Configuration
|
||||
node. Build the index with `scripts/build_embedding_index.py`.
|
||||
- **Automatic Knowledge Base builder & validation tooling (epic #35).** A single,
|
||||
fully-automated pipeline (`scripts/build_knowledge_base.py`, zero human input)
|
||||
regenerates the generated KB deterministically from committed snapshots: generate
|
||||
|
||||
@@ -7,3 +7,5 @@ recursive-include prompts *.md
|
||||
recursive-include config *.json
|
||||
recursive-include examples *.json *.md
|
||||
recursive-include docs *.md
|
||||
# The embedding index is a runtime artifact for the optional semantic fallback.
|
||||
include data/kb_embedding_index.json
|
||||
|
||||
@@ -126,6 +126,29 @@ class PromptBuilderConfig:
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SemanticConfig:
|
||||
"""Semantic/embedding fallback options (epic #34).
|
||||
|
||||
Off by default: when ``enabled`` is false the Resolver behaves exactly as
|
||||
before. When enabled, an unknown concept that misses deterministic lookup may
|
||||
fall back to the nearest Knowledge Base entry in the embedding index, accepted
|
||||
only when similarity is at least ``min_similarity``. The default backend is a
|
||||
deterministic, offline character-n-gram embedding, so results are reproducible.
|
||||
"""
|
||||
|
||||
enabled: bool = False
|
||||
min_similarity: float = 0.5
|
||||
backend: str = "char_ngram"
|
||||
|
||||
def to_json(self) -> dict[str, Any]:
|
||||
return {
|
||||
"enabled": self.enabled,
|
||||
"min_similarity": self.min_similarity,
|
||||
"backend": self.backend,
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class DebugConfig:
|
||||
"""Debug/logging options (§23.2)."""
|
||||
@@ -146,6 +169,7 @@ class Config:
|
||||
resolver: ResolverConfig = ResolverConfig()
|
||||
validator: ValidatorConfig = ValidatorConfig()
|
||||
prompt_builder: PromptBuilderConfig = PromptBuilderConfig()
|
||||
semantic: SemanticConfig = SemanticConfig()
|
||||
debug: DebugConfig = DebugConfig()
|
||||
|
||||
@classmethod
|
||||
@@ -169,6 +193,7 @@ class Config:
|
||||
resolver=_section_from_json(ResolverConfig, data.get("resolver")),
|
||||
validator=_section_from_json(ValidatorConfig, data.get("validator")),
|
||||
prompt_builder=_section_from_json(PromptBuilderConfig, data.get("prompt_builder")),
|
||||
semantic=_section_from_json(SemanticConfig, data.get("semantic")),
|
||||
debug=_section_from_json(DebugConfig, data.get("debug")),
|
||||
)
|
||||
|
||||
@@ -179,6 +204,7 @@ class Config:
|
||||
"resolver": self.resolver.to_json(),
|
||||
"validator": self.validator.to_json(),
|
||||
"prompt_builder": self.prompt_builder.to_json(),
|
||||
"semantic": self.semantic.to_json(),
|
||||
"debug": self.debug.to_json(),
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,85 @@
|
||||
"""Deterministic, offline embedding backend (MASTER_SPEC §30.2, epic #34).
|
||||
|
||||
The semantic fallback needs to compare an unknown concept against Knowledge Base
|
||||
entries **without any network call, extra dependency, or nondeterminism**. This
|
||||
module provides a small ``embed(texts) -> vectors`` abstraction and a default
|
||||
character-n-gram implementation that satisfies those constraints:
|
||||
|
||||
- **Offline & dependency-free** — pure standard library; no model download.
|
||||
- **Deterministic** — the same text always maps to the same vector (no random
|
||||
seed, no hardware-dependent floats beyond IEEE arithmetic).
|
||||
- **Sparse** — a vector is a mapping of character n-gram → weight, so the committed
|
||||
Knowledge Base index (epic #34, #115) stays small.
|
||||
|
||||
Because the index embeds each entry's canonical text **and its aliases**, a query
|
||||
matches an entry when it is lexically close to any of the entry's surface forms
|
||||
(e.g. ``stockings`` matches the ``thighhighs`` entry via its ``stockings`` alias).
|
||||
Retrieval therefore never invents a tag — it can only return an existing entry.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
import unicodedata
|
||||
from typing import Protocol
|
||||
|
||||
# A sparse embedding: character n-gram -> L2-normalized weight.
|
||||
SparseVector = dict[str, float]
|
||||
|
||||
DEFAULT_BACKEND = "char_ngram"
|
||||
_NGRAM_SIZE = 3
|
||||
_PAD = "\x02" # a boundary marker so prefixes/suffixes get their own n-grams
|
||||
|
||||
|
||||
class EmbeddingBackend(Protocol):
|
||||
"""A text embedding backend: turn texts into comparable sparse vectors."""
|
||||
|
||||
def embed(self, texts: list[str]) -> list[SparseVector]: ...
|
||||
|
||||
|
||||
def _normalize_text(text: str) -> str:
|
||||
"""Lowercase, NFKC-normalize, and collapse whitespace/underscores."""
|
||||
folded = unicodedata.normalize("NFKC", text).replace("_", " ").lower()
|
||||
return " ".join(folded.split())
|
||||
|
||||
|
||||
class CharNGramEmbeddingBackend:
|
||||
"""Character-n-gram bag-of-features embedding (deterministic, offline)."""
|
||||
|
||||
def __init__(self, ngram_size: int = _NGRAM_SIZE) -> None:
|
||||
self._n = ngram_size
|
||||
|
||||
def embed(self, texts: list[str]) -> list[SparseVector]:
|
||||
return [self._embed_one(text) for text in texts]
|
||||
|
||||
def _embed_one(self, text: str) -> SparseVector:
|
||||
normalized = _normalize_text(text)
|
||||
if not normalized:
|
||||
return {}
|
||||
padded = _PAD + normalized + _PAD
|
||||
counts: SparseVector = {}
|
||||
for i in range(len(padded) - self._n + 1):
|
||||
gram = padded[i : i + self._n]
|
||||
counts[gram] = counts.get(gram, 0.0) + 1.0
|
||||
norm = math.sqrt(sum(weight * weight for weight in counts.values()))
|
||||
return {gram: weight / norm for gram, weight in counts.items()}
|
||||
|
||||
|
||||
def get_backend(backend_id: str = DEFAULT_BACKEND) -> EmbeddingBackend:
|
||||
"""Return the embedding backend for a config id.
|
||||
|
||||
Raises:
|
||||
ValueError: If ``backend_id`` is not a known backend.
|
||||
"""
|
||||
if backend_id == DEFAULT_BACKEND:
|
||||
return CharNGramEmbeddingBackend()
|
||||
raise ValueError(f"Unknown embedding backend '{backend_id}'.")
|
||||
|
||||
|
||||
def cosine_similarity(a: SparseVector, b: SparseVector) -> float:
|
||||
"""Cosine similarity of two L2-normalized sparse vectors (their dot product)."""
|
||||
if not a or not b:
|
||||
return 0.0
|
||||
# Iterate the smaller vector for speed; both are unit-normalized.
|
||||
small, large = (a, b) if len(a) <= len(b) else (b, a)
|
||||
return sum(weight * large.get(gram, 0.0) for gram, weight in small.items())
|
||||
@@ -0,0 +1,133 @@
|
||||
"""Knowledge Base embedding index (MASTER_SPEC §30.2, epic #34, #115).
|
||||
|
||||
A committed snapshot that maps embedding vectors back to **existing** Knowledge
|
||||
Base entry ids, so the semantic fallback can only ever return a real entry — it
|
||||
can never invent a tag. Each entry contributes one row per surface form (its
|
||||
canonical id-as-words, its tags, and its aliases), so a query matches an entry
|
||||
when it is close to any of that entry's surfaces (``stockings`` → the
|
||||
``thighhighs`` entry via its ``stockings`` alias).
|
||||
|
||||
The index is built offline by ``scripts/build_embedding_index.py`` and committed
|
||||
under ``data/`` so CI never recomputes embeddings. :class:`EmbeddingIndex` reads
|
||||
it once and reuses it for nearest-neighbour search.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
from compiler.common.embedding import (
|
||||
DEFAULT_BACKEND,
|
||||
EmbeddingBackend,
|
||||
SparseVector,
|
||||
cosine_similarity,
|
||||
get_backend,
|
||||
)
|
||||
from compiler.common.knowledge_base import KnowledgeBase
|
||||
|
||||
# Weights are rounded when serialized to keep the committed artifact compact and
|
||||
# its regeneration byte-stable.
|
||||
_WEIGHT_PRECISION = 6
|
||||
|
||||
|
||||
def _entry_surfaces(entry) -> list[str]:
|
||||
"""The distinct surface texts of an entry: id-as-words, tags, and aliases."""
|
||||
surfaces = [entry.id.replace("_", " "), *entry.tags, *entry.aliases]
|
||||
seen: set[str] = set()
|
||||
ordered: list[str] = []
|
||||
for surface in surfaces:
|
||||
if surface not in seen:
|
||||
seen.add(surface)
|
||||
ordered.append(surface)
|
||||
return ordered
|
||||
|
||||
|
||||
def build_index_rows(
|
||||
knowledge_base: KnowledgeBase, backend: EmbeddingBackend | None = None
|
||||
) -> list[dict]:
|
||||
"""Build the index rows ``[{"id", "surface", "vector"}, ...]`` deterministically.
|
||||
|
||||
Entries are visited in sorted id order and surfaces in their listed order, so
|
||||
the artifact is byte-stable for a given Knowledge Base.
|
||||
"""
|
||||
backend = backend or get_backend(DEFAULT_BACKEND)
|
||||
rows: list[dict] = []
|
||||
for entry_id in sorted(knowledge_base.by_id):
|
||||
entry = knowledge_base.by_id[entry_id]
|
||||
surfaces = _entry_surfaces(entry)
|
||||
vectors = backend.embed(surfaces)
|
||||
for surface, vector in zip(surfaces, vectors, strict=True):
|
||||
if not vector:
|
||||
continue
|
||||
rows.append(
|
||||
{
|
||||
"id": entry_id,
|
||||
"surface": surface,
|
||||
"vector": {
|
||||
gram: round(weight, _WEIGHT_PRECISION) for gram, weight in vector.items()
|
||||
},
|
||||
}
|
||||
)
|
||||
return rows
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class NeighbourMatch:
|
||||
"""The nearest Knowledge Base entry to a query, with its similarity score."""
|
||||
|
||||
entry_id: str
|
||||
surface: str
|
||||
score: float
|
||||
|
||||
|
||||
class EmbeddingIndex:
|
||||
"""An in-memory embedding index over Knowledge Base entry surfaces."""
|
||||
|
||||
def __init__(self, rows: list[dict], backend: EmbeddingBackend | None = None) -> None:
|
||||
self._rows: list[tuple[str, str, SparseVector]] = [
|
||||
(row["id"], row["surface"], row["vector"]) for row in rows
|
||||
]
|
||||
self._backend = backend or get_backend(DEFAULT_BACKEND)
|
||||
|
||||
@property
|
||||
def backend(self) -> EmbeddingBackend:
|
||||
return self._backend
|
||||
|
||||
def nearest(self, text: str) -> NeighbourMatch | None:
|
||||
"""Return the nearest entry surface to ``text`` (highest cosine), or None."""
|
||||
(query,) = self._backend.embed([text])
|
||||
if not query:
|
||||
return None
|
||||
best: NeighbourMatch | None = None
|
||||
for entry_id, surface, vector in self._rows:
|
||||
score = cosine_similarity(query, vector)
|
||||
# Strict '>' keeps the first (sorted-id, listed-surface) winner on ties,
|
||||
# so results are deterministic regardless of scan order.
|
||||
if best is None or score > best.score:
|
||||
best = NeighbourMatch(entry_id=entry_id, surface=surface, score=score)
|
||||
return best
|
||||
|
||||
def __len__(self) -> int:
|
||||
return len(self._rows)
|
||||
|
||||
|
||||
def load_index(path: str | Path, backend: EmbeddingBackend | None = None) -> EmbeddingIndex:
|
||||
"""Load a committed embedding-index artifact once into an :class:`EmbeddingIndex`."""
|
||||
rows = json.loads(Path(path).read_text(encoding="utf-8"))
|
||||
return EmbeddingIndex(rows, backend)
|
||||
|
||||
|
||||
# The shipped index lives next to the package data; resolved relative to the repo
|
||||
# root so it works regardless of the process working directory.
|
||||
DEFAULT_INDEX_PATH = Path(__file__).resolve().parents[2] / "data" / "kb_embedding_index.json"
|
||||
_default_index_cache: EmbeddingIndex | None = None
|
||||
|
||||
|
||||
def load_default_index() -> EmbeddingIndex | None:
|
||||
"""Load and cache the shipped embedding index once, or None when it is absent."""
|
||||
global _default_index_cache
|
||||
if _default_index_cache is None and DEFAULT_INDEX_PATH.is_file():
|
||||
_default_index_cache = load_index(DEFAULT_INDEX_PATH)
|
||||
return _default_index_cache
|
||||
@@ -50,6 +50,7 @@ _EXTENSIONS: dict[str, CodeSpec] = {
|
||||
"SC0017": CodeSpec(Severity.WARNING, "Interaction Dropped"),
|
||||
"SC0018": CodeSpec(Severity.ERROR, "Analyzer Unexpected Response"),
|
||||
"SC0019": CodeSpec(Severity.WARNING, "Concept Reduced"),
|
||||
"SC0020": CodeSpec(Severity.WARNING, "Semantic Fallback"),
|
||||
"SC0021": CodeSpec(Severity.WARNING, "List Under-Transcription"),
|
||||
"SC0022": CodeSpec(Severity.WARNING, "Candidate Rejected"),
|
||||
}
|
||||
|
||||
@@ -17,6 +17,7 @@ from collections.abc import Iterator
|
||||
|
||||
from compiler.common.categories import is_valid_category
|
||||
from compiler.common.config import Config
|
||||
from compiler.common.embedding_index import EmbeddingIndex
|
||||
from compiler.common.knowledge_base import KnowledgeBase, KnowledgeBaseEntry
|
||||
from compiler.common.log import StructuredLogger
|
||||
from compiler.common.message_codes import message
|
||||
@@ -51,9 +52,15 @@ def resolve_scene(
|
||||
knowledge_base: KnowledgeBase,
|
||||
config: Config,
|
||||
logger: StructuredLogger | None = None,
|
||||
embedding_index: EmbeddingIndex | None = None,
|
||||
) -> CompilerResult:
|
||||
"""Resolve a validated Scene into an ordered tuple of Resolved Tags.
|
||||
|
||||
When ``config.semantic.enabled`` and an ``embedding_index`` is supplied, a
|
||||
concept that misses deterministic lookup **and** the head-noun reduction may
|
||||
fall back to its nearest Knowledge Base entry (epic #34). Deterministic lookup
|
||||
always wins; with the feature disabled (or no index) behaviour is unchanged.
|
||||
|
||||
Returns:
|
||||
A CompilerResult whose ``data`` is the tuple of :class:`ResolvedTag` on
|
||||
success, or ``None`` with error messages when an error condition
|
||||
@@ -61,7 +68,7 @@ def resolve_scene(
|
||||
"""
|
||||
include_nsfw = config.resolver.include_nsfw
|
||||
index = _build_normalized_index(knowledge_base, include_nsfw)
|
||||
resolver = _ConceptResolver(knowledge_base, index, config, logger)
|
||||
resolver = _ConceptResolver(knowledge_base, index, config, logger, embedding_index)
|
||||
|
||||
tags: list[ResolvedTag] = []
|
||||
warnings: list[Message] = []
|
||||
@@ -154,6 +161,7 @@ class _ConceptResolver:
|
||||
index: dict[str, KnowledgeBaseEntry],
|
||||
config: Config,
|
||||
logger: StructuredLogger | None,
|
||||
embedding_index: EmbeddingIndex | None = None,
|
||||
) -> None:
|
||||
self._kb = knowledge_base
|
||||
self._index = index
|
||||
@@ -162,6 +170,8 @@ class _ConceptResolver:
|
||||
self._max_depth = config.resolver.max_expansion_depth
|
||||
self._include_nsfw = config.resolver.include_nsfw
|
||||
self._logger = logger
|
||||
self._min_similarity = config.semantic.min_similarity
|
||||
self._embedding_index = embedding_index if config.semantic.enabled else None
|
||||
|
||||
def resolve(self, concept_name: str) -> _Resolution:
|
||||
outcome = _Resolution()
|
||||
@@ -174,6 +184,8 @@ class _ConceptResolver:
|
||||
tokens = key.split()
|
||||
entry, reduced_key, start = self._reduce(tokens)
|
||||
if entry is None:
|
||||
if self._semantic_fallback(concept_name, outcome):
|
||||
return outcome
|
||||
outcome.warnings.append(
|
||||
message(
|
||||
"SC0001",
|
||||
@@ -202,6 +214,41 @@ class _ConceptResolver:
|
||||
self._recover_modifiers(tokens, start, concept_name, outcome)
|
||||
return outcome
|
||||
|
||||
def _semantic_fallback(self, concept_name: str, outcome: _Resolution) -> bool:
|
||||
"""Resolve an otherwise-dropped concept via nearest-neighbour (epic #34, #116).
|
||||
|
||||
Runs only when the feature is enabled and an index is present, i.e. after
|
||||
deterministic lookup and the head-noun reduction both miss — deterministic
|
||||
resolution always wins. Accepts the neighbour only when its similarity is at
|
||||
least ``min_similarity`` and it survives NSFW gating; the matched entry is a
|
||||
real Knowledge Base entry, so nothing is invented. Returns True when the
|
||||
concept was resolved (emitting ``SC0020``), False to keep the drop/warn path.
|
||||
"""
|
||||
if self._embedding_index is None:
|
||||
return False
|
||||
match = self._embedding_index.nearest(concept_name)
|
||||
if match is None or match.score < self._min_similarity:
|
||||
return False
|
||||
entry = self._kb.get(match.entry_id)
|
||||
if entry is None:
|
||||
return False
|
||||
if entry.rating == "explicit" and not self._include_nsfw:
|
||||
return False # gated NSFW neighbour: keep the drop/warn path
|
||||
outcome.warnings.append(
|
||||
message(
|
||||
"SC0020",
|
||||
(
|
||||
f"Concept '{concept_name}' resolved via semantic fallback to "
|
||||
f"'{entry.id}' (similarity {match.score:.3f})."
|
||||
),
|
||||
concept=concept_name,
|
||||
entry=entry.id,
|
||||
score=match.score,
|
||||
)
|
||||
)
|
||||
self._expand(entry, concept_name, depth=0, path=(), outcome=outcome)
|
||||
return True
|
||||
|
||||
def _reduce(self, tokens: list[str]) -> tuple[KnowledgeBaseEntry | None, str | None, int]:
|
||||
"""Fall back to the concept's head noun when the full key has no entry (§17.2).
|
||||
|
||||
|
||||
@@ -24,6 +24,11 @@
|
||||
"separator": ",",
|
||||
"remove_duplicate_tags": true
|
||||
},
|
||||
"semantic": {
|
||||
"enabled": false,
|
||||
"min_similarity": 0.5,
|
||||
"backend": "char_ngram"
|
||||
},
|
||||
"debug": {
|
||||
"enabled": false,
|
||||
"level": "basic"
|
||||
|
||||
File diff suppressed because one or more lines are too long
@@ -44,6 +44,7 @@ constant before running) and are occasional dev steps, not part of runtime or CI
|
||||
|
||||
## Related tools
|
||||
|
||||
- `python scripts/build_embedding_index.py` — rebuild the committed semantic-fallback embedding index (`data/kb_embedding_index.json`) from the current KB.
|
||||
- `python scripts/detect_kb_conflicts.py knowledge_base/` — standalone conflict scan (CI-gateable).
|
||||
- `python scripts/coverage_benchmark.py` — Knowledge Base coverage report.
|
||||
- `python scripts/validate_knowledge_base.py knowledge_base/` — full KB validation.
|
||||
|
||||
@@ -53,6 +53,12 @@ class ConfigurationNode:
|
||||
"optional": {
|
||||
"analyzer_system_prompt": ("STRING", {"multiline": True, "default": ""}),
|
||||
"resolver_knowledge_base_version": ("STRING", {"default": ""}),
|
||||
"semantic_enabled": ("BOOLEAN", {"default": False}),
|
||||
"semantic_min_similarity": (
|
||||
"FLOAT",
|
||||
{"default": 0.5, "min": 0.0, "max": 1.0, "step": 0.05},
|
||||
),
|
||||
"semantic_backend": ("STRING", {"default": "char_ngram"}),
|
||||
},
|
||||
}
|
||||
|
||||
@@ -76,6 +82,9 @@ class ConfigurationNode:
|
||||
debug_level: str,
|
||||
analyzer_system_prompt: str = "",
|
||||
resolver_knowledge_base_version: str = "",
|
||||
semantic_enabled: bool = False,
|
||||
semantic_min_similarity: float = 0.5,
|
||||
semantic_backend: str = "char_ngram",
|
||||
) -> tuple[Any, str]:
|
||||
analyzer: dict[str, Any] = {
|
||||
"model": analyzer_model,
|
||||
@@ -102,6 +111,11 @@ class ConfigurationNode:
|
||||
),
|
||||
},
|
||||
"validator": {"allow_unknown_fields": validator_allow_unknown_fields},
|
||||
"semantic": {
|
||||
"enabled": semantic_enabled,
|
||||
"min_similarity": semantic_min_similarity,
|
||||
"backend": semantic_backend,
|
||||
},
|
||||
"prompt_builder": {
|
||||
"target": prompt_target,
|
||||
"separator": prompt_separator,
|
||||
|
||||
@@ -12,6 +12,7 @@ from __future__ import annotations
|
||||
from typing import Any
|
||||
|
||||
from compiler.common.config import Config
|
||||
from compiler.common.embedding_index import load_default_index
|
||||
from compiler.resolver.illustrious_resolver import resolve_scene, tags_to_prompt
|
||||
|
||||
from .adapters import format_messages, to_raw_json, upstream_failure_message
|
||||
@@ -44,7 +45,10 @@ class ResolverNode:
|
||||
if scene is None or knowledge_base is None:
|
||||
return ("", "", upstream_failure_message("scene or knowledge base"), "")
|
||||
config = config or Config()
|
||||
result = resolve_scene(scene, knowledge_base, config)
|
||||
# The (optional) semantic fallback reuses the shipped embedding index,
|
||||
# loaded once and cached; disabled configs never touch it.
|
||||
embedding_index = load_default_index() if config.semantic.enabled else None
|
||||
result = resolve_scene(scene, knowledge_base, config, embedding_index=embedding_index)
|
||||
separator = config.prompt_builder.separator
|
||||
prompt = "" if result.data is None else tags_to_prompt(result.data, separator)
|
||||
return (
|
||||
|
||||
@@ -49,6 +49,15 @@
|
||||
"remove_duplicate_tags": { "type": "boolean" }
|
||||
}
|
||||
},
|
||||
"semantic": {
|
||||
"type": "object",
|
||||
"additionalProperties": false,
|
||||
"properties": {
|
||||
"enabled": { "type": "boolean" },
|
||||
"min_similarity": { "type": "number", "minimum": 0, "maximum": 1 },
|
||||
"backend": { "type": "string" }
|
||||
}
|
||||
},
|
||||
"debug": {
|
||||
"type": "object",
|
||||
"additionalProperties": false,
|
||||
|
||||
@@ -0,0 +1,46 @@
|
||||
"""Build the committed Knowledge Base embedding index (issue #115, epic #34).
|
||||
|
||||
Embeds each Knowledge Base entry's surface forms (canonical id, tags, aliases)
|
||||
with the deterministic offline backend and writes the index snapshot to
|
||||
``data/kb_embedding_index.json``. The snapshot maps vectors back to existing entry
|
||||
ids only, so the semantic fallback can never invent a tag. It is committed so CI
|
||||
never recomputes embeddings.
|
||||
|
||||
Paths are hard-coded constants (no CLI/untrusted input) matching the style of
|
||||
``scripts/build_vocab.py`` (SonarCloud S8707).
|
||||
|
||||
Usage:
|
||||
python scripts/build_embedding_index.py
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from compiler.common.embedding_index import build_index_rows # noqa: E402
|
||||
from compiler.common.knowledge_base import load_knowledge_base # noqa: E402
|
||||
|
||||
_REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
_KB_DIR = _REPO_ROOT / "knowledge_base"
|
||||
_OUTPUT = _REPO_ROOT / "data" / "kb_embedding_index.json"
|
||||
|
||||
|
||||
def build() -> int:
|
||||
kb = load_knowledge_base(_KB_DIR)
|
||||
rows = build_index_rows(kb)
|
||||
_OUTPUT.parent.mkdir(parents=True, exist_ok=True)
|
||||
_OUTPUT.write_text(
|
||||
json.dumps(rows, ensure_ascii=False, separators=(",", ":")) + "\n", encoding="utf-8"
|
||||
)
|
||||
print(
|
||||
f"wrote {len(rows)} index rows for {len(kb)} entries to {_OUTPUT.relative_to(_REPO_ROOT)}"
|
||||
)
|
||||
return len(rows)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
build()
|
||||
@@ -64,7 +64,7 @@ def test_resolves_to_flat_prompt_with_real_resolver() -> None:
|
||||
|
||||
|
||||
def test_delegates_and_surfaces_messages(monkeypatch) -> None:
|
||||
def fake_resolve(scene_arg, knowledge_base, config):
|
||||
def fake_resolve(scene_arg, knowledge_base, config, embedding_index=None):
|
||||
return CompilerResult(data=()).add_warning(message("SC0001", "unknown 'x'"))
|
||||
|
||||
monkeypatch.setattr(node_module, "resolve_scene", fake_resolve)
|
||||
@@ -76,7 +76,7 @@ def test_delegates_and_surfaces_messages(monkeypatch) -> None:
|
||||
def test_default_config_used_when_absent(monkeypatch) -> None:
|
||||
captured = {}
|
||||
|
||||
def fake_resolve(scene_arg, knowledge_base, config):
|
||||
def fake_resolve(scene_arg, knowledge_base, config, embedding_index=None):
|
||||
captured["config"] = config
|
||||
return CompilerResult(data=())
|
||||
|
||||
|
||||
@@ -0,0 +1,125 @@
|
||||
"""Tests for the semantic nearest-neighbour fallback (issue #116, epic #34)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from compiler.common.config import Config
|
||||
from compiler.common.embedding_index import EmbeddingIndex, build_index_rows
|
||||
from compiler.common.knowledge_base import KnowledgeBase, KnowledgeBaseEntry
|
||||
from compiler.resolver.illustrious_resolver import resolve_scene
|
||||
from schemas.models import Scene
|
||||
|
||||
|
||||
def entry(id_, tags, category="clothing", aliases=(), rating="general"):
|
||||
return KnowledgeBaseEntry(
|
||||
id=id_, tags=tuple(tags), category=category, aliases=tuple(aliases), rating=rating
|
||||
)
|
||||
|
||||
|
||||
def kb() -> KnowledgeBase:
|
||||
return KnowledgeBase(
|
||||
[
|
||||
entry("thighhighs", ["thighhighs"], aliases=["stockings"]),
|
||||
entry("dress", ["dress"]),
|
||||
entry("pussy", ["pussy"], category="body", rating="explicit"),
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
def index(knowledge: KnowledgeBase) -> EmbeddingIndex:
|
||||
return EmbeddingIndex(build_index_rows(knowledge))
|
||||
|
||||
|
||||
def scene_with(concept: str) -> Scene:
|
||||
return Scene.from_json(
|
||||
{
|
||||
"characters": [
|
||||
{
|
||||
"id": 0,
|
||||
"identity": [concept],
|
||||
"appearance": [],
|
||||
"clothing": [],
|
||||
"accessories": [],
|
||||
"pose": [],
|
||||
"expression": [],
|
||||
"actions": [],
|
||||
}
|
||||
],
|
||||
"interactions": [],
|
||||
"objects": [],
|
||||
"environment": [],
|
||||
"camera": [],
|
||||
"lighting": [],
|
||||
"metadata": {"schema_version": "1.0"},
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def semantic_config(min_similarity: float = 0.5, include_nsfw: bool = False) -> Config:
|
||||
return Config.from_json(
|
||||
{
|
||||
"semantic": {"enabled": True, "min_similarity": min_similarity},
|
||||
"resolver": {"include_nsfw": include_nsfw},
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def tags(result):
|
||||
return [t.tag for t in result.data]
|
||||
|
||||
|
||||
def codes(messages):
|
||||
return [m.code for m in messages]
|
||||
|
||||
|
||||
def test_near_concept_resolves_via_fallback_and_emits_sc0020() -> None:
|
||||
knowledge = kb()
|
||||
result = resolve_scene(
|
||||
scene_with("thighighs"), knowledge, semantic_config(), embedding_index=index(knowledge)
|
||||
)
|
||||
assert tags(result) == ["thighhighs"]
|
||||
assert "SC0020" in codes(result.warnings)
|
||||
|
||||
|
||||
def test_far_concept_still_drops() -> None:
|
||||
knowledge = kb()
|
||||
result = resolve_scene(
|
||||
scene_with("quantum flux inverter"),
|
||||
knowledge,
|
||||
semantic_config(min_similarity=0.6),
|
||||
embedding_index=index(knowledge),
|
||||
)
|
||||
assert tags(result) == []
|
||||
assert "SC0001" in codes(result.warnings)
|
||||
assert "SC0020" not in codes(result.warnings)
|
||||
|
||||
|
||||
def test_disabled_feature_is_byte_identical() -> None:
|
||||
knowledge = kb()
|
||||
disabled = resolve_scene(scene_with("thighighs"), knowledge, Config())
|
||||
with_index_but_disabled = resolve_scene(
|
||||
scene_with("thighighs"), knowledge, Config(), embedding_index=index(knowledge)
|
||||
)
|
||||
assert tags(disabled) == tags(with_index_but_disabled) == []
|
||||
assert "SC0020" not in codes(with_index_but_disabled.warnings)
|
||||
|
||||
|
||||
def test_deterministic_lookup_always_wins() -> None:
|
||||
knowledge = kb()
|
||||
result = resolve_scene(
|
||||
scene_with("dress"), knowledge, semantic_config(), embedding_index=index(knowledge)
|
||||
)
|
||||
assert tags(result) == ["dress"]
|
||||
assert "SC0020" not in codes(result.warnings) # exact hit, fallback not used
|
||||
|
||||
|
||||
def test_gated_nsfw_neighbour_is_not_used() -> None:
|
||||
knowledge = kb()
|
||||
# "pusy" is closest to the explicit "pussy" entry, but NSFW is gated off.
|
||||
result = resolve_scene(
|
||||
scene_with("pusy"),
|
||||
knowledge,
|
||||
semantic_config(include_nsfw=False),
|
||||
embedding_index=index(knowledge),
|
||||
)
|
||||
assert "SC0020" not in codes(result.warnings)
|
||||
assert "SC0001" in codes(result.warnings)
|
||||
@@ -0,0 +1,102 @@
|
||||
"""Determinism & no-invention guarantees for the semantic fallback (issue #117)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from compiler.common.config import Config
|
||||
from compiler.common.embedding_index import EmbeddingIndex, build_index_rows
|
||||
from compiler.common.knowledge_base import KnowledgeBase, KnowledgeBaseEntry
|
||||
from compiler.common.message_codes import CODES
|
||||
from compiler.common.result import Severity
|
||||
from compiler.resolver.illustrious_resolver import resolve_scene
|
||||
from schemas.models import Scene
|
||||
|
||||
|
||||
def entry(id_, tags, aliases=()):
|
||||
return KnowledgeBaseEntry(id=id_, tags=tuple(tags), category="clothing", aliases=tuple(aliases))
|
||||
|
||||
|
||||
def kb() -> KnowledgeBase:
|
||||
return KnowledgeBase(
|
||||
[
|
||||
entry("thighhighs", ["thighhighs"], aliases=["stockings"]),
|
||||
entry("dress", ["dress"]),
|
||||
entry("jacket", ["jacket"]),
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
def scene_with(*concepts: str) -> Scene:
|
||||
return Scene.from_json(
|
||||
{
|
||||
"characters": [
|
||||
{
|
||||
"id": 0,
|
||||
"identity": list(concepts),
|
||||
"appearance": [],
|
||||
"clothing": [],
|
||||
"accessories": [],
|
||||
"pose": [],
|
||||
"expression": [],
|
||||
"actions": [],
|
||||
}
|
||||
],
|
||||
"interactions": [],
|
||||
"objects": [],
|
||||
"environment": [],
|
||||
"camera": [],
|
||||
"lighting": [],
|
||||
"metadata": {"schema_version": "1.0"},
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def enabled_config() -> Config:
|
||||
return Config.from_json({"semantic": {"enabled": True, "min_similarity": 0.3}})
|
||||
|
||||
|
||||
class _ExplodingIndex(EmbeddingIndex):
|
||||
"""An index that fails if queried — proves the fallback was never reached."""
|
||||
|
||||
def nearest(self, text: str): # type: ignore[override]
|
||||
raise AssertionError(f"embedding fallback must not run for '{text}'")
|
||||
|
||||
|
||||
def test_sc0020_is_registered() -> None:
|
||||
assert "SC0020" in CODES
|
||||
assert CODES["SC0020"].severity is Severity.WARNING
|
||||
|
||||
|
||||
def test_exact_hit_never_routes_through_fallback() -> None:
|
||||
knowledge = kb()
|
||||
# Every concept is an exact id/alias hit, so nearest() must never be called
|
||||
# even with the feature enabled — deterministic resolution always wins.
|
||||
result = resolve_scene(
|
||||
scene_with("dress", "jacket", "stockings"),
|
||||
knowledge,
|
||||
enabled_config(),
|
||||
embedding_index=_ExplodingIndex(build_index_rows(knowledge)),
|
||||
)
|
||||
assert [t.tag for t in result.data] == ["dress", "jacket", "thighhighs"]
|
||||
|
||||
|
||||
def test_fallback_only_emits_existing_kb_tags() -> None:
|
||||
knowledge = kb()
|
||||
index = EmbeddingIndex(build_index_rows(knowledge))
|
||||
kb_tags = {tag for e in knowledge.by_id.values() for tag in e.tags}
|
||||
# A spread of misspellings/near-misses; whatever resolves must be a real tag.
|
||||
for query in ["thighighs", "jackt", "dres", "stockngs", "zzzz"]:
|
||||
result = resolve_scene(
|
||||
scene_with(query), knowledge, enabled_config(), embedding_index=index
|
||||
)
|
||||
for resolved in result.data:
|
||||
assert resolved.tag in kb_tags # no invention
|
||||
|
||||
|
||||
def test_same_scene_twice_is_identical() -> None:
|
||||
knowledge = kb()
|
||||
index = EmbeddingIndex(build_index_rows(knowledge))
|
||||
scene = scene_with("thighighs", "dres", "jacket")
|
||||
first = resolve_scene(scene, knowledge, enabled_config(), embedding_index=index)
|
||||
second = resolve_scene(scene, knowledge, enabled_config(), embedding_index=index)
|
||||
assert [t.tag for t in first.data] == [t.tag for t in second.data]
|
||||
assert [m.code for m in first.warnings] == [m.code for m in second.warnings]
|
||||
@@ -34,6 +34,7 @@ EXAMPLE_CONFIG = {
|
||||
"separator": ",",
|
||||
"remove_duplicate_tags": True,
|
||||
},
|
||||
"semantic": {"enabled": False, "min_similarity": 0.5, "backend": "char_ngram"},
|
||||
"debug": {"enabled": False, "level": "basic"},
|
||||
}
|
||||
|
||||
@@ -81,6 +82,20 @@ def test_partial_config_fills_defaults() -> None:
|
||||
assert c.resolver.max_expansion_depth == 8
|
||||
|
||||
|
||||
def test_semantic_defaults_off() -> None:
|
||||
c = Config()
|
||||
assert c.semantic.enabled is False
|
||||
assert c.semantic.min_similarity == pytest.approx(0.5)
|
||||
assert c.semantic.backend == "char_ngram"
|
||||
|
||||
|
||||
def test_semantic_section_roundtrips() -> None:
|
||||
c = Config.from_json({"semantic": {"enabled": True, "min_similarity": 0.8}})
|
||||
assert c.semantic.enabled is True
|
||||
assert c.semantic.min_similarity == pytest.approx(0.8)
|
||||
assert c.to_json()["semantic"]["enabled"] is True
|
||||
|
||||
|
||||
def test_knowledge_base_version_defaults_to_none_and_is_omitted() -> None:
|
||||
c = Config()
|
||||
assert c.resolver.knowledge_base_version is None
|
||||
|
||||
@@ -25,6 +25,9 @@ def default_inputs() -> dict:
|
||||
"prompt_remove_duplicate_tags": True,
|
||||
"debug_enabled": False,
|
||||
"debug_level": "basic",
|
||||
"semantic_enabled": False,
|
||||
"semantic_min_similarity": 0.5,
|
||||
"semantic_backend": "char_ngram",
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,51 @@
|
||||
"""Tests for the deterministic offline embedding backend (issue #114, epic #34)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from compiler.common.embedding import (
|
||||
CharNGramEmbeddingBackend,
|
||||
cosine_similarity,
|
||||
get_backend,
|
||||
)
|
||||
|
||||
|
||||
def test_embedding_is_deterministic() -> None:
|
||||
backend = CharNGramEmbeddingBackend()
|
||||
assert backend.embed(["thighhighs"]) == backend.embed(["thighhighs"])
|
||||
|
||||
|
||||
def test_vectors_are_l2_normalized() -> None:
|
||||
(vector,) = CharNGramEmbeddingBackend().embed(["blue eyes"])
|
||||
magnitude = sum(weight * weight for weight in vector.values())
|
||||
assert magnitude == pytest.approx(1.0)
|
||||
|
||||
|
||||
def test_identical_text_has_self_similarity_one() -> None:
|
||||
(vector,) = CharNGramEmbeddingBackend().embed(["long hair"])
|
||||
assert cosine_similarity(vector, vector) == pytest.approx(1.0)
|
||||
|
||||
|
||||
def test_similar_text_scores_higher_than_dissimilar() -> None:
|
||||
backend = CharNGramEmbeddingBackend()
|
||||
(thighhighs,) = backend.embed(["thighhighs"])
|
||||
(thigh_highs,) = backend.embed(["thigh highs"])
|
||||
(unrelated,) = backend.embed(["castle"])
|
||||
assert cosine_similarity(thighhighs, thigh_highs) > cosine_similarity(thighhighs, unrelated)
|
||||
|
||||
|
||||
def test_normalization_ignores_case_and_underscores() -> None:
|
||||
backend = CharNGramEmbeddingBackend()
|
||||
assert backend.embed(["Long_Hair"]) == backend.embed(["long hair"])
|
||||
|
||||
|
||||
def test_empty_text_yields_empty_vector() -> None:
|
||||
assert CharNGramEmbeddingBackend().embed([""]) == [{}]
|
||||
assert cosine_similarity({}, {"abc": 1.0}) == pytest.approx(0.0)
|
||||
|
||||
|
||||
def test_get_backend_returns_default_and_rejects_unknown() -> None:
|
||||
assert isinstance(get_backend("char_ngram"), CharNGramEmbeddingBackend)
|
||||
with pytest.raises(ValueError):
|
||||
get_backend("nope")
|
||||
@@ -0,0 +1,65 @@
|
||||
"""Tests for the Knowledge Base embedding index (issue #115, epic #34)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from compiler.common.embedding_index import EmbeddingIndex, build_index_rows, load_index
|
||||
from compiler.common.knowledge_base import KnowledgeBase, KnowledgeBaseEntry, load_knowledge_base
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
KB_DIR = REPO_ROOT / "knowledge_base"
|
||||
INDEX_FILE = REPO_ROOT / "data" / "kb_embedding_index.json"
|
||||
|
||||
|
||||
def _entry(id_, tags, aliases=()):
|
||||
return KnowledgeBaseEntry(id=id_, tags=tuple(tags), category="clothing", aliases=tuple(aliases))
|
||||
|
||||
|
||||
# --- committed reference index ---------------------------------------------
|
||||
|
||||
|
||||
def test_committed_index_maps_known_concept_to_expected_entry() -> None:
|
||||
index = load_index(INDEX_FILE)
|
||||
match = index.nearest("stockings")
|
||||
assert match.entry_id == "thighhighs" # via the ingested alias surface
|
||||
assert match.score > 0.99 # near-exact (weights are rounded in the snapshot)
|
||||
|
||||
|
||||
def test_index_only_references_real_kb_entries() -> None:
|
||||
kb = load_knowledge_base(KB_DIR)
|
||||
index = load_index(INDEX_FILE)
|
||||
# Every row id is a real entry: a fallback can never invent a tag.
|
||||
match = index.nearest("blue eyes")
|
||||
assert kb.get(match.entry_id) is not None
|
||||
|
||||
|
||||
# --- build determinism ------------------------------------------------------
|
||||
|
||||
|
||||
def test_build_is_deterministic() -> None:
|
||||
kb = KnowledgeBase(
|
||||
[
|
||||
_entry("thighhighs", ["thighhighs"], aliases=["thigh highs", "stockings"]),
|
||||
_entry("pantyhose", ["pantyhose"], aliases=["tights"]),
|
||||
]
|
||||
)
|
||||
assert build_index_rows(kb) == build_index_rows(kb)
|
||||
|
||||
|
||||
def test_nearest_returns_expected_surface() -> None:
|
||||
kb = KnowledgeBase(
|
||||
[
|
||||
_entry("thighhighs", ["thighhighs"], aliases=["stockings"]),
|
||||
_entry("dress", ["dress"]),
|
||||
]
|
||||
)
|
||||
index = EmbeddingIndex(build_index_rows(kb))
|
||||
match = index.nearest("stockings")
|
||||
assert match.entry_id == "thighhighs"
|
||||
assert match.surface == "stockings"
|
||||
|
||||
|
||||
def test_empty_query_has_no_neighbour() -> None:
|
||||
index = EmbeddingIndex(build_index_rows(KnowledgeBase([_entry("dress", ["dress"])])))
|
||||
assert index.nearest("") is None
|
||||
Reference in New Issue
Block a user