davide/rag-from-scratch
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: rimuove chunker legacy, aggiunge dipendenze AST-based pipeline

Browse Source 
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Davide Grilli
2026-06-08 16:16:07 +02:00
parent b321a51c5c
commit be8ae9f6b8
6 changed files with 2 additions and 970 deletions
Download Patch File Download Diff File Expand all files Collapse all files
chunks/chunker.py
-413
View File
@@ -1,413 +0,0 @@
#!/usr/bin/env python3
"""
chunker.py — Chunking semantico da Markdown pulito
Input: sources/<stem>.md — Markdown già strutturato (H1/H2/H3 + paragrafi)
Output: chunks/<stem>/chunks.json
chunks/<stem>/meta.json
Regole:
- ogni paragrafo diventa un chunk; paragrafi di contesti diversi non si mescolano
- se un paragrafo supera MAX_CHARS viene spezzato a confine di frase (mai a metà)
- paragrafi brevi (< MIN_CHARS) vengono fusi col successivo finché non raggiungono
MIN_CHARS, a patto che abbiano lo stesso contesto heading
- tabelle, liste e codice sono blocchi atomici (non si spezzano)
Uso:
python chunks/chunker.py --stem <stem>
python chunks/chunker.py # tutti gli stem con .md in sources/
python chunks/chunker.py --stem <stem> --force
"""
import argparse
import json
import re
import sys
from pathlib import Path
_HERE = Path(__file__).resolve().parent
if str(_HERE) not in sys.path:
sys.path.insert(0, str(_HERE))
import config as cfg
# ─── Utilità ──────────────────────────────────────────────────────────────────
def split_sentences(text: str) -> list[str]:
parts = re.split(cfg.SENTENCE_SPLIT_RE, text.strip())
return [p.strip() for p in parts if p.strip()]
def context_to_meta(context: str) -> tuple[str, str]:
parts = [p.strip() for p in context.split(" > ") if p.strip()]
if len(parts) >= 2:
return " > ".join(parts[:-1]), parts[-1]
return (parts[0] if parts else ""), ""
# ─── Parser Markdown ──────────────────────────────────────────────────────────
_SKIP_HEADINGS_LOWER = {h.lower() for h in cfg.SKIP_HEADINGS}
def _is_skip_heading(title: str) -> bool:
t = title.lower()
return any(t == s or t.startswith(s) for s in _SKIP_HEADINGS_LOWER)
def parse_paragraphs(text: str) -> list[dict]:
"""Estrae blocchi dal Markdown con il loro contesto heading.
Restituisce: [{"context": "H1 > H2 > H3", "text": "...", "kind": "text|table|list|code"}]
- Heading senza corpo non emettono chunk: aggiornano solo il contesto.
- Tabelle (righe |), liste (righe -/*) e code block (```) sono atomici.
- Sezioni in SKIP_HEADINGS vengono saltate completamente.
- Contenuto pre-heading saltato se SKIP_PRE_HEADING è True.
"""
headings = ["", "", ""] # H1, H2, H3
result: list[dict] = []
buf: list[str] = []
cur_kind = "text"
in_code = False
skip_level: int | None = None # livello heading che ha attivato lo skip
def current_context() -> str:
parts = [h for h in headings if h]
return " > ".join(parts[:cfg.CONTEXT_DEPTH]) if parts else "documento"
def is_skipping() -> bool:
return skip_level is not None
def flush() -> None:
if is_skipping():
buf.clear()
return
if cfg.SKIP_PRE_HEADING and current_context() == "documento":
buf.clear()
return
body = "\n".join(buf).strip()
if body:
result.append({"context": current_context(), "text": body, "kind": cur_kind})
buf.clear()
for line in text.splitlines():
# ── Code block toggle ─────────────────────────────────────────────────
if line.strip().startswith("```"):
if not in_code:
flush()
cur_kind = "code"
in_code = True
if not is_skipping():
buf.append(line)
else:
if not is_skipping():
buf.append(line)
in_code = False
flush()
cur_kind = "text"
continue
if in_code:
if not is_skipping():
buf.append(line)
continue
# ── Heading ───────────────────────────────────────────────────────────
m = re.match(r"^(#{1,3}) (.+)", line)
if m:
flush()
level = len(m.group(1))
title = m.group(2).strip()
# chiudi skip se torniamo a livello pari o superiore
if skip_level is not None and level <= skip_level:
skip_level = None
headings[level - 1] = title
for i in range(level, 3):
headings[i] = ""
cur_kind = "text"
# apri skip se questo heading è nella lista
if _is_skip_heading(title):
skip_level = level
continue
# ── Tabella ───────────────────────────────────────────────────────────
if line.strip().startswith("|"):
if cur_kind != "table":
flush()
cur_kind = "table"
buf.append(line)
continue
# ── Lista ─────────────────────────────────────────────────────────────
if re.match(r"^\s*[-*]\s", line):
if cur_kind != "list":
flush()
cur_kind = "list"
buf.append(line)
continue
# ── Riga vuota: chiude il paragrafo corrente ──────────────────────────
if line.strip() == "":
flush()
cur_kind = "text"
continue
# ── Testo normale ─────────────────────────────────────────────────────
if cur_kind in ("table", "list", "code"):
flush()
cur_kind = "text"
buf.append(line)
flush()
return result
# ─── Merge paragrafi brevi ────────────────────────────────────────────────────
def merge_short(paragraphs: list[dict]) -> list[dict]:
"""Fonde paragrafi di testo consecutivi sotto MIN_CHARS con il successivo,
purché condividano lo stesso contesto heading."""
if not cfg.MERGE_SHORT_PARAGRAPHS:
return paragraphs
result: list[dict] = []
buf_para: dict | None = None
buf_text: str = ""
def flush_buf() -> None:
nonlocal buf_para, buf_text
if buf_para is not None:
result.append({**buf_para, "text": buf_text})
buf_para = None
buf_text = ""
for para in paragraphs:
if para["kind"] != "text":
flush_buf()
result.append(para)
continue
if buf_para is None:
buf_para = para
buf_text = para["text"]
elif buf_para["context"] == para["context"] and len(buf_text) < cfg.MIN_CHARS:
buf_text = buf_text + "\n\n" + para["text"]
else:
flush_buf()
buf_para = para
buf_text = para["text"]
flush_buf()
return result
# ─── Chunking ─────────────────────────────────────────────────────────────────
def make_chunks(paragraphs: list[dict]) -> list[dict]:
"""Genera chunk da parse_paragraphs (dopo eventuale merge).
- Blocchi atomici (table, list, code): un chunk, mai spezzato.
- Testo: un chunk per paragrafo; se supera MAX_CHARS, spezza a confine frase.
"""
chunks: list[dict] = []
idx = 0
for para in paragraphs:
text = para["text"]
context = para["context"]
kind = para["kind"]
sezione, titolo = context_to_meta(context)
def emit(body: str) -> None:
nonlocal idx
chunk_text = f"[{context}]\n{body}"
chunks.append({
"chunk_id": f"c{idx}",
"text": chunk_text,
"sezione": sezione,
"titolo": titolo,
"context": context,
"n_chars": len(chunk_text),
})
idx += 1
# ── Atomici ───────────────────────────────────────────────────────────
if kind in ("table", "list", "code"):
emit(text)
continue
# ── Testo: split a confine di frase se supera MAX_CHARS ───────────────
sents = split_sentences(text)
if not sents:
continue
current: list[str] = []
for sent in sents:
projected = len(" ".join(current + [sent]))
if projected <= cfg.MAX_CHARS or not current:
current.append(sent)
else:
emit(" ".join(current))
current = [sent]
if current:
emit(" ".join(current))
return chunks
# ─── Merge frasi spezzate ─────────────────────────────────────────────────────
_SENT_END = re.compile(
"[.!?;:"
+ chr(0xBB) + ")\\\\"
+ chr(0x2018) + chr(0x2019)
+ chr(0x201C) + chr(0x201D)
+ "\"'"
+ chr(0x2014) + chr(0x2013) + chr(0x2026) + chr(0xB7)
+ "]$"
+ r"|\d[\d.,/]*$" # numero o versione
+ r"|\$$" # formula LaTeX inline
+ r"|\}$" # blocco LaTeX
+ r"|>$" # tag HTML
+ r"|\\\\$" # \\ LaTeX
+ r"|\|$" # riga tabella
)
def _body(chunk: dict) -> str:
text = chunk["text"]
nl = text.find("\n")
return text[nl + 1:] if nl != -1 else text
def merge_broken_sentences(chunks: list[dict]) -> list[dict]:
"""Fonde chunk consecutivi con lo stesso contesto quando il primo termina
senza punteggiatura di fine frase (frase spezzata dal sorgente)."""
result: list[dict] = []
i = 0
while i < len(chunks):
c = dict(chunks[i])
body = _body(c)
while (
i + 1 < len(chunks)
and chunks[i + 1]["context"] == c["context"]
and not _SENT_END.search(body.rstrip())
):
i += 1
next_body = _body(chunks[i])
body = body.rstrip() + " " + next_body.lstrip()
chunk_text = f"[{c['context']}]\n{body}"
c["text"] = chunk_text
c["n_chars"] = len(chunk_text)
result.append(c)
i += 1
for idx, c in enumerate(result):
c["chunk_id"] = f"c{idx}"
return result
# ─── Pipeline per documento ───────────────────────────────────────────────────
def process_stem(stem: str, project_root: Path, force: bool) -> bool:
md_path = project_root / "sources" / f"{stem}_output" / "auto" / f"{stem}.md"
if not md_path.exists():
md_path = project_root / "sources" / f"{stem}.md"
if not md_path.exists():
print(f" ✗ {stem}.md non trovato (cercato in sources/{stem}_output/auto/ e sources/)")
return False
out_dir = project_root / "chunks" / stem
out_file = out_dir / "chunks.json"
if out_file.exists() and not force:
print(f" ↩ chunks/{stem}/chunks.json già presente — skip (usa --force per rigenerare)")
return True
print(f"[chunker] {stem}")
text = md_path.read_text(encoding="utf-8")
paragraphs = parse_paragraphs(text)
if not paragraphs:
print(f" ✗ Nessun paragrafo estratto da {md_path.name}")
return False
if cfg.MERGE_SHORT_PARAGRAPHS:
paragraphs = merge_short(paragraphs)
chunks = make_chunks(paragraphs)
chunks = merge_broken_sentences(chunks)
if not chunks:
print(f" ✗ Nessun chunk generato")
return False
out_dir.mkdir(parents=True, exist_ok=True)
out_file.write_text(
json.dumps(chunks, ensure_ascii=False, indent=2), encoding="utf-8"
)
(out_dir / "meta.json").write_text(
json.dumps({
"stem": stem,
"source": str(md_path.relative_to(project_root)),
"max_chars": cfg.MAX_CHARS,
"min_chars": cfg.MIN_CHARS,
"merge_short": cfg.MERGE_SHORT_PARAGRAPHS,
"strategy": "one_paragraph_per_chunk",
}, ensure_ascii=False, indent=2),
encoding="utf-8",
)
lengths = [c["n_chars"] for c in chunks]
over_max = sum(1 for l in lengths if l > cfg.MAX_CHARS)
under_min = sum(1 for l in lengths if l < cfg.MIN_CHARS)
avg = int(sum(lengths) / len(lengths))
print(f" ✅ {len(chunks)} chunk | media {avg} char | max {max(lengths)} char")
if over_max:
print(f" ⚠️ {over_max} chunk superano MAX_CHARS={cfg.MAX_CHARS}")
if under_min:
print(f" ℹ️ {under_min} chunk sotto MIN_CHARS={cfg.MIN_CHARS}")
print(f" → chunks/{stem}/chunks.json")
return True
# ─── Entry point ──────────────────────────────────────────────────────────────
if __name__ == "__main__":
project_root = Path(__file__).parent.parent
sources_dir = project_root / "sources"
parser = argparse.ArgumentParser(description="Markdown pulito → chunks.json")
parser.add_argument("--stem", help="Nome documento (es. analisi2)")
parser.add_argument("--force", action="store_true",
help="Rigenera chunks.json anche se già presente")
args = parser.parse_args()
if args.stem:
stems = [args.stem]
else:
found = set()
for p in sources_dir.glob("*_output/auto/*.md"):
found.add(p.stem)
for p in sources_dir.glob("*.md"):
found.add(p.stem)
stems = sorted(found)
if not stems:
print("Nessun file .md trovato in sources/")
sys.exit(1)
results = [process_stem(s, project_root, args.force) for s in stems]
ok = sum(results)
print(f"\n{'✅' if all(results) else '⚠️ '} {ok}/{len(results)} documenti processati")
sys.exit(0 if all(results) else 1)
chunks/config.py
-64
View File
@@ -1,64 +0,0 @@
#!/usr/bin/env python3
"""
Parametri della pipeline di chunking.
Input atteso: sources/<stem>/<stem>.md — Markdown già pulito e ben strutturato.
"""
# ─── Dimensione chunk ─────────────────────────────────────────────────────────
# Caratteri massimi per chunk (prefisso di contesto incluso).
# Paragrafi più lunghi vengono spezzati a confine di frase.
# Una singola frase che supera MAX_CHARS non viene mai spezzata.
MAX_CHARS: int = 1200
# Soglia minima attesa (usata da verify_chunks come warning, non blocca).
MIN_CHARS: int = 80
# ─── Spezzatura frasi ─────────────────────────────────────────────────────────
# Regex per rilevare il confine di fine frase.
# Split solo prima di lettera maiuscola o virgolette — evita split su abbreviazioni.
SENTENCE_SPLIT_RE: str = r"(?<=[.!?»])\s+(?=[A-ZÀÈÉÌÒÙ\"])"
# ─── Blocchi atomici ──────────────────────────────────────────────────────────
# Blocchi Markdown che non vengono mai spezzati, anche se superano MAX_CHARS.
ATOMIC_TYPES: set[str] = {"table", "code", "list"}
# ─── Contesto heading ─────────────────────────────────────────────────────────
# Profondità massima del percorso heading incluso nel prefisso di ogni chunk.
# 1 = solo H1, 2 = H1 > H2, 3 = H1 > H2 > H3.
CONTEXT_DEPTH: int = 3
# ─── Sezioni da escludere ────────────────────────────────────────────────────
# Heading (case-insensitive) le cui sezioni vengono saltate completamente.
# Il match è su prefisso: "indice" salta anche "Indice delle figure".
# Lasciare vuoto per non escludere nulla.
SKIP_HEADINGS: set[str] = {
"indice",
"sommario",
"bibliografia",
"ringraziamenti",
"abbreviazioni",
}
# Se True, salta il contenuto che precede il primo heading (frontespizio, copertina).
SKIP_PRE_HEADING: bool = True
# ─── Merge paragrafi corti ────────────────────────────────────────────────────
# Paragrafi consecutivi più corti di MIN_CHARS vengono fusi fino a raggiungerlo,
# purché appartengano allo stesso contesto heading.
MERGE_SHORT_PARAGRAPHS: bool = True
# ─── verify_chunks ────────────────────────────────────────────────────────────
# Numero minimo di simboli matematici perché un chunk incompleto sia classificato
# come "matematico" (warning meno grave rispetto a frase spezzata normale).
MATH_SYMS_MIN: int = 3
PROTECT_TABLES: bool = True
PROTECT_MATH: bool = True
chunks/verify_chunks.py
-493
View File
@@ -1,493 +0,0 @@
#!/usr/bin/env python3
"""
Verifica chunk
Analizza chunks/<stem>/chunks.json e segnala ogni anomalia che potrebbe
degradare la qualità del retrieval. Non modifica nulla.
Input: chunks/<stem>/chunks.json
Output: report a schermo + chunks/<stem>/report.json + exit code (0 = OK, 1 = problemi)
Uso:
python chunks/verify_chunks.py --stem documento
python chunks/verify_chunks.py # tutti i documenti in chunks/
python chunks/verify_chunks.py --min 200 --max 800
"""
import argparse
import json
import re
import sys
from collections import Counter
from pathlib import Path
_HERE = Path(__file__).resolve().parent
if str(_HERE) not in sys.path:
sys.path.insert(0, str(_HERE))
import config as cfg
# ─── Soglie ───────────────────────────────────────────────────────────────────
MIN_CHARS = cfg.MIN_CHARS
MAX_CHARS = cfg.MAX_CHARS
_PUNCT_CLS = (
"[.!?"
+ chr(0xBB) + ")\\]"
+ chr(0x2018) + chr(0x2019)
+ chr(0x201C) + chr(0x201D)
+ "\"'"
+ chr(0x2014) + chr(0x2013) + chr(0x2026)
+ chr(0xB7) # punto centrato LaTeX
+ "]$"
)
PUNCT_END = re.compile(
_PUNCT_CLS
+ r"|/$"
+ r"|\|$"
+ r"|;$"
+ r"|:$"
+ r"|\d[\d.,/]*$"
+ r"|\$$"
+ r"|\}$"
+ r"|>$"
+ r"|\\\\$"
)
_HEX_END = re.compile(r"[0-9a-fA-F]{8,}$")
_URL_TAIL = re.compile(r"(https?://|www\.)\S+(\s+\S+){0,3}$")
_MATH_SYMS = re.compile(r"[∈∑≤≥≠∀∃∫√∞∂±×÷→←↔⊂⊃⊆⊇∩∪·°]")
_ROMAN_END = re.compile(r"\b(I{1,3}|IV|VI{0,3}|IX|XI{0,2}|XIV|XV|XVI{0,2}|XIX|XX{0,2})$")
_TABLE_SEP = re.compile(r"^\s*\|[\s\-|:]+\|\s*$")
def _load_thresholds(stem_dir: Path) -> tuple[int, int]:
meta = stem_dir / "meta.json"
if meta.exists():
m = json.loads(meta.read_text(encoding="utf-8"))
return m["min_chars"], m["max_chars"]
return MIN_CHARS, MAX_CHARS
def _strip_prefix(text: str) -> str:
text = text.lstrip()
if text.startswith("["):
end = text.find("]")
if end != -1:
return text[end + 1:].lstrip("\n")
return text
# ─── Checks ───────────────────────────────────────────────────────────────────
def is_empty(chunk: dict) -> bool:
return not chunk.get("text", "").strip()
def has_prefix(chunk: dict) -> bool:
return chunk.get("text", "").lstrip().startswith("[")
def is_prefix_malformed(chunk: dict) -> bool:
"""Inizia con [ ma il prefisso non chiude con ] o ha contenuto vuoto."""
text = chunk.get("text", "").lstrip()
if not text.startswith("["):
return False
first_line = text.split("\n")[0]
end = first_line.find("]")
if end == -1:
return True
return len(first_line[1:end].strip()) == 0
def is_body_empty(chunk: dict) -> bool:
"""Prefisso valido ma nessun testo nel corpo."""
text = chunk.get("text", "").lstrip()
if not text.startswith("["):
return False
end = text.find("]")
if end == -1:
return False
return len(text[end + 1:].strip()) == 0
def is_too_short(chunk: dict, min_chars: int) -> bool:
return chunk.get("n_chars", 0) < min_chars
def is_too_long(chunk: dict, max_chars: int) -> bool:
return chunk.get("n_chars", 0) > max_chars
def ends_incomplete(chunk: dict) -> bool:
text = chunk.get("text", "").rstrip()
if not text:
return False
text_check = re.sub(r"[_*]+$", "", text).rstrip()
if not text_check:
return False
if PUNCT_END.search(text_check):
return False
if _HEX_END.search(text_check):
return False
if _ROMAN_END.search(text_check):
return False
if _URL_TAIL.search(text_check[-200:]):
return False
return True
def is_math_incomplete(chunk: dict) -> bool:
return ends_incomplete(chunk) and len(_MATH_SYMS.findall(chunk.get("text", ""))) >= cfg.MATH_SYMS_MIN
def is_table_broken(chunk: dict) -> bool:
"""Tabella Markdown (≥2 righe con |) senza riga separatore |---|."""
text = chunk.get("text", "")
pipe_lines = [l for l in text.splitlines() if "|" in l and l.strip().startswith("|")]
if len(pipe_lines) < 2:
return False
return not any(_TABLE_SEP.match(l) for l in pipe_lines)
def find_duplicate_bodies(chunks: list[dict]) -> list[dict]:
"""Chunk con testo body identico (prefisso escluso). Ignora corpi < 30 char."""
seen: dict[str, str] = {}
dupes = []
for c in chunks:
body = _strip_prefix(c.get("text", "")).strip()
if len(body) < 30:
continue
cid = c["chunk_id"]
if body in seen:
dupes.append({
"chunk_id": cid,
"duplicate_of": seen[body],
"sezione": c.get("sezione", ""),
"titolo": c.get("titolo", ""),
"n_chars": c.get("n_chars", 0),
"last_text": body[:120],
})
else:
seen[body] = cid
return dupes
# ─── Istogramma ───────────────────────────────────────────────────────────────
def _ascii_histogram(lengths: list[int], min_t: int, max_t: int,
n_bins: int = 10, bar_width: int = 28) -> list[str]:
if not lengths:
return []
lo, hi = min(lengths), max(lengths)
if lo == hi:
return [f" {lo:>5}–{hi:<5} │{'█' * bar_width}│ {len(lengths)}"]
step = (hi - lo) / n_bins
bins = [0] * n_bins
for l in lengths:
idx = min(int((l - lo) / step), n_bins - 1)
bins[idx] += 1
max_count = max(bins) or 1
lines = []
for i, count in enumerate(bins):
lo_b = int(lo + i * step)
hi_b = int(lo + (i + 1) * step)
bar = "█" * round(count / max_count * bar_width)
note = ""
if lo_b <= min_t < hi_b:
note = " ← MIN"
elif lo_b <= max_t < hi_b:
note = " ← MAX"
lines.append(f" {lo_b:>5}–{hi_b:<5} │{bar:<{bar_width}}│ {count}{note}")
return lines
# ─── Helpers output ───────────────────────────────────────────────────────────
def _fmt_chunk(c: dict) -> str:
cid = c.get("chunk_id", "?")
n = c.get("n_chars", 0)
preview = c.get("text", "")[:60].replace("\n", " ")
return f" [{cid}] ({n} char) «{preview}»"
def _chunk_entry(c: dict) -> dict:
return {
"chunk_id": c.get("chunk_id", ""),
"sezione": c.get("sezione", ""),
"titolo": c.get("titolo", ""),
"n_chars": c.get("n_chars", 0),
"last_text": c.get("text", "").rstrip().split("\n")[-1][-120:],
}
def _print_list(items: list[dict], limit: int = 5) -> None:
for c in items[:limit]:
print(_fmt_chunk(c))
if len(items) > limit:
print(f" ... e altri {len(items) - limit}")
# ─── Core ─────────────────────────────────────────────────────────────────────
def verify_stem(stem: str, project_root: Path, min_chars: int, max_chars: int) -> bool:
stem_dir = project_root / "chunks" / stem
chunks_path = stem_dir / "chunks.json"
min_chars, max_chars = _load_thresholds(stem_dir)
print(f"\nDocumento: {stem}")
if not chunks_path.exists():
print(f" ✗ chunks/{stem}/chunks.json non trovato")
print(f" Esegui prima: python chunks/chunker.py --stem {stem}")
return False
chunks: list[dict] = json.loads(chunks_path.read_text(encoding="utf-8"))
if not chunks:
print(f" ✗ chunks.json è vuoto")
return False
# ── Raccogli problemi ──────────────────────────────────────────────────────
empty_chunks = [c for c in chunks if is_empty(c)]
no_prefix = [c for c in chunks if not is_empty(c) and not has_prefix(c)]
malformed_prefix = [c for c in chunks
if not is_empty(c) and has_prefix(c) and is_prefix_malformed(c)]
body_empty = [c for c in chunks
if not is_empty(c) and has_prefix(c)
and not is_prefix_malformed(c) and is_body_empty(c)]
too_short = [c for c in chunks if is_too_short(c, min_chars)]
too_long = [c for c in chunks if is_too_long(c, max_chars)]
_incomplete_all = [c for c in chunks if not is_empty(c) and ends_incomplete(c)]
incomplete_math = [c for c in _incomplete_all if is_math_incomplete(c)]
incomplete = [c for c in _incomplete_all if not is_math_incomplete(c)]
broken_tables = [c for c in chunks if is_table_broken(c)]
duplicates = find_duplicate_bodies(chunks)
# ── Statistiche ───────────────────────────────────────────────────────────
lengths = [c.get("n_chars", 0) for c in chunks]
n_total = len(chunks)
blocker_ids = set(
c["chunk_id"]
for lst in [empty_chunks, no_prefix, malformed_prefix, body_empty, incomplete]
for c in lst
)
n_ok = n_total - len(blocker_ids)
min_l = min(lengths)
max_l = max(lengths)
avg_l = int(sum(lengths) / n_total)
p50 = sorted(lengths)[n_total // 2]
n_under = sum(1 for l in lengths if l < min_chars)
n_norm = sum(1 for l in lengths if min_chars <= l <= max_chars)
n_over = sum(1 for l in lengths if l > max_chars)
section_counts = Counter(c.get("sezione", "—") or "—" for c in chunks)
# ── Output statistiche ────────────────────────────────────────────────────
print(f" Totale: {n_total} | ✅ OK: {n_ok}")
print()
print(f" Lunghezze — min {min_l} p50 {p50} media {avg_l} max {max_l}")
print(f" Fasce — <{min_chars}: {n_under} | {min_chars}–{max_chars}: {n_norm} | >{max_chars}: {n_over}")
print()
print(" Istogramma:")
for line in _ascii_histogram(lengths, min_chars, max_chars):
print(line)
print()
print(" Top sezioni:")
for sezione, count in section_counts.most_common(5):
bar = "▪" * min(count, 35)
print(f" {bar} {count:>4} {sezione[:65]}")
# ── Blockers ──────────────────────────────────────────────────────────────
if empty_chunks:
print(f"\n 🔴 {len(empty_chunks)} chunk VUOTI:")
for c in empty_chunks[:5]:
print(f" [{c.get('chunk_id', '?')}]")
if len(empty_chunks) > 5:
print(f" ... e altri {len(empty_chunks) - 5}")
if no_prefix:
print(f"\n 🔴 {len(no_prefix)} chunk SENZA PREFISSO DI CONTESTO:")
_print_list(no_prefix)
print(f" → Causa probabile: heading mancanti nel clean.md")
if malformed_prefix:
print(f"\n 🔴 {len(malformed_prefix)} chunk con PREFISSO MALFORMATO ([ senza ] o vuoto):")
_print_list(malformed_prefix)
print(f" → Causa probabile: heading con caratteri speciali nel clean.md")
if body_empty:
print(f"\n 🔴 {len(body_empty)} chunk con CORPO VUOTO (solo prefisso):")
_print_list(body_empty)
print(f" → Causa probabile: sezioni senza testo nel clean.md")
if incomplete:
print(f"\n 🟡 {len(incomplete)} chunk con FRASE SPEZZATA (warning):")
for c in incomplete[:5]:
last_line = c.get("text", "").rstrip().split("\n")[-1][-80:]
print(f" [{c.get('chunk_id', '?')}] ...{last_line!r}")
if len(incomplete) > 5:
print(f" ... e altri {len(incomplete) - 5}")
print(f" → Soluzione: correggi il sorgente .md e rigenera con chunker.py --force")
# ── Warnings ──────────────────────────────────────────────────────────────
if too_short:
print(f"\n 🟡 {len(too_short)} chunk SOTTO MIN_CHARS ({min_chars}):")
_print_list(too_short)
if too_long:
print(f"\n 🟡 {len(too_long)} chunk SOPRA MAX ({max_chars}):")
_print_list(too_long)
print(f" → Causa: frasi non suddivisibili o blocchi atomici (tabelle/liste)")
if incomplete_math:
print(f"\n 🟡 {len(incomplete_math)} chunk MATEMATICI senza punteggiatura finale:")
for c in incomplete_math[:3]:
last_line = c.get("text", "").rstrip().split("\n")[-1][-80:]
print(f" [{c.get('chunk_id', '?')}] ...{last_line!r}")
if len(incomplete_math) > 3:
print(f" ... e altri {len(incomplete_math) - 3}")
if broken_tables:
print(f"\n 🟡 {len(broken_tables)} TABELLE senza riga separatore |---|:")
_print_list(broken_tables, limit=3)
print(f" → Le tabelle potrebbero non renderizzarsi nel retrieval")
if duplicates:
print(f"\n 🟡 {len(duplicates)} DUPLICATI (corpo identico):")
for e in duplicates[:5]:
print(f" [{e['chunk_id']}] ≡ [{e['duplicate_of']}] «{e['last_text'][:60]}»")
if len(duplicates) > 5:
print(f" ... e altri {len(duplicates) - 5}")
print(f" → Causa probabile: sezioni ripetute nel sorgente .md")
# ── Report.json ───────────────────────────────────────────────────────────
blockers = empty_chunks + no_prefix + malformed_prefix + body_empty
warnings = too_short + too_long + incomplete + incomplete_math + broken_tables
verdict = "blocked" if blockers else ("warnings_only" if (warnings or duplicates) else "ok")
report = {
"stem": stem,
"verdict": verdict,
"stats": {
"total": n_total,
"ok": n_ok,
"min_chars": min_l,
"max_chars": max_l,
"avg_chars": avg_l,
"p50_chars": p50,
"under_min": n_under,
"in_range": n_norm,
"over_max": n_over,
"sections": [{"sezione": s, "n_chunks": n}
for s, n in section_counts.most_common()],
},
"thresholds": {
"min_chars": min_chars,
"max_chars": max_chars,
"target_chars": cfg.MAX_CHARS,
},
"blockers": {
"empty": [_chunk_entry(c) for c in empty_chunks],
"no_prefix": [_chunk_entry(c) for c in no_prefix],
"malformed_prefix": [_chunk_entry(c) for c in malformed_prefix],
"body_empty": [_chunk_entry(c) for c in body_empty],
"incomplete": [_chunk_entry(c) for c in incomplete],
},
"warnings": {
"too_short": [_chunk_entry(c) for c in too_short],
"too_long": [_chunk_entry(c) for c in too_long],
"incomplete_math": [_chunk_entry(c) for c in incomplete_math],
"broken_tables": [_chunk_entry(c) for c in broken_tables],
"duplicate_bodies": duplicates,
},
}
out_dir = project_root / "chunks" / stem
out_dir.mkdir(parents=True, exist_ok=True)
(out_dir / "report.json").write_text(
json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8"
)
print(f"\n report.json → chunks/{stem}/")
# ── Prossimi passi ────────────────────────────────────────────────────────
print(f"\n {'─' * 50}")
print(f" Verdict: {verdict.upper()}")
print(f" {'─' * 50}")
if verdict == "ok":
print(f" ✅ Tutto OK — procedi alla vettorizzazione:")
print(f" python ingestion/ingest.py --stem {stem}")
elif verdict == "warnings_only":
print(f" 🟡 Solo avvisi — puoi procedere alla vettorizzazione:")
print(f" python ingestion/ingest.py --stem {stem}")
if too_short or too_long:
print()
print(f" Per ottimizzare: correggi il sorgente .md e rigenera con --force")
else:
print(f" 🔴 {len(blockers)} problemi bloccanti — correggi prima di procedere:")
if empty_chunks or body_empty:
print(f" • chunk vuoti/senza corpo → controlla sources/{stem}/auto/{stem}_clean.md")
if no_prefix or malformed_prefix:
print(f" • prefisso mancante/malformato → controlla gli heading in sources/{stem}.md")
if incomplete:
print(f" • frasi spezzate → correggi il sorgente e rigenera con --force")
print()
print(f" Dopo le correzioni:")
print(f" python chunks/chunker.py --stem {stem} --force")
print(f" python chunks/verify_chunks.py --stem {stem}")
if warnings:
print()
print(f" 🟡 Hai anche {len(warnings)} avvisi — affrontali dopo aver risolto i 🔴.")
return not blockers
# ─── Entry point ──────────────────────────────────────────────────────────────
if __name__ == "__main__":
project_root = Path(__file__).parent.parent
parser = argparse.ArgumentParser(description="Verifica chunk")
parser.add_argument("--stem", help="Nome del documento (sottocartella di chunks/)")
parser.add_argument(
"--min", type=int, default=cfg.MIN_CHARS,
help=f"Soglia minima caratteri (default: {cfg.MIN_CHARS})"
)
parser.add_argument(
"--max", type=int, default=cfg.MAX_CHARS,
help=f"Soglia massima caratteri (default: {cfg.MAX_CHARS})"
)
args = parser.parse_args()
if args.stem:
stems = [args.stem]
else:
chunks_dir = project_root / "chunks"
if not chunks_dir.exists():
print(f"Errore: cartella chunks/ non trovata in {project_root}")
sys.exit(1)
stems = sorted(
p.name for p in chunks_dir.iterdir()
if p.is_dir() and (p / "chunks.json").exists()
)
if not stems:
print("Errore: nessun chunks.json trovato in chunks/")
sys.exit(1)
results = [verify_stem(s, project_root, args.min, args.max) for s in stems]
ok = sum(results)
total = len(results)
print(f"\n{'✅' if all(results) else '⚠️ '} {ok}/{total} documenti senza problemi bloccanti")
sys.exit(0 if all(results) else 1)
requirements.txt
+2
View File
@@ -2,3 +2,5 @@ pdfplumber==0.11.9
PyMuPDF>=1.24.0 PyMuPDF>=1.24.0
chromadb chromadb
pytest>=8.0 pytest>=8.0
markdown-it-py>=4.0
mdit-py-plugins>=0.4
tests/__init__.py
View File
tests/chunks/__init__.py
View File