Healthcare document Q&A: chunking clinical PDFs correctly

Clinical PDFs break naive chunking. A discharge summary has structured sections — HPI, medications, allergies, plan — each with distinct semantics. A lab report embeds tables that lose meaning when split across chunks. Radiology reports nest findings and impressions in ways that demand section-aware boundaries. If you chunk by fixed token count or simple paragraph splits, you destroy the clinical context that makes retrieval useful.

This guide walks through a production-grade pipeline for chunking clinical PDFs: PDF parsing with layout awareness, section detection using medical document structure, table extraction and serialization, and chunk assembly with metadata that supports filtered retrieval. The code runs end-to-end with open-source tools.

Step 1: Parse with layout preservation

PyMuPDF (fitz) gives you block-level layout data — coordinates, font sizes, reading order — which is essential for detecting section headers and table boundaries. pdfplumber is an alternative; both work. We use fitz here for speed and coordinate fidelity.

# pdf_parser.py
import fitz  # pymupdf
from dataclasses import dataclass
from typing import List, Optional
import json

@dataclass
class TextBlock:
    text: str
    bbox: tuple  # (x0, y0, x1, y1)
    page_num: int
    font_size: float
    font_name: str
    is_bold: bool
    block_type: str  # "text", "table", "image"

def extract_blocks(pdf_path: str) -> List[TextBlock]:
    doc = fitz.open(pdf_path)
    blocks = []
    
    for page_num, page in enumerate(doc):
        # Get raw text blocks with font info
        raw_blocks = page.get_text("dict")["blocks"]
        
        for block in raw_blocks:
            if block["type"] == 0:  # text block
                for line in block["lines"]:
                    line_text = " ".join(span["text"] for span in line["spans"])
                    if not line_text.strip():
                        continue
                    
                    # Use first span for font metadata (good enough for headers)
                    first_span = line["spans"][0]
                    blocks.append(TextBlock(
                        text=line_text.strip(),
                        bbox=tuple(line["bbox"]),
                        page_num=page_num,
                        font_size=first_span["size"],
                        font_name=first_span["font"],
                        is_bold="bold" in first_span["font"].lower(),
                        block_type="text"
                    ))
            elif block["type"] == 1:  # image block - could be a table rendered as image
                blocks.append(TextBlock(
                    text="[IMAGE]",
                    bbox=tuple(block["bbox"]),
                    page_num=page_num,
                    font_size=0,
                    font_name="",
                    is_bold=False,
                    block_type="image"
                ))
    
    doc.close()
    return blocks

Verify: run on a sample discharge summary. You should see blocks with increasing y-coordinates per page, font sizes clustering around 10-12pt for body text and 14-18pt for headers.

Step 2: Detect clinical section headers

Clinical documents follow loose but recognizable patterns. Section headers are typically bold, larger font, often all-caps or title case, and appear at left margin. We build a heuristic classifier tuned for common clinical sections.

# section_detector.py
import re
from typing import List, Dict, Optional
from pdf_parser import TextBlock

CLINICAL_SECTIONS = {
    # History & physical
    "chief_complaint": ["chief complaint", "cc:", "presenting complaint"],
    "hpi": ["history of present illness", "hpi:", "present illness"],
    "pmh": ["past medical history", "pmh:", "medical history"],
    "psh": ["past surgical history", "psh:", "surgical history"],
    "medications": ["medications", "current medications", "meds:", "home medications"],
    "allergies": ["allergies", "drug allergies", "allergy:"],
    "family_history": ["family history", "fh:", "family hx"],
    "social_history": ["social history", "sh:", "social hx"],
    "review_of_systems": ["review of systems", "ros:", "systems review"],
    "physical_exam": ["physical exam", "physical examination", "pe:", "exam:"],
    "vitals": ["vital signs", "vitals:", "vs:"],
    "assessment": ["assessment", "impression:", "diagnosis:"],
    "plan": ["plan", "treatment plan", "disposition:", "follow-up:"],
    # Labs / radiology
    "results": ["results", "findings:", "result:"],
    "impression": ["impression", "conclusion:"],
    "recommendations": ["recommendations", "recommended:"],
}

HEADER_PATTERNS = [
    re.compile(r"^[A-Z][A-Z\s]{2,}:?$"),  # ALL CAPS header
    re.compile(r"^[A-Z][a-z]+(?:\s+[A-Z][a-z]+)*:?$"),  # Title Case
]

def is_likely_header(block: TextBlock, prev_block: Optional[TextBlock] = None) -> bool:
    text = block.text.strip()
    if len(text) > 80:  # headers are short
        return False
    if not (block.is_bold or block.font_size >= 13):
        return False
    if block.bbox[0] > 100:  # not left-aligned (adjust for your PDF margins)
        return False
    
    # Match known clinical sections
    text_lower = text.lower().rstrip(":")
    for section_key, aliases in CLINICAL_SECTIONS.items():
        if any(alias in text_lower for alias in aliases):
            return True
    
    # Fallback: pattern match
    return any(p.match(text) for p in HEADER_PATTERNS)

def classify_section(header_text: str) -> str:
    text_lower = header_text.lower().rstrip(":")
    for section_key, aliases in CLINICAL_SECTIONS.items():
        if any(alias in text_lower for alias in aliases):
            return section_key
    return "other"

def segment_blocks(blocks: List[TextBlock]) -> List[Dict]:
    """Group blocks into sections. Returns list of {section, blocks, page_start, page_end}."""
    segments = []
    current_section = "preamble"
    current_blocks = []
    section_start_page = blocks[0].page_num if blocks else 0
    
    for i, block in enumerate(blocks):
        prev = blocks[i-1] if i > 0 else None
        
        if is_likely_header(block, prev):
            # Save previous section
            if current_blocks:
                segments.append({
                    "section": current_section,
                    "blocks": current_blocks,
                    "page_start": section_start_page,
                    "page_end": prev.page_num if prev else section_start_page
                })
            # Start new section
            current_section = classify_section(block.text)
            current_blocks = [block]
            section_start_page = block.page_num
        else:
            current_blocks.append(block)
    
    # Final section
    if current_blocks:
        segments.append({
            "section": current_section,
            "blocks": current_blocks,
            "page_start": section_start_page,
            "page_end": blocks[-1].page_num if blocks else 0
        })
    
    return segments

Verify: print segments for a known PDF. You should see section keys like “hpi”, “medications”, “assessment” with block counts that match visual inspection.

Step 3: Extract and serialize tables

Clinical PDFs contain medication tables, lab result grids, vital sign tables. These must stay intact. pdfplumber extracts tables with cell alignment; we serialize each table to markdown for embedding compatibility.

# table_extractor.py
import pdfplumber
from typing import List, Dict, Optional
from pdf_parser import TextBlock

def extract_tables(pdf_path: str) -> List[Dict]:
    """Returns list of {page_num, bbox, markdown, raw_rows}."""
    tables = []
    with pdfplumber.open(pdf_path) as pdf:
        for page_num, page in enumerate(pdf.pages):
            page_tables = page.extract_tables({
                "vertical_strategy": "lines",
                "horizontal_strategy": "lines",
                "intersection_tolerance": 5,
                "snap_tolerance": 3,
            })
            
            for table_idx, table in enumerate(page_tables):
                if not table or len(table) < 2:
                    continue
                
                # Clean cells
                cleaned = [[cell.strip() if cell else "" for cell in row] for row in table]
                
                # Skip empty tables
                if not any(any(cell for cell in row) for row in cleaned):
                    continue
                
                # Convert to markdown
                markdown = table_to_markdown(cleaned)
                
                tables.append({
                    "page_num": page_num,
                    "table_index": table_idx,
                    "markdown": markdown,
                    "raw_rows": cleaned,
                    "bbox": None  # pdfplumber doesn't easily give bbox; approximate from page
                })
    return tables

def table_to_markdown(rows: List[List[str]]) -> str:
    if not rows:
        return ""
    
    # Assume first row is header
    header = rows[0]
    body = rows[1:]
    
    # Escape pipes in cells
    def escape(cell):
        return cell.replace("|", "\\|")
    
    header_row = "| " + " | ".join(escape(c) for c in header) + " |"
    separator = "| " + " | ".join(["---"] * len(header)) + " |"
    body_rows = ["| " + " | ".join(escape(c) for c in row) + " |" for row in body]
    
    return "\n".join([header_row, separator] + body_rows)

def associate_tables_with_sections(tables: List[Dict], segments: List[Dict]) -> List[Dict]:
    """Attach each table to the nearest preceding section."""
    for table in tables:
        table_page = table["page_num"]
        # Find section that contains this page
        assigned = False
        for seg in segments:
            if seg["page_start"] <= table_page <= seg["page_end"]:
                if "tables" not in seg:
                    seg["tables"] = []
                seg["tables"].append(table)
                assigned = True
                break
        if not assigned:
            # Fallback: create synthetic section
            segments.append({
                "section": "table_only",
                "blocks": [],
                "tables": [table],
                "page_start": table_page,
                "page_end": table_page
            })
    return segments

Verify: check markdown output for a medication table. Columns should align, headers preserved, no cell truncation.

Step 4: Build semantic chunks with overlap control

Now we assemble chunks. Strategy: each clinical section becomes one or more chunks. If a section exceeds token budget, split at paragraph boundaries with 100-token overlap. Tables become their own chunks (they’re dense). Preserve section metadata for filtered retrieval.

# chunker.py
from dataclasses import dataclass, asdict
from typing import List, Dict, Optional
import tiktoken
from pdf_parser import TextBlock

@dataclass
class Chunk:
    text: str
    section: str
    page_start: int
    page_end: int
    token_count: int
    chunk_index: int
    total_chunks_in_section: int
    document_id: str
    metadata: Dict  # extensible: patient_id, encounter_date, doc_type, etc.

def count_tokens(text: str, model: str = "text-embedding-3-small") -> int:
    enc = tiktoken.encoding_for_model(model)
    return len(enc.encode(text))

def split_long_section(text: str, max_tokens: int, overlap_tokens: int, model: str) -> List[str]:
    """Split at paragraph boundaries with token overlap."""
    enc = tiktoken.encoding_for_model(model)
    paragraphs = text.split("\n\n")
    
    chunks = []
    current_chunk_paras = []
    current_tokens = 0
    
    for para in paragraphs:
        para_tokens = len(enc.encode(para))
        
        if current_tokens + para_tokens > max_tokens and current_chunk_paras:
            # Emit current chunk
            chunks.append("\n\n".join(current_chunk_paras))
            
            # Start new chunk with overlap from tail of previous
            overlap_text = "\n\n".join(current_chunk_paras)
            overlap_token_ids = enc.encode(overlap_text)[-overlap_tokens:]
            overlap_str = enc.decode(overlap_token_ids)
            
            current_chunk_paras = [overlap_str, para]
            current_tokens = len(enc.encode(overlap_str)) + para_tokens
        else:
            current_chunk_paras.append(para)
            current_tokens += para_tokens
    
    if current_chunk_paras:
        chunks.append("\n\n".join(current_chunk_paras))
    
    return chunks

def build_chunks(
    segments: List[Dict],
    document_id: str,
    max_tokens: int = 512,
    overlap_tokens: int = 100,
    model: str = "text-embedding-3-small"
) -> List[Chunk]:
    chunks = []
    
    for seg in segments:
        section = seg["section"]
        page_start = seg["page_start"]
        page_end = seg["page_end"]
        
        # Combine text blocks
        section_text = "\n".join(b.text for b in seg["blocks"] if b.block_type == "text")
        
        # Handle tables: each table becomes its own chunk
        tables = seg.get("tables", [])
        
        # First, chunk the section text
        if section_text.strip():
            text_chunks = split_long_section(section_text, max_tokens, overlap_tokens, model)
            
            for idx, chunk_text in enumerate(text_chunks):
                chunks.append(Chunk(
                    text=chunk_text,
                    section=section,
                    page_start=page_start,
                    page_end=page_end,
                    token_count=count_tokens(chunk_text, model),
                    chunk_index=idx,
                    total_chunks_in_section=len(text_chunks),
                    document_id=document_id,
                    metadata={
                        "has_tables": len(tables) > 0,
                        "section": section,
                    }
                ))
        
        # Then, table chunks
        for table_idx, table in enumerate(tables):
            table_text = f"[TABLE: {section}]\n{table['markdown']}"
            chunks.append(Chunk(
                text=table_text,
                section=f"{section}_table",
                page_start=table["page_num"],
                page_end=table["page_num"],
                token_count=count_tokens(table_text, model),
                chunk_index=table_idx,
                total_chunks_in_section=len(tables),
                document_id=document_id,
                metadata={
                    "is_table": True,
                    "table_index": table_idx,
                    "raw_rows": table["raw_rows"],
                }
            ))
    
    return chunks

Verify: token counts should cluster near max_tokens (512) with few outliers. Table chunks may exceed budget — that’s acceptable; they’re atomic.

Step 5: Enrich with clinical metadata

Retrieval quality improves dramatically when you can filter by patient, encounter, document type, or section. Extract what you can from the PDF text and filename conventions.

# metadata_enricher.py
import re
from datetime import datetime
from typing import Dict, Optional
from chunker import Chunk

# Common patterns in clinical filenames: MRN_encounterDate_docType.pdf
FILENAME_PATTERN = re.compile(
    r"(?P<mrn>\d{6,10})_?(?P<encounter_date>\d{8})_?(?P<doc_type>[a-zA-Z]+)\.pdf$",
    re.IGNORECASE
)

DOC_TYPE_MAP = {
    "ds": "discharge_summary",
    "h&p": "history_physical",
    "hp": "history_physical",
    "op": "operative_note",
    "consult": "consultation",
    "rad": "radiology",
    "lab": "laboratory",
    "path": "pathology",
    "ed": "ed_note",
    "progress": "progress_note",
}

SECTION_WEIGHTS = {
    "chief_complaint": 1.5,
    "hpi": 1.3,
    "assessment": 1.4,
    "plan": 1.3,
    "medications": 1.2,
    "allergies": 1.5,
    "impression": 1.4,
    "results": 1.1,
    "table_only": 0.8,
}

def extract_metadata_from_filename(filename: str) -> Dict:
    match = FILENAME_PATTERN.search(filename)
    if not match:
        return {}
    
    mrn = match.group("mrn")
    date_str = match.group("encounter_date")
    doc_type_raw = match.group("doc_type").lower()
    
    try:
        encounter_date = datetime.strptime(date_str, "%Y%m%d").date().isoformat()
    except ValueError:
        encounter_date = None
    
    doc_type = DOC_TYPE_MAP.get(doc_type_raw, doc_type_raw)
    
    return {
        "mrn": mrn,
        "encounter_date": encounter_date,
        "document_type": doc_type,
    }

def extract_patient_demographics(text: str) -> Dict:
    """Best-effort extraction from first few pages."""
    meta = {}
    
    # DOB pattern
    dob_match = re.search(r"(?:DOB|Date of Birth|Born)[:\s]+(\d{1,2}[/-]\d{1,2}[/-]\d{2,4})", text, re.IGNORECASE)
    if dob_match:
        meta["patient_dob"] = dob_match.group(1)
    
    # Sex/Gender
    sex_match = re.search(r"\b(?:Sex|Gender)[:\s]+(Male|Female|M|F)\b", text, re.IGNORECASE)
    if sex_match:
        meta["patient_sex"] = sex_match.group(1)[0].upper()
    
    return meta

def enrich_chunks(chunks: List[Chunk], filename: str, full_text: str) -> List[Chunk]:
    file_meta = extract_metadata_from_filename(filename)
    demo_meta = extract_patient_demographics(full_text[:5000])  # first 5k chars
    
    for chunk in chunks:
        # Merge metadata
        chunk.metadata.update(file_meta)
        chunk.metadata.update(demo_meta)
        
        # Add section weight for retrieval boosting
        chunk.metadata["section_weight"] = SECTION_WEIGHTS.get(chunk.section, 1.0)
        
        # Document-level ID for deduplication
        chunk.metadata["document_id"] = chunk.document_id
    
    return chunks

Verify: inspect chunk.metadata for a known PDF. You should see mrn, encounter_date, document_type, section_weight populated.

Step 6: Embed and index with section-aware retrieval

Embed chunks. At query time, boost by section_weight and filter by metadata. This is where n4n.ai’s routing directives help — you can send clinical queries to a model tuned for medical reasoning while keeping the same embedding index.

# embed_and_index.py
import json
from typing import List
from chunker import Chunk
from openai import OpenAI

client = OpenAI()  # or your preferred embedding provider

EMBED_MODEL = "text-embedding-3-small"
BATCH_SIZE = 100

def embed_chunks(chunks: List[Chunk]) -> List[Dict]:
    """Returns list of {chunk, embedding} dicts."""
    records = []
    
    for i in range(0, len(chunks), BATCH_SIZE):
        batch = chunks[i:i+BATCH_SIZE]
        texts = [c.text for c in batch]
        
        response = client.embeddings.create(
            model=EMBED_MODEL,
            input=texts,
            encoding_format="float"
        )
        
        for chunk, emb_data in zip(batch, response.data):
            records.append({
                "id": f"{chunk.document_id}_chunk_{chunk.chunk_index}",
                "text": chunk.text,
                "embedding": emb_data.embedding,
                "metadata": chunk.metadata
            })
    
    return records

def save_index(records: List[Dict], output_path: str):
    with open(output_path, "w") as f:
        for r in records:
            f.write(json.dumps(r) + "\n")

# Example query-time filter construction
def build_retrieval_filter(
    mrn: Optional[str] = None,
    encounter_date: Optional[str] = None,
    document_type: Optional[str] = None,
    sections: Optional[List[str]] = None
) -> Dict:
    """Build a metadata filter for your vector DB (Pinecone, Weaviate, Qdrant, etc.)."""
    filter_dict = {}
    if mrn:
        filter_dict["mrn"] = mrn
    if encounter_date:
        filter_dict["encounter_date"] = encounter_date
    if document_type:
        filter_dict["document_type"] = document_type
    if sections:
        filter_dict["section"] = {"$in": sections}
    return filter_dict

Verify: load the JSONL, spot-check embeddings dimension (1536 for text-embedding-3-small), confirm metadata fields present.

Step 7: End-to-end pipeline script

Wire it together. This runs in ~30 seconds for a 50-page PDF on a modern laptop.

# pipeline.py
import sys
import os
from pdf_parser import extract_blocks
from section_detector import segment_blocks
from table_extractor import extract_tables, associate_tables_with_sections
from chunker import build_chunks
from metadata_enricher import enrich_chunks
from embed_and_index import embed_chunks, save_index

def process_pdf(pdf_path: str, output_path: str):
    document_id = os.path.splitext(os.path.basename(pdf_path))[0]
    
    print(f"[{document_id}] Extracting blocks...")
    blocks = extract_blocks(pdf_path)
    
    print(f"[{document_id}] Detecting sections...")
    segments = segment_blocks(blocks)
    
    print(f"[{document_id}] Extracting tables...")
    tables = extract_tables(pdf_path)
    segments = associate_tables_with_sections(tables, segments)
    
    print(f"[{document_id}] Building chunks...")
    chunks = build_chunks(segments, document_id)
    
    # Full text for demographic extraction
    full_text = "\n".join(b.text for b in blocks)
    chunks = enrich_chunks(chunks, os.path.basename(pdf_path), full_text)
    
    print(f"[{document_id}] Embedding {len(chunks)} chunks...")
    records = embed_chunks(chunks)
    
    print(f"[{document_id}] Saving index to {output_path}...")
    save_index(records, output_path)
    
    # Stats
    sections_seen = set(c.metadata.get("section") for c in chunks)
    table_chunks = sum(1 for c in chunks if c.metadata.get("is_table"))
    print(f"[{document_id}] Done. Sections: {sections_seen}, Table chunks: {table_chunks}, Total: {len(chunks)}")

if __name__ == "__main__":
    if len(sys.argv) != 3:
        print("Usage: python pipeline.py <input.pdf> <output.jsonl>")
        sys.exit(1)
    
    process_pdf(sys.argv[1], sys.argv[2])

Run it:

python pipeline.py "data/123456_20240115_ds.pdf" "index/123456_20240115_ds.jsonl"

Verify: output JSONL loads, each record has id, text, embedding (1536 floats), metadata with mrn, encounter_date, document_type, section, section_weight. Spot-check a medication table chunk — markdown renders cleanly.

Common failure modes and fixes

What this buys you

• Section-filtered retrieval: Query “current medications” → filter section=medications → precise hits.
• Table integrity: Lab values, med doses stay in structured markdown, not fragmented across chunks.
• Patient-scoped search: Filter by MRN + encounter date before vector search — reduces noise, respects privacy boundaries.
• Re-ranking signal: section_weight lets you boost assessment/plan over review-of-systems at query time.

The pipeline is modular. Swap fitz for pdfplumber, OpenAI embeddings for local sentence-transformers, JSONL for your vector DB’s bulk loader. The chunking logic — section-aware, table-preserving, metadata-rich — stays the same.