* feat(kb): automatic KB builder, ingestion & validation tooling (#35) Closes #118, #119, #120, #121, #122. - Ingest committed Danbooru alias/implication snapshots during generation: synonyms reach canonicals (long black stockings -> thighhighs), implications become expand relations. Curated-wins preserved. - Automatic candidate validator (structural rules reused from kb_validation + confidence heuristic, no approval gate; SC0022). - Duplicate/conflict detection tool (scripts/detect_kb_conflicts.py) + reusable detect_conflicts; clean reference KB reports zero. - KB coverage benchmark harness (scripts/coverage_benchmark.py) + JSON summary. - Single fully-automated, CI-runnable build pipeline (scripts/build_knowledge_base.py) that fails fast on rejections/conflicts and never prompts. Docs added. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * test(kb): use pytest.approx for confidence float assertions (SonarCloud reliability) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> --------- 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
Claude Opus 4.8
kunail0804
Claude Opus 4.8
kunail0804
parent
bb846c8aa0
commit
50e7b0a8a4
@@ -7,6 +7,16 @@ All notable changes to Scene Compiler are documented here. The project follows
|
||||
|
||||
### Added
|
||||
|
||||
- **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
|
||||
candidates → ingest Danbooru aliases/implications (`data/danbooru_aliases.txt`,
|
||||
`data/danbooru_implications.txt`) → auto-validate (structural rules + confidence,
|
||||
no approval gate) → conflict-scan → write. Synonyms now reach their canonical
|
||||
(`long black stockings` → `thighhighs`). New standalone tools:
|
||||
`scripts/detect_kb_conflicts.py` (CI-gateable conflict scan) and
|
||||
`scripts/coverage_benchmark.py` (KB coverage report). See
|
||||
`docs/knowledge_base_build.md`.
|
||||
- **Knowledge Base versioning (epic #36).** A Knowledge Base directory can now
|
||||
ship a `manifest.json` (new `knowledge_base_manifest` schema) recording a
|
||||
dataset `version`, a deterministic `content_hash`, the `source_vocab` digest,
|
||||
|
||||
@@ -0,0 +1,135 @@
|
||||
"""Automatic acceptance validator for generated Knowledge Base candidates (#119).
|
||||
|
||||
Per the epic #35 direction (**zero user input, no approval gate**), generated
|
||||
candidate entries are accepted or rejected **automatically** — there is no
|
||||
human-in-the-loop prompt. This module combines:
|
||||
|
||||
- **Structural rules** — reused from :mod:`compiler.common.kb_validation`
|
||||
(valid id pattern, known category, per-entry schema) plus cross-entry checks a
|
||||
candidate must satisfy against the existing dataset: no id/alias collision with
|
||||
curated entries, expansion targets exist, no self-cycle.
|
||||
- **Confidence heuristics** — a deterministic score in ``[0, 1]``: a candidate
|
||||
must be a real source-vocab tag (no invention) and map to a known category. A
|
||||
candidate below :data:`DEFAULT_CONFIDENCE_THRESHOLD` is rejected.
|
||||
|
||||
A rejected candidate carries structured :class:`Message` reasons for logging; the
|
||||
pipeline (#122) records them and never prompts.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Collection, Mapping
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
|
||||
from compiler.common.categories import is_valid_category
|
||||
from compiler.common.kb_validation import validate_entry
|
||||
from compiler.common.message_codes import message
|
||||
from compiler.common.result import Message
|
||||
|
||||
# Weighting: source-vocab membership dominates (no invention), a known category
|
||||
# completes the score. The default threshold requires source-vocab membership.
|
||||
_VOCAB_WEIGHT = 0.7
|
||||
_CATEGORY_WEIGHT = 0.3
|
||||
DEFAULT_CONFIDENCE_THRESHOLD = 0.7
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CandidateDecision:
|
||||
"""The automatic accept/reject outcome for one candidate entry."""
|
||||
|
||||
entry: Mapping[str, Any]
|
||||
accepted: bool
|
||||
confidence: float
|
||||
reasons: list[Message] = field(default_factory=list)
|
||||
|
||||
|
||||
def compute_confidence(entry: Mapping[str, Any], source_vocab: Collection[str]) -> float:
|
||||
"""Deterministic confidence in ``[0, 1]`` for a candidate entry."""
|
||||
score = 0.0
|
||||
if entry.get("id") in source_vocab:
|
||||
score += _VOCAB_WEIGHT
|
||||
if is_valid_category(entry.get("category")):
|
||||
score += _CATEGORY_WEIGHT
|
||||
return round(score, 6)
|
||||
|
||||
|
||||
def validate_candidate(
|
||||
entry: Mapping[str, Any],
|
||||
*,
|
||||
curated_ids: Collection[str],
|
||||
curated_aliases: Collection[str],
|
||||
known_ids: Collection[str],
|
||||
source_vocab: Collection[str],
|
||||
confidence_threshold: float = DEFAULT_CONFIDENCE_THRESHOLD,
|
||||
) -> CandidateDecision:
|
||||
"""Decide whether a generated candidate entry is auto-accepted.
|
||||
|
||||
Args:
|
||||
entry: The candidate entry (parsed dict).
|
||||
curated_ids / curated_aliases: Ids/aliases owned by curated entries
|
||||
(curated always wins — a candidate colliding with them is rejected).
|
||||
known_ids: All canonical ids that will exist (curated + generated), used to
|
||||
check that expansion targets resolve.
|
||||
source_vocab: The committed source-vocab tag set; the candidate id must be
|
||||
a member (no invention).
|
||||
confidence_threshold: Minimum confidence to accept.
|
||||
"""
|
||||
reasons: list[Message] = list(validate_entry(entry))
|
||||
entry_id = entry.get("id", "<unknown>")
|
||||
|
||||
if entry_id in curated_ids:
|
||||
reasons.append(
|
||||
message(
|
||||
"SC0005",
|
||||
f"Candidate id '{entry_id}' collides with a curated entry (curated wins).",
|
||||
id=entry_id,
|
||||
)
|
||||
)
|
||||
for alias in entry.get("aliases", ()):
|
||||
if alias in curated_ids or alias in curated_aliases:
|
||||
reasons.append(
|
||||
message(
|
||||
"SC0004",
|
||||
f"Candidate alias '{alias}' collides with a curated id/alias.",
|
||||
id=entry_id,
|
||||
alias=alias,
|
||||
)
|
||||
)
|
||||
for target in entry.get("expand", ()):
|
||||
if target == entry_id:
|
||||
reasons.append(
|
||||
message(
|
||||
"SC0003",
|
||||
f"Candidate '{entry_id}' expands to itself.",
|
||||
id=entry_id,
|
||||
)
|
||||
)
|
||||
elif target not in known_ids:
|
||||
reasons.append(
|
||||
message(
|
||||
"SC0004",
|
||||
f"Candidate '{entry_id}' expands to unknown id '{target}'.",
|
||||
id=entry_id,
|
||||
target=target,
|
||||
)
|
||||
)
|
||||
|
||||
confidence = compute_confidence(entry, source_vocab)
|
||||
if confidence < confidence_threshold:
|
||||
reasons.append(
|
||||
message(
|
||||
"SC0022",
|
||||
(
|
||||
f"Candidate '{entry_id}' rejected: confidence {confidence:.2f} is below "
|
||||
f"threshold {confidence_threshold:.2f} (not a real source-vocab tag?)."
|
||||
),
|
||||
id=entry_id,
|
||||
confidence=confidence,
|
||||
threshold=confidence_threshold,
|
||||
)
|
||||
)
|
||||
|
||||
return CandidateDecision(
|
||||
entry=entry, accepted=not reasons, confidence=confidence, reasons=reasons
|
||||
)
|
||||
@@ -0,0 +1,116 @@
|
||||
"""Knowledge Base duplicate & conflict detection (issue #120, epic #35).
|
||||
|
||||
As the Knowledge Base grows automatically, silent duplicates or conflicting
|
||||
entries degrade resolution. This module scans a whole Knowledge Base and reports —
|
||||
**detection only, it never edits** — four conflict classes:
|
||||
|
||||
- ``duplicate_ids`` — a canonical id defined by more than one entry.
|
||||
- ``duplicate_aliases`` — an alias string claimed by more than one entry.
|
||||
- ``tag_multiple_owners`` — a tag emitted by more than one entry with **no
|
||||
curated owner** to disambiguate it. The curated-wins additive-merge is respected:
|
||||
a tag shared between a curated entry and generated ones (or between two curated
|
||||
entries — an intentional cross-category synonym) is *not* a conflict, because the
|
||||
curated entry is authoritative.
|
||||
- ``contradictory_expansions`` — a mutual expansion pair (``a`` expands to ``b`` and
|
||||
``b`` expands to ``a``), which pulls each concept into the other.
|
||||
|
||||
The report is machine-readable (:meth:`ConflictReport.to_json`) with a human
|
||||
summary count (:attr:`ConflictReport.total`), so it can drive a CI gate.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import defaultdict
|
||||
from collections.abc import Collection, Iterable, Mapping
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ConflictReport:
|
||||
"""The result of a Knowledge Base conflict scan."""
|
||||
|
||||
duplicate_ids: list[dict[str, Any]] = field(default_factory=list)
|
||||
duplicate_aliases: list[dict[str, Any]] = field(default_factory=list)
|
||||
tag_multiple_owners: list[dict[str, Any]] = field(default_factory=list)
|
||||
contradictory_expansions: list[dict[str, Any]] = field(default_factory=list)
|
||||
|
||||
@property
|
||||
def total(self) -> int:
|
||||
return (
|
||||
len(self.duplicate_ids)
|
||||
+ len(self.duplicate_aliases)
|
||||
+ len(self.tag_multiple_owners)
|
||||
+ len(self.contradictory_expansions)
|
||||
)
|
||||
|
||||
def to_json(self) -> dict[str, Any]:
|
||||
return {
|
||||
"total": self.total,
|
||||
"duplicate_ids": self.duplicate_ids,
|
||||
"duplicate_aliases": self.duplicate_aliases,
|
||||
"tag_multiple_owners": self.tag_multiple_owners,
|
||||
"contradictory_expansions": self.contradictory_expansions,
|
||||
}
|
||||
|
||||
|
||||
def detect_conflicts(
|
||||
entries: Iterable[Mapping[str, Any]],
|
||||
curated_ids: Collection[str] = (),
|
||||
) -> ConflictReport:
|
||||
"""Scan a Knowledge Base for duplicate/conflict classes.
|
||||
|
||||
Args:
|
||||
entries: All Knowledge Base entries (parsed dicts) across every file.
|
||||
curated_ids: Ids owned by the hand-curated files; used so a tag with a
|
||||
curated owner is treated as authoritative (not a conflict).
|
||||
"""
|
||||
entries = list(entries)
|
||||
curated = set(curated_ids)
|
||||
|
||||
id_counts: dict[str, int] = defaultdict(int)
|
||||
alias_owners: dict[str, list[str]] = defaultdict(list)
|
||||
tag_owners: dict[str, list[str]] = defaultdict(list)
|
||||
expand: dict[str, set[str]] = {}
|
||||
|
||||
for entry in entries:
|
||||
entry_id = entry.get("id")
|
||||
if not isinstance(entry_id, str):
|
||||
continue
|
||||
id_counts[entry_id] += 1
|
||||
for alias in entry.get("aliases", ()):
|
||||
alias_owners[alias].append(entry_id)
|
||||
for tag in entry.get("tags", ()):
|
||||
tag_owners[tag].append(entry_id)
|
||||
expand[entry_id] = set(entry.get("expand", ()))
|
||||
|
||||
duplicate_ids = [
|
||||
{"id": entry_id, "count": count}
|
||||
for entry_id, count in sorted(id_counts.items())
|
||||
if count > 1
|
||||
]
|
||||
duplicate_aliases = [
|
||||
{"alias": alias, "owners": sorted(owners)}
|
||||
for alias, owners in sorted(alias_owners.items())
|
||||
if len(owners) > 1
|
||||
]
|
||||
|
||||
tag_multiple_owners = [
|
||||
{"tag": tag, "owners": sorted(owners)}
|
||||
for tag, owners in sorted(tag_owners.items())
|
||||
if len(owners) > 1 and not any(owner in curated for owner in owners)
|
||||
]
|
||||
|
||||
contradictory_expansions = [
|
||||
{"pair": [a, b]}
|
||||
for a in sorted(expand)
|
||||
for b in sorted(expand[a])
|
||||
if a < b and a in expand.get(b, set())
|
||||
]
|
||||
|
||||
return ConflictReport(
|
||||
duplicate_ids=duplicate_ids,
|
||||
duplicate_aliases=duplicate_aliases,
|
||||
tag_multiple_owners=tag_multiple_owners,
|
||||
contradictory_expansions=contradictory_expansions,
|
||||
)
|
||||
@@ -197,6 +197,17 @@ def _find_cycles(graph: Mapping[str, list[str]], ids: set[str]) -> list[list[str
|
||||
return cycles
|
||||
|
||||
|
||||
def validate_entry(entry: Mapping[str, Any]) -> list[Message]:
|
||||
"""Validate a single entry's structure (schema + category), no cross-entry rules.
|
||||
|
||||
Reuses the same schema/category checks as :func:`validate_knowledge_base` so
|
||||
per-entry tooling (e.g. the automatic candidate validator, #119) stays in sync
|
||||
with the authoritative rules instead of duplicating them.
|
||||
"""
|
||||
entry_id = entry.get("id", "<unknown>")
|
||||
return _schema_messages(entry, entry_id) + _category_messages(entry, entry_id)
|
||||
|
||||
|
||||
def validate_knowledge_base(entries: Iterable[Mapping[str, Any]]) -> list[Message]:
|
||||
"""Validate a whole Knowledge Base and return every problem found.
|
||||
|
||||
|
||||
@@ -51,6 +51,7 @@ _EXTENSIONS: dict[str, CodeSpec] = {
|
||||
"SC0018": CodeSpec(Severity.ERROR, "Analyzer Unexpected Response"),
|
||||
"SC0019": CodeSpec(Severity.WARNING, "Concept Reduced"),
|
||||
"SC0021": CodeSpec(Severity.WARNING, "List Under-Transcription"),
|
||||
"SC0022": CodeSpec(Severity.WARNING, "Candidate Rejected"),
|
||||
}
|
||||
|
||||
CODES: dict[str, CodeSpec] = {**_APPENDIX_B, **_EXTENSIONS}
|
||||
|
||||
@@ -0,0 +1,34 @@
|
||||
# Representative concept phrases for the KB coverage benchmark (issue #121).
|
||||
# One phrase per line; blank lines and lines starting with '#' are ignored.
|
||||
# A deterministic, curated corpus spanning categories and resolution paths.
|
||||
1girl
|
||||
long hair
|
||||
blue eyes
|
||||
smile
|
||||
blush
|
||||
school uniform
|
||||
thighhighs
|
||||
stockings
|
||||
jacket
|
||||
open white shirt
|
||||
white summer dress
|
||||
black pleated skirt
|
||||
sitting
|
||||
standing
|
||||
walking
|
||||
holding hands
|
||||
sword
|
||||
book
|
||||
umbrella
|
||||
classroom
|
||||
forest
|
||||
castle
|
||||
beach
|
||||
sunset
|
||||
moonlight
|
||||
close-up
|
||||
wide shot
|
||||
cat ears
|
||||
armor
|
||||
quantum flux inverter
|
||||
eldritch nonexistent artifact
|
||||
@@ -0,0 +1,3 @@
|
||||
spectacles glasses
|
||||
stockings thighhighs
|
||||
tights pantyhose
|
||||
@@ -0,0 +1,4 @@
|
||||
elbow_gloves gloves
|
||||
katana weapon
|
||||
mini_hat hat
|
||||
thigh_boots boots
|
||||
@@ -0,0 +1,26 @@
|
||||
{
|
||||
"coverage_rate": 0.9032,
|
||||
"covered": 28,
|
||||
"hit_rate": {
|
||||
"dropped": 3,
|
||||
"exact": 25,
|
||||
"fallback": 0,
|
||||
"head_noun": 3
|
||||
},
|
||||
"per_category": {
|
||||
"action": 1,
|
||||
"body": 1,
|
||||
"camera": 1,
|
||||
"character": 1,
|
||||
"clothing": 8,
|
||||
"environment": 4,
|
||||
"expression": 2,
|
||||
"eyes": 1,
|
||||
"hair": 1,
|
||||
"interaction": 1,
|
||||
"lighting": 2,
|
||||
"objects": 3,
|
||||
"pose": 2
|
||||
},
|
||||
"total": 31
|
||||
}
|
||||
@@ -0,0 +1,49 @@
|
||||
# Rebuilding the Knowledge Base
|
||||
|
||||
The generated Knowledge Base is produced by a single, fully-automated pipeline
|
||||
with **zero human input** (epic #35). It regenerates the `gen_*.json` files and the
|
||||
`manifest.json` deterministically from the committed source snapshots under
|
||||
`data/`.
|
||||
|
||||
## One command
|
||||
|
||||
```bash
|
||||
python scripts/build_knowledge_base.py
|
||||
```
|
||||
|
||||
The pipeline runs:
|
||||
|
||||
1. **build vocab** (dev-only) — rebuilds `data/danbooru_vocab.txt` from an external
|
||||
Danbooru/Raffle export. Skipped automatically when that source is absent (e.g.
|
||||
in CI), so the committed snapshot is used and no network access is needed.
|
||||
2. **generate candidates** from the vocab snapshot.
|
||||
3. **ingest aliases/implications** from `data/danbooru_aliases.txt` and
|
||||
`data/danbooru_implications.txt` (synonyms → canonical aliases, implications →
|
||||
`expand`).
|
||||
4. **auto-validate** every candidate (structural rules reused from
|
||||
`compiler/common/kb_validation.py` + a confidence heuristic — no approval gate).
|
||||
5. **conflict-scan** the whole Knowledge Base (duplicate ids/aliases, ambiguous tag
|
||||
ownership, contradictory expansions).
|
||||
6. **write** `gen_*.json` and stamp `manifest.json` — only if steps 4–5 are clean.
|
||||
|
||||
On any rejected candidate or detected conflict the pipeline writes a structured
|
||||
JSON log to stderr and exits non-zero **without writing** — it never prompts. The
|
||||
curated-wins additive merge is preserved end-to-end: a candidate colliding with a
|
||||
curated id/alias is rejected.
|
||||
|
||||
## Source snapshots (committed)
|
||||
|
||||
| File | Built by | Contents |
|
||||
|---|---|---|
|
||||
| `data/danbooru_vocab.txt` | `scripts/build_vocab.py` | `raffle_category<TAB>tag` |
|
||||
| `data/danbooru_aliases.txt` | `scripts/build_aliases.py` | `alias<TAB>canonical_id` |
|
||||
| `data/danbooru_implications.txt` | `scripts/build_aliases.py` | `antecedent_id<TAB>consequent_id` |
|
||||
|
||||
The `build_*` scripts read an external export from a hard-coded path (edit the
|
||||
constant before running) and are occasional dev steps, not part of runtime or CI.
|
||||
|
||||
## Related tools
|
||||
|
||||
- `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.
|
||||
@@ -25,7 +25,7 @@
|
||||
{ "id": "shoes", "tags": ["shoes"], "category": "clothing" },
|
||||
{ "id": "gloves", "tags": ["gloves"], "category": "clothing" },
|
||||
{ "id": "socks", "tags": ["socks"], "category": "clothing" },
|
||||
{ "id": "thighhighs", "aliases": ["thigh highs"], "tags": ["thighhighs"], "category": "clothing" },
|
||||
{ "id": "thighhighs", "aliases": ["thigh highs", "stockings"], "tags": ["thighhighs"], "category": "clothing" },
|
||||
|
||||
{ "id": "hat", "tags": ["hat"], "category": "accessories" },
|
||||
{ "id": "glasses", "aliases": ["eyeglasses", "spectacles"], "tags": ["glasses"], "category": "accessories" },
|
||||
|
||||
@@ -9824,7 +9824,10 @@
|
||||
"tags": [
|
||||
"elbow gloves"
|
||||
],
|
||||
"category": "clothing"
|
||||
"category": "clothing",
|
||||
"expand": [
|
||||
"gloves"
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": "elbow_pads",
|
||||
@@ -18886,7 +18889,10 @@
|
||||
"tags": [
|
||||
"mini hat"
|
||||
],
|
||||
"category": "clothing"
|
||||
"category": "clothing",
|
||||
"expand": [
|
||||
"hat"
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": "mini_santa_hat",
|
||||
@@ -22164,7 +22170,10 @@
|
||||
"tags": [
|
||||
"pantyhose"
|
||||
],
|
||||
"category": "clothing"
|
||||
"category": "clothing",
|
||||
"aliases": [
|
||||
"tights"
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": "pantyhose_around_one_leg",
|
||||
@@ -32568,7 +32577,10 @@
|
||||
"tags": [
|
||||
"thigh boots"
|
||||
],
|
||||
"category": "clothing"
|
||||
"category": "clothing",
|
||||
"expand": [
|
||||
"boots"
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": "thigh_bow",
|
||||
|
||||
@@ -14179,7 +14179,10 @@
|
||||
"tags": [
|
||||
"katana"
|
||||
],
|
||||
"category": "objects"
|
||||
"category": "objects",
|
||||
"expand": [
|
||||
"weapon"
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": "kayari_buta",
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
{
|
||||
"content_hash": "sha256:a9f132307076670a20999a5be768537c2a6b93f7a0afdd41ce46a3da08175b68",
|
||||
"content_hash": "sha256:88f8c3ce21e3433570b88c38841f5b356e0516b86477b740682c896274005372",
|
||||
"manifest_schema_version": "1.0",
|
||||
"source_vocab": "sha256:d9238cecb70736aa81e1297b174d2ed0c4555c0b214651d33c955d9e5ce0e88b",
|
||||
"version": "1.0.0"
|
||||
|
||||
@@ -0,0 +1,70 @@
|
||||
"""Build the committed Danbooru alias/implication snapshots.
|
||||
|
||||
Danbooru publishes tag **aliases** (synonym → canonical, e.g. ``stockings`` →
|
||||
``thighhighs``) and **implications** (a tag implies another, e.g. ``elbow_gloves``
|
||||
implies ``gloves``). Ingesting them lets synonyms reach their canonical entry and
|
||||
turns implications into Knowledge Base ``expand`` relations
|
||||
(``scripts/generate_kb_from_vocab.py``).
|
||||
|
||||
Like ``build_vocab.py``, this is an occasional dev step, not part of runtime/CI.
|
||||
It reads a Danbooru export from a hard-coded path (no CLI/untrusted input) and
|
||||
writes two compact, deterministic, committed snapshots:
|
||||
|
||||
- ``data/danbooru_aliases.txt`` — ``alias<TAB>canonical_id`` per line
|
||||
- ``data/danbooru_implications.txt`` — ``antecedent_id<TAB>consequent_id`` per line
|
||||
|
||||
Only rows whose names are valid Knowledge Base ids (lowercase alphanumerics joined
|
||||
by single underscores) are kept, so no symbol/emoticon tag can enter the KB. The
|
||||
generator consumes only these committed snapshots, never the external export, so
|
||||
the repository stays self-contained and CI never needs network access.
|
||||
|
||||
Edit ``_ALIAS_SOURCE`` / ``_IMPLICATION_SOURCE`` to point at your local Danbooru
|
||||
export (CSV with ``antecedent_name,consequent_name`` columns) before running.
|
||||
|
||||
Usage:
|
||||
python scripts/build_aliases.py
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import csv
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
_REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
_ALIAS_SOURCE = Path("/home/kunail/Documents/AIStuff/danbooru/tag_aliases.csv")
|
||||
_IMPLICATION_SOURCE = Path("/home/kunail/Documents/AIStuff/danbooru/tag_implications.csv")
|
||||
_ALIAS_OUTPUT = _REPO_ROOT / "data" / "danbooru_aliases.txt"
|
||||
_IMPLICATION_OUTPUT = _REPO_ROOT / "data" / "danbooru_implications.txt"
|
||||
|
||||
_ID_PATTERN = re.compile(r"^[a-z0-9]+(_[a-z0-9]+)*$")
|
||||
|
||||
|
||||
def _read_pairs(source: Path) -> list[tuple[str, str]]:
|
||||
"""Return kept ``(antecedent, consequent)`` rows, sorted and deduplicated."""
|
||||
kept: dict[str, str] = {} # antecedent -> consequent (first wins)
|
||||
with source.open(encoding="utf-8", newline="") as handle:
|
||||
for row in csv.DictReader(handle):
|
||||
antecedent = (row.get("antecedent_name") or "").strip()
|
||||
consequent = (row.get("consequent_name") or "").strip()
|
||||
if not _ID_PATTERN.match(antecedent) or not _ID_PATTERN.match(consequent):
|
||||
continue
|
||||
if antecedent == consequent:
|
||||
continue
|
||||
kept.setdefault(antecedent, consequent)
|
||||
return sorted(kept.items())
|
||||
|
||||
|
||||
def _write(pairs: list[tuple[str, str]], output: Path) -> None:
|
||||
output.parent.mkdir(parents=True, exist_ok=True)
|
||||
output.write_text("".join(f"{a}\t{b}\n" for a, b in pairs), encoding="utf-8")
|
||||
print(f"wrote {len(pairs)} rows to {output.relative_to(_REPO_ROOT)}")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
_write(_read_pairs(_ALIAS_SOURCE), _ALIAS_OUTPUT)
|
||||
_write(_read_pairs(_IMPLICATION_SOURCE), _IMPLICATION_OUTPUT)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,131 @@
|
||||
"""Fully-automated Knowledge Base build pipeline (issue #122, epic #35).
|
||||
|
||||
One command runs the whole generation flow with **zero human input**:
|
||||
|
||||
build vocab (dev-only) → generate candidates → ingest aliases/implications
|
||||
→ auto-validate candidates → conflict-scan → write gen_*.json + manifest
|
||||
|
||||
- The vocab build reads an external Danbooru/Raffle export and is skipped when that
|
||||
source is absent (e.g. in CI); the committed ``data/danbooru_vocab.txt`` snapshot
|
||||
is used instead, so the pipeline is CI-runnable with no network access.
|
||||
- Candidates are validated automatically (structural rules + confidence, #119) and
|
||||
the resulting Knowledge Base is conflict-scanned (#120). On any rejection or
|
||||
conflict the pipeline writes a structured JSON log to stderr and exits non-zero
|
||||
**without writing** — it never prompts.
|
||||
- The curated-wins additive merge is preserved end-to-end (candidates colliding
|
||||
with curated ids/aliases are rejected).
|
||||
|
||||
Usage:
|
||||
python scripts/build_knowledge_base.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.kb_autovalidate import validate_candidate # noqa: E402
|
||||
from compiler.common.kb_conflicts import detect_conflicts # noqa: E402
|
||||
from scripts import build_vocab # noqa: E402
|
||||
from scripts import generate_kb_from_vocab as gen # noqa: E402
|
||||
|
||||
_REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
_KB_DIR = _REPO_ROOT / "knowledge_base"
|
||||
_VOCAB = _REPO_ROOT / "data" / "danbooru_vocab.txt"
|
||||
_GEN_PREFIX = "gen_"
|
||||
_MANIFEST = "manifest.json"
|
||||
|
||||
|
||||
def _source_vocab() -> set[str]:
|
||||
"""The committed source-vocab tag set (candidate ids must be members)."""
|
||||
tags: set[str] = set()
|
||||
for line in _VOCAB.read_text(encoding="utf-8").splitlines():
|
||||
if line.strip():
|
||||
tags.add(line.split("\t", 1)[1].strip())
|
||||
return tags
|
||||
|
||||
|
||||
def _curated() -> tuple[list[dict], set[str], set[str]]:
|
||||
"""Return (curated entries, curated ids, curated aliases)."""
|
||||
entries: list[dict] = []
|
||||
ids: set[str] = set()
|
||||
aliases: set[str] = set()
|
||||
for path in sorted(_KB_DIR.glob("*.json")):
|
||||
if path.name.startswith(_GEN_PREFIX) or path.name == _MANIFEST:
|
||||
continue
|
||||
file_entries = json.loads(path.read_text(encoding="utf-8"))
|
||||
entries.extend(file_entries)
|
||||
for entry in file_entries:
|
||||
ids.add(entry["id"])
|
||||
aliases.update(entry.get("aliases", ()))
|
||||
return entries, ids, aliases
|
||||
|
||||
|
||||
def validate_candidates(
|
||||
by_category: dict[str, list[dict]],
|
||||
curated_ids: set[str],
|
||||
curated_aliases: set[str],
|
||||
source_vocab: set[str],
|
||||
) -> list[dict]:
|
||||
"""Auto-validate every candidate; return structured rejection records."""
|
||||
candidates = [entry for entries in by_category.values() for entry in entries]
|
||||
generated_ids = {entry["id"] for entry in candidates}
|
||||
known_ids = curated_ids | generated_ids
|
||||
rejections: list[dict] = []
|
||||
for entry in candidates:
|
||||
decision = validate_candidate(
|
||||
entry,
|
||||
curated_ids=curated_ids,
|
||||
curated_aliases=curated_aliases,
|
||||
known_ids=known_ids,
|
||||
source_vocab=source_vocab,
|
||||
)
|
||||
if not decision.accepted:
|
||||
rejections.append(
|
||||
{"id": entry.get("id"), "reasons": [m.description for m in decision.reasons]}
|
||||
)
|
||||
return rejections
|
||||
|
||||
|
||||
def run_pipeline(write: bool = True) -> int:
|
||||
"""Run the full pipeline. Return 0 on success, 1 on rejection/conflict."""
|
||||
if build_vocab._SOURCE.is_file():
|
||||
build_vocab.main()
|
||||
else:
|
||||
print(f"[pipeline] external vocab source absent; using committed {_VOCAB.name}")
|
||||
|
||||
by_category = gen.build_generated_entries()
|
||||
curated_entries, curated_ids, curated_aliases = _curated()
|
||||
source_vocab = _source_vocab()
|
||||
|
||||
rejections = validate_candidates(by_category, curated_ids, curated_aliases, source_vocab)
|
||||
candidates = [entry for entries in by_category.values() for entry in entries]
|
||||
report = detect_conflicts(curated_entries + candidates, curated_ids)
|
||||
|
||||
if rejections or report.total:
|
||||
sys.stderr.write(
|
||||
json.dumps(
|
||||
{
|
||||
"status": "failed",
|
||||
"rejected_candidates": rejections,
|
||||
"conflicts": report.to_json(),
|
||||
},
|
||||
indent=2,
|
||||
)
|
||||
+ "\n"
|
||||
)
|
||||
return 1
|
||||
|
||||
if write:
|
||||
counts = gen.write_generated(by_category)
|
||||
print(f"[pipeline] wrote {sum(counts.values())} generated entries; manifest stamped.")
|
||||
else:
|
||||
print(f"[pipeline] dry run OK: {len(candidates)} candidates, no conflicts.")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(run_pipeline())
|
||||
@@ -0,0 +1,140 @@
|
||||
"""Knowledge Base coverage benchmark (issue #121, epic #35).
|
||||
|
||||
Runs a fixed corpus of representative concept phrases through the deterministic
|
||||
Resolver and measures how well the Knowledge Base covers them:
|
||||
|
||||
- **hit rate** by resolution path — ``exact`` (direct id/alias hit), ``head_noun``
|
||||
(recovered via the head-noun reduction, ``SC0019``), ``fallback`` (semantic
|
||||
nearest-neighbour, ``SC0020``, when that feature is enabled), and ``dropped``
|
||||
(no entry, ``SC0001``).
|
||||
- **per-category coverage** — the category of the first resolved tag for each
|
||||
covered phrase.
|
||||
|
||||
The report is informational (no hard threshold gate — one can be added later). It
|
||||
is deterministic: a fixed corpus and no live model. Prints a human summary and
|
||||
writes a JSON summary.
|
||||
|
||||
Usage:
|
||||
python scripts/coverage_benchmark.py
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sys
|
||||
from collections import Counter
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from compiler.common.config import Config # noqa: E402
|
||||
from compiler.common.knowledge_base import KnowledgeBase, load_knowledge_base # noqa: E402
|
||||
from compiler.resolver.illustrious_resolver import resolve_scene # noqa: E402
|
||||
from schemas.models import Scene # noqa: E402
|
||||
|
||||
_REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
_KB_DIR = _REPO_ROOT / "knowledge_base"
|
||||
_CORPUS = _REPO_ROOT / "data" / "coverage_corpus.txt"
|
||||
_REPORT = _REPO_ROOT / "docs" / "kb_coverage.json"
|
||||
|
||||
_PATHS = ("exact", "head_noun", "fallback", "dropped")
|
||||
|
||||
|
||||
def load_corpus(path: Path = _CORPUS) -> list[str]:
|
||||
"""Read the corpus file: one phrase per line, ignoring blanks and comments."""
|
||||
phrases = []
|
||||
for line in path.read_text(encoding="utf-8").splitlines():
|
||||
stripped = line.strip()
|
||||
if stripped and not stripped.startswith("#"):
|
||||
phrases.append(stripped)
|
||||
return phrases
|
||||
|
||||
|
||||
def _single_concept_scene(phrase: str) -> Scene:
|
||||
return Scene.from_json(
|
||||
{
|
||||
"characters": [
|
||||
{
|
||||
"id": 0,
|
||||
"identity": [phrase],
|
||||
"appearance": [],
|
||||
"clothing": [],
|
||||
"accessories": [],
|
||||
"pose": [],
|
||||
"expression": [],
|
||||
"actions": [],
|
||||
}
|
||||
],
|
||||
"interactions": [],
|
||||
"objects": [],
|
||||
"environment": [],
|
||||
"camera": [],
|
||||
"lighting": [],
|
||||
"metadata": {"schema_version": "1.0"},
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _classify(phrase: str, kb: KnowledgeBase, config: Config) -> tuple[str, str | None]:
|
||||
"""Return (resolution path, first resolved tag category) for one phrase."""
|
||||
result = resolve_scene(_single_concept_scene(phrase), kb, config)
|
||||
codes = {warning.code for warning in result.warnings}
|
||||
tags = result.data or ()
|
||||
category = tags[0].category if tags else None
|
||||
if "SC0001" in codes and not tags:
|
||||
return "dropped", None
|
||||
if "SC0020" in codes:
|
||||
return "fallback", category
|
||||
if "SC0019" in codes:
|
||||
return "head_noun", category
|
||||
return "exact", category
|
||||
|
||||
|
||||
def measure_coverage(kb: KnowledgeBase, phrases: list[str], config: Config | None = None) -> dict:
|
||||
"""Measure resolution hit-rate and per-category coverage over a phrase corpus."""
|
||||
config = config or Config()
|
||||
hits: Counter[str] = Counter({path: 0 for path in _PATHS})
|
||||
per_category: Counter[str] = Counter()
|
||||
for phrase in phrases:
|
||||
path, category = _classify(phrase, kb, config)
|
||||
hits[path] += 1
|
||||
if category is not None:
|
||||
per_category[category] += 1
|
||||
|
||||
total = len(phrases)
|
||||
covered = total - hits["dropped"]
|
||||
return {
|
||||
"total": total,
|
||||
"covered": covered,
|
||||
"coverage_rate": round(covered / total, 4) if total else 0.0,
|
||||
"hit_rate": {path: hits[path] for path in _PATHS},
|
||||
"per_category": dict(sorted(per_category.items())),
|
||||
}
|
||||
|
||||
|
||||
def run() -> dict:
|
||||
kb = load_knowledge_base(_KB_DIR)
|
||||
return measure_coverage(kb, load_corpus())
|
||||
|
||||
|
||||
def _format_human(summary: dict) -> str:
|
||||
lines = [
|
||||
f"KB coverage: {summary['covered']}/{summary['total']} "
|
||||
f"({summary['coverage_rate'] * 100:.1f}%)",
|
||||
" by path: " + ", ".join(f"{path}={count}" for path, count in summary["hit_rate"].items()),
|
||||
" by category: "
|
||||
+ ", ".join(f"{cat}={count}" for cat, count in summary["per_category"].items()),
|
||||
]
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
summary = run()
|
||||
_REPORT.write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
||||
print(_format_human(summary))
|
||||
print(f"wrote JSON summary to {_REPORT.relative_to(_REPO_ROOT)}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,58 @@
|
||||
"""CLI to scan a Knowledge Base for duplicates and conflicts (issue #120).
|
||||
|
||||
Loads every entry file under a Knowledge Base directory (curated ids are taken
|
||||
from the non-``gen_*`` files so curated-wins semantics are respected), runs the
|
||||
conflict detector, writes a machine-readable JSON report to stdout, and prints a
|
||||
one-line human summary to stderr. Exit code is 0 when the Knowledge Base is clean,
|
||||
1 when any conflict is found — so it can serve as a CI gate.
|
||||
|
||||
Usage:
|
||||
python scripts/detect_kb_conflicts.py knowledge_base/
|
||||
"""
|
||||
|
||||
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.kb_conflicts import detect_conflicts # noqa: E402
|
||||
|
||||
_GEN_PREFIX = "gen_"
|
||||
_MANIFEST = "manifest.json"
|
||||
|
||||
|
||||
def _load(directory: Path) -> tuple[list[dict], set[str]]:
|
||||
"""Return (all entries, curated ids) from a Knowledge Base directory."""
|
||||
entries: list[dict] = []
|
||||
curated_ids: set[str] = set()
|
||||
for path in sorted(directory.glob("*.json")):
|
||||
if path.name == _MANIFEST:
|
||||
continue
|
||||
file_entries = json.loads(path.read_text(encoding="utf-8"))
|
||||
entries.extend(file_entries)
|
||||
if not path.name.startswith(_GEN_PREFIX):
|
||||
curated_ids.update(e["id"] for e in file_entries if isinstance(e.get("id"), str))
|
||||
return entries, curated_ids
|
||||
|
||||
|
||||
def main(argv: list[str]) -> int:
|
||||
if len(argv) != 1:
|
||||
sys.stderr.write("usage: detect_kb_conflicts.py <knowledge_base_dir>\n")
|
||||
return 2
|
||||
directory = Path(argv[0])
|
||||
if not directory.is_dir():
|
||||
sys.stderr.write(f"not a directory: {directory}\n")
|
||||
return 2
|
||||
|
||||
entries, curated_ids = _load(directory)
|
||||
report = detect_conflicts(entries, curated_ids)
|
||||
sys.stdout.write(json.dumps(report.to_json(), indent=2, sort_keys=True) + "\n")
|
||||
sys.stderr.write(f"{report.total} conflict(s) found across {len(entries)} entries.\n")
|
||||
return 0 if report.total == 0 else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main(sys.argv[1:]))
|
||||
@@ -29,6 +29,8 @@ from compiler.common.kb_manifest import write_manifest # noqa: E402
|
||||
|
||||
_REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
_VOCAB = _REPO_ROOT / "data" / "danbooru_vocab.txt"
|
||||
_ALIASES = _REPO_ROOT / "data" / "danbooru_aliases.txt"
|
||||
_IMPLICATIONS = _REPO_ROOT / "data" / "danbooru_implications.txt"
|
||||
_KB_DIR = _REPO_ROOT / "knowledge_base"
|
||||
_GEN_PREFIX = "gen_"
|
||||
|
||||
@@ -96,7 +98,7 @@ def _load_hand_reserved() -> set[str]:
|
||||
"""Collect every id and alias from the hand-curated (non-generated) KB files."""
|
||||
reserved: set[str] = set()
|
||||
for path in sorted(_KB_DIR.glob("*.json")):
|
||||
if path.name.startswith(_GEN_PREFIX):
|
||||
if path.name.startswith(_GEN_PREFIX) or path.name == "manifest.json":
|
||||
continue
|
||||
for entry in json.loads(path.read_text(encoding="utf-8")):
|
||||
reserved.add(entry["id"])
|
||||
@@ -104,6 +106,46 @@ def _load_hand_reserved() -> set[str]:
|
||||
return reserved
|
||||
|
||||
|
||||
def _load_curated_ids() -> set[str]:
|
||||
"""Canonical ids from the hand-curated files (valid expansion targets)."""
|
||||
ids: set[str] = set()
|
||||
for path in sorted(_KB_DIR.glob("*.json")):
|
||||
if path.name.startswith(_GEN_PREFIX) or path.name == "manifest.json":
|
||||
continue
|
||||
for entry in json.loads(path.read_text(encoding="utf-8")):
|
||||
ids.add(entry["id"])
|
||||
return ids
|
||||
|
||||
|
||||
def _read_tsv_pairs(path: Path) -> list[tuple[str, str]]:
|
||||
"""Read ``a<TAB>b`` lines (deterministic order); empty when the file is absent."""
|
||||
if not path.is_file():
|
||||
return []
|
||||
pairs: list[tuple[str, str]] = []
|
||||
for line in path.read_text(encoding="utf-8").splitlines():
|
||||
if not line.strip():
|
||||
continue
|
||||
left, right = line.split("\t", 1)
|
||||
pairs.append((left.strip(), right.strip()))
|
||||
return pairs
|
||||
|
||||
|
||||
def _load_aliases_by_canonical() -> dict[str, list[str]]:
|
||||
"""Danbooru aliases grouped by canonical id: ``canonical -> [alias, ...]``."""
|
||||
by_canonical: dict[str, list[str]] = {}
|
||||
for alias, canonical in _read_tsv_pairs(_ALIASES):
|
||||
by_canonical.setdefault(canonical, []).append(alias)
|
||||
return by_canonical
|
||||
|
||||
|
||||
def _load_implications() -> dict[str, list[str]]:
|
||||
"""Danbooru implications grouped by antecedent: ``antecedent -> [consequent, ...]``."""
|
||||
by_antecedent: dict[str, list[str]] = {}
|
||||
for antecedent, consequent in _read_tsv_pairs(_IMPLICATIONS):
|
||||
by_antecedent.setdefault(antecedent, []).append(consequent)
|
||||
return by_antecedent
|
||||
|
||||
|
||||
def _read_vocab() -> list[tuple[str, str]]:
|
||||
pairs: list[tuple[str, str]] = []
|
||||
for line in _VOCAB.read_text(encoding="utf-8").splitlines():
|
||||
@@ -114,18 +156,61 @@ def _read_vocab() -> list[tuple[str, str]]:
|
||||
return pairs
|
||||
|
||||
|
||||
def generate() -> dict[str, int]:
|
||||
def _ingest_aliases_and_implications(
|
||||
by_category: dict[str, list[dict]],
|
||||
reserved: set[str],
|
||||
curated_ids: set[str],
|
||||
) -> None:
|
||||
"""Attach Danbooru aliases/implications to the generated entries (issue #118).
|
||||
|
||||
- Aliases whose canonical is a generated entry are attached to it, so synonyms
|
||||
resolve (``tights`` → ``pantyhose``). Aliases colliding with a curated
|
||||
id/alias or any generated id are skipped (curated always wins; no chains).
|
||||
- Implications whose antecedent is a generated entry become ``expand`` targets
|
||||
when the consequent exists (curated id or generated id). This is additive and
|
||||
deterministic; nothing is invented (data comes only from the committed
|
||||
Danbooru snapshots). Aliases whose canonical is a *curated* entry are owned by
|
||||
the curated files and left untouched here.
|
||||
"""
|
||||
generated_ids = {entry["id"] for entries in by_category.values() for entry in entries}
|
||||
known_targets = curated_ids | generated_ids
|
||||
aliases_by_canonical = _load_aliases_by_canonical()
|
||||
implications = _load_implications()
|
||||
|
||||
for entries in by_category.values():
|
||||
for entry in entries:
|
||||
entry_id = entry["id"]
|
||||
|
||||
new_aliases = sorted(
|
||||
alias
|
||||
for alias in aliases_by_canonical.get(entry_id, ())
|
||||
if alias not in reserved and alias not in generated_ids and alias != entry_id
|
||||
)
|
||||
if new_aliases:
|
||||
entry["aliases"] = new_aliases
|
||||
|
||||
new_expand = sorted(
|
||||
target
|
||||
for target in implications.get(entry_id, ())
|
||||
if target in known_targets and target != entry_id
|
||||
)
|
||||
if new_expand:
|
||||
entry["expand"] = new_expand
|
||||
|
||||
|
||||
def build_generated_entries() -> dict[str, list[dict]]:
|
||||
"""Build the generated entries (with alias/implication ingestion), sorted.
|
||||
|
||||
Pure and side-effect-free: it reads the committed snapshots and returns the
|
||||
``category -> [entry, ...]`` mapping without touching disk, so the pipeline
|
||||
(#122) can validate the candidates before anything is written.
|
||||
"""
|
||||
reserved = _load_hand_reserved()
|
||||
by_category: dict[str, list[dict]] = {}
|
||||
skipped_reserved = skipped_unmapped = 0
|
||||
|
||||
for raffle_category, tag in _read_vocab():
|
||||
sc_category = _CATEGORY_MAP.get(raffle_category)
|
||||
if sc_category is None:
|
||||
skipped_unmapped += 1
|
||||
continue
|
||||
if tag in reserved:
|
||||
skipped_reserved += 1
|
||||
if sc_category is None or tag in reserved:
|
||||
continue
|
||||
entry = {
|
||||
"id": tag,
|
||||
@@ -136,23 +221,33 @@ def generate() -> dict[str, int]:
|
||||
entry["rating"] = "explicit"
|
||||
by_category.setdefault(sc_category, []).append(entry)
|
||||
|
||||
_ingest_aliases_and_implications(by_category, reserved, _load_curated_ids())
|
||||
for entries in by_category.values():
|
||||
entries.sort(key=lambda e: e["id"])
|
||||
return by_category
|
||||
|
||||
|
||||
def write_generated(by_category: dict[str, list[dict]]) -> dict[str, int]:
|
||||
"""Write the generated entry files and stamp the manifest deterministically."""
|
||||
# Remove any stale generated files before writing the fresh set.
|
||||
for path in _KB_DIR.glob(f"{_GEN_PREFIX}*.json"):
|
||||
path.unlink()
|
||||
|
||||
counts: dict[str, int] = {}
|
||||
for sc_category, entries in sorted(by_category.items()):
|
||||
entries.sort(key=lambda e: e["id"])
|
||||
out = _KB_DIR / f"{_GEN_PREFIX}{sc_category}.json"
|
||||
out.write_text(json.dumps(entries, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
|
||||
counts[sc_category] = len(entries)
|
||||
|
||||
manifest = write_manifest(_KB_DIR, _DATASET_VERSION, _VOCAB)
|
||||
write_manifest(_KB_DIR, _DATASET_VERSION, _VOCAB)
|
||||
return counts
|
||||
|
||||
|
||||
def generate() -> dict[str, int]:
|
||||
by_category = build_generated_entries()
|
||||
counts = write_generated(by_category)
|
||||
total = sum(counts.values())
|
||||
print(f"generated {total} entries across {len(counts)} category files")
|
||||
print(f" stamped manifest version {manifest.version} ({manifest.content_hash[:19]}...)")
|
||||
print(f" skipped {skipped_reserved} (already hand-curated), {skipped_unmapped} (unmapped)")
|
||||
for cat, n in sorted(counts.items(), key=lambda x: -x[1]):
|
||||
print(f" {n:6} gen_{cat}.json")
|
||||
return counts
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
"""Tests for Danbooru alias/implication ingestion (issue #118, epic #35)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from compiler.common.knowledge_base import load_knowledge_base
|
||||
from scripts import generate_kb_from_vocab as gen
|
||||
|
||||
KB_DIR = Path(__file__).resolve().parent.parent / "knowledge_base"
|
||||
|
||||
|
||||
# --- dataset level: the shipped, regenerated KB ----------------------------
|
||||
|
||||
|
||||
def test_synonym_reaches_canonical_via_ingested_alias() -> None:
|
||||
kb = load_knowledge_base(KB_DIR)
|
||||
# The known V1 gap: "long black stockings" must reach "thighhighs".
|
||||
assert kb.lookup("stockings") is not None
|
||||
assert kb.lookup("stockings").id == "thighhighs"
|
||||
# An alias attached to a *generated* canonical (exercises ingestion on gen_*).
|
||||
assert kb.lookup("tights") is not None
|
||||
assert kb.lookup("tights").id == "pantyhose"
|
||||
|
||||
|
||||
def test_implication_becomes_expand_relation() -> None:
|
||||
kb = load_knowledge_base(KB_DIR)
|
||||
assert "gloves" in kb.get("elbow_gloves").expand
|
||||
assert "weapon" in kb.get("katana").expand
|
||||
|
||||
|
||||
# --- unit level: the ingestion function ------------------------------------
|
||||
|
||||
|
||||
def test_ingestion_attaches_alias_and_skips_reserved(monkeypatch) -> None:
|
||||
monkeypatch.setattr(
|
||||
gen, "_load_aliases_by_canonical", lambda: {"pantyhose": ["tights", "curated_word"]}
|
||||
)
|
||||
monkeypatch.setattr(gen, "_load_implications", lambda: {})
|
||||
by_category = {"clothing": [{"id": "pantyhose", "tags": ["pantyhose"], "category": "clothing"}]}
|
||||
# "curated_word" is reserved by a curated entry, so curated wins → skipped.
|
||||
gen._ingest_aliases_and_implications(by_category, {"curated_word"}, curated_ids=set())
|
||||
assert by_category["clothing"][0]["aliases"] == ["tights"]
|
||||
|
||||
|
||||
def test_ingestion_adds_expand_only_for_known_targets(monkeypatch) -> None:
|
||||
monkeypatch.setattr(gen, "_load_aliases_by_canonical", lambda: {})
|
||||
monkeypatch.setattr(
|
||||
gen, "_load_implications", lambda: {"elbow_gloves": ["gloves", "does_not_exist"]}
|
||||
)
|
||||
by_category = {
|
||||
"clothing": [{"id": "elbow_gloves", "tags": ["elbow gloves"], "category": "clothing"}]
|
||||
}
|
||||
gen._ingest_aliases_and_implications(by_category, set(), curated_ids={"gloves"})
|
||||
assert by_category["clothing"][0]["expand"] == ["gloves"] # unknown target dropped
|
||||
@@ -0,0 +1,46 @@
|
||||
"""Tests for the automated KB build pipeline (issue #122, epic #35)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from scripts import build_knowledge_base as pipe
|
||||
from scripts import generate_kb_from_vocab as gen
|
||||
|
||||
|
||||
def test_dry_run_pipeline_succeeds(monkeypatch) -> None:
|
||||
# Mirror CI: no external vocab source, so the committed snapshot is used.
|
||||
monkeypatch.setattr(pipe.build_vocab, "_SOURCE", Path("/nonexistent/source.txt"))
|
||||
assert pipe.run_pipeline(write=False) == 0 # clean: no rejections, no conflicts
|
||||
|
||||
|
||||
def test_candidate_generation_is_deterministic() -> None:
|
||||
assert gen.build_generated_entries() == gen.build_generated_entries()
|
||||
|
||||
|
||||
def test_validation_rejects_bad_candidates() -> None:
|
||||
by_category = {
|
||||
"objects": [
|
||||
{"id": "real_tag", "tags": ["real tag"], "category": "objects"},
|
||||
{"id": "bad_category", "tags": ["bad"], "category": "not_a_category"},
|
||||
]
|
||||
}
|
||||
rejections = pipe.validate_candidates(
|
||||
by_category,
|
||||
curated_ids=set(),
|
||||
curated_aliases=set(),
|
||||
source_vocab={"real_tag", "bad_category"},
|
||||
)
|
||||
ids = {r["id"] for r in rejections}
|
||||
assert ids == {"bad_category"} # only the malformed candidate is rejected
|
||||
|
||||
|
||||
def test_collision_with_curated_is_rejected() -> None:
|
||||
by_category = {"objects": [{"id": "sword", "tags": ["sword"], "category": "objects"}]}
|
||||
rejections = pipe.validate_candidates(
|
||||
by_category,
|
||||
curated_ids={"sword"}, # curated already owns "sword"
|
||||
curated_aliases=set(),
|
||||
source_vocab={"sword"},
|
||||
)
|
||||
assert [r["id"] for r in rejections] == ["sword"]
|
||||
@@ -0,0 +1,31 @@
|
||||
"""Smoke tests for the KB coverage benchmark harness (issue #121, epic #35)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from scripts.coverage_benchmark import load_corpus, measure_coverage, run
|
||||
from tests.regression.golden_scenes import load_reference_knowledge_base
|
||||
|
||||
|
||||
def test_run_returns_expected_metric_keys() -> None:
|
||||
summary = run()
|
||||
assert set(summary) == {"total", "covered", "coverage_rate", "hit_rate", "per_category"}
|
||||
assert set(summary["hit_rate"]) == {"exact", "head_noun", "fallback", "dropped"}
|
||||
assert summary["total"] > 0
|
||||
assert 0.0 <= summary["coverage_rate"] <= 1.0
|
||||
|
||||
|
||||
def test_hit_rate_paths_sum_to_total() -> None:
|
||||
kb = load_reference_knowledge_base()
|
||||
summary = measure_coverage(kb, load_corpus())
|
||||
assert sum(summary["hit_rate"].values()) == summary["total"]
|
||||
assert summary["covered"] == summary["total"] - summary["hit_rate"]["dropped"]
|
||||
|
||||
|
||||
def test_known_paths_are_classified() -> None:
|
||||
kb = load_reference_knowledge_base()
|
||||
# "long hair" resolves exactly; "white summer dress" via head-noun reduction;
|
||||
# a nonsense phrase is dropped.
|
||||
summary = measure_coverage(kb, ["long hair", "white summer dress", "zzz nonexistent thing"])
|
||||
assert summary["hit_rate"]["exact"] >= 1
|
||||
assert summary["hit_rate"]["head_noun"] >= 1
|
||||
assert summary["hit_rate"]["dropped"] >= 1
|
||||
@@ -0,0 +1,90 @@
|
||||
"""Tests for the automatic candidate validator (issue #119, epic #35)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from compiler.common.kb_autovalidate import (
|
||||
DEFAULT_CONFIDENCE_THRESHOLD,
|
||||
compute_confidence,
|
||||
validate_candidate,
|
||||
)
|
||||
|
||||
VOCAB = {"pantyhose", "elbow_gloves", "gloves", "katana"}
|
||||
|
||||
|
||||
def decide(entry, **overrides):
|
||||
kwargs = dict(
|
||||
curated_ids={"glasses", "thighhighs"},
|
||||
curated_aliases={"spectacles", "thigh highs"},
|
||||
known_ids={"pantyhose", "elbow_gloves", "gloves", "katana", "weapon"},
|
||||
source_vocab=VOCAB,
|
||||
)
|
||||
kwargs.update(overrides)
|
||||
return validate_candidate(entry, **kwargs)
|
||||
|
||||
|
||||
def codes(decision):
|
||||
return [m.code for m in decision.reasons]
|
||||
|
||||
|
||||
def test_well_formed_candidate_is_accepted() -> None:
|
||||
decision = decide({"id": "pantyhose", "tags": ["pantyhose"], "category": "clothing"})
|
||||
assert decision.accepted
|
||||
assert decision.confidence == pytest.approx(1.0)
|
||||
assert decision.reasons == []
|
||||
|
||||
|
||||
def test_id_collision_with_curated_is_rejected() -> None:
|
||||
decision = decide({"id": "thighhighs", "tags": ["thighhighs"], "category": "clothing"})
|
||||
assert not decision.accepted
|
||||
assert "SC0005" in codes(decision)
|
||||
|
||||
|
||||
def test_alias_collision_with_curated_is_rejected() -> None:
|
||||
decision = decide(
|
||||
{
|
||||
"id": "pantyhose",
|
||||
"tags": ["pantyhose"],
|
||||
"category": "clothing",
|
||||
"aliases": ["spectacles"],
|
||||
}
|
||||
)
|
||||
assert not decision.accepted
|
||||
assert "SC0004" in codes(decision)
|
||||
|
||||
|
||||
def test_unknown_category_is_rejected() -> None:
|
||||
decision = decide({"id": "pantyhose", "tags": ["pantyhose"], "category": "not_a_category"})
|
||||
assert not decision.accepted
|
||||
assert "SC0008" in codes(decision)
|
||||
|
||||
|
||||
def test_self_cycle_expansion_is_rejected() -> None:
|
||||
decision = decide(
|
||||
{"id": "katana", "tags": ["katana"], "category": "objects", "expand": ["katana"]}
|
||||
)
|
||||
assert not decision.accepted
|
||||
assert "SC0003" in codes(decision)
|
||||
|
||||
|
||||
def test_unknown_expand_target_is_rejected() -> None:
|
||||
decision = decide(
|
||||
{"id": "katana", "tags": ["katana"], "category": "objects", "expand": ["nonexistent"]}
|
||||
)
|
||||
assert not decision.accepted
|
||||
assert "SC0004" in codes(decision)
|
||||
|
||||
|
||||
def test_below_confidence_threshold_is_rejected() -> None:
|
||||
# A tag not in the source vocab scores below threshold (no invention).
|
||||
decision = decide({"id": "invented_tag", "tags": ["invented tag"], "category": "clothing"})
|
||||
assert not decision.accepted
|
||||
assert "SC0022" in codes(decision)
|
||||
assert decision.confidence < DEFAULT_CONFIDENCE_THRESHOLD
|
||||
|
||||
|
||||
def test_confidence_scoring() -> None:
|
||||
assert compute_confidence({"id": "katana", "category": "objects"}, VOCAB) == pytest.approx(1.0)
|
||||
assert compute_confidence({"id": "katana", "category": "bogus"}, VOCAB) == pytest.approx(0.7)
|
||||
assert compute_confidence({"id": "unknown", "category": "objects"}, VOCAB) == pytest.approx(0.3)
|
||||
@@ -0,0 +1,50 @@
|
||||
"""Tests for Knowledge Base conflict detection (issue #120, epic #35)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from compiler.common.kb_conflicts import detect_conflicts
|
||||
from scripts.detect_kb_conflicts import _load
|
||||
|
||||
KB_DIR = Path(__file__).resolve().parent.parent / "knowledge_base"
|
||||
|
||||
|
||||
def test_clean_reference_kb_has_no_conflicts() -> None:
|
||||
entries, curated_ids = _load(KB_DIR)
|
||||
report = detect_conflicts(entries, curated_ids)
|
||||
assert report.total == 0, report.to_json()
|
||||
|
||||
|
||||
def test_fixture_reports_each_conflict_type_and_only_those() -> None:
|
||||
entries = [
|
||||
# duplicate id
|
||||
{"id": "dup", "tags": ["dup a"], "category": "objects"},
|
||||
{"id": "dup", "tags": ["dup b"], "category": "objects"},
|
||||
# duplicate alias across two entries
|
||||
{"id": "a1", "tags": ["a1"], "category": "objects", "aliases": ["shared_alias"]},
|
||||
{"id": "a2", "tags": ["a2"], "category": "objects", "aliases": ["shared_alias"]},
|
||||
# same tag owned by multiple NON-curated entries
|
||||
{"id": "g1", "tags": ["overlap tag"], "category": "objects"},
|
||||
{"id": "g2", "tags": ["overlap tag"], "category": "objects"},
|
||||
# contradictory (mutual) expansion
|
||||
{"id": "x", "tags": ["x"], "category": "objects", "expand": ["y"]},
|
||||
{"id": "y", "tags": ["y"], "category": "objects", "expand": ["x"]},
|
||||
]
|
||||
report = detect_conflicts(entries, curated_ids=set())
|
||||
assert [d["id"] for d in report.duplicate_ids] == ["dup"]
|
||||
assert [d["alias"] for d in report.duplicate_aliases] == ["shared_alias"]
|
||||
assert [d["tag"] for d in report.tag_multiple_owners] == ["overlap tag"]
|
||||
assert report.contradictory_expansions == [{"pair": ["x", "y"]}]
|
||||
assert report.total == 4
|
||||
|
||||
|
||||
def test_curated_owner_disambiguates_shared_tag() -> None:
|
||||
# A tag shared between a curated entry and a generated one is NOT a conflict
|
||||
# (curated wins). Same tag between two generated entries IS.
|
||||
entries = [
|
||||
{"id": "curated_one", "tags": ["t"], "category": "objects"},
|
||||
{"id": "generated_one", "tags": ["t"], "category": "objects"},
|
||||
]
|
||||
assert detect_conflicts(entries, curated_ids={"curated_one"}).tag_multiple_owners == []
|
||||
assert detect_conflicts(entries, curated_ids=set()).tag_multiple_owners != []
|
||||
Reference in New Issue
Block a user