embedding-pipeline

Files (1)

• SKILL.md

SKILL.md (first lines)

# Embedding Pipeline

## Overview

An embedding pipeline converts text into dense vector representations that capture semantic meaning, enabling similarity search, RAG (Retrieval-Augmented Generation), recommendation systems, and clustering. This skill covers the full pipeline: text preprocessing and chunking, batched embedding generation with multiple providers, vector database storage, incremental updates for new content, and monitoring for embedding drift and quality degradation.

## When to Use

- Building a RAG system that needs to retrieve relevant context for LLM prompts
- Implementing semantic search over a document corpus (not just keyword matching)
- Recommendation systems based on content similarity
- Deduplication of large document sets using embedding similarity
- Clustering and categorizing unstructured text at scale

## Step-by-Step Workflow

### 1. Text Chunking — Critical Foundation
```python
from typing import List, Iterator
import re

def chunk_by_tokens(text: str, max_tokens: int = 512, overlap: int = 50) -> List[str]:
    """
    Sliding window chunking with token overlap.
    Overlap preserves context across chunk boundaries.
    Rough estimate: 1 token ≈ 4 chars for English.
    """
    words = text.split()
    chunk_size_words = max_tokens  # 1 token ≈ 1 word (rough)
    overlap_words = overlap
    
    chunks = []
    start = 0
    while start < len(words):
        end = min(start + chunk_size_words, len(words))
        chunk = " ".join(words[start:end])
        chunks.append(chunk)
        start += chunk_size_words - overlap_words
    
    return chunks

def chunk_by_sentences(text: str, max_chars: int = 2000, overlap_sentences: int = 2) -> List[str]:
    """
    Sentence-aware chunking — never splits mid-sentence.
    Better for documents where sentence boundaries matter.
    """
    # Split on sentence boundaries
    sentences = re.split(r'(?<=[.!?])\s+', text.strip())
    
    chunks = []
    current_chunk = []
    current_length = 0
    
    for sentence in sentences:
        if current_length + len(sentence) > max_chars and current_chunk:
            chunks.append(" ".join(current_chunk))
            # Overlap: keep last N sentences for context
            current_chunk = current_chunk[-overlap_sentences:]
            current_length = sum(len(s) for s in current_chunk)
        
        current_chunk.append(sentence)
        current_length += len(sentence)
    
    if current_chunk:
        chunks.append(" ".join(current_chunk))
    
    return chunks

def chunk_markdown(text: str, max_chars: int = 2000) -> List[dict]:
    """
    Structure-aware chunking for markdown documents.
    Respects heading hierarchy for better retrieval context.
    """
    chunks = []
    current_section = {"heading": "", "content": [], "level": 0}
    
    for line in text.split("\n"):
        heading_match = re.match(r"^(#{1,6})\s+(.+)", line)
        
        if heading_match:
            # Save current section before starting new one
            if current_section["content"]:
                content = "\n".join(current_section["content"])
                if content.strip():
                    chunks.append({
                        "heading": current_section["heading"],
                        "content": content,
                        "full_text": f"{current_section['heading']}\n{content}",
                    })
            
            level = len(heading_match.group(1))
            heading_text = heading_match.group(2)
            current_section = {"heading": f"{'#' * level} {heading_text}", "content": [], "level": level}
        else:
            current_section["content"].append(line)
    
    # Don't forget the last section
    if current_section["content"]:
        content = "\n".join(current_section["content"])
        if content.strip():
            chunks.append({
                "heading": current_section["heading"],
                "content": content,
                "full_text": f"{current_section['heading']}\n{content}",
            })
    
    return chunks
```

### 2. Batch Embedding Generation
```python
import asyncio
import time
from typing import List, Optional
import numpy as np
from openai import AsyncOpenAI

client = AsyncOpenAI()

From UltronCore/claude-skill-vault, MIT.

More from UltronCore/claude-skill-vault