# Design — Markdown Chunker Phase 1 **Data:** 2026-06-08 **Scope:** Riscrittura completa di `chunks/` seguendo il blueprint. Solo Fase 1 (core strutturale). Immagini/asset registry rinviati a Fase 2. --- ## Obiettivo Sostituire il parser regex attuale con una pipeline modulare basata su AST reale (`markdown-it-py`). Output: `chunks.json` con schema blueprint, `meta.json`, `report.json`. --- ## Struttura file ``` chunks/ models.py # dataclass: Block, Chunk, ChunkingResult config.py # ChunkerConfig dataclass con valori default parser.py # Markdown → token stream + source map segmenter.py # token stream → Section tree → Block[] packer.py # Block[] → Chunk[] (packing min/target/max) validator.py # Chunk[] → invarianti + diagnostics chunker.py # CLI + orchestrazione pipeline ``` `verify_chunks.py` eliminato — validazione integrata in `validator.py`. --- ## Dipendenze ``` markdown-it-py[linkify]>=3.0 mdit-py-plugins>=0.4 # per math (dollarmath), footnotes ``` Nessun tokenizer esterno (tiktoken non disponibile con Ollama). I limiti sono in caratteri; il campo `chars` nei modelli è il proxy per token. --- ## Modelli (`models.py`) ```python @dataclass class Block: id: str # "blk_0001" kind: str # paragraph|heading|table|code|list|math|blockquote|html|thematic_break content: str # Markdown originale plain_text: str # testo pulito (senza sintassi MD) per embedding atomic: bool start_line: int end_line: int header_path: list[dict] # [{"level": 1, "text": "Titolo"}, ...] chars: int @dataclass class Chunk: chunk_id: str # "chk_000001" chunk_index: int content_original: str content_for_embedding: str # "H1 > H2 > H3\n\n" + content_original content_type: str # section_fragment | atomic_block | overflow chars: int start_line: int end_line: int header_path: list[dict] block_ids: list[str] flags: dict # has_code, has_table, has_math, is_overflow, is_sparse neighbors: dict # previous_chunk_id, next_chunk_id (popolato post-packing) assets: list # vuoto in Fase 1, pronto per Fase 2 @dataclass class ChunkingResult: stem: str source_path: str chunks: list[Chunk] diagnostics: Diagnostics @dataclass class Diagnostics: errors: list[str] warnings: list[str] metrics: dict ``` --- ## Config (`config.py`) ```python @dataclass class ChunkerConfig: # Dimensioni max_chars: int = 1200 min_chars: int = 80 target_chars: int = 800 # Heading context context_depth: int = 3 # 1-3 livelli nel prefisso embedding # Sezioni da saltare skip_headings: set[str] = field(default_factory=lambda: { "indice", "sommario", "bibliografia", "ringraziamenti", "abbreviazioni" }) skip_pre_heading: bool = True # Merge merge_short: bool = True # fonde paragrafi < min_chars consecutivi # Atomicità atomic_types: set[str] = field(default_factory=lambda: { "table", "code", "list", "math", "html" }) # Validazione fail_on_broken_fence: bool = True fail_on_content_loss: bool = False # warning invece di errore ``` Default caricabili da `chunks/config.yaml` se presente, altrimenti hardcoded. --- ## Parser (`parser.py`) **Responsabilità:** Markdown string → lista di token `markdown-it-py` con `map` (line ranges). ```python def parse(source: str) -> tuple[list[Token], list[str]]: """Ritorna (tokens, lines) con source map popolata.""" ``` - Configura `markdown-it` con plugin: `table`, `dollarmath` (math opzionale), `front_matter`. - Ogni token ha `token.map = [start_line, end_line]`. - Restituisce anche le righe sorgente per ricostruzione testo esatto. --- ## Segmenter (`segmenter.py`) **Responsabilità:** token stream → `Block[]` con heading stack e atomicità marcata. ```python def segment(tokens: list[Token], lines: list[str], config: ChunkerConfig) -> list[Block]: ``` Algoritmo: 1. Mantieni heading stack `[H1, H2, H3, H4, H5, H6]`. 2. Per ogni token di apertura (`is_opening`): - `heading_open` → aggiorna stack, non emette Block. - `fence` / `code_block` → Block atomico `kind=code`. - `table_open` → consuma fino a `table_close`, Block atomico `kind=table`. - `bullet_list_open` / `ordered_list_open` → consuma fino a close, Block atomico `kind=list`. - `math_block` → Block atomico `kind=math`. - `html_block` → Block atomico `kind=html`. - `hr` (thematic break) → Block `kind=thematic_break`. - `paragraph_open` → Block `kind=paragraph`. - `blockquote_open` → Block `kind=blockquote`. 3. Applica `skip_headings` e `skip_pre_heading`. 4. Calcola `header_path` per ogni Block dallo stack corrente. 5. `plain_text`: strip link syntax, immagini, codice inline, bold/italic. --- ## Packer (`packer.py`) **Responsabilità:** `Block[]` → `Chunk[]` rispettando `min/target/max_chars`. ```python def pack(blocks: list[Block], config: ChunkerConfig, stem: str) -> list[Chunk]: ``` Algoritmo: 1. Raggruppa Block per `header_path` (cambio header_path = confine obbligatorio). 2. All'interno del gruppo, packing greedy: - Accumula Block finché `chars < target_chars`. - Se aggiungere il prossimo Block supera `max_chars`: - Se il Block è atomico → flush chunk corrente, Block diventa chunk dedicato (marcato `atomic_block` o `overflow` se > max_chars). - Se il Block è `paragraph` → split a confine di frase (regex `(?<=[.!?»])\s+(?=[A-ZÀÈÉÌÒÙ\"])`); ogni frammento ≥ min_chars diventa sotto-Block. - Se il chunk corrente non raggiunge `min_chars` e c'è un Block successivo con stesso header_path → merge. 3. Heading orfani (solo heading senza body): uniti al chunk successivo o marcati `is_sparse=true`. 4. Popola `content_for_embedding = header_path_prefix + "\n\n" + content_original`. 5. Popola `neighbors` in un secondo passaggio. **header_path_prefix:** `"H1 > H2 > H3\n\n"` con `context_depth` livelli. --- ## Validator (`validator.py`) **Responsabilità:** controlla invarianti, produce `Diagnostics`. ```python def validate(result: ChunkingResult, source: str, config: ChunkerConfig) -> Diagnostics: ``` Invarianti controllati: - Nessun code fence aperto/chiuso male nel `content_original`. - Nessun chunk con solo heading (heading orfano). - Tutti i chunk rispettano `max_chars` (salvo `is_overflow=true`). - Copertura: le righe sorgente significative sono coperte da almeno un chunk (warning se non `fail_on_content_loss`). - Nessun `chunk_id` duplicato. Metriche emesse: - `total_chunks`, `total_chars`, `avg_chars`, `min_chars_actual`, `max_chars_actual` - `overflow_count`, `sparse_count`, `atomic_count` - `size_compliance`: % chunk entro `[min_chars, max_chars]` --- ## CLI (`chunker.py`) ``` python chunks/chunker.py --stem # singolo documento python chunks/chunker.py # tutti gli stem in sources/ python chunks/chunker.py --stem --force # rigenera anche se presente ``` Ricerca sorgente in ordine: 1. `sources/_output/auto/.md` 2. `sources/.md` Output: ``` chunks//chunks.json # lista Chunk serializzata chunks//meta.json # stem, source_path, total_chunks, created_at, config snapshot chunks//report.json # Diagnostics (errors, warnings, metrics) ``` --- ## Output schema — `chunks.json` (array di oggetti) ```json { "chunk_id": "chk_000001", "chunk_index": 1, "content_original": "...", "content_for_embedding": "H1 > H2\n\nIl refresh token...", "content_type": "section_fragment", "chars": 742, "start_line": 12, "end_line": 31, "header_path": [ {"level": 1, "text": "Titolo documento"}, {"level": 2, "text": "Sezione"} ], "block_ids": ["blk_0010", "blk_0011"], "flags": { "has_code": false, "has_table": false, "has_math": false, "is_overflow": false, "is_sparse": false }, "neighbors": { "previous_chunk_id": null, "next_chunk_id": "chk_000002" }, "assets": [] } ``` --- ## Cosa NON è in scope (Fase 2) - Asset registry immagini (campo `assets` presente ma vuoto). - Fetch remoto. - Token counting con tiktoken. - Table splitting con header ripetuto (tabelle rimangono atomiche). - Tree-sitter per code splitting. - Metriche retrieval (Recall@k, MRR).