mirror of
https://github.com/willmiao/ComfyUI-Lora-Manager.git
synced 2026-07-06 09:21:16 -03:00
170c8068c5
- extract_relevant_section(): trim README to model-filename-matching section for collection repos (download link, anchor ID, heading strategies) - _strip_standalone_images(): preserve markdown image URLs so LLM can extract preview_url; strip only HTML <img> tags - extract_simple_markdown_images(): extract civitai.images from ![]() body - extract_html_img_tags(): extract from <img src="..."> (deadman44-style) - extract_gallery_images(): fix widget parser for YAML - output: dash prefix - _is_heading: exclude </hN> closing tags from boundary detection - _extract_section: start at matching heading when match IS a heading line - _try_salvage_json(): recover truncated JSON (close braces/brackets in LIFO order, close unterminated strings, strip trailing commas) - PostProcessor: store _llm_confidence, add instance_prompt YAML fallback - agent_service: pass model_basename to prompt, trim README via extract_relevant_section before clean_readme_for_llm - Add tests/enrich_hf_validation/ suite: 100-model pipeline with progress checkpoint/resume, per-field scoring, markdown+JSON reporting - Fix evaluation_engine: read _llm_confidence (not _llm_response)
353 lines
12 KiB
Python
353 lines
12 KiB
Python
"""Evaluate enriched ``.metadata.json`` quality across multiple dimensions.
|
|
|
|
Scoring rubric (per field):
|
|
|
|
- **Completeness**: Is the field populated with meaningful content?
|
|
- **Validity**: Does the value conform to expected constraints (controlled
|
|
vocab, non-placeholder, parsable JSON)?
|
|
- **Accuracy**: (sub-sample only — requires manual verification against
|
|
the HF README).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
from typing import Any, Dict, List, Optional, Set
|
|
|
|
from .config import (
|
|
CIVITAI_MODEL_TAGS,
|
|
PLACEHOLDER_VALUES,
|
|
SUPPORTED_BASE_MODELS,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Scoring helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_MIN_TAGS = 1
|
|
_MAX_TAGS = 8
|
|
_MIN_DESC_LENGTH = 20
|
|
_MIN_NOTES_LENGTH = 30
|
|
|
|
# Tags that the LLM sometimes emits but which are not meaningful content tags.
|
|
_TECH_TAGS = frozenset({
|
|
"lora", "dreambooth", "text-to-image", "diffusers", "flux",
|
|
"sdxl", "checkpoint", "pytorch", "safetensors", "fine-tuning",
|
|
"stable-diffusion", "training", "stablediffusion",
|
|
})
|
|
|
|
|
|
def _is_placeholder(val: str) -> bool:
|
|
return val.strip().lower() in PLACEHOLDER_VALUES
|
|
|
|
|
|
def _is_valid_trigger_words(words: List[str]) -> bool:
|
|
"""Return True if *words* is a non-empty list of real trigger words."""
|
|
if not words:
|
|
return False
|
|
cleaned = [w.strip() for w in words if w.strip()]
|
|
if not cleaned:
|
|
return False
|
|
# Reject if ALL entries are placeholders
|
|
non_placeholder = [w for w in cleaned if not _is_placeholder(w)]
|
|
return len(non_placeholder) > 0
|
|
|
|
|
|
def _is_valid_tags(tags: List[str]) -> bool:
|
|
"""Return True if *tags* is a reasonable list of content tags."""
|
|
if not tags:
|
|
return False
|
|
cleaned = [t.strip().lower() for t in tags if t.strip()]
|
|
if not cleaned:
|
|
return False
|
|
# At least one tag that isn't a technical keyword
|
|
meaningful = [t for t in cleaned if t not in _TECH_TAGS]
|
|
return len(meaningful) >= _MIN_TAGS
|
|
|
|
|
|
def _tag_priority_coverage(tags: List[str]) -> float:
|
|
"""Fraction of tags that align with the user's priority tag vocabulary."""
|
|
if not tags:
|
|
return 0.0
|
|
priority_lower = {t.lower() for t in CIVITAI_MODEL_TAGS}
|
|
matched = sum(1 for t in tags if t.strip().lower() in priority_lower)
|
|
return matched / len(tags)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Per-model evaluation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# Type alias for a score record
|
|
ScoreRecord = Dict[str, Any]
|
|
|
|
|
|
def evaluate_model(
|
|
metadata: Dict[str, Any],
|
|
model_path: str,
|
|
repo_id: str,
|
|
*,
|
|
enrichment_success: bool,
|
|
enrichment_errors: List[str],
|
|
) -> ScoreRecord:
|
|
"""Score a single enriched model's metadata.
|
|
|
|
Returns a dict with per-field scores, a total score, and a list of
|
|
flagged issues.
|
|
"""
|
|
civitai = metadata.get("civitai") or {}
|
|
trained_words: List[str] = civitai.get("trainedWords") or metadata.get("trainedWords") or []
|
|
short_desc: str = civitai.get("description") or ""
|
|
tags: List[str] = metadata.get("tags") or []
|
|
notes: str = metadata.get("notes") or ""
|
|
usage_tips_raw: str = metadata.get("usage_tips") or "{}"
|
|
model_description: str = metadata.get("modelDescription") or ""
|
|
base_model: str = metadata.get("base_model") or ""
|
|
preview_url: str = metadata.get("preview_url") or ""
|
|
confidence: str = metadata.get("_llm_confidence") or ""
|
|
|
|
# --- base_model ---
|
|
base_model_valid = base_model in SUPPORTED_BASE_MODELS
|
|
base_model_filled = bool(base_model) and base_model != "Unknown"
|
|
|
|
# --- trigger_words (trainedWords) ---
|
|
triggers_valid = _is_valid_trigger_words(trained_words)
|
|
|
|
# --- short_description (civitai.description) ---
|
|
desc_filled = len(short_desc.strip()) >= _MIN_DESC_LENGTH
|
|
|
|
# --- tags ---
|
|
tags_valid = _is_valid_tags(tags)
|
|
tags_priority_coverage = _tag_priority_coverage(tags)
|
|
tags_no_technical = (
|
|
sum(1 for t in tags if t.strip().lower() not in _TECH_TAGS) >= _MIN_TAGS
|
|
if tags else False
|
|
)
|
|
|
|
# --- notes ---
|
|
notes_filled = len(notes.strip()) >= _MIN_NOTES_LENGTH
|
|
|
|
# --- usage_tips ---
|
|
usage_tips_valid = False
|
|
if usage_tips_raw.strip() and usage_tips_raw.strip() != "{}":
|
|
try:
|
|
parsed = json.loads(usage_tips_raw)
|
|
if isinstance(parsed, dict) and len(parsed) > 0:
|
|
usage_tips_valid = True
|
|
except (json.JSONDecodeError, TypeError):
|
|
pass
|
|
|
|
# --- modelDescription (README → HTML) ---
|
|
desc_html_filled = len(model_description.strip()) > 100
|
|
|
|
# --- preview_url ---
|
|
preview_filled = bool(preview_url) and os.path.exists(preview_url)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Composite score (0-100)
|
|
# ------------------------------------------------------------------
|
|
|
|
field_scores = {
|
|
"base_model": _score_bool(base_model_filled and base_model_valid, weight=15),
|
|
"trigger_words": _score_bool(triggers_valid, weight=15),
|
|
"short_description": _score_bool(desc_filled, weight=10),
|
|
"tags": _score_bool(tags_valid, weight=15),
|
|
"tags_priority_coverage": _score_continuous(tags_priority_coverage, weight=5),
|
|
"notes": _score_bool(notes_filled, weight=5),
|
|
"usage_tips": _score_bool(usage_tips_valid, weight=5),
|
|
"modelDescription_html": _score_bool(desc_html_filled, weight=10),
|
|
"preview_downloaded": _score_bool(preview_filled, weight=10),
|
|
}
|
|
|
|
# Deduct points for enrichment-level failures
|
|
penalty = 0
|
|
if enrichment_errors:
|
|
penalty += 10
|
|
if not enrichment_success:
|
|
penalty += 20
|
|
|
|
total_raw = sum(field_scores.values())
|
|
total = max(0, min(100, total_raw - penalty))
|
|
|
|
# ------------------------------------------------------------------
|
|
# Flagged issues
|
|
# ------------------------------------------------------------------
|
|
|
|
issues: List[str] = []
|
|
if not base_model_filled:
|
|
issues.append("base_model is empty or 'Unknown'")
|
|
elif not base_model_valid:
|
|
issues.append(f"base_model '{base_model}' not in SUPPORTED_BASE_MODELS")
|
|
if not triggers_valid:
|
|
issues.append("trigger_words are missing or contain only placeholders")
|
|
if not desc_filled:
|
|
issues.append("short_description is too short or empty")
|
|
if not tags_valid:
|
|
issues.append("tags are missing, too few, or purely technical")
|
|
if tags_valid and tags_priority_coverage < 0.5:
|
|
issues.append("tags have low overlap with priority_tags (< 50%)")
|
|
if not notes_filled:
|
|
issues.append("notes are too short or empty")
|
|
if not usage_tips_valid:
|
|
issues.append("usage_tips is empty or invalid JSON")
|
|
if not desc_html_filled:
|
|
issues.append("modelDescription is too short (README may not have been converted)")
|
|
if not preview_filled:
|
|
issues.append("preview image not downloaded (URL missing or download failed)")
|
|
|
|
return {
|
|
"repo_id": repo_id,
|
|
"model_path": model_path,
|
|
"enrichment_success": enrichment_success,
|
|
"total_score": total,
|
|
"field_scores": field_scores,
|
|
"issues": issues,
|
|
"confidence_from_llm": confidence,
|
|
"raw_values": {
|
|
"base_model": base_model,
|
|
"trigger_words": trained_words,
|
|
"short_description": short_desc,
|
|
"tags": tags,
|
|
"notes": notes,
|
|
"usage_tips": usage_tips_raw,
|
|
"preview_url": preview_url,
|
|
"has_modelDescription": len(model_description) > 0,
|
|
},
|
|
}
|
|
|
|
|
|
def _score_bool(condition: bool, weight: int = 10) -> int:
|
|
return weight if condition else 0
|
|
|
|
|
|
def _score_continuous(value: float, weight: int = 10) -> int:
|
|
"""Linear interpolation: value 0.0 → 0, value 1.0 → *weight*."""
|
|
return int(round(value * weight))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Batch evaluation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def evaluate_batch(
|
|
enriched: List[Dict[str, Any]],
|
|
) -> List[ScoreRecord]:
|
|
"""Evaluate a list of enrichment results.
|
|
|
|
Each entry in *enriched* should have keys:
|
|
``repo_id``, ``model_path``, ``metadata`` (the enriched dict),
|
|
``success``, ``errors``.
|
|
"""
|
|
scores: List[ScoreRecord] = []
|
|
for entry in enriched:
|
|
record = evaluate_model(
|
|
metadata=entry.get("metadata", {}),
|
|
model_path=entry.get("model_path", ""),
|
|
repo_id=entry.get("repo_id", ""),
|
|
enrichment_success=entry.get("success", False),
|
|
enrichment_errors=entry.get("errors", []),
|
|
)
|
|
scores.append(record)
|
|
return scores
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Aggregate statistics
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def aggregate_scores(scores: List[ScoreRecord]) -> Dict[str, Any]:
|
|
"""Compute aggregate stats across all scored models."""
|
|
n = len(scores)
|
|
if n == 0:
|
|
return {"error": "no scores to aggregate"}
|
|
|
|
field_names = [
|
|
"base_model", "trigger_words", "short_description", "tags",
|
|
"tags_priority_coverage", "notes", "usage_tips",
|
|
"modelDescription_html", "preview_downloaded",
|
|
]
|
|
possible = {f: 15 if f == "base_model" or f == "trigger_words" or f == "tags" else
|
|
10 if f == "short_description" or f == "modelDescription_html" or f == "preview_downloaded" else
|
|
5
|
|
for f in field_names}
|
|
|
|
# Per-field aggregate
|
|
field_agg: Dict[str, Any] = {}
|
|
for fn in field_names:
|
|
vals = [s["field_scores"].get(fn, 0) for s in scores]
|
|
max_per_field = possible[fn]
|
|
field_agg[fn] = {
|
|
"mean": round(sum(vals) / n, 1) if n else 0,
|
|
"fill_rate_pct": round(
|
|
sum(1 for v in vals if v >= max_per_field) / n * 100, 1
|
|
) if n else 0.0,
|
|
"partial_rate_pct": round(
|
|
sum(1 for v in vals if 0 < v < max_per_field) / n * 100, 1
|
|
) if n else 0.0,
|
|
"empty_rate_pct": round(
|
|
sum(1 for v in vals if v == 0) / n * 100, 1
|
|
) if n else 0.0,
|
|
}
|
|
|
|
# Total score distribution
|
|
total_scores = [s["total_score"] for s in scores]
|
|
total_agg = {
|
|
"mean": round(sum(total_scores) / n, 1) if n else 0,
|
|
"median": _median(total_scores),
|
|
"min": min(total_scores) if total_scores else 0,
|
|
"max": max(total_scores) if total_scores else 0,
|
|
"bins": {
|
|
"excellent_80+": sum(1 for s in total_scores if s >= 80),
|
|
"good_60_79": sum(1 for s in total_scores if 60 <= s < 80),
|
|
"fair_40_59": sum(1 for s in total_scores if 40 <= s < 60),
|
|
"poor_20_39": sum(1 for s in total_scores if 20 <= s < 40),
|
|
"bad_0_19": sum(1 for s in total_scores if s < 20),
|
|
},
|
|
}
|
|
|
|
# Issue frequency
|
|
issue_counter: Dict[str, int] = {}
|
|
for s in scores:
|
|
for issue in s["issues"]:
|
|
issue_counter[issue] = issue_counter.get(issue, 0) + 1
|
|
top_issues = sorted(issue_counter.items(), key=lambda x: -x[1])
|
|
|
|
# Confidence distribution
|
|
conf_counter: Dict[str, int] = {"high": 0, "medium": 0, "low": 0, "": 0}
|
|
for s in scores:
|
|
c = (s.get("confidence_from_llm") or "").strip().lower()
|
|
if c in conf_counter:
|
|
conf_counter[c] += 1
|
|
else:
|
|
conf_counter[""] += 1
|
|
|
|
# Success / timeout / failure stats
|
|
success_count = sum(1 for s in scores if s["enrichment_success"])
|
|
fail_count = n - success_count
|
|
|
|
return {
|
|
"model_count": n,
|
|
"success_count": success_count,
|
|
"fail_count": fail_count,
|
|
"total_score": total_agg,
|
|
"field_aggregates": field_agg,
|
|
"top_issues": top_issues[:15],
|
|
"confidence_distribution": conf_counter,
|
|
}
|
|
|
|
|
|
def _median(values: List[float]) -> float:
|
|
if not values:
|
|
return 0.0
|
|
sorted_v = sorted(values)
|
|
m = len(sorted_v) // 2
|
|
if len(sorted_v) % 2 == 0:
|
|
return round((sorted_v[m - 1] + sorted_v[m]) / 2, 1)
|
|
return round(sorted_v[m], 1)
|