feat(chunks): pipeline unificata Stage 1+2 con md_optimizer

chunker.py ora esegue in sequenza:
  - Stage 1 (md_optimizer.py): _content_list_v2.json + _model.json → _clean.md
    con pulizia TOC, frontespizio, sommari interni, merge titoli capitolo
  - Stage 2: _clean.md → chunks.json (paragraph-overlap, atomici tabelle/liste)

config.py esteso con CHAPTER_PREFIX_PATTERNS, SOMMARIO_PATTERNS,
MODEL_SKIP_LABELS, MODEL_ABSTRACT_LABELS, MIN_CONTENT_CHARS.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-05-20 16:07:40 +02:00
parent 52e881d09c
commit e96c510e1f
4 changed files with 855 additions and 480 deletions
+205 -395
View File
@@ -1,17 +1,29 @@
#!/usr/bin/env python3
"""
Chunking adattivo
Pipeline di chunking unificata (Stage 1 + Stage 2)
Divide il Markdown revisionato in chunk semantici pronti per la
vettorizzazione. La strategia dipende dal profilo strutturale del documento.
Stage 1 — Ottimizzazione Markdown (md_optimizer):
Legge _content_list_v2.json + _model.json di MinerU e produce _clean.md
con gerarchia H1/H2/H3 pulita (TOC, frontespizi e sommari rimossi).
Input: conversione/<stem>/clean.md + conversione/<stem>/structure_profile.json
Output: chunks/<stem>/chunks.json
Stage 2 — Chunking semantico:
Divide il _clean.md in chunk semantici:
- un chunk per paragrafo (mai due paragrafi nello stesso chunk)
- split a confine di frase se il paragrafo supera MAX_CHARS
- overlap di OVERLAP_SENTENCES frasi tra chunk consecutivi
- tabelle e liste sono blocchi atomici (non si spezzano)
Input: sources/<stem>/auto/<stem>_content_list_v2.json
sources/<stem>/auto/<stem>_model.json (opzionale)
Output: sources/<stem>/auto/<stem>_clean.md
chunks/<stem>/chunks.json
chunks/<stem>/meta.json
Uso:
python chunks/chunker.py # tutti i documenti in conversione/
python chunks/chunker.py --stem documento # un solo documento
python chunks/chunker.py --stem documento --force
python chunks/chunker.py --stem <stem>
python chunks/chunker.py # tutti gli stem in sources/
python chunks/chunker.py --stem <stem> --force
python chunks/chunker.py --stem <stem> --skip-optimize # salta Stage 1
"""
import argparse
@@ -24,474 +36,272 @@ _HERE = Path(__file__).resolve().parent
if str(_HERE) not in sys.path:
sys.path.insert(0, str(_HERE))
import config as cfg
from md_optimizer import optimize as _optimize_md
# ─── Utilità ──────────────────────────────────────────────────────────────────
def split_sentences(text: str) -> list[str]:
parts = re.split(r'(?<=[.!?»])\s+(?=[A-ZÀÈÉÌÒÙA-Z\"])', text.strip())
if len(parts) <= 1:
parts = re.split(r'(?<=[.!?»])\s+', text.strip())
parts = re.split(cfg.SENTENCE_SPLIT_RE, text.strip())
return [p.strip() for p in parts if p.strip()]
def slugify(s: str, max_len: int = 60) -> str:
s = s.lower()
s = re.sub(r'[^\w\s-]', '', s)
s = re.sub(r'[\s_-]+', '_', s).strip('_')
return s[:max_len] if s else "section"
def _is_table_block(text: str) -> bool:
"""True se il testo è prevalentemente una tabella Markdown (≥50% righe con |)."""
lines = [l for l in text.strip().splitlines() if l.strip()]
if not lines:
return False
table_lines = sum(1 for l in lines if l.strip().startswith("|"))
return table_lines / len(lines) >= 0.5
def _ov(strategy: str) -> tuple[int, float, int]:
"""Legge (target_chars, tolerance, overlap) dagli override di strategia."""
ov = cfg.STRATEGY_OVERRIDES.get(strategy, {})
target = ov.get("target_chars", cfg.TARGET_CHARS)
tolerance = ov.get("tolerance", cfg.CHUNK_TOLERANCE)
overlap = ov.get("overlap", cfg.OVERLAP_SENTENCES)
return target, tolerance, overlap
# ─── Core: split in sotto-chunk orientato al target ───────────────────────────
def make_sub_chunks(
body: str,
prefix: str,
sezione: str,
titolo: str,
target: int,
tolerance: float,
overlap_s: int,
) -> list[dict]:
"""Divide body in chunk il più vicini possibile a `target` char.
Logica:
lower = target × (1 tolerance) → soglia minima per emettere
upper = target × (1 + tolerance) → limite massimo
Si accumulano frasi intere finché la successiva farebbe superare `upper`.
A quel punto si emette (siamo vicini al target) e si riparte con overlap.
Ogni chunk termina sempre su un confine di frase; non attraversa mai
il boundary dell'header corrente.
"""
if cfg.PROTECT_TABLES and _is_table_block(body):
chunk_text = prefix + body
return [{
"chunk_id": f"{slugify(sezione)}__{slugify(titolo)}__s0",
"text": chunk_text,
"sezione": sezione,
"titolo": titolo,
"sub_index": 0,
"n_chars": len(chunk_text),
}]
# Soglia calcolata sul corpo (n_chars finale = prefix_len + body_len).
prefix_len = len(prefix)
upper_body = max(1, int(target * (1 + tolerance)) - prefix_len)
sentences = split_sentences(body)
if not sentences:
return []
chunks: list[dict] = []
current: list[str] = []
current_len = 0
sub_index = 0
def _emit() -> None:
nonlocal current, current_len, sub_index
chunk_text = prefix + " ".join(current)
chunks.append({
"chunk_id": f"{slugify(sezione)}__{slugify(titolo)}__s{sub_index}",
"text": chunk_text,
"sezione": sezione,
"titolo": titolo,
"sub_index": sub_index,
"n_chars": len(chunk_text),
})
overlap = current[-overlap_s:] if overlap_s and len(current) > overlap_s else []
current = overlap[:]
# Lunghezza corretta dell'overlap (n-1 spazi tra n frasi).
current_len = sum(len(s) for s in current) + max(0, len(current) - 1)
sub_index += 1
for sent in sentences:
sep = 1 if current else 0
new_len = current_len + sep + len(sent)
if new_len <= upper_body:
# Ancora entro il limite del corpo: aggiungi e continua.
current.append(sent)
current_len = new_len
elif current:
# La frase successiva sfora il limite: emetti il chunk corrente
# (che termina su frase completa) poi inizia il nuovo con questa frase.
_emit()
current.append(sent)
current_len += (1 if current[:-1] else 0) + len(sent)
else:
# Chunk vuoto: la singola frase supera già il limite — emettiamo così com'è.
current.append(sent)
current_len = len(sent)
_emit()
if current:
chunk_text = prefix + " ".join(current)
chunks.append({
"chunk_id": f"{slugify(sezione)}__{slugify(titolo)}__s{sub_index}",
"text": chunk_text,
"sezione": sezione,
"titolo": titolo,
"sub_index": sub_index,
"n_chars": len(chunk_text),
})
return chunks
def context_to_meta(context: str) -> tuple[str, str]:
"""Divide 'H1 > H2 > H3' in (sezione, titolo) per ingest/verify."""
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 ──────────────────────────────────────────────────────────
def parse_h3_sections(text: str) -> list[dict]:
sections = []
current_h2 = ""
current_h3 = ""
current_body_lines: list[str] = []
def parse_paragraphs(text: str) -> list[dict]:
"""Estrae blocchi dal _clean.md con il loro contesto heading.
def flush():
body = "\n".join(current_body_lines).strip()
Restituisce: [{"context": "H1 > H2 > H3", "text": "...", "kind": "text|table|list"}]
Ogni riga vuota chiude il paragrafo corrente. Tabelle (righe con |) e
liste (righe con -) vengono accumulate come blocchi atomici.
"""
h1 = h2 = h3 = ""
result: list[dict] = []
buf: list[str] = []
cur_kind = "text"
def flush() -> None:
body = "\n".join(buf).strip()
if body:
sections.append({
"sezione": current_h2,
"titolo": current_h3,
"body": body,
})
parts = [p for p in [h1, h2, h3] if p]
context = " > ".join(parts) if parts else "documento"
result.append({"context": context, "text": body, "kind": cur_kind})
buf.clear()
for line in text.splitlines():
if re.match(r"^# ", line):
flush()
current_h2 = line[2:].strip()
current_h3 = ""
current_body_lines = []
h1, h2, h3 = line[2:].strip(), "", ""
cur_kind = "text"
elif re.match(r"^## ", line):
flush()
current_h2 = line[3:].strip()
current_h3 = ""
current_body_lines = []
h2, h3 = line[3:].strip(), ""
cur_kind = "text"
elif re.match(r"^### ", line):
flush()
current_h3 = line[4:].strip()
current_body_lines = []
h3 = line[4:].strip()
cur_kind = "text"
elif line.strip().startswith("|"):
if cur_kind != "table":
flush()
cur_kind = "table"
buf.append(line)
elif line.strip().startswith("- "):
if cur_kind != "list":
flush()
cur_kind = "list"
buf.append(line)
elif line.strip() == "":
flush()
cur_kind = "text"
else:
current_body_lines.append(line)
if cur_kind in ("table", "list"):
flush()
cur_kind = "text"
buf.append(line)
flush()
return sections
return result
def parse_h2_sections(text: str) -> list[dict]:
sections = []
current_h2 = ""
current_body_lines: list[str] = []
# ─── Chunking ─────────────────────────────────────────────────────────────────
def flush():
body = "\n".join(current_body_lines).strip()
if body:
sections.append({"sezione": current_h2, "body": body})
def make_chunks(paragraphs: list[dict]) -> list[dict]:
"""Genera chunk dal risultato di parse_paragraphs.
for line in text.splitlines():
if re.match(r"^## ", line):
flush()
current_h2 = line[3:].strip()
current_body_lines = []
elif re.match(r"^# ", line):
flush()
current_h2 = line[2:].strip()
current_body_lines = []
else:
current_body_lines.append(line)
Regole:
- un chunk = un paragrafo (o sotto-parte se > MAX_CHARS)
- split solo a confine di frase; una frase che supera MAX_CHARS è emessa intera
- l'ultima frase del chunk N viene preposta al chunk N+1 (overlap)
- tabelle e liste: blocco atomico (mai spezzato)
"""
chunks: list[dict] = []
overlap_tail: list[str] = []
idx = 0
flush()
return sections
for para in paragraphs:
text = para["text"]
context = para["context"]
kind = para["kind"]
sezione, titolo = context_to_meta(context)
# ─── Strategie di chunking ────────────────────────────────────────────────────
def chunk_h3_aware(text: str, stem: str) -> list[dict]:
target, tolerance, overlap = _ov("h3_aware")
lower = int(target * (1 - tolerance))
sections = parse_h3_sections(text)
merged: list[dict] = []
pending: dict | None = None
for sec in sections:
if pending is None:
pending = dict(sec)
# ── Blocchi atomici (tabelle, liste) ──────────────────────────────────
if kind in ("table", "list"):
prefix = " ".join(overlap_tail) + " " if overlap_tail else ""
body = (prefix + text).strip()
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
sents = split_sentences(text)
overlap_tail = sents[-cfg.OVERLAP_SENTENCES:] if cfg.OVERLAP_SENTENCES else []
continue
if (pending["sezione"] == sec["sezione"]
and len(pending["body"]) < lower):
sep_title = " / ".join(filter(None, [pending["titolo"], sec["titolo"]]))
pending = {
"sezione": pending["sezione"],
"titolo": sep_title or pending["titolo"],
"body": pending["body"] + "\n\n" + sec["body"],
}
else:
merged.append(pending)
pending = dict(sec)
# ── Paragrafo testo: split a confine di frase ─────────────────────────
sents = split_sentences(text)
if not sents:
continue
if pending:
merged.append(pending)
current: list[str] = list(overlap_tail)
has_primary: bool = False
chunks = []
for sec in merged:
sezione = sec["sezione"] or stem
titolo = sec["titolo"] or ""
body = sec["body"]
prefix = f"[{sezione} > {titolo}]\n" if titolo else f"[{sezione}]\n"
chunks.extend(make_sub_chunks(body, prefix, sezione, titolo, target, tolerance, overlap))
for sent in sents:
candidate_len = len(" ".join(current + [sent]))
return chunks
def chunk_h2_paragraph_split(text: str, stem: str) -> list[dict]:
target, tolerance, overlap = _ov("h2_paragraph_split")
lower = int(target * (1 - tolerance))
sections = parse_h2_sections(text)
chunks = []
for sec in sections:
sezione = sec["sezione"] or stem
body = sec["body"]
prefix = f"[{sezione}]\n"
paragraphs = [
p.strip()
for p in re.split(r"\n{2,}", body)
if p.strip() and not re.match(r"^#+\s", p.strip())
]
merged_pars: list[str] = []
pending = ""
for par in paragraphs:
if pending and len(pending) < lower:
pending = pending + "\n\n" + par
if candidate_len <= cfg.MAX_CHARS or not has_primary:
current.append(sent)
has_primary = True
else:
if pending:
merged_pars.append(pending)
pending = par
if pending:
merged_pars.append(pending)
body = " ".join(current)
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
overlap_tail = current[-cfg.OVERLAP_SENTENCES:] if cfg.OVERLAP_SENTENCES else []
current = list(overlap_tail) + [sent]
has_primary = True
for idx, par in enumerate(merged_pars):
sub = make_sub_chunks(par, prefix, sezione, f"par{idx}", target, tolerance, overlap)
for c in sub:
c["chunk_id"] = f"{slugify(sezione)}__p{idx}__s{c['sub_index']}"
chunks.extend(sub)
if has_primary:
body = " ".join(current)
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
overlap_tail = current[-cfg.OVERLAP_SENTENCES:] if cfg.OVERLAP_SENTENCES else []
return chunks
def chunk_paragraph(text: str, stem: str) -> list[dict]:
target, tolerance, overlap = _ov("paragraph")
lower = int(target * (1 - tolerance))
# ─── Pipeline per documento ───────────────────────────────────────────────────
paragraphs = [
p.strip()
for p in re.split(r"\n{2,}", text)
if p.strip() and not re.match(r"^#+\s", p.strip())
]
prefix = f"[Documento: {stem}]\n"
def process_stem(stem: str, project_root: Path,
force: bool, skip_optimize: bool) -> bool:
"""Esegue Stage 1 (ottimizzazione MD) + Stage 2 (chunking) per un documento."""
merged: list[str] = []
pending = ""
for par in paragraphs:
if pending and len(pending) < lower:
pending = pending + "\n\n" + par
else:
if pending:
merged.append(pending)
pending = par
if pending:
merged.append(pending)
# ── Stage 1: ottimizzazione Markdown ──────────────────────────────────────
if not skip_optimize:
ok = _optimize_md(stem, project_root, force=force)
if not ok:
return False
else:
print(f"\n[Stage 1] skip (--skip-optimize)")
chunks = []
for idx, par in enumerate(merged):
sub = make_sub_chunks(par, prefix, stem, f"par{idx}", target, tolerance, overlap)
for c in sub:
c["chunk_id"] = f"para__{idx}__s{c['sub_index']}"
chunks.extend(sub)
# ── Stage 2: chunking ─────────────────────────────────────────────────────
clean_md = project_root / "sources" / stem / "auto" / f"{stem}_clean.md"
out_dir = project_root / "chunks" / stem
out_file = out_dir / "chunks.json"
return chunks
def chunk_sliding_window(text: str, stem: str) -> list[dict]:
target, tolerance, overlap = _ov("sliding_window")
upper = int(target * (1 + tolerance))
sentences = split_sentences(text)
prefix = f"[Documento: {stem}]\n"
chunks = []
i = 0
win_idx = 0
while i < len(sentences):
window: list[str] = []
cur_len = 0
j = i
while j < len(sentences):
s = sentences[j]
sep = 1 if window else 0
if window and cur_len + sep + len(s) > upper:
break
window.append(s)
cur_len += sep + len(s)
j += 1
if not window:
window = [sentences[i]]
j = i + 1
chunk_text = prefix + " ".join(window)
chunks.append({
"chunk_id": f"win__{win_idx}",
"text": chunk_text,
"sezione": stem,
"titolo": f"finestra {win_idx}",
"sub_index": win_idx,
"n_chars": len(chunk_text),
})
win_idx += 1
i += max(1, len(window) - overlap)
return chunks
# ─── Dispatcher ───────────────────────────────────────────────────────────────
_STRATEGIES: dict[str, callable] = {
"h3_aware": chunk_h3_aware,
"h2_paragraph_split": chunk_h2_paragraph_split,
"paragraph": chunk_paragraph,
"sliding_window": chunk_sliding_window,
}
def chunk_document(clean_md: Path, profile: dict, stem: str) -> list[dict]:
text = clean_md.read_text(encoding="utf-8")
strategia = profile.get("strategia_chunking", "paragraph")
fn = _STRATEGIES.get(strategia, chunk_paragraph)
return fn(text, stem)
# ─── Per-document processing ──────────────────────────────────────────────────
def process_stem(stem: str, project_root: Path, force: bool) -> bool:
conv_dir = project_root / "conversione" / stem
out_dir = project_root / "chunks" / stem
clean_md = conv_dir / "clean.md"
profile_path = conv_dir / "structure_profile.json"
out_file = out_dir / "chunks.json"
print(f"\nDocumento: {stem}")
print(f"[Stage 2] Chunking: {stem}")
if not clean_md.exists():
print(f" ✗ clean.md non trovato in conversione/{stem}/ — skip")
return False
if not profile_path.exists():
print(f" ✗ structure_profile.json non trovato in conversione/{stem}/ — skip")
print(f"{stem}_clean.md non trovato")
return False
if out_file.exists() and not force:
print(f" ⚠️ chunks.json già presente — skip")
print(f" (usa --force per rieseguire)")
print(f" chunks.json già presente — skip chunking")
return True
profile = json.loads(profile_path.read_text(encoding="utf-8"))
strategia = profile.get("strategia_chunking", "paragraph")
print(f" Strategia: {strategia}")
text = clean_md.read_text(encoding="utf-8")
paragraphs = parse_paragraphs(text)
chunks = chunk_document(clean_md, profile, stem)
if not paragraphs:
print(f" ✗ Nessun paragrafo estratto da {clean_md.name}")
return False
chunks = make_chunks(paragraphs)
if not chunks:
print(f" ✗ Nessun chunk generato — controlla clean.md")
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"
)
target, tolerance, _ = _ov(strategia)
lower = int(target * (1 - tolerance))
upper = int(target * (1 + tolerance))
meta = {"strategy": strategia, "target_chars": target,
"min_chars": lower, "max_chars": upper}
(out_dir / "meta.json").write_text(
json.dumps(meta, ensure_ascii=False), encoding="utf-8"
json.dumps({
"min_chars": cfg.MIN_CHARS,
"max_chars": cfg.MAX_CHARS,
"target_chars": cfg.MAX_CHARS,
"overlap": cfg.OVERLAP_SENTENCES,
"strategy": "paragraph_overlap",
}, ensure_ascii=False),
encoding="utf-8",
)
lengths = [c["n_chars"] for c in chunks]
min_c = min(lengths)
max_c = max(lengths)
avg_c = int(sum(lengths) / len(lengths))
short = sum(1 for l in lengths if l < lower)
long_ = sum(1 for l in lengths if l > upper)
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" Target: {target} char ±{int(tolerance*100)}% "
f"→ range [{lower}, {upper}]")
print(f" Chunk totali: {len(chunks)}")
print(f" Min: {min_c} char Max: {max_c} char Media: {avg_c} char")
if short:
print(f" ⚠️ {short} chunk sotto lower ({lower})")
if long_:
print(f" ⚠️ {long_} chunk sopra upper ({upper})")
print(f" ✅ chunks.json salvato in chunks/{stem}/")
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 ─────────────────────────────────────────────────────────────
# ─── Entry point ─────────────────────────────────────────────────────────────
if __name__ == "__main__":
project_root = Path(__file__).parent.parent
parser = argparse.ArgumentParser(description="Chunking adattivo")
parser.add_argument("--stem", help="Nome del documento (sottocartella di conversione/)")
parser.add_argument("--force", action="store_true", help="Riesegui anche se già presente")
parser = argparse.ArgumentParser(
description="Pipeline unificata MinerU → _clean.md → chunks.json"
)
parser.add_argument("--stem", help="Nome documento (sottocartella di sources/)")
parser.add_argument("--force", action="store_true",
help="Rigenera _clean.md e chunks.json anche se esistono")
parser.add_argument("--skip-optimize", action="store_true",
help="Salta Stage 1 (usa _clean.md già presente)")
args = parser.parse_args()
if args.stem:
stems = [args.stem]
else:
conv_dir = project_root / "conversione"
if not conv_dir.exists():
print(f"Errore: cartella conversione/ non trovata in {project_root}")
sys.exit(1)
sources_dir = project_root / "sources"
stems = sorted(
p.name for p in conv_dir.iterdir()
if p.is_dir() and (p / "clean.md").exists()
p.name for p in sources_dir.iterdir()
if p.is_dir()
and (p / "auto" / f"{p.name}_content_list_v2.json").exists()
)
if not stems:
print(f"Errore: nessun documento trovato in conversione/")
print("Errore: nessun documento MinerU trovato in sources/")
sys.exit(1)
results = [process_stem(s, project_root, args.force) for s in stems]
ok = sum(results)
total = len(results)
print(f"\n{'' if all(results) else '⚠️ '} {ok}/{total} documenti processati")
results = [
process_stem(s, project_root, args.force, args.skip_optimize)
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)