| name | embedding-pipeline-builder |
| description | Builds document embedding pipelines with text chunking, embedding generation, indexing, and retrieval optimization. Use when users request "embedding pipeline", "document indexing", "text chunking", "RAG preprocessing", or "semantic indexing". |
Embedding Pipeline Builder
Build production-ready document embedding and retrieval pipelines.
Core Workflow
- Load documents: Ingest from various sources
- Preprocess text: Clean and normalize
- Chunk documents: Split into optimal sizes
- Generate embeddings: Create vector representations
- Index vectors: Store in vector database
- Optimize retrieval: Tune for accuracy
Pipeline Architecture
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Loader │───▶│ Preprocessor │───▶│ Chunker │
└─────────────┘ └─────────────┘ └─────────────┘
│
▼
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Retriever │◀───│ Indexer │◀───│ Embedder │
└─────────────┘ └─────────────┘ └─────────────┘
Document Loading
Multi-Source Loader
import { readFile, readdir } from 'fs/promises';
import { join, extname } from 'path';
import pdf from 'pdf-parse';
import mammoth from 'mammoth';
interface LoadedDocument {
id: string;
content: string;
metadata: {
source: string;
type: string;
title?: string;
createdAt?: Date;
[key: string]: any;
};
}
export class DocumentLoader {
async loadFile(filePath: string): Promise<LoadedDocument> {
const ext = extname(filePath).toLowerCase();
const content = await this.extractContent(filePath, ext);
return {
id: this.generateId(filePath),
content,
metadata: {
source: filePath,
type: ext.slice(1),
},
};
}
async loadDirectory(dirPath: string): Promise<LoadedDocument[]> {
const files = await readdir(dirPath, { recursive: true });
const documents: LoadedDocument[] = [];
for (const file of files) {
const filePath = join(dirPath, file);
try {
const doc = await this.loadFile(filePath);
documents.push(doc);
} catch (error) {
console.error(`Failed to load ${filePath}:`, error);
}
}
return documents;
}
private async extractContent(filePath: string, ext: string): Promise<string> {
const buffer = await readFile(filePath);
switch (ext) {
case '.txt':
case '.md':
return buffer.toString('utf-8');
case '.pdf':
const pdfData = await pdf(buffer);
return pdfData.text;
case '.docx':
const result = await mammoth.extractRawText({ buffer });
return result.value;
case '.json':
const json = JSON.parse(buffer.toString('utf-8'));
return this.flattenJson(json);
default:
throw new Error(`Unsupported file type: ${ext}`);
}
}
private flattenJson(obj: any, prefix = ''): string {
const parts: string[] = [];
for (const [key, value] of Object.entries(obj)) {
const path = prefix ? `${prefix}.${key}` : key;
if (typeof value === 'object' && value !== null) {
parts.push(this.flattenJson(value, path));
} else {
parts.push(`${path}: ${value}`);
}
}
return parts.join('\n');
}
private generateId(source: string): string {
return `doc_${Buffer.from(source).toString('base64url').slice(0, 16)}`;
}
}
Web Loader
import { JSDOM } from 'jsdom';
import { Readability } from '@mozilla/readability';
export class WebLoader {
async loadUrl(url: string): Promise<LoadedDocument> {
const response = await fetch(url);
const html = await response.text();
const dom = new JSDOM(html, { url });
const reader = new Readability(dom.window.document);
const article = reader.parse();
return {
id: this.generateId(url),
content: article?.textContent || '',
metadata: {
source: url,
type: 'webpage',
title: article?.title,
byline: article?.byline,
},
};
}
async loadSitemap(: ): <[]> {
response = (sitemapUrl);
xml = response.();
urlMatches = xml.() || [];
urls = urlMatches.(
match.(, ).(, )
);
: [] = [];
( url urls.(, )) {
{
doc = .(url);
documents.(doc);
( (r, ));
} (error) {
.(, error);
}
}
documents;
}
}
Text Preprocessing
export class TextPreprocessor {
process(text: string): string {
return this.pipeline(text, [
this.normalizeWhitespace,
this.removeSpecialCharacters,
this.normalizeUnicode,
this.removeExcessiveNewlines,
]);
}
private pipeline(text: string, transforms: Array<(t: string) => string>): string {
return transforms.reduce((t, fn) => fn(t), text);
}
private normalizeWhitespace(text: string): string {
return text
.replace(/\t/g, ' ')
.replace(/[ ]+/g, ' ')
.trim();
}
private removeSpecialCharacters(: ): {
text
.(, )
.(, );
}
(: ): {
text.();
}
(: ): {
text.(, );
}
}
Text Chunking
Smart Chunker
export interface Chunk {
id: string;
content: string;
metadata: {
documentId: string;
chunkIndex: number;
startChar: number;
endChar: number;
[key: string]: any;
};
}
export interface ChunkerOptions {
chunkSize: number;
chunkOverlap: number;
separators?: string[];
}
export class RecursiveChunker {
private options: ChunkerOptions;
private separators: string[];
constructor(options: ChunkerOptions) {
this.options = options;
this.separators = options.separators || [
'\n\n',
'\n',
'. ',
,
,
,
];
}
(: ): [] {
: [] = [];
textChunks = .(., );
textChunks.( {
chunks.({
: ,
: text.,
: {
: .,
: index,
: text.,
: text.,
....,
},
});
});
chunks;
}
(
: ,
:
): <{ : ; : ; : }> {
separator = .[separatorIndex];
{ chunkSize, chunkOverlap } = .;
(text. <= chunkSize) {
[{ : text, : , : text. }];
}
parts = separator ? text.(separator) : text.();
: <{ : ; : ; : }> = [];
currentChunk = ;
currentStart = ;
position = ;
( part parts) {
partWithSep = part + (separator || );
((currentChunk + partWithSep). > chunkSize) {
(currentChunk) {
(currentChunk. > chunkSize && separatorIndex < .. - ) {
subChunks = .(currentChunk, separatorIndex + );
results.(...subChunks.( ({
...c,
: c. + currentStart,
: c. + currentStart,
})));
} {
results.({
: currentChunk.(),
: currentStart,
: position,
});
}
overlapText = .(currentChunk, chunkOverlap);
currentChunk = overlapText + partWithSep;
currentStart = position - overlapText.;
} {
currentChunk = partWithSep;
}
} {
currentChunk += partWithSep;
}
position += partWithSep.;
}
(currentChunk.()) {
results.({
: currentChunk.(),
: currentStart,
: position,
});
}
results;
}
(: , : ): {
(overlapSize === ) ;
text.(-overlapSize);
}
}
Semantic Chunker
import { generateEmbedding } from '../embeddings';
export class SemanticChunker {
private similarityThreshold: number;
private minChunkSize: number;
private maxChunkSize: number;
constructor(options: {
similarityThreshold?: number;
minChunkSize?: number;
maxChunkSize?: number;
} = {}) {
this.similarityThreshold = options.similarityThreshold || 0.8;
this.minChunkSize = options.minChunkSize || 100;
this.maxChunkSize = options.maxChunkSize || 2000;
}
async chunk(document: LoadedDocument): Promise<Chunk[]> {
const sentences = this.splitIntoSentences(document.content);
const embeddings = await .(
sentences.( (s))
);
groups = .(sentences, embeddings);
groups.( ({
: ,
: group.(),
: {
: .,
: index,
: ,
....,
},
}));
}
(: ): [] {
text
.()
.( s. > );
}
(: [], : [][]): [][] {
: [][] = [];
: [] = [];
currentLength = ;
( i = ; i < sentences.; i++) {
sentence = sentences[i];
(currentGroup. === ) {
currentGroup.(sentence);
currentLength = sentence.;
;
}
similarity = .(embeddings[i], embeddings[i - ]);
(
similarity > . &&
currentLength + sentence. < .
) {
currentGroup.(sentence);
currentLength += sentence.;
} {
(currentLength >= .) {
groups.(currentGroup);
}
currentGroup = [sentence];
currentLength = sentence.;
}
}
(currentGroup. > ) {
groups.(currentGroup);
}
groups;
}
(: [], : []): {
dotProduct = ;
normA = ;
normB = ;
( i = ; i < a.; i++) {
dotProduct += a[i] * b[i];
normA += a[i] * a[i];
normB += b[i] * b[i];
}
dotProduct / (.(normA) * .(normB));
}
}
Embedding Generation
import OpenAI from 'openai';
import pLimit from 'p-limit';
const openai = new OpenAI();
export interface EmbeddedChunk extends Chunk {
embedding: number[];
}
export class Embedder {
private model: string;
private batchSize: number;
private concurrency: number;
constructor(options: {
model?: string;
batchSize?: number;
concurrency?: number;
} = {}) {
this.model = options.model || 'text-embedding-3-small';
this.batchSize = options.batchSize || 100;
this.concurrency = options.concurrency || 5;
}
async embedChunks(chunks: Chunk[]): <[]> {
limit = (.);
batches = .(chunks);
embeddedBatches = .(
batches.(
( .(batch))
)
);
embeddedBatches.();
}
(: []): [][] {
: [][] = [];
( i = ; i < chunks.; i += .) {
batches.(chunks.(i, i + .));
}
batches;
}
(: []): <[]> {
response = openai..({
: .,
: chunks.( c.),
});
chunks.( ({
...chunk,
: response.[i].,
}));
}
}
Complete Pipeline
import { DocumentLoader } from './loaders';
import { TextPreprocessor } from './preprocessor';
import { RecursiveChunker } from './chunker';
import { Embedder } from './embedder';
import { VectorStore } from './vectorstore';
export interface PipelineOptions {
chunkSize?: number;
chunkOverlap?: number;
embeddingModel?: string;
namespace?: string;
}
export class EmbeddingPipeline {
private loader: DocumentLoader;
private preprocessor: TextPreprocessor;
private chunker: RecursiveChunker;
private embedder: Embedder;
private vectorStore: VectorStore;
constructor(options: = {}) {
. = ();
. = ();
. = ({
: options. || ,
: options. || ,
});
. = ({
: options.,
});
. = (options.);
}
(: ): <{ : }> {
= ..(filePath);
.();
}
(: ): <{ : ; : }> {
documents = ..(dirPath);
totalChunks = ;
( doc documents) {
result = .(doc);
totalChunks += result.;
}
{ : documents., : totalChunks };
}
(: ): <{ : }> {
webLoader = ();
= webLoader.(url);
.();
}
(: ): <{ : }> {
. = ..(.);
chunks = ..();
embeddedChunks = ..(chunks);
..(embeddedChunks);
{ : embeddedChunks. };
}
(: , topK = ): <[]> {
..(query, topK);
}
}
pipeline = ({
: ,
: ,
: ,
});
pipeline.();
pipeline.();
results = pipeline.();
Best Practices
- Chunk size: Balance context vs noise (500-1500 chars)
- Overlap: 10-20% prevents context loss
- Preprocessing: Clean but preserve meaning
- Batch embeddings: Reduce API calls
- Add metadata: Enable filtering
- Semantic chunking: For high-quality retrieval
- Hybrid search: Combine vector and keyword
- Monitor quality: Test retrieval accuracy
Output Checklist
Every embedding pipeline should include: