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

feat(chunks): target-based chunking con config centralizzata

Browse Source 
Introduce chunks/config.py come unica fonte di verità per tutti i
parametri della pipeline di chunking. TARGET_CHARS + CHUNK_TOLERANCE
sostituiscono MIN_CHARS/MAX_CHARS: il chunker mira a una dimensione
target e si avvicina il più possibile rispettando il vincolo assoluto
di terminare ogni chunk su un confine di frase (punto/punteggiatura).

- config.py: TARGET_CHARS, CHUNK_TOLERANCE, SPLIT_THRESHOLD_FACTOR,
  PROTECT_TABLES, FIX_MAX_ITERATIONS, STRATEGY_OVERRIDES per strategia
- chunker.py: algoritmo target-based (emit quando frase successiva
  sfora upper_body = upper - prefix_len), table protection atomica,
  override MIN/MAX/overlap per ciascuna delle 4 strategie
- verify_chunks.py: soglie derivate da target*(1±tolerance)
- fix_chunks.py: _split_at_boundary sempre su punteggiatura finale,
  loop ricorsivo fix→verify fino a FIX_MAX_ITERATIONS, split solo
  per chunk > upper × SPLIT_THRESHOLD_FACTOR

Risultato su bitcoin: 694 chunk, 0 incompleti, 83% in range [450,750],
tutti terminanti su punteggiatura indipendentemente dalla dimensione.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Davide Grilli
2026-05-11 15:45:24 +02:00
parent d590346539
commit e264bc52b1
4 changed files with 342 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files
chunks/fix_chunks.py
+96 -32
View File
@@ -26,7 +26,13 @@ import re
import sys
from pathlib import Path
MAX_CHARS = 800
_HERE = Path(__file__).resolve().parent
if str(_HERE) not in sys.path:
sys.path.insert(0, str(_HERE))
import config as cfg
from verify_chunks import verify_stem as _verify_stem
MAX_CHARS = int(cfg.TARGET_CHARS * (1 + cfg.CHUNK_TOLERANCE))
PUNCT_END = re.compile(r"[.!?»)\]'\u2019\"\u201c\u201d\u2018\u2014\u2013-]$")
@@ -53,7 +59,11 @@ def _rebuild_text(chunk: dict, body: str) -> str:
return f"{_prefix(chunk)}\n{body}"
_SENT_END = re.compile(r'[.!?»)\]\'"’”…]')
def _split_at_boundary(text: str, max_chars: int) -> list[str]:
"""Spezza text in parti ≤ max_chars, ciascuna terminante su punteggiatura."""
if len(text) <= max_chars:
return [text]
@@ -62,20 +72,28 @@ def _split_at_boundary(text: str, max_chars: int) -> list[str]:
while len(remaining) > max_chars:
candidate = remaining[:max_chars]
split_pos = candidate.rfind("\n\n")
if split_pos == -1:
m = None
for m in re.finditer(r"[.!?»]\s+", candidate):
pass
split_pos = m.end() if m else None
# Cerca l'ultima punteggiatura finale entro max_chars.
last_punct = -1
for m in _SENT_END.finditer(candidate):
last_punct = m.end() # posizione dopo il carattere di punteggiatura
if split_pos is None or split_pos == 0:
sp = remaining.find(" ", max_chars)
split_pos = sp if sp != -1 else len(remaining)
if last_punct > 0:
# Taglia dopo la punteggiatura; il resto inizia alla parola successiva.
first = remaining[:last_punct].rstrip()
remaining = remaining[last_punct:].lstrip()
else:
# Nessuna punteggiatura: taglia all'ultimo spazio disponibile.
sp = candidate.rfind(" ")
if sp > 0:
first = remaining[:sp].rstrip()
remaining = remaining[sp:].lstrip()
else:
first = remaining[:max_chars]
remaining = remaining[max_chars:]
parts.append(remaining[:split_pos].rstrip())
remaining = remaining[split_pos:].lstrip()
if first:
parts.append(first)
if remaining:
parts.append(remaining)
@@ -202,7 +220,15 @@ def fix_stem(stem: str, project_root: Path, max_chars: int, dry_run: bool) -> bo
no_prefix_ids = {e["chunk_id"] for e in report.get("blockers", {}).get("no_prefix", [])}
incomplete_ids = {e["chunk_id"] for e in report.get("blockers", {}).get("incomplete", [])}
too_short_ids = {e["chunk_id"] for e in report.get("warnings", {}).get("too_short", [])}
too_long_ids = {e["chunk_id"] for e in report.get("warnings", {}).get("too_long", [])}
# Spezza solo chunk che superano upper × SPLIT_THRESHOLD_FACTOR,
# non quelli appena oltre upper (che causerebbero split con chunk incompleti).
_split_limit = max_chars * cfg.SPLIT_THRESHOLD_FACTOR
too_long_ids = {
e["chunk_id"]
for e in report.get("warnings", {}).get("too_long", [])
if e.get("n_chars", 0) > _split_limit
}
ops: list[str] = []
if empty_ids:
@@ -230,24 +256,54 @@ def fix_stem(stem: str, project_root: Path, max_chars: int, dry_run: bool) -> bo
n_before = len(chunks)
if empty_ids:
chunks, n = fix_empty(chunks, empty_ids)
print(f"\n 🗑 Rimossi {n} chunk vuoti.")
def _apply_fixes(chunks: list[dict], report: dict) -> list[dict]:
empty_ids_ = {e["chunk_id"] for e in report.get("blockers", {}).get("empty", [])}
no_prefix_ids_ = {e["chunk_id"] for e in report.get("blockers", {}).get("no_prefix", [])}
incomplete_ids_= {e["chunk_id"] for e in report.get("blockers", {}).get("incomplete", [])}
too_short_ids_ = {e["chunk_id"] for e in report.get("warnings", {}).get("too_short", [])}
too_long_ids_ = {
e["chunk_id"]
for e in report.get("warnings", {}).get("too_long", [])
if e.get("n_chars", 0) > max_chars * cfg.SPLIT_THRESHOLD_FACTOR
}
if no_prefix_ids:
chunks, n = fix_no_prefix(chunks, no_prefix_ids)
print(f" 🔧 Aggiunto prefisso a {n} chunk.")
if empty_ids_:
chunks, n = fix_empty(chunks, empty_ids_)
print(f" 🗑 Rimossi {n} chunk vuoti.")
if no_prefix_ids_:
chunks, n = fix_no_prefix(chunks, no_prefix_ids_)
print(f" 🔧 Aggiunto prefisso a {n} chunk.")
merge_ids_ = incomplete_ids_ | too_short_ids_
if merge_ids_:
chunks, n = fix_incomplete_and_short(chunks, merge_ids_)
print(f" 🔗 Fusi {n} chunk (incompleti + corti).")
if too_long_ids_:
chunks, n = fix_too_long(chunks, too_long_ids_, max_chars)
print(f" ✂️ Spezzati {n} chunk lunghi.")
return renumber_ids(chunks)
merge_ids = incomplete_ids | too_short_ids
if merge_ids:
chunks, n = fix_incomplete_and_short(chunks, merge_ids)
print(f" 🔗 Fusi {n} chunk (incompleti + corti).")
chunks = _apply_fixes(chunks, report)
if too_long_ids:
chunks, n = fix_too_long(chunks, too_long_ids, max_chars)
print(f" ✂️ Spezzati {n} chunk lunghi.")
chunks = renumber_ids(chunks)
for iteration in range(1, cfg.FIX_MAX_ITERATIONS):
chunks_path.write_text(
json.dumps(chunks, ensure_ascii=False, indent=2), encoding="utf-8"
)
project_root = chunks_path.parent.parent.parent
_min = int(cfg.TARGET_CHARS * (1 - cfg.CHUNK_TOLERANCE))
_max = int(cfg.TARGET_CHARS * (1 + cfg.CHUNK_TOLERANCE))
_verify_stem(stem, project_root, _min, _max)
report = json.loads(report_path.read_text(encoding="utf-8"))
new_verdict = report.get("verdict", "ok")
if new_verdict in ("ok", "warnings_only"):
break
remaining_blockers = sum(
len(v) for v in report.get("blockers", {}).values()
)
if remaining_blockers == 0:
break
print(f"\n Iterazione {iteration + 1}/{cfg.FIX_MAX_ITERATIONS} "
f"— {remaining_blockers} bloccer residui:")
chunks = _apply_fixes(chunks, report)
n_after = len(chunks)
print(f"\n Totale chunk: {n_before} → {n_after}")
@@ -256,8 +312,15 @@ def fix_stem(stem: str, project_root: Path, max_chars: int, dry_run: bool) -> bo
json.dumps(chunks, ensure_ascii=False, indent=2), encoding="utf-8"
)
print(f" ✅ Salvato: chunks/{stem}/chunks.json")
print(f"\n Riesegui la verifica:")
print(f" python chunks/verify_chunks.py --stem {stem}")
final_verdict = report.get("verdict", "?")
if final_verdict == "ok":
print(f" ✅ Verdict finale: ok — procedi alla vettorizzazione.")
elif final_verdict == "warnings_only":
print(f" 🟡 Verdict finale: warnings_only — puoi procedere.")
else:
print(f" 🔴 Verdict finale: {final_verdict} — rilancia la verifica manualmente:")
print(f" python chunks/verify_chunks.py --stem {stem}")
return True
@@ -269,9 +332,10 @@ if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Fix chunk")
parser.add_argument("--stem", required=True, help="Nome del documento (sottocartella di chunks/)")
_max_def = int(cfg.TARGET_CHARS * (1 + cfg.CHUNK_TOLERANCE))
parser.add_argument(
"--max", type=int, default=MAX_CHARS,
help=f"Soglia massima caratteri per lo split (default: {MAX_CHARS})"
"--max", type=int, default=_max_def,
help=f"Soglia massima caratteri per lo split (default: TARGET×(1+TOL) = {_max_def})"
)
parser.add_argument(
"--dry-run", action="store_true",