Add compiler/resolver/illustrious_resolver.py: resolve_scene(scene, knowledge_base, config, logger=None) -> CompilerResult, per MASTER_SPEC §17. No LLM; no concept or tag is invented. - normalize_concept (§17.3): NFKC + lowercase, unifying underscores/spaces so natural concepts, snake_case ids, and spaced aliases share one lookup key. The resolver builds its own normalized index over the Knowledge Base. - Per-concept pipeline in scene discovery order (§17.8): normalize -> direct alias resolution (no chains; skipped when allow_aliases is false) -> canonical lookup -> recursive expansion -> tag generation. - Expansion (§17.5): own tags first, then depth-first in list order per expand target, inserted immediately after; capped by max_expansion_depth; skipped when expansion_enabled is false; defensive path-based cycle detection -> SC0003. - Deduplication after expansion keeps the first occurrence (SC0007). - Unknown concept -> SC0001 warning (continue); deprecated -> SC0006 (continue); invalid category -> SC0008 (stop); circular expansion -> SC0003 (stop). - Every ResolvedTag keeps its entry's category and full traceability (source_concept = originating scene concept, knowledge_base_entry = generating entry). Verbose logging of each decision. Tests (no Ollama): normalization, alias resolution + disable, expansion order, nested/depth-capped/disabled expansion, traceability, dedup, unknown/deprecated, circular/invalid-category stops, category preservation, and determinism. 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
3f0358bfa9
commit
76d440fbaf
@@ -0,0 +1,250 @@
|
||||
"""The Illustrious Resolver (MASTER_SPEC §17).
|
||||
|
||||
Deterministically converts the semantic concepts of a validated Scene into
|
||||
Illustrious Resolved Tags using the Knowledge Base. No language model is involved
|
||||
and no concept or tag is ever invented.
|
||||
|
||||
Per-concept pipeline (§17.2): normalize -> alias resolution (direct only) ->
|
||||
canonical lookup -> expansion -> tag generation. Concepts are processed in scene
|
||||
discovery order (§17.8) and every generated tag keeps the category of its
|
||||
Knowledge Base Entry.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import unicodedata
|
||||
from collections.abc import Iterator
|
||||
|
||||
from compiler.common.categories import is_valid_category
|
||||
from compiler.common.config import Config
|
||||
from compiler.common.knowledge_base import KnowledgeBase, KnowledgeBaseEntry
|
||||
from compiler.common.log import StructuredLogger
|
||||
from compiler.common.message_codes import message
|
||||
from compiler.common.result import CompilerResult, Message
|
||||
from schemas.models import ResolvedTag, Scene
|
||||
|
||||
_CHARACTER_CONCEPT_FIELDS = (
|
||||
"identity",
|
||||
"appearance",
|
||||
"clothing",
|
||||
"accessories",
|
||||
"pose",
|
||||
"expression",
|
||||
"actions",
|
||||
)
|
||||
|
||||
|
||||
def normalize_concept(name: str) -> str:
|
||||
"""Normalize a concept for lookup (§17.3).
|
||||
|
||||
Applies Unicode (NFKC) normalization and lowercasing, and unifies underscores
|
||||
with spaces before collapsing whitespace, so natural-language concepts
|
||||
("holding hands"), snake_case ids ("holding_hands"), and spaced aliases all
|
||||
map to the same key.
|
||||
"""
|
||||
text = unicodedata.normalize("NFKC", name).replace("_", " ").lower()
|
||||
return " ".join(text.split())
|
||||
|
||||
|
||||
def resolve_scene(
|
||||
scene: Scene,
|
||||
knowledge_base: KnowledgeBase,
|
||||
config: Config,
|
||||
logger: StructuredLogger | None = None,
|
||||
) -> CompilerResult:
|
||||
"""Resolve a validated Scene into an ordered tuple of Resolved Tags.
|
||||
|
||||
Returns:
|
||||
A CompilerResult whose ``data`` is the tuple of :class:`ResolvedTag` on
|
||||
success, or ``None`` with error messages when an error condition
|
||||
(circular expansion SC0003, invalid category SC0008) stops resolution.
|
||||
"""
|
||||
index = _build_normalized_index(knowledge_base)
|
||||
resolver = _ConceptResolver(knowledge_base, index, config, logger)
|
||||
|
||||
tags: list[ResolvedTag] = []
|
||||
warnings: list[Message] = []
|
||||
errors: list[Message] = []
|
||||
|
||||
for concept_name in _iter_scene_concepts(scene):
|
||||
resolved = resolver.resolve(concept_name)
|
||||
tags.extend(resolved.tags)
|
||||
warnings.extend(resolved.warnings)
|
||||
errors.extend(resolved.errors)
|
||||
|
||||
if errors:
|
||||
result = CompilerResult()
|
||||
for error in errors:
|
||||
result = result.add_error(error)
|
||||
return result
|
||||
|
||||
deduped, dedup_warnings = _deduplicate(tags)
|
||||
warnings.extend(dedup_warnings)
|
||||
|
||||
result = CompilerResult(data=tuple(deduped))
|
||||
for warning in warnings:
|
||||
result = result.add_warning(warning)
|
||||
if logger is not None:
|
||||
logger.basic("scene_resolved", tags=len(deduped), warnings=len(warnings))
|
||||
return result
|
||||
|
||||
|
||||
def _build_normalized_index(
|
||||
knowledge_base: KnowledgeBase,
|
||||
) -> dict[str, KnowledgeBaseEntry]:
|
||||
"""Map every normalized canonical id and alias to its entry (aliases direct)."""
|
||||
index: dict[str, KnowledgeBaseEntry] = {}
|
||||
for entry in knowledge_base.by_id.values():
|
||||
index[normalize_concept(entry.id)] = entry
|
||||
for alias in entry.aliases:
|
||||
index[normalize_concept(alias)] = entry
|
||||
return index
|
||||
|
||||
|
||||
def _iter_scene_concepts(scene: Scene) -> Iterator[str]:
|
||||
"""Yield concept names in scene discovery order (§17.8)."""
|
||||
for character in scene.characters:
|
||||
for field in _CHARACTER_CONCEPT_FIELDS:
|
||||
for concept in getattr(character, field):
|
||||
yield concept.name
|
||||
for interaction in scene.interactions:
|
||||
yield interaction.concept
|
||||
for field in ("objects", "environment", "camera", "lighting"):
|
||||
for concept in getattr(scene, field):
|
||||
yield concept.name
|
||||
|
||||
|
||||
class _Resolution:
|
||||
"""The outcome of resolving one concept."""
|
||||
|
||||
__slots__ = ("tags", "warnings", "errors")
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.tags: list[ResolvedTag] = []
|
||||
self.warnings: list[Message] = []
|
||||
self.errors: list[Message] = []
|
||||
|
||||
|
||||
class _ConceptResolver:
|
||||
"""Resolves a single concept (with expansion) against a normalized index."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
knowledge_base: KnowledgeBase,
|
||||
index: dict[str, KnowledgeBaseEntry],
|
||||
config: Config,
|
||||
logger: StructuredLogger | None,
|
||||
) -> None:
|
||||
self._kb = knowledge_base
|
||||
self._index = index
|
||||
self._allow_aliases = config.resolver.allow_aliases
|
||||
self._expansion_enabled = config.resolver.expansion_enabled
|
||||
self._max_depth = config.resolver.max_expansion_depth
|
||||
self._logger = logger
|
||||
|
||||
def resolve(self, concept_name: str) -> _Resolution:
|
||||
outcome = _Resolution()
|
||||
key = normalize_concept(concept_name)
|
||||
entry = self._lookup(key)
|
||||
if entry is None:
|
||||
outcome.warnings.append(
|
||||
message(
|
||||
"SC0001",
|
||||
f"Unknown concept '{concept_name}' has no Knowledge Base entry; ignored.",
|
||||
concept=concept_name,
|
||||
)
|
||||
)
|
||||
if self._logger is not None:
|
||||
self._logger.verbose("concept_unknown", concept=concept_name, normalized=key)
|
||||
return outcome
|
||||
|
||||
self._expand(entry, concept_name, depth=0, path=(), outcome=outcome)
|
||||
return outcome
|
||||
|
||||
def _lookup(self, key: str) -> KnowledgeBaseEntry | None:
|
||||
entry = self._index.get(key)
|
||||
if entry is None:
|
||||
return None
|
||||
# A key that matches an alias but not the entry's own id is an alias hit.
|
||||
if not self._allow_aliases and key != normalize_concept(entry.id):
|
||||
return None
|
||||
return entry
|
||||
|
||||
def _expand(
|
||||
self,
|
||||
entry: KnowledgeBaseEntry,
|
||||
source_concept: str,
|
||||
depth: int,
|
||||
path: tuple[str, ...],
|
||||
outcome: _Resolution,
|
||||
) -> None:
|
||||
if entry.id in path:
|
||||
outcome.errors.append(
|
||||
message(
|
||||
"SC0003",
|
||||
f"Circular expansion detected at '{entry.id}' (path: {' -> '.join(path)}).",
|
||||
entry=entry.id,
|
||||
)
|
||||
)
|
||||
return
|
||||
if not is_valid_category(entry.category):
|
||||
outcome.errors.append(
|
||||
message(
|
||||
"SC0008",
|
||||
f"Entry '{entry.id}' has invalid category '{entry.category}'.",
|
||||
entry=entry.id,
|
||||
)
|
||||
)
|
||||
return
|
||||
if entry.deprecated:
|
||||
outcome.warnings.append(
|
||||
message(
|
||||
"SC0006",
|
||||
f"Concept '{entry.id}' is deprecated; migration is encouraged.",
|
||||
entry=entry.id,
|
||||
)
|
||||
)
|
||||
|
||||
for tag in entry.tags:
|
||||
outcome.tags.append(
|
||||
ResolvedTag(
|
||||
tag=tag,
|
||||
category=entry.category,
|
||||
source_concept=source_concept,
|
||||
knowledge_base_entry=entry.id,
|
||||
)
|
||||
)
|
||||
if self._logger is not None:
|
||||
self._logger.verbose(
|
||||
"concept_resolved",
|
||||
concept=source_concept,
|
||||
canonical=entry.id,
|
||||
tags=list(entry.tags),
|
||||
)
|
||||
|
||||
if not self._expansion_enabled or depth >= self._max_depth:
|
||||
return
|
||||
for target_id in entry.expand:
|
||||
target = self._kb.get(target_id)
|
||||
if target is not None:
|
||||
self._expand(target, source_concept, depth + 1, (*path, entry.id), outcome)
|
||||
|
||||
|
||||
def _deduplicate(tags: list[ResolvedTag]) -> tuple[list[ResolvedTag], list[Message]]:
|
||||
"""Remove duplicate tags after expansion, keeping the first occurrence (§17.7)."""
|
||||
seen: set[str] = set()
|
||||
kept: list[ResolvedTag] = []
|
||||
warnings: list[Message] = []
|
||||
for resolved in tags:
|
||||
if resolved.tag in seen:
|
||||
warnings.append(
|
||||
message(
|
||||
"SC0007",
|
||||
f"Duplicate tag '{resolved.tag}' removed after expansion.",
|
||||
tag=resolved.tag,
|
||||
)
|
||||
)
|
||||
continue
|
||||
seen.add(resolved.tag)
|
||||
kept.append(resolved)
|
||||
return kept, warnings
|
||||
@@ -0,0 +1,229 @@
|
||||
"""Tests for the Illustrious Resolver (issue #12, MASTER_SPEC §17). No Ollama."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from compiler.common.config import Config
|
||||
from compiler.common.knowledge_base import KnowledgeBase, KnowledgeBaseEntry
|
||||
from compiler.resolver.illustrious_resolver import resolve_scene
|
||||
from schemas.models import Scene
|
||||
|
||||
|
||||
def kb(*entries: KnowledgeBaseEntry) -> KnowledgeBase:
|
||||
return KnowledgeBase(list(entries))
|
||||
|
||||
|
||||
def entry(id_, tags, category="hair", aliases=(), expand=(), deprecated=False):
|
||||
return KnowledgeBaseEntry(
|
||||
id=id_,
|
||||
tags=tuple(tags),
|
||||
category=category,
|
||||
aliases=tuple(aliases),
|
||||
expand=tuple(expand),
|
||||
deprecated=deprecated,
|
||||
)
|
||||
|
||||
|
||||
def scene_with(identity=(), **fields) -> Scene:
|
||||
character = {
|
||||
"id": 0,
|
||||
"identity": list(identity),
|
||||
"appearance": list(fields.get("appearance", [])),
|
||||
"clothing": list(fields.get("clothing", [])),
|
||||
"accessories": [],
|
||||
"pose": [],
|
||||
"expression": [],
|
||||
"actions": [],
|
||||
}
|
||||
return Scene.from_json(
|
||||
{
|
||||
"characters": [character],
|
||||
"interactions": [],
|
||||
"objects": list(fields.get("objects", [])),
|
||||
"environment": list(fields.get("environment", [])),
|
||||
"camera": [],
|
||||
"lighting": [],
|
||||
"metadata": {"schema_version": "1.0"},
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def default_config() -> Config:
|
||||
return Config()
|
||||
|
||||
|
||||
def tags(result) -> list[str]:
|
||||
return [t.tag for t in result.data]
|
||||
|
||||
|
||||
def codes(messages) -> list[str]:
|
||||
return [m.code for m in messages]
|
||||
|
||||
|
||||
# --- normalization ---------------------------------------------------------
|
||||
|
||||
|
||||
def test_normalization_matches_snake_case_id() -> None:
|
||||
knowledge = kb(entry("long_hair", ["long hair"]))
|
||||
for written in ["long hair", "Long Hair", " LONG_HAIR ", "long_hair"]:
|
||||
result = resolve_scene(scene_with(identity=[written]), knowledge, default_config())
|
||||
assert tags(result) == ["long hair"], written
|
||||
|
||||
|
||||
# --- alias resolution ------------------------------------------------------
|
||||
|
||||
|
||||
def test_alias_resolves_directly() -> None:
|
||||
knowledge = kb(entry("female", ["1girl"], category="character", aliases=["girl", "woman"]))
|
||||
result = resolve_scene(scene_with(identity=["girl"]), knowledge, default_config())
|
||||
assert tags(result) == ["1girl"]
|
||||
assert result.data[0].knowledge_base_entry == "female"
|
||||
assert result.data[0].source_concept == "girl"
|
||||
|
||||
|
||||
def test_alias_ignored_when_disabled() -> None:
|
||||
knowledge = kb(entry("female", ["1girl"], category="character", aliases=["girl"]))
|
||||
config = Config.from_json({"resolver": {"allow_aliases": False}})
|
||||
result = resolve_scene(scene_with(identity=["girl"]), knowledge, config)
|
||||
assert tags(result) == []
|
||||
assert "SC0001" in codes(result.warnings)
|
||||
|
||||
|
||||
# --- expansion order -------------------------------------------------------
|
||||
|
||||
|
||||
def test_expansion_order_is_normative() -> None:
|
||||
knowledge = kb(
|
||||
entry(
|
||||
"school_uniform", ["school uniform"], category="clothing", expand=["blazer", "skirt"]
|
||||
),
|
||||
entry("blazer", ["blazer"], category="clothing"),
|
||||
entry("skirt", ["pleated skirt"], category="clothing"),
|
||||
)
|
||||
result = resolve_scene(scene_with(clothing=["school uniform"]), knowledge, default_config())
|
||||
assert tags(result) == ["school uniform", "blazer", "pleated skirt"]
|
||||
|
||||
|
||||
def test_nested_expansion_is_depth_first() -> None:
|
||||
knowledge = kb(
|
||||
entry("a", ["ta"], expand=["b"]),
|
||||
entry("b", ["tb"], expand=["c"]),
|
||||
entry("c", ["tc"]),
|
||||
)
|
||||
result = resolve_scene(scene_with(identity=["a"]), knowledge, default_config())
|
||||
assert tags(result) == ["ta", "tb", "tc"]
|
||||
|
||||
|
||||
def test_expansion_disabled_emits_only_own_tags() -> None:
|
||||
knowledge = kb(
|
||||
entry("a", ["ta"], expand=["b"]),
|
||||
entry("b", ["tb"]),
|
||||
)
|
||||
config = Config.from_json({"resolver": {"expansion_enabled": False}})
|
||||
result = resolve_scene(scene_with(identity=["a"]), knowledge, config)
|
||||
assert tags(result) == ["ta"]
|
||||
|
||||
|
||||
def test_max_expansion_depth_caps_recursion() -> None:
|
||||
# depth 1 allows one expansion level (a -> b) but not the next (b -> c).
|
||||
knowledge = kb(
|
||||
entry("a", ["ta"], expand=["b"]),
|
||||
entry("b", ["tb"], expand=["c"]),
|
||||
entry("c", ["tc"]),
|
||||
)
|
||||
config = Config.from_json({"resolver": {"max_expansion_depth": 1}})
|
||||
result = resolve_scene(scene_with(identity=["a"]), knowledge, config)
|
||||
assert tags(result) == ["ta", "tb"]
|
||||
|
||||
|
||||
def test_expanded_tag_traceability() -> None:
|
||||
knowledge = kb(
|
||||
entry("school_uniform", ["school uniform"], category="clothing", expand=["blazer"]),
|
||||
entry("blazer", ["blazer"], category="clothing"),
|
||||
)
|
||||
result = resolve_scene(scene_with(clothing=["school uniform"]), knowledge, default_config())
|
||||
blazer_tag = result.data[1]
|
||||
assert blazer_tag.tag == "blazer"
|
||||
assert blazer_tag.knowledge_base_entry == "blazer"
|
||||
assert blazer_tag.source_concept == "school uniform" # originating scene concept
|
||||
|
||||
|
||||
# --- deduplication ---------------------------------------------------------
|
||||
|
||||
|
||||
def test_duplicate_tags_removed_keeping_first() -> None:
|
||||
knowledge = kb(
|
||||
entry("a", ["shared"]),
|
||||
entry("b", ["shared"]),
|
||||
)
|
||||
result = resolve_scene(scene_with(identity=["a", "b"]), knowledge, default_config())
|
||||
assert tags(result) == ["shared"]
|
||||
assert "SC0007" in codes(result.warnings)
|
||||
|
||||
|
||||
# --- unknown / deprecated --------------------------------------------------
|
||||
|
||||
|
||||
def test_unknown_concept_warns_and_continues() -> None:
|
||||
knowledge = kb(entry("female", ["1girl"], category="character"))
|
||||
result = resolve_scene(scene_with(identity=["dragon", "female"]), knowledge, default_config())
|
||||
assert tags(result) == ["1girl"]
|
||||
assert "SC0001" in codes(result.warnings)
|
||||
|
||||
|
||||
def test_deprecated_concept_warns_but_emits_tags() -> None:
|
||||
knowledge = kb(entry("old_style", ["old style"], deprecated=True))
|
||||
result = resolve_scene(scene_with(identity=["old_style"]), knowledge, default_config())
|
||||
assert tags(result) == ["old style"]
|
||||
assert "SC0006" in codes(result.warnings)
|
||||
|
||||
|
||||
# --- error conditions stop -------------------------------------------------
|
||||
|
||||
|
||||
def test_circular_expansion_stops_with_sc0003() -> None:
|
||||
knowledge = kb(
|
||||
entry("a", ["ta"], expand=["b"]),
|
||||
entry("b", ["tb"], expand=["a"]),
|
||||
)
|
||||
result = resolve_scene(scene_with(identity=["a"]), knowledge, default_config())
|
||||
assert not result.success
|
||||
assert result.data is None
|
||||
assert "SC0003" in codes(result.errors)
|
||||
|
||||
|
||||
def test_invalid_category_stops_with_sc0008() -> None:
|
||||
knowledge = kb(entry("a", ["ta"], category="not_a_category"))
|
||||
result = resolve_scene(scene_with(identity=["a"]), knowledge, default_config())
|
||||
assert not result.success
|
||||
assert "SC0008" in codes(result.errors)
|
||||
|
||||
|
||||
# --- category preservation & determinism -----------------------------------
|
||||
|
||||
|
||||
def test_tag_keeps_entry_category() -> None:
|
||||
knowledge = kb(entry("blue_eyes", ["blue eyes"], category="eyes"))
|
||||
result = resolve_scene(scene_with(appearance=["blue_eyes"]), knowledge, default_config())
|
||||
assert result.data[0].category == "eyes"
|
||||
|
||||
|
||||
def test_discovery_order_preserved_across_scene() -> None:
|
||||
knowledge = kb(
|
||||
entry("female", ["1girl"], category="character"),
|
||||
entry("forest", ["forest"], category="environment"),
|
||||
)
|
||||
scene = scene_with(identity=["female"], environment=["forest"])
|
||||
result = resolve_scene(scene, knowledge, default_config())
|
||||
assert tags(result) == ["1girl", "forest"]
|
||||
|
||||
|
||||
def test_output_is_deterministic() -> None:
|
||||
knowledge = kb(
|
||||
entry("school_uniform", ["school uniform"], category="clothing", expand=["blazer"]),
|
||||
entry("blazer", ["blazer"], category="clothing"),
|
||||
)
|
||||
scene = scene_with(clothing=["school uniform", "school uniform"])
|
||||
a = resolve_scene(scene, knowledge, default_config())
|
||||
b = resolve_scene(scene, knowledge, default_config())
|
||||
assert tags(a) == tags(b)
|
||||
assert codes(a.warnings) == codes(b.warnings)
|
||||
Reference in New Issue
Block a user