A multi-format extraction pipeline is an ingestion system that accepts heterogeneous file types—text documents, source code, PDFs, images, audio, and video—and produces uniformly structured, searchable records. The core challenge is routing each file to the correct extraction strategy while maintaining consistent metadata across all output documents, regardless of the original format.
A ProcessFile function embodies this pattern. It enforces per-file size limits, computes content hashes for deduplication, classifies the input into format categories (text, code, PDF, image, audio/video), and dispatches each file to a specialized processor—a PDF extraction service for PDFs, an image analysis service for OCR/captioning/embeddings, and a speech recognition service for audio and video. Every output document is tagged with the source filename and the watch_dir identifier that triggered ingestion, enabling downstream systems to track provenance and trigger reindexing when source files change.
This architecture is essential for building search engines over heterogeneous document collections. Without a unified extraction layer, each consumer would need to understand every input format, leading to duplicated parsing logic and inconsistent metadata. The pipeline centralizes format-specific complexity behind a single interface that always produces indexed documents with predictable fields.
Rather than relying solely on file extensions, robust extraction pipelines use a combination of extension matching, MIME type detection, and magic byte inspection to classify files. The extraction service groups files into five categories: plain text, source code (which receives language-aware chunking), PDF documents, images, and audio/video media. This classification determines which downstream processor handles the file and what metadata fields will be populated in the resulting document.
File classification is the critical first step in any multi-format pipeline. Misclassifying a binary image as text produces garbage output that pollutes search results. The classification function acts as a router: it inspects the file, selects the appropriate processor, and guarantees that downstream code always receives data in the format it expects.
Before processing, the pipeline computes a cryptographic hash (typically SHA-256) of the file contents. This hash serves two purposes: deduplication (skip reprocessing if the hash matches an already-indexed document) and change detection (reindex when the hash changes). Hashing is cheap compared to the extraction cost of running OCR or transcription, so the dedup check provides significant savings in pipelines that re-scan directories on a schedule.
Content hashing makes the pipeline idempotent. Running the pipeline twice on the same file produces the same hash, so the second run can short-circuit without invoking expensive processors. This is the same principle behind Git’s object store and content-addressable storage systems—the content’s identity is its hash.
Extraction pipelines enforce maximum file sizes to prevent resource exhaustion. A single multi-gigabyte video file could monopolize memory and CPU for minutes, starving other files in the queue. The pipeline checks file size before reading contents and rejects files that exceed the configured threshold, logging a warning so operators can adjust limits or pre-process oversized files externally.
Each ingested document is tagged with its watch_dir—the monitored directory that triggered the file’s discovery. This tag enables scoped reindexing: when a watch directory is rescanned, only documents with that watch_dir tag need to be checked for updates or deletions. Without this tag, reindexing would require scanning the entire document store.
Every document produced by the pipeline carries metadata about where it came from: filename, watch directory, content hash, and extraction timestamp. This provenance chain enables debugging (which file produced this bad search result?), reprocessing (reindex everything from this directory), and auditing (when was this document last updated?).
The extraction pipeline processes each file through a five-stage sequence: validation, hashing, classification, extraction, and tagging. Each stage can reject the file (too large, duplicate, unsupported format) or pass it to the next stage with accumulated metadata.
// Go: Multi-format file processor with routing
package extraction
import (
"crypto/sha256"
"fmt"
"io"
"os"
"path/filepath"
"strings"
)
type FileCategory int
const (
CategoryText FileCategory = iota
CategoryCode
CategoryPDF
CategoryImage
CategoryAudioVideo
CategoryUnsupported
)
const MaxFileSize = 100 * 1024 * 1024 // 100 MB
// ClassifyFile determines the processing strategy for a file.
func ClassifyFile(path string) FileCategory {
ext := strings.ToLower(filepath.Ext(path))
switch ext {
case ".txt", ".md", ".rst", ".csv", ".tsv", ".log":
return CategoryText
case ".go", ".py", ".js", ".ts", ".rs", ".zig", ".java", ".c", ".cpp":
return CategoryCode
case ".pdf":
return CategoryPDF
case ".png", ".jpg", ".jpeg", ".gif", ".bmp", ".tiff", ".webp":
return CategoryImage
case ".mp3", ".wav", ".flac", ".mp4", ".mkv", ".webm", ".m4a":
return CategoryAudioVideo
default:
return CategoryUnsupported
}
}
// ProcessFile is the main entry point for the extraction pipeline.
func ProcessFile(path string, watchDir string) (*Document, error) {
// Stage 1: Validate file size
info, err := os.Stat(path)
if err != nil {
return nil, fmt.Errorf("stat %s: %w", path, err)
}
if info.Size() > MaxFileSize {
return nil, fmt.Errorf("file %s exceeds max size (%d bytes)", path, MaxFileSize)
}
// Stage 2: Hash contents for dedup
hash, err := hashFile(path)
if err != nil {
return nil, fmt.Errorf("hash %s: %w", path, err)
}
// Stage 3: Classify
category := ClassifyFile(path)
if category == CategoryUnsupported {
return nil, fmt.Errorf("unsupported file type: %s", path)
}
// Stage 4: Extract based on category
var doc *Document
switch category {
case CategoryText, CategoryCode:
doc, err = extractText(path, category)
case CategoryPDF:
doc, err = extractPDF(path) // delegates to PDF extraction service
case CategoryImage:
doc, err = extractImage(path) // delegates to image analysis service
case CategoryAudioVideo:
doc, err = extractAudio(path) // delegates to Whisper
}
if err != nil {
return nil, fmt.Errorf("extract %s: %w", path, err)
}
// Stage 5: Tag with provenance metadata
doc.Metadata["filename"] = filepath.Base(path)
doc.Metadata["watch_dir"] = watchDir
doc.Metadata["content_hash"] = hash
return doc, nil
}
func hashFile(path string) (string, error) {
f, err := os.Open(path)
if err != nil {
return "", err
}
defer f.Close()
h := sha256.New()
if _, err := io.Copy(h, f); err != nil {
return "", err
}
return fmt.Sprintf("%x", h.Sum(nil)), nil
}// Go: Text extraction with language-aware chunking
func extractText(path string, category FileCategory) (*Document, error) {
content, err := os.ReadFile(path)
if err != nil {
return nil, err
}
doc := &Document{
Content: string(content),
Metadata: make(map[string]string),
}
if category == CategoryCode {
lang := detectLanguage(path)
doc.Metadata["language"] = lang
doc.Metadata["type"] = "code"
// Code files get function-level chunking
doc.Chunks = chunkByFunction(string(content), lang)
} else {
doc.Metadata["type"] = "text"
// Text files get paragraph-level chunking
doc.Chunks = chunkByParagraph(string(content))
}
return doc, nil
}Organizations accumulate documents in dozens of formats across file servers, cloud drives, and code repositories. A multi-format extraction pipeline feeds a single search index from all of these sources. Users search for “quarterly revenue” and get results from PDFs, spreadsheets, slide decks, and meeting transcripts—all ranked by the same relevance model because all content was normalized to the same document schema at ingestion.
This pipeline pattern can be used to build searchable knowledge bases from heterogeneous project artifacts. A single repository might contain Go source code, markdown documentation, PDF design specs, architecture diagram images, and recorded standup videos. The extraction pipeline processes all of these into indexed documents, enabling semantic search across the entire project’s knowledge—not just its text files.
By tagging each document with its watch_dir, the system supports efficient incremental updates. When a directory is rescanned, only files with changed content hashes are reprocessed. Deleted files are detected by comparing the scan results against existing documents with the same watch_dir tag. This makes the pipeline practical for monitoring live directories where files change frequently.
Each file category maps to a processor that implements a common interface. New format support is added by registering a new processor—the routing logic and metadata tagging remain unchanged. This is the classic Strategy pattern applied to document extraction.
// Go: Processor interface and registry
type Processor interface {
Extract(path string) (*Document, error)
SupportedExtensions() []string
}
type ProcessorRegistry struct {
processors map[string]Processor
}
func (r *ProcessorRegistry) Register(p Processor) {
for _, ext := range p.SupportedExtensions() {
r.processors[ext] = p
}
}
func (r *ProcessorRegistry) Process(path string) (*Document, error) {
ext := strings.ToLower(filepath.Ext(path))
p, ok := r.processors[ext]
if !ok {
return nil, fmt.Errorf("no processor for extension: %s", ext)
}
return p.Extract(path)
}When scanning large directories, files can be processed concurrently using a worker pool. CPU-bound operations (text chunking, hashing) and I/O-bound operations (OCR API calls, transcription) benefit differently from parallelism: CPU tasks are bounded by core count, while I/O tasks can run many more concurrent workers. A two-tier pool—fast workers for text/code and slow workers for media—prevents lightweight files from waiting behind expensive transcription jobs.
Separate worker pools for fast (text/code) and slow (PDF/image/audio) extraction prevent head-of-line blocking. A 500-byte README should not wait behind a 20-minute video transcription. Route files to the appropriate pool based on their classification, and size each pool according to its bottleneck (CPU cores vs. API rate limits).
When an external service (OCR, Whisper) is unavailable, the pipeline should log the failure and continue processing other files rather than aborting the entire batch. Failed files are recorded with an error status so they can be retried later. This ensures that a transient API outage does not block indexing of the majority of files that do not depend on that service.
.txt file might contain binary data, or a file might have no extension at all. Combine extension checks with MIME type detection or magic byte inspection for robust classification.io.Copy into the hasher, not os.ReadFile followed by h.Write.