xbe.at

Miscellany / Embeddings and vectors

Semantic Chunking for Intelligent Document Processing

14 min read, 14 sections

Contents
  1. Introduction to Semantic Text Chunking
  2. Vector Representation for Semantic Analysis
  3. Dynamic Chunk Size Optimization
  4. Context-Aware Boundary Detection
  5. Semantic Clustering for Document Segmentation
  6. Advanced Semantic Similarity Metrics
  7. Content-Aware Chunk Refinement
  8. Coherence Evaluation Metrics
  9. Semantic Chunk Visualization
  10. Adaptive Chunk Size Control
  11. Real-Time Chunk Processing Pipeline
  12. Chunk Optimization with Reinforcement Learning
  13. Performance Optimization and Caching
  14. Additional Resources

Introduction to Semantic Text Chunking

Semantic chunking transforms document processing by intelligently segmenting text based on meaning rather than arbitrary character counts. This implementation demonstrates the core concept using natural language processing techniques to identify coherent boundaries.

import nltk
from nltk.tokenize import sent_tokenize
from sklearn.feature_extraction.text import TfidfVectorizer
import numpy as np

class SemanticChunker:
    def __init__(self, similarity_threshold=0.3):
        self.vectorizer = TfidfVectorizer()
        self.threshold = similarity_threshold
    
    def chunk_text(self, text):
        sentences = sent_tokenize(text)
        vectors = self.vectorizer.fit_transform(sentences)
        similarities = (vectors * vectors.T).toarray()
        
        chunks = []
        current_chunk = [sentences[0]]
        
        for i in range(1, len(sentences)):
            if similarities[i][i-1] > self.threshold:
                current_chunk.append(sentences[i])
            else:
                chunks.append(' '.join(current_chunk))
                current_chunk = [sentences[i]]
        
        chunks.append(' '.join(current_chunk))
        return chunks

# Example usage
text = "Semantic chunking is powerful. It helps preserve context. Traditional methods often fail."
chunker = SemanticChunker()
chunks = chunker.chunk_text(text)
print(f"Generated chunks: {chunks}")

Vector Representation for Semantic Analysis

Understanding semantic relationships requires converting text into numerical representations. This implementation uses advanced embedding techniques to capture contextual meaning through dense vector representations.

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
from sentence_transformers import SentenceTransformer

class SemanticVectorizer:
    def __init__(self, model_name='all-MiniLM-L6-v2'):
        self.model = SentenceTransformer(model_name)
    
    def vectorize(self, text_chunks):
        embeddings = self.model.encode(text_chunks)
        return embeddings
    
    def calculate_semantic_similarity(self, chunk1, chunk2):
        vec1 = self.model.encode([chunk1])[0]
        vec2 = self.model.encode([chunk2])[0]
        return cosine_similarity([vec1], [vec2])[0][0]

# Example usage
vectorizer = SemanticVectorizer()
chunks = ["Semantic analysis is crucial", "Analysis is important for NLP"]
embeddings = vectorizer.vectorize(chunks)
similarity = vectorizer.calculate_semantic_similarity(chunks[0], chunks[1])
print(f"Semantic similarity: {similarity:.4f}")

Dynamic Chunk Size Optimization

Optimizing chunk size dynamically based on content complexity enhances the quality of semantic segmentation. This algorithm adjusts boundaries using statistical measures of semantic coherence.

import numpy as np
from sklearn.metrics import silhouette_score
from scipy.cluster.hierarchy import linkage, fcluster

class DynamicChunker:
    def __init__(self, min_chunk_size=3, max_chunk_size=10):
        self.min_size = min_chunk_size
        self.max_size = max_chunk_size
    
    def optimize_chunk_size(self, embeddings):
        scores = []
        chunk_sizes = range(self.min_size, self.max_size + 1)
        
        for size in chunk_sizes:
            chunks = len(embeddings) // size
            if chunks < 2:
                continue
                
            linkage_matrix = linkage(embeddings, method='ward')
            labels = fcluster(linkage_matrix, chunks, criterion='maxclust')
            
            if len(np.unique(labels)) < 2:
                continue
                
            score = silhouette_score(embeddings, labels)
            scores.append((size, score))
        
        optimal_size = max(scores, key=lambda x: x[1])[0]
        return optimal_size

# Example usage
embeddings = np.random.rand(100, 768)  # Simulated embeddings
chunker = DynamicChunker()
optimal_size = chunker.optimize_chunk_size(embeddings)
print(f"Optimal chunk size: {optimal_size}")

Context-Aware Boundary Detection

The boundary detection system utilizes semantic similarity scores and natural language understanding to identify optimal splitting points. This implementation considers both local and global context to maintain coherence.

import numpy as np
from transformers import AutoTokenizer, AutoModel
import torch

class BoundaryDetector:
    def __init__(self, model_name='bert-base-uncased'):
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModel.from_pretrained(model_name)
        
    def detect_boundaries(self, text, window_size=3):
        # Tokenize and get embeddings
        inputs = self.tokenizer(text, return_tensors='pt', truncation=True)
        with torch.no_grad():
            outputs = self.model(**inputs)
        embeddings = outputs.last_hidden_state.squeeze(0)
        
        # Calculate semantic boundaries
        scores = []
        for i in range(window_size, len(embeddings) - window_size):
            left_context = embeddings[i-window_size:i].mean(0)
            right_context = embeddings[i:i+window_size].mean(0)
            similarity = torch.cosine_similarity(left_context, right_context, dim=0)
            scores.append(float(similarity))
            
        # Find local minima as boundary candidates
        boundaries = []
        for i in range(1, len(scores)-1):
            if scores[i] < scores[i-1] and scores[i] < scores[i+1]:
                boundaries.append(i + window_size)
                
        return boundaries, scores

# Example usage
detector = BoundaryDetector()
text = "This is a sample text. We need to find boundaries. Here is another sentence."
boundaries, scores = detector.detect_boundaries(text)
print(f"Detected boundaries at positions: {boundaries}")

Semantic Clustering for Document Segmentation

Document segmentation benefits from clustering semantically similar content. This implementation uses hierarchical clustering to group related text segments while maintaining contextual relationships.

from sklearn.cluster import AgglomerativeClustering
from scipy.spatial.distance import pdist, squareform

class SemanticClusterer:
    def __init__(self, n_clusters=None, distance_threshold=0.5):
        self.clustering = AgglomerativeClustering(
            n_clusters=n_clusters,
            distance_threshold=distance_threshold,
            linkage='ward'
        )
    
    def cluster_segments(self, embeddings):
        # Calculate distance matrix
        distances = pdist(embeddings, metric='euclidean')
        distance_matrix = squareform(distances)
        
        # Perform clustering
        labels = self.clustering.fit_predict(embeddings)
        
        # Calculate cluster coherence
        cluster_coherence = {}
        for label in set(labels):
            mask = labels == label
            cluster_embeddings = embeddings[mask]
            if len(cluster_embeddings) > 1:
                internal_distances = pdist(cluster_embeddings)
                coherence = np.mean(internal_distances)
                cluster_coherence[label] = coherence
                
        return labels, cluster_coherence

# Example usage
embeddings = np.random.rand(20, 768)  # Simulated embeddings
clusterer = SemanticClusterer()
labels, coherence = clusterer.cluster_segments(embeddings)
print(f"Cluster assignments: {labels}")
print(f"Cluster coherence scores: {coherence}")

Advanced Semantic Similarity Metrics

Implementing sophisticated similarity metrics enhances chunking accuracy. This approach combines multiple similarity measures including cosine similarity, Euclidean distance, and contextual relevance scores.

import numpy as np
from scipy.spatial.distance import cdist
from sklearn.preprocessing import normalize

class SemanticSimilarityMetrics:
    def __init__(self, weights={'cosine': 0.4, 'euclidean': 0.3, 'contextual': 0.3}):
        self.weights = weights
        
    def compute_similarity_matrix(self, embeddings):
        # Normalize embeddings
        normalized_embeddings = normalize(embeddings)
        
        # Cosine similarity
        cosine_sim = 1 - cdist(normalized_embeddings, normalized_embeddings, 'cosine')
        
        # Euclidean similarity (inversed and normalized)
        euclidean_dist = cdist(embeddings, embeddings, 'euclidean')
        euclidean_sim = 1 / (1 + euclidean_dist)
        
        # Contextual similarity using sliding window
        contextual_sim = np.zeros((len(embeddings), len(embeddings)))
        window_size = 3
        for i in range(len(embeddings)):
            for j in range(max(0, i-window_size), min(len(embeddings), i+window_size+1)):
                contextual_sim[i,j] = 1 - abs(i-j)/window_size
        
        # Combine similarities
        combined_sim = (self.weights['cosine'] * cosine_sim +
                       self.weights['euclidean'] * euclidean_sim +
                       self.weights['contextual'] * contextual_sim)
        
        return combined_sim

# Example usage
metrics = SemanticSimilarityMetrics()
embeddings = np.random.rand(10, 768)  # Simulated embeddings
similarity_matrix = metrics.compute_similarity_matrix(embeddings)
print(f"Combined similarity matrix shape: {similarity_matrix.shape}")

Content-Aware Chunk Refinement

The refinement process adjusts chunk boundaries based on semantic coherence and contextual relevance. This implementation uses dynamic programming to optimize chunk boundaries while maintaining semantic integrity.

import numpy as np
from scipy.optimize import linear_sum_assignment

class ChunkRefiner:
    def __init__(self, min_chunk_size=50, max_chunk_size=200):
        self.min_size = min_chunk_size
        self.max_size = max_chunk_size
        
    def refine_chunks(self, text, embeddings, initial_boundaries):
        n = len(embeddings)
        cost_matrix = np.zeros((n, n))
        
        # Calculate cost matrix for all possible chunks
        for i in range(n):
            for j in range(i + 1, min(n, i + self.max_size)):
                chunk_size = j - i
                if chunk_size < self.min_size:
                    cost_matrix[i,j] = np.inf
                else:
                    chunk_embeddings = embeddings[i:j]
                    coherence = np.mean(np.std(chunk_embeddings, axis=0))
                    size_penalty = abs(chunk_size - (self.min_size + self.max_size)/2)
                    cost_matrix[i,j] = coherence + 0.01 * size_penalty
        
        # Dynamic programming to find optimal boundaries
        dp = np.full(n + 1, np.inf)
        dp[0] = 0
        prev = np.zeros(n + 1, dtype=int)
        
        for j in range(1, n + 1):
            for i in range(max(0, j - self.max_size), j):
                if dp[i] + cost_matrix[i,j-1] < dp[j]:
                    dp[j] = dp[i] + cost_matrix[i,j-1]
                    prev[j] = i
        
        # Reconstruct optimal boundaries
        refined_boundaries = []
        pos = n
        while pos > 0:
            refined_boundaries.append(prev[pos])
            pos = prev[pos]
        
        return sorted(refined_boundaries)

# Example usage
text = "Long document text..."
embeddings = np.random.rand(1000, 768)  # Simulated embeddings
initial_boundaries = [100, 250, 400, 600]
refiner = ChunkRefiner()
refined_boundaries = refiner.refine_chunks(text, embeddings, initial_boundaries)
print(f"Refined chunk boundaries: {refined_boundaries}")

Coherence Evaluation Metrics

Implementing robust evaluation metrics ensures chunk quality through quantitative assessment of semantic coherence and boundary appropriateness using multiple statistical measures.

from sklearn.metrics import silhouette_score, calinski_harabasz_score
import numpy as np

class ChunkEvaluator:
    def __init__(self):
        self.metrics = {}
    
    def evaluate_chunks(self, embeddings, boundaries):
        # Convert boundaries to labels for clustering metrics
        labels = np.zeros(len(embeddings), dtype=int)
        for i, boundary in enumerate(boundaries[:-1]):
            labels[boundary:boundaries[i+1]] = i
            
        # Calculate internal coherence
        chunk_coherences = []
        for i in range(len(boundaries)-1):
            chunk_emb = embeddings[boundaries[i]:boundaries[i+1]]
            if len(chunk_emb) > 1:
                variance = np.mean(np.var(chunk_emb, axis=0))
                chunk_coherences.append(variance)
        
        self.metrics['avg_coherence'] = np.mean(chunk_coherences)
        
        # Calculate inter-chunk separation
        if len(np.unique(labels)) > 1:
            self.metrics['silhouette'] = silhouette_score(embeddings, labels)
            self.metrics['calinski'] = calinski_harabasz_score(embeddings, labels)
        
        # Calculate boundary strength
        boundary_strengths = []
        for boundary in boundaries[1:-1]:
            left_emb = embeddings[boundary-1]
            right_emb = embeddings[boundary]
            strength = np.linalg.norm(left_emb - right_emb)
            boundary_strengths.append(strength)
        
        self.metrics['avg_boundary_strength'] = np.mean(boundary_strengths)
        
        return self.metrics

    def get_detailed_report(self):
        report = "Chunk Evaluation Report:\n"
        report += f"Average Coherence: {self.metrics['avg_coherence']:.4f}\n"
        if 'silhouette' in self.metrics:
            report += f"Silhouette Score: {self.metrics['silhouette']:.4f}\n"
        report += f"Average Boundary Strength: {self.metrics['avg_boundary_strength']:.4f}"
        return report

# Example usage
embeddings = np.random.rand(500, 768)  # Simulated embeddings
boundaries = [0, 100, 250, 400, 500]
evaluator = ChunkEvaluator()
metrics = evaluator.evaluate_chunks(embeddings, boundaries)
print(evaluator.get_detailed_report())

Semantic Chunk Visualization

Implementing visualization tools helps analyze chunk quality and distribution. This implementation creates detailed visualizations of semantic relationships and chunk boundaries using dimensionality reduction techniques.

import matplotlib.pyplot as plt
from sklearn.manifold import TSNE
import seaborn as sns
import numpy as np

class ChunkVisualizer:
    def __init__(self, embeddings, boundaries):
        self.embeddings = embeddings
        self.boundaries = boundaries
        self.tsne = TSNE(n_components=2, random_state=42)
        
    def visualize_chunks(self):
        # Reduce dimensionality for visualization
        reduced_embeddings = self.tsne.fit_transform(self.embeddings)
        
        # Create labels for chunks
        labels = np.zeros(len(self.embeddings))
        for i in range(len(self.boundaries)-1):
            labels[self.boundaries[i]:self.boundaries[i+1]] = i
            
        # Setup visualization
        plt.figure(figsize=(12, 8))
        scatter = plt.scatter(reduced_embeddings[:, 0], 
                            reduced_embeddings[:, 1],
                            c=labels, 
                            cmap='tab20',
                            alpha=0.6)
        
        # Add boundary lines
        for boundary in self.boundaries[1:-1]:
            plt.axvline(x=reduced_embeddings[boundary, 0], 
                       color='red', 
                       linestyle='--', 
                       alpha=0.3)
            
        plt.colorbar(scatter, label='Chunk Index')
        plt.title('Semantic Chunk Distribution')
        plt.xlabel('TSNE Dimension 1')
        plt.ylabel('TSNE Dimension 2')
        
        return plt.gcf()
        
    def plot_similarity_heatmap(self):
        # Calculate similarity matrix
        similarity = np.corrcoef(self.embeddings)
        
        plt.figure(figsize=(10, 10))
        sns.heatmap(similarity, 
                   cmap='viridis',
                   xticklabels=False,
                   yticklabels=False)
        
        # Add boundary lines
        for boundary in self.boundaries[1:-1]:
            plt.axhline(y=boundary, color='red', linestyle='--')
            plt.axvline(x=boundary, color='red', linestyle='--')
            
        plt.title('Chunk Similarity Heatmap')
        return plt.gcf()

# Example usage
embeddings = np.random.rand(300, 768)  # Simulated embeddings
boundaries = [0, 75, 150, 225, 300]
visualizer = ChunkVisualizer(embeddings, boundaries)

# Generate visualizations
chunk_distribution = visualizer.visualize_chunks()
similarity_heatmap = visualizer.plot_similarity_heatmap()

plt.close('all')  # Clean up plots

Adaptive Chunk Size Control

This implementation dynamically adjusts chunk sizes based on content complexity and semantic density, ensuring optimal segmentation for varying document structures.

import numpy as np
from scipy.signal import find_peaks
from sklearn.preprocessing import StandardScaler

class AdaptiveChunker:
    def __init__(self, min_size=50, max_size=500):
        self.min_size = min_size
        self.max_size = max_size
        self.scaler = StandardScaler()
        
    def compute_complexity_score(self, embeddings, window_size=10):
        # Calculate local variance in embeddings
        complexity = []
        for i in range(0, len(embeddings) - window_size + 1):
            window = embeddings[i:i+window_size]
            local_var = np.var(window, axis=0).mean()
            complexity.append(local_var)
            
        return np.array(complexity)
    
    def find_adaptive_boundaries(self, embeddings):
        # Compute complexity scores
        complexity_scores = self.compute_complexity_score(embeddings)
        normalized_scores = self.scaler.fit_transform(
            complexity_scores.reshape(-1, 1)
        ).ravel()
        
        # Find peaks in complexity
        peaks, properties = find_peaks(
            normalized_scores,
            distance=self.min_size,
            prominence=0.5
        )
        
        # Adjust boundaries based on complexity
        boundaries = [0]
        current_pos = 0
        
        for peak in peaks:
            chunk_size = peak - current_pos
            if self.min_size <= chunk_size <= self.max_size:
                boundaries.append(peak)
                current_pos = peak
                
        if embeddings.shape[0] - current_pos >= self.min_size:
            boundaries.append(embeddings.shape[0])
            
        return np.array(boundaries), normalized_scores

# Example usage
embeddings = np.random.rand(1000, 768)  # Simulated embeddings
chunker = AdaptiveChunker()
boundaries, scores = chunker.find_adaptive_boundaries(embeddings)

print(f"Number of chunks: {len(boundaries)-1}")
print(f"Chunk sizes: {np.diff(boundaries)}")
print(f"Average complexity score: {np.mean(scores):.4f}")

Real-Time Chunk Processing Pipeline

This implementation demonstrates a complete pipeline for processing streaming text data with semantic chunking, incorporating real-time updates and adaptive threshold adjustment.

import numpy as np
from collections import deque
from threading import Lock
import time

class StreamingChunker:
    def __init__(self, buffer_size=1000, update_interval=0.1):
        self.buffer = deque(maxlen=buffer_size)
        self.lock = Lock()
        self.embeddings_buffer = []
        self.update_interval = update_interval
        self.threshold_history = []
        
    def process_stream(self, text_stream, embedding_model):
        current_chunk = []
        last_update = time.time()
        
        for text in text_stream:
            with self.lock:
                self.buffer.append(text)
                embedding = embedding_model.encode([text])[0]
                self.embeddings_buffer.append(embedding)
                
                # Dynamic threshold update
                if time.time() - last_update > self.update_interval:
                    self._update_threshold()
                    last_update = time.time()
                
                # Check for chunk boundary
                if self._should_split_chunk(current_chunk, embedding):
                    yield self._process_chunk(current_chunk)
                    current_chunk = []
                
                current_chunk.append(text)
    
    def _update_threshold(self):
        if len(self.embeddings_buffer) > 1:
            similarities = np.array([
                np.dot(self.embeddings_buffer[-1], emb)
                for emb in self.embeddings_buffer[-10:]
            ])
            new_threshold = np.mean(similarities) - np.std(similarities)
            self.threshold_history.append(new_threshold)
            
            # Adaptive threshold smoothing
            if len(self.threshold_history) > 10:
                self.current_threshold = np.median(self.threshold_history[-10:])
    
    def _should_split_chunk(self, current_chunk, new_embedding):
        if not current_chunk:
            return False
            
        chunk_embedding = np.mean([
            self.embeddings_buffer[i] 
            for i in range(-len(current_chunk), 0)
        ], axis=0)
        
        similarity = np.dot(chunk_embedding, new_embedding)
        return similarity < self.current_threshold
    
    def _process_chunk(self, chunk):
        return {
            'text': ' '.join(chunk),
            'size': len(chunk),
            'timestamp': time.time()
        }

# Example usage
class MockEmbeddingModel:
    def encode(self, texts):
        return np.random.rand(len(texts), 768)

# Simulate streaming data
def text_stream_generator():
    sentences = [
        "This is a test sentence.",
        "Another sentence for testing.",
        "Yet another test sentence.",
        "This sentence changes the topic.",
        "Now we're talking about something else."
    ]
    for sentence in sentences:
        yield sentence
        time.sleep(0.1)

# Process stream
model = MockEmbeddingModel()
chunker = StreamingChunker()
for chunk in chunker.process_stream(text_stream_generator(), model):
    print(f"New chunk detected: {chunk['size']} sentences")
    print(f"Chunk text: {chunk['text'][:100]}...")

Chunk Optimization with Reinforcement Learning

This advanced implementation uses reinforcement learning to optimize chunk boundaries based on semantic coherence rewards and downstream task performance.

import numpy as np
from collections import namedtuple
import torch
import torch.nn as nn
import torch.optim as optim

Experience = namedtuple('Experience', 
    ['state', 'action', 'reward', 'next_state', 'done'])

class ChunkingPolicy(nn.Module):
    def __init__(self, state_dim=768, hidden_dim=256):
        super().__init__()
        self.network = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, 2)  # Split or continue
        )
        
    def forward(self, x):
        return self.network(x)

class RLChunker:
    def __init__(self, embedding_dim=768, learning_rate=1e-4):
        self.policy = ChunkingPolicy(embedding_dim)
        self.optimizer = optim.Adam(self.policy.parameters(), 
                                  lr=learning_rate)
        self.memory = []
        self.gamma = 0.99
        
    def get_action(self, state):
        with torch.no_grad():
            logits = self.policy(torch.FloatTensor(state))
            prob = torch.softmax(logits, dim=-1)
            action = torch.multinomial(prob, 1).item()
        return action
        
    def compute_reward(self, chunk_embeddings):
        if len(chunk_embeddings) < 2:
            return 0
        
        # Compute coherence reward
        coherence = np.mean([
            np.dot(chunk_embeddings[i], chunk_embeddings[i-1])
            for i in range(1, len(chunk_embeddings))
        ])
        
        # Compute size penalty
        size_penalty = -0.1 * abs(len(chunk_embeddings) - 10)
        
        return coherence + size_penalty
        
    def update_policy(self, batch_size=32):
        if len(self.memory) < batch_size:
            return
            
        batch = np.random.choice(self.memory, batch_size)
        states = torch.FloatTensor([e.state for e in batch])
        actions = torch.LongTensor([e.action for e in batch])
        rewards = torch.FloatTensor([e.reward for e in batch])
        next_states = torch.FloatTensor([e.next_state for e in batch])
        dones = torch.FloatTensor([e.done for e in batch])
        
        # Compute target values
        with torch.no_grad():
            next_values = self.policy(next_states).max(1)[0]
            targets = rewards + self.gamma * next_values * (1 - dones)
        
        # Update policy
        self.optimizer.zero_grad()
        values = self.policy(states).gather(1, actions.unsqueeze(1))
        loss = nn.MSELoss()(values.squeeze(), targets)
        loss.backward()
        self.optimizer.step()
        
        return loss.item()

# Example usage
chunker = RLChunker()
embeddings = np.random.rand(100, 768)  # Simulated embeddings
current_chunk = []
total_reward = 0

for embedding in embeddings:
    state = np.mean(current_chunk + [embedding], axis=0) if current_chunk else embedding
    action = chunker.get_action(state)
    
    if action == 1:  # Split chunk
        reward = chunker.compute_reward([embedding] + current_chunk)
        total_reward += reward
        current_chunk = []
    else:
        current_chunk.append(embedding)
        
print(f"Total reward accumulated: {total_reward:.4f}")

Performance Optimization and Caching

This implementation focuses on optimizing semantic chunking performance through intelligent caching mechanisms and parallel processing strategies for large-scale document processing.

import numpy as np
from functools import lru_cache
from concurrent.futures import ThreadPoolExecutor
import hashlib
from typing import List, Dict, Tuple

class OptimizedChunker:
    def __init__(self, cache_size: int = 1000, max_workers: int = 4):
        self.cache_size = cache_size
        self.max_workers = max_workers
        self.embedding_cache = {}
        
    @lru_cache(maxsize=1000)
    def _compute_embedding_hash(self, text: str) -> str:
        return hashlib.md5(text.encode()).hexdigest()
    
    def _process_chunk_parallel(self, chunk: str) -> Tuple[str, np.ndarray]:
        chunk_hash = self._compute_embedding_hash(chunk)
        if chunk_hash in self.embedding_cache:
            return chunk, self.embedding_cache[chunk_hash]
            
        # Simulate embedding computation
        embedding = np.random.rand(768)  # Replace with actual embedding
        self.embedding_cache[chunk_hash] = embedding
        
        if len(self.embedding_cache) > self.cache_size:
            # Remove oldest entries
            oldest_key = next(iter(self.embedding_cache))
            del self.embedding_cache[oldest_key]
            
        return chunk, embedding
    
    def process_document(self, text: str, chunk_size: int = 1000) -> List[Dict]:
        # Pre-split text into rough chunks
        initial_chunks = [
            text[i:i + chunk_size] 
            for i in range(0, len(text), chunk_size)
        ]
        
        # Process chunks in parallel
        processed_chunks = []
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = [
                executor.submit(self._process_chunk_parallel, chunk)
                for chunk in initial_chunks
            ]
            
            # Collect results
            for future in futures:
                chunk, embedding = future.result()
                processed_chunks.append({
                    'text': chunk,
                    'embedding': embedding,
                    'hash': self._compute_embedding_hash(chunk)
                })
        
        return self._optimize_boundaries(processed_chunks)
    
    def _optimize_boundaries(self, chunks: List[Dict]) -> List[Dict]:
        optimized_chunks = []
        current_chunk = []
        
        for i in range(len(chunks)):
            current_chunk.append(chunks[i])
            
            if i < len(chunks) - 1:
                current_embedding = np.mean([c['embedding'] for c in current_chunk], axis=0)
                next_embedding = chunks[i + 1]['embedding']
                
                similarity = np.dot(current_embedding, next_embedding)
                
                if similarity < 0.5:  # Threshold for new chunk
                    optimized_chunks.append({
                        'text': ' '.join(c['text'] for c in current_chunk),
                        'embedding': current_embedding,
                        'size': len(current_chunk)
                    })
                    current_chunk = []
        
        # Add remaining chunk
        if current_chunk:
            optimized_chunks.append({
                'text': ' '.join(c['text'] for c in current_chunk),
                'embedding': np.mean([c['embedding'] for c in current_chunk], axis=0),
                'size': len(current_chunk)
            })
        
        return optimized_chunks

# Example usage
sample_text = """
Long document text that needs to be processed efficiently...
""" * 100  # Simulate large document

chunker = OptimizedChunker()
result = chunker.process_document(sample_text)

print(f"Processed {len(result)} optimized chunks")
print(f"Cache size: {len(chunker.embedding_cache)}")
print(f"Average chunk size: {np.mean([chunk['size'] for chunk in result]):.2f}")

Additional Resources

Note: These are example URLs and may not represent actual papers. Please verify the exact URLs for the most current research.

More in Embeddings and vectors

  1. Structural Limits of the London Eye's Spin6 min

    The London Eye, a iconic Ferris wheel on the South Bank of the River Thames in London, stands as a symbol of modern engineering. This presentation explores a...

  2. Multilingual Sentence Encoding with PyTorch4 min

    The Multilingual Universal Sentence Encoder (mUSE) is a powerful pre-trained model that can encode text from various languages into high-dimensional vector...

  3. Text Vectorization Transforming Words into Numbers10 min

    Text vectorization transforms human-readable text into numerical representations that machine learning models can process. The foundation begins with...

  4. Mathematics The Language of the Universe11 min

    Mathematics is often described as the universal language that unveils the secrets of the universe. This powerful tool allows us to describe, analyze, and...

  5. Vector Embeddings The AI's Secret to Comparing Anything6 min

    Vector embeddings are numerical representations of real-world objects or concepts. They allow AI systems to compare and analyze diverse items, from fruits to...

  6. Leonhard Euler Pioneering Mathematician and Physicist:8 min

    Leonhard Euler, a Swiss mathematician and physicist, revolutionized multiple fields of mathematics and laid the groundwork for modern mathematical notation....