rag_pipeline.py

"""
RAG Pipeline - Step-by-step with verbose output so you can watch the data flow.

Usage:
  python rag_pipeline.py --step load        # Step 1: Load .md files
  python rag_pipeline.py --step chunk       # Step 2: Chunk documents
  python rag_pipeline.py --step embed       # Step 3: Embed chunks (saves to runs/)
  python rag_pipeline.py --step all         # Run all steps sequentially

  --limit N         Only process first N files (default: 5, use 0 for all)
  --chunk-size N    Chunk size in characters (default: 1000)
  --overlap N       Chunk overlap in characters (default: 200)
  --run-name NAME   Custom name for this embed run (default: auto-generated)
  --list-runs       List all saved runs and exit
  --scan            Run boilerplate scan after chunking (produces boilerplate_report_N.json)
"""

import argparse
import glob
import os
import sys
import json
import pickle
import threading
import time
import math
from datetime import datetime

# ============================================================================
# LOAD .env (equivalent to: set -a && source .env && set +a)
# ============================================================================
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
_env_path = os.path.join(SCRIPT_DIR, ".env")
if os.path.isfile(_env_path):
    with open(_env_path) as _f:
        for _line in _f:
            _line = _line.strip()
            if not _line or _line.startswith("#"):
                continue
            key, _, val = _line.partition("=")
            os.environ.setdefault(key.strip(), val.strip().strip("'\""))

# ============================================================================
# CONFIG
# ============================================================================
JOBS_DIR = os.path.join(SCRIPT_DIR, "jobs/jobs-no-boilerplate")
RUNS_DIR = os.path.join(SCRIPT_DIR, "runs")
os.makedirs(RUNS_DIR, exist_ok=True)

# Colors for terminal output
C = {
    "RESET": "\033[0m",
    "DIM": "\033[2m",
    "BOLD": "\033[1m",
    "CYAN": "\033[36m",
    "GREEN": "\033[32m",
    "YELLOW": "\033[33m",
    "MAGENTA": "\033[35m",
    "RED": "\033[31m",
}

LOG_COLORS = {"LOAD": C["CYAN"], "CHUNK": C["MAGENTA"], "EMBED": C["GREEN"], "QUERY": C["YELLOW"], "INFO": C["DIM"], "ERROR": C["RED"]}

def log(tag, msg):
    color = LOG_COLORS.get(tag, C["RESET"])
    ts = datetime.now().strftime("%H:%M:%S")
    print(f"{C['DIM']}{ts}{C['RESET']} {color}[{tag}]{C['RESET']}  {msg}")

def separator(title):
    print(f"\n{C['BOLD']}{'='*60}")
    print(f"  {title}")
    print(f"{'='*60}{C['RESET']}\n")


# ============================================================================
# BRAILLE BOUNCING ANIMATION (transpiled from index.js startDotAnimation)
# ============================================================================
TRAIL = [
    {"char": "⣿", "color": "\033[1m\033[96m"},   # bold bright cyan - head
    {"char": "⣶", "color": "\033[96m"},           # bright cyan
    {"char": "⣤", "color": "\033[36m"},           # cyan
    {"char": "⠤", "color": "\033[2m\033[36m"},    # dim cyan
]
TRACK_LEN = 7

def start_dot_animation(tag, message):
    color = LOG_COLORS.get(tag, C["RESET"])
    fixed_ts = datetime.now().strftime("%H:%M:%S")
    start_time = time.time()
    head_pos = 0
    direction = 1
    last_advance = time.time()
    stop_event = threading.Event()
    current_msg = [message]  # mutable so render_loop sees updates

    def render_loop():
        nonlocal head_pos, direction, last_advance
        sys.stdout.write("\033[?25l")
        sys.stdout.flush()

        while not stop_event.is_set():
            now = time.time()
            if now - last_advance >= 0.1:
                head_pos += direction
                if head_pos >= TRACK_LEN - 1:
                    head_pos = TRACK_LEN - 1
                    direction = -1
                elif head_pos <= 0:
                    head_pos = 0
                    direction = 1
                last_advance = now

            elapsed = int(now - start_time)
            anim = ""
            for i in range(TRACK_LEN):
                if direction == 1:
                    dist = head_pos - i
                else:
                    dist = i - head_pos

                if 0 <= dist < len(TRAIL):
                    t = TRAIL[dist]
                    anim += t["color"] + t["char"] + C["RESET"]
                else:
                    anim += " "

            line = f"{C['DIM']}{fixed_ts}{C['RESET']} {color}[{tag}]{C['RESET']}  {current_msg[0]} {C['DIM']}({elapsed}s){C['RESET']} {anim}"
            sys.stdout.write(f"\r\033[K{line}")
            sys.stdout.flush()
            time.sleep(0.02)

    thread = threading.Thread(target=render_loop, daemon=True)
    thread.start()

    class Spinner:
        def update(self, new_message):
            current_msg[0] = new_message

        def stop(self):
            stop_event.set()
            thread.join()
            elapsed = f"{time.time() - start_time:.1f}"
            line = f"{C['DIM']}{fixed_ts}{C['RESET']} {color}[{tag}]{C['RESET']}  {current_msg[0]} {C['DIM']}({elapsed}s){C['RESET']}"
            sys.stdout.write(f"\r\033[K{line}\n")
            sys.stdout.write("\033[?25h")
            sys.stdout.flush()

    return Spinner()

# ============================================================================
# RUN PERSISTENCE (pickle)
# ============================================================================
def make_run_name(chunk_size, overlap, limit, num_chunks):
    """Auto-generate a run name from parameters."""
    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
    lim = "all" if limit == 0 else str(limit)
    return f"md_cs{chunk_size}_ol{overlap}_n{num_chunks}_{ts}"


def save_run(name, vectors, texts, metadatas, params, est_cost=0, embed_seconds=0):
    """Save an embed run to runs/<name>.pkl"""
    path = os.path.join(RUNS_DIR, f"{name}.pkl")
    data = {
        "name": name,
        "saved_at": datetime.now().isoformat(),
        "params": params,
        "num_vectors": len(vectors),
        "dims": len(vectors[0]) if vectors else 0,
        "est_cost_usd": round(est_cost, 6),
        "embed_seconds": round(embed_seconds, 1),
        "vectors": vectors,
        "texts": texts,
        "metadatas": metadatas,
    }
    with open(path, "wb") as f:
        pickle.dump(data, f)
    size_mb = os.path.getsize(path) / (1024 * 1024)
    log("EMBED", f"Run saved: runs/{name}.pkl ({size_mb:.1f} MB)")
    return path


def load_run(name):
    """Load a saved run from runs/<name>.pkl"""
    path = os.path.join(RUNS_DIR, f"{name}.pkl")
    if not os.path.exists(path):
        log("ERROR", f"Run not found: runs/{name}.pkl")
        available = [f[:-4] for f in os.listdir(RUNS_DIR) if f.endswith(".pkl")]
        if available:
            log("INFO", f"Available runs: {', '.join(available)}")
        sys.exit(1)
    with open(path, "rb") as f:
        data = pickle.load(f)
    log("QUERY", f"Loaded run: {name} ({data['num_vectors']} vectors, {data['dims']} dims)")
    log("QUERY", f"  params: {json.dumps(data['params'], indent=None)}")
    log("QUERY", f"  cost: ${data.get('est_cost_usd', 0):.6f} | embed time: {data.get('embed_seconds', 0):.1f}s")
    log("QUERY", f"  saved at: {data['saved_at']}")
    return data["vectors"], data["texts"], data["metadatas"]


def list_runs():
    """List all saved runs with their metadata."""
    files = sorted(glob.glob(os.path.join(RUNS_DIR, "*.pkl")))
    if not files:
        log("INFO", "No saved runs yet. Run --step embed to create one.")
        return
    separator("SAVED RUNS")
    for path in files:
        name = os.path.basename(path)[:-4]
        size_mb = os.path.getsize(path) / (1024 * 1024)
        try:
            with open(path, "rb") as f:
                data = pickle.load(f)
            params = data.get("params", {})
            cost = data.get('est_cost_usd', 0)
            secs = data.get('embed_seconds', 0)
            print(f"  {C['BOLD']}{name}{C['RESET']}")
            print(f"    {data['num_vectors']} vectors | {data['dims']} dims | {size_mb:.1f} MB")
            print(f"    params: fmt={params.get('format','?')} chunk_size={params.get('chunk_size','?')} overlap={params.get('overlap','?')} limit={params.get('limit','?')}")
            print(f"    cost: ${cost:.6f} | time: {secs:.1f}s")
            print(f"    saved: {data.get('saved_at', '?')}")
            print()
        except Exception:
            print(f"  {C['BOLD']}{name}{C['RESET']}  ({size_mb:.1f} MB) {C['RED']}[corrupt]{C['RESET']}")
            print()


# ============================================================================
# BOILERPLATE SCAN
# ============================================================================
BOILERPLATE_PATTERNS = [
    "AI REASONING", "Copilot:", "[Copilot:", "DESCRIPTION\nAbout the job",
    "vision", "disability", "linkedin.com/redir", "sexual orientation",
    "WEAK_MATCH", "PTO", "STRONG_MATCH", "national origin", "equal opportunity",
    "gender identity", "dental", "paid time off", "race, color", "without regard to",
    "gpt-4.1]", "religion, sex", "marital status", "health insurance",
    "protected veteran", "reasonable accommodation", "genetic information", "EEO",
    "affirmative action", "we do not discriminate", "salary range", "work authorization",
    "benefits include", "base salary", "compensation range", "background check",
    "pay range", "e-verify", "right to work", "accommodations due to",
    "401k", "401(k)", "EEOC", "Americans with Disabilities", "employment eligibility",
]


def next_report_number():
    """Find the next N for boilerplate_report_N.json by scanning existing files."""
    existing = glob.glob(os.path.join(SCRIPT_DIR, "boilerplate_report_*.json"))
    if not existing:
        return 1
    nums = []
    for path in existing:
        name = os.path.basename(path)
        # boilerplate_report_3.json -> "3"
        part = name.replace("boilerplate_report_", "").replace(".json", "")
        try:
            nums.append(int(part))
        except ValueError:
            pass
    return max(nums) + 1 if nums else 1


def scan_boilerplate(chunks, params):
    """Scan chunks for boilerplate patterns and save a numbered report."""
    log("CHUNK", "Scanning chunks for boilerplate patterns...")

    pattern_counts = {p: 0 for p in BOILERPLATE_PATTERNS}
    flagged = []
    total_chars = 0

    for i, chunk in enumerate(chunks):
        text = chunk.page_content
        total_chars += len(text)
        flags = []
        text_lower = text.lower()
        for pattern in BOILERPLATE_PATTERNS:
            if pattern.lower() in text_lower:
                pattern_counts[pattern] += 1
                flags.append(pattern)
        if len(flags) >= 3:
            flagged.append({
                "index": i,
                "source": chunk.metadata.get("source", "?"),
                "char_count": len(text),
                "flags": flags,
            })

    # Remove zero-count patterns for cleaner output
    pattern_counts = {k: v for k, v in pattern_counts.items() if v > 0}
    pattern_counts = dict(sorted(pattern_counts.items(), key=lambda x: x[1], reverse=True))

    n = next_report_number()
    fmt = params.get("format", "?")
    cs = params.get("chunk_size", "?")
    ol = params.get("overlap", "?")
    lim = params.get("limit", "?")
    pct = f"{len(flagged)/len(chunks)*100:.1f}%" if chunks else "0%"

    report = {
        "report_number": n,
        "description": f"format={fmt} | chunk_size={cs} | overlap={ol} | limit={lim} | {len(chunks)} chunks | {len(flagged)} flagged ({pct})",
        "generated_at": datetime.now().isoformat(),
        "pipeline_params": params,
        "total_chunks": len(chunks),
        "total_chars": total_chars,
        "pattern_counts": pattern_counts,
        "flagged_chunks_count": len(flagged),
        "flagged_chunks_pct": pct,
        "flagged_chunks": flagged[:200],
    }

    path = os.path.join(SCRIPT_DIR, f"boilerplate_report_{args.run_name}.json")
    with open(path, "w") as f:
        json.dump(report, f, indent=2)

    log("CHUNK", f"Boilerplate: {len(flagged)}/{len(chunks)} chunks flagged ({report['flagged_chunks_pct']}) -> boilerplate_report_{n}.json")
    return n


# ============================================================================
# STEP 1: LOAD DOCUMENTS
# ============================================================================
def step_load(limit):
    separator("STEP 1: LOAD DOCUMENTS")

    md_files = sorted(glob.glob(os.path.join(JOBS_DIR, "*.md")))
    log("LOAD", f"Found {len(md_files)} .md files in jobs/")

    if limit > 0:
        md_files = md_files[:limit]

    log("LOAD", f"Using {len(md_files)} files")

    documents = []
    total_chars = 0
    for filepath in md_files:
        filename = os.path.basename(filepath)
        with open(filepath, "r", encoding="utf-8") as f:
            content = f.read()
        documents.append({
            "filename": filename,
            "filepath": filepath,
            "content": content,
            "char_count": len(content),
            "line_count": content.count("\n") + 1,
        })
        total_chars += len(content)

    log("LOAD", f"Loaded {len(documents)} docs ({total_chars:,} chars total)")
    return documents


# ============================================================================
# STEP 2: CHUNK DOCUMENTS
# ============================================================================
def step_chunk(documents, chunk_size, overlap):
    separator("STEP 2: CHUNK DOCUMENTS")

    from langchain_text_splitters import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter

    headers_to_split_on = [
        ("#", "title"),
        ("##", "section"),
        ("###", "subsection"),
    ]
    md_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=overlap,
        length_function=len,
        separators=["\n\n", "\n", ". ", " ", ""],
    )

    log("CHUNK", f"MarkdownHeaderTextSplitter -> RecursiveCharacterTextSplitter(size={chunk_size}, overlap={overlap})")

    all_chunks = []
    for doc in documents:
        header_chunks = md_splitter.split_text(doc["content"])
        final_chunks = text_splitter.split_documents(header_chunks)
        for chunk in final_chunks:
            chunk.metadata["source"] = doc["filename"]
            all_chunks.append(chunk)

    avg = len(all_chunks) / len(documents) if documents else 0
    log("CHUNK", f"Total: {len(all_chunks)} chunks from {len(documents)} docs (avg {avg:.1f} chunks/file)")

    return all_chunks


# ============================================================================
# STEP 3: EMBED CHUNKS
# ============================================================================
def step_embed(chunks, run_name=None, params=None, embed_model_name="gemini-embedding-001"):
    separator("STEP 3: EMBED CHUNKS")

    from embed_model import get_embeddings, get_model_info

    model_info = get_model_info(embed_model_name)
    if not model_info:
        log("ERROR", f"Unknown embedding model: {embed_model_name}. Run: python embed_model.py --list")
        sys.exit(1)

    log("EMBED", f"Model: {embed_model_name} | provider: {model_info['provider']} | dims: {model_info['dims']}")
    embeddings_model = get_embeddings(embed_model_name)

    texts = [chunk.page_content for chunk in chunks]
    metadatas = [chunk.metadata for chunk in chunks]

    COST_PER_1M_TOKENS = model_info["cost_per_1m_tokens"]
    total_chars = sum(len(t) for t in texts)
    est_tokens = total_chars // 4
    est_cost = (est_tokens / 1_000_000) * COST_PER_1M_TOKENS

    log("EMBED", f"embed_documents() -> {len(texts)} texts, ~{est_tokens:,} tokens, est ${est_cost:.6f}")

    # Build batches -- HF free API has tiny limits, paid APIs can handle more
    CHARS_PER_TOKEN = 4
    provider = model_info["provider"]
    if provider == "huggingface_api":
        TOKEN_BATCH_LIMIT = 100_000     # HF inference API: moderate batches
        BATCH_DELAY = 0.5               # pause between batches to avoid 429s
    else:
        TOKEN_BATCH_LIMIT = 500_000     # paid APIs (Google, OpenAI)
        BATCH_DELAY = 0.0
    CHAR_BATCH_LIMIT = TOKEN_BATCH_LIMIT * CHARS_PER_TOKEN
    MAX_RETRIES = 8

    batches = []
    current_batch = []
    current_chars = 0
    for i, t in enumerate(texts):
        tc = len(t)
        if current_batch and current_chars + tc > CHAR_BATCH_LIMIT:
            batches.append(current_batch)
            current_batch = []
            current_chars = 0
        current_batch.append(i)
        current_chars += tc
    if current_batch:
        batches.append(current_batch)

    batch_tokens = [sum(len(texts[i]) for i in b) // 4 for b in batches]
    log("EMBED", f"Split into {len(batches)} batches (~{TOKEN_BATCH_LIMIT:,} tokens each)")

    # Check for existing checkpoint
    checkpoint_dir = os.path.join(RUNS_DIR, "checkpoints")
    os.makedirs(checkpoint_dir, exist_ok=True)
    ckpt_name = run_name or "current"
    ckpt_path = os.path.join(checkpoint_dir, f"{ckpt_name}.ckpt.pkl")
    vectors = [None] * len(texts)
    start_batch = 0

    if os.path.exists(ckpt_path):
        with open(ckpt_path, "rb") as f:
            ckpt = pickle.load(f)
        start_batch = ckpt["next_batch"]
        for idx, vec in zip(ckpt["indices"], ckpt["vectors"]):
            vectors[idx] = vec
        log("EMBED", f"Resuming from checkpoint: batch {start_batch}/{len(batches)} ({sum(1 for v in vectors if v is not None)} vectors cached)")

    embed_start = time.time()
    completed_indices = [i for i, v in enumerate(vectors) if v is not None]

    first_batch = batches[start_batch] if start_batch < len(batches) else batches[0]
    first_tok = batch_tokens[start_batch] if start_batch < len(batches) else 0
    spinner = start_dot_animation("EMBED", f"batch {start_batch+1}/{len(batches)} | {len(first_batch)} texts ~{first_tok:,} tok")

    for bi in range(start_batch, len(batches)):
        batch_indices = batches[bi]
        batch_texts = [texts[i] for i in batch_indices]
        batch_tok = batch_tokens[bi]
        batch_cost = (batch_tok / 1_000_000) * COST_PER_1M_TOKENS
        done_count = sum(1 for v in vectors if v is not None)

        spinner.update(f"batch {bi+1}/{len(batches)} | {len(batch_texts)} texts ~{batch_tok:,} tok | {done_count}/{len(texts)} embedded")

        for attempt in range(MAX_RETRIES):
            try:
                # Suppress any stdout/stderr from LangChain internals to protect spinner line
                import io as _io
                _devnull = _io.StringIO()
                _old_stderr = sys.stderr
                sys.stderr = _devnull
                try:
                    batch_vectors = embeddings_model.embed_documents(batch_texts)
                finally:
                    sys.stderr = _old_stderr
                break
            except Exception as e:
                if attempt < MAX_RETRIES - 1:
                    # 429 rate limit: wait longer (40s), other errors: exponential backoff
                    if "429" in str(e) or "ResourceExhausted" in str(e) or "quota" in str(e).lower():
                        wait = 90
                    else:
                        wait = 2 ** (attempt + 1)
                    err_msg = str(e)[:200]
                    spinner.stop()
                    log("ERROR", f"Batch {bi+1} attempt {attempt+1}/{MAX_RETRIES}: {err_msg}")
                    log("ERROR", f"Waiting {wait}s before retry...")
                    spinner = start_dot_animation("EMBED", f"batch {bi+1}/{len(batches)} | retry {attempt+1}/{MAX_RETRIES} in {wait}s...")
                    time.sleep(wait)
                else:
                    spinner.stop()
                    log("ERROR", f"Batch {bi+1} failed after {MAX_RETRIES} retries: {e}")
                    # Save checkpoint before dying
                    ckpt_vecs = [vectors[i] for i in completed_indices]
                    with open(ckpt_path, "wb") as f:
                        pickle.dump({"next_batch": bi, "indices": completed_indices, "vectors": ckpt_vecs}, f)
                    log("EMBED", f"Checkpoint saved: {ckpt_path} ({len(completed_indices)} vectors)")
                    raise

        for idx, vec in zip(batch_indices, batch_vectors):
            vectors[idx] = vec
            completed_indices.append(idx)

        # Throttle between batches for rate-limited APIs
        if BATCH_DELAY > 0:
            time.sleep(BATCH_DELAY)

        # Checkpoint after every batch
        ckpt_vecs = [vectors[i] for i in completed_indices]
        with open(ckpt_path, "wb") as f:
            pickle.dump({"next_batch": bi + 1, "indices": completed_indices, "vectors": ckpt_vecs}, f)

    spinner.stop()
    embed_duration = time.time() - embed_start

    # Clean up checkpoint on success
    if os.path.exists(ckpt_path):
        os.remove(ckpt_path)

    log("EMBED", f"Received {len(vectors)} vectors, {len(vectors[0])} dimensions each")

    # Show vectors for each chunk (only for small runs)
    if len(vectors) <= 50:
        for i, (text, vector, meta) in enumerate(zip(texts, vectors, metadatas)):
            text_preview = text[:60].replace("\n", " ")
            print(f"  {C['DIM']}chunk[{i}]{C['RESET']} src={meta.get('source','?')}  dims={len(vector)}  norm={sum(v*v for v in vector)**0.5:.4f}  \"{text_preview}...\"")

        if len(vectors) >= 2:
            print(f"\n  {C['BOLD']}cosine similarity:{C['RESET']}")
            for a in range(min(len(vectors), 4)):
                for b in range(a + 1, min(len(vectors), 4)):
                    dot = sum(x*y for x, y in zip(vectors[a], vectors[b]))
                    na = math.sqrt(sum(x*x for x in vectors[a]))
                    nb = math.sqrt(sum(x*x for x in vectors[b]))
                    sim = dot / (na * nb) if na and nb else 0
                    print(f"    [{a}]<->[{b}] = {sim:.6f}")
            print()

    # Persist run to pickle
    if params is None:
        params = {}
    name = run_name or make_run_name(
        params.get("chunk_size", "?"),
        params.get("overlap", "?"),
        params.get("limit", "?"),
        len(vectors),
    )
    save_run(name, vectors, texts, metadatas, params, est_cost=est_cost, embed_seconds=embed_duration)

    return vectors, texts, metadatas



# ============================================================================
# GRACEFUL SHUTDOWN
# ============================================================================
import signal

def _shutdown(signum, frame):
    name = signal.Signals(signum).name
    # Restore cursor in case a spinner hid it
    sys.stdout.write("\033[?25h")
    sys.stdout.flush()
    log("INFO", f"Received {name}, exiting.")
    sys.exit(0)

signal.signal(signal.SIGINT, _shutdown)
signal.signal(signal.SIGTERM, _shutdown)

# ============================================================================
# MAIN
# ============================================================================
if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="RAG Pipeline - step by step")
    parser.add_argument("--step", choices=["load", "chunk", "embed", "all"],
                        help="Which step to run")
    parser.add_argument("--limit", type=int, default=5, help="Number of files to process (0=all)")
    parser.add_argument("--embed-model", type=str, default="gemini-embedding-001", help="Embedding model name (see: python embed_model.py --list)")
    parser.add_argument("--chunk-size", type=int, default=1000, help="Chunk size in characters")
    parser.add_argument("--overlap", type=int, default=200, help="Chunk overlap in characters")
    parser.add_argument("--run-name", type=str, default=None, help="Custom name for this embed run")
    parser.add_argument("--list-runs", action="store_true", help="List all saved runs and exit")
    parser.add_argument("--scan", action="store_true", help="Run boilerplate scan after chunking")
    args = parser.parse_args()

    # --list-runs: show saved runs and exit
    if args.list_runs:
        list_runs()
        sys.exit(0)

    if not args.step:
        parser.error("--step is required (unless using --list-runs)")

    # Build params dict for run metadata
    run_params = {
        "format": "md",
        "chunk_size": args.chunk_size,
        "overlap": args.overlap,
        "limit": "all" if args.limit == 0 else args.limit,
    }

    log("INFO", f"step={args.step} limit={'all' if args.limit == 0 else args.limit} chunk_size={args.chunk_size} overlap={args.overlap}")

    if args.step in ("load", "all"):
        documents = step_load(args.limit)

    if args.step in ("chunk", "all"):
        if args.step == "chunk":
            documents = step_load(args.limit)
        chunks = step_chunk(documents, args.chunk_size, args.overlap)
        if args.scan:
            scan_boilerplate(chunks, run_params)

    if args.step in ("embed", "all"):
        if args.step == "embed":
            documents = step_load(args.limit)
            chunks = step_chunk(documents, args.chunk_size, args.overlap)
        if args.scan:
            scan_boilerplate(chunks, run_params)
        vectors, texts, metadatas = step_embed(chunks, run_name=args.run_name, params=run_params, embed_model_name=args.embed_model)