chunking-strategy — Skillopedia

Chunking Strategy for RAG Systems Overview Provides chunking strategies for RAG systems, vector databases, and document processing. Recommends chunk sizes, overlap percentages, and boundary detection methods; validates semantic coherence; evaluates retrieval metrics. When to Use Use when building or optimizing RAG systems, vector search pipelines, document chunking workflows, or performance-tuning existing systems with poor retrieval quality. Instructions Choose Chunking Strategy Select based on document type and use case: 1. Fixed-Size Chunking (Level 1) - Use for simple documents without cl…

in text):\n return \"data_heavy\"\n else:\n return \"prose\"\n```\n\n## 4. Sliding Window Chunking\n\n### Overview\nCreate overlapping chunks using a sliding window approach for maximum context preservation.\n\n### Implementation\n```python\nfrom typing import List, Iterator\nimport numpy as np\n\nclass SlidingWindowChunker:\n def __init__(self, window_size=1000, step_size=500, unit=\"tokens\"):\n self.window_size = window_size\n self.step_size = step_size\n self.unit = unit\n\n def sliding_chunk_tokens(self, text, encoding_name=\"cl100k_base\"):\n \"\"\"Create sliding window chunks by tokens\"\"\"\n import tiktoken\n encoding = tiktoken.get_encoding(encoding_name)\n tokens = encoding.encode(text)\n\n chunks = []\n for start in range(0, len(tokens), self.step_size):\n end = min(start + self.window_size, len(tokens))\n window_tokens = tokens[start:end]\n chunk_text = encoding.decode(window_tokens)\n\n chunks.append({\n \"text\": chunk_text,\n \"start_token\": start,\n \"end_token\": end - 1,\n \"token_count\": len(window_tokens),\n \"overlap\": self.window_size - self.step_size\n })\n\n if end >= len(tokens):\n break\n\n return chunks\n\n def sliding_chunk_characters(self, text):\n \"\"\"Create sliding window chunks by characters\"\"\"\n chunks = []\n for start in range(0, len(text), self.step_size):\n end = min(start + self.window_size, len(text))\n chunk_text = text[start:end]\n\n chunks.append({\n \"text\": chunk_text,\n \"start_char\": start,\n \"end_char\": end - 1,\n \"char_count\": len(chunk_text),\n \"overlap\": self.window_size - self.step_size\n })\n\n if end >= len(text):\n break\n\n return chunks\n\n def adaptive_sliding_window(self, text, min_overlap=0.1, max_overlap=0.5):\n \"\"\"Adaptive sliding window based on content density\"\"\"\n if self.unit == \"tokens\":\n base_chunks = self.sliding_chunk_tokens(text)\n else:\n base_chunks = self.sliding_chunk_characters(text)\n\n # Analyze content density\n adaptive_chunks = []\n for i, chunk in enumerate(base_chunks):\n text_content = chunk[\"text\"]\n density = self._calculate_content_density(text_content)\n\n # Adjust overlap based on density\n if density > 0.8: # High density - more overlap\n adjusted_overlap = int(self.window_size * max_overlap)\n elif density \u003c 0.3: # Low density - less overlap\n adjusted_overlap = int(self.window_size * min_overlap)\n else:\n adjusted_overlap = self.window_size - self.step_size\n\n chunk[\"content_density\"] = density\n chunk[\"adjusted_overlap\"] = adjusted_overlap\n adaptive_chunks.append(chunk)\n\n return adaptive_chunks\n\n def _calculate_content_density(self, text):\n \"\"\"Calculate content density (information per unit)\"\"\"\n # Simple heuristic: unique words / total words\n words = text.split()\n if not words:\n return 0.0\n\n unique_words = set(word.lower().strip('.,!?;:()[]{}\"\\'') for word in words)\n density = len(unique_words) / len(words)\n\n # Adjust for punctuation and special characters\n special_chars = sum(1 for char in text if not char.isalnum() and not char.isspace())\n density += special_chars / len(text) * 0.1\n\n return min(density, 1.0)\n\n def semantic_sliding_window(self, text, embedding_model, similarity_threshold=0.7):\n \"\"\"Sliding window with semantic boundary detection\"\"\"\n from sentence_transformers import SentenceTransformer\n from sklearn.metrics.pairwise import cosine_similarity\n import numpy as np\n\n # Split into sentences\n sentences = self._split_into_sentences(text)\n if len(sentences) \u003c 2:\n return [{\"text\": text, \"method\": \"single_sentence\"}]\n\n # Generate sentence embeddings\n sentence_embeddings = embedding_model.encode(sentences)\n\n chunks = []\n current_window_sentences = []\n current_window_start = 0\n\n for i, sentence in enumerate(sentences):\n current_window_sentences.append(sentence)\n\n # Check if we should create a boundary\n should_create_boundary = (\n len(current_window_sentences) >= 10 or # Max sentences per window\n (i \u003c len(sentences) - 1 and # Not the last sentence\n self._should_create_semantic_boundary(\n sentence_embeddings, i, similarity_threshold\n ))\n )\n\n if should_create_boundary:\n chunk_text = \" \".join(current_window_sentences)\n chunks.append({\n \"text\": chunk_text,\n \"sentence_count\": len(current_window_sentences),\n \"start_sentence\": current_window_start,\n \"end_sentence\": i,\n \"method\": \"semantic_sliding_window\"\n })\n\n # Start new window with overlap\n overlap_size = min(2, len(current_window_sentences) // 2)\n current_window_sentences = current_window_sentences[-overlap_size:]\n current_window_start = i + 1 - overlap_size\n\n # Add final chunk\n if current_window_sentences:\n chunk_text = \" \".join(current_window_sentences)\n chunks.append({\n \"text\": chunk_text,\n \"sentence_count\": len(current_window_sentences),\n \"start_sentence\": current_window_start,\n \"end_sentence\": len(sentences) - 1,\n \"method\": \"semantic_sliding_window\"\n })\n\n return chunks\n\n def _split_into_sentences(self, text):\n \"\"\"Split text into sentences\"\"\"\n import re\n # Simple sentence splitting\n sentences = re.split(r'[.!?]+', text)\n return [s.strip() for s in sentences if s.strip()]\n\n def _should_create_semantic_boundary(self, embeddings, current_idx, threshold):\n \"\"\"Determine if semantic boundary should be created\"\"\"\n if current_idx >= len(embeddings) - 1:\n return True\n\n # Calculate similarity with next sentence\n current_embedding = embeddings[current_idx].reshape(1, -1)\n next_embedding = embeddings[current_idx + 1].reshape(1, -1)\n\n similarity = cosine_similarity(current_embedding, next_embedding)[0][0]\n\n return similarity \u003c threshold\n```\n\n## 5. Semantic Chunking\n\n### Overview\nUse semantic similarity to identify natural boundaries in text.\n\n### Implementation\n```python\nfrom sentence_transformers import SentenceTransformer\nfrom sklearn.metrics.pairwise import cosine_similarity\nimport numpy as np\nfrom typing import List, Dict\n\nclass SemanticChunker:\n def __init__(self, model_name=\"all-MiniLM-L6-v2\",\n similarity_threshold=0.8,\n min_chunk_size=2,\n max_chunk_size=10):\n self.model = SentenceTransformer(model_name)\n self.similarity_threshold = similarity_threshold\n self.min_chunk_size = min_chunk_size\n self.max_chunk_size = max_chunk_size\n\n def semantic_chunk_sentences(self, text):\n \"\"\"Chunk text based on semantic similarity between sentences\"\"\"\n # Split into sentences\n sentences = self._split_into_sentences(text)\n if len(sentences) \u003c= self.min_chunk_size:\n return [{\"text\": text, \"sentence_count\": len(sentences), \"method\": \"single_chunk\"}]\n\n # Generate embeddings for all sentences\n sentence_embeddings = self.model.encode(sentences)\n\n # Find semantic boundaries\n boundaries = self._find_semantic_boundaries(sentence_embeddings)\n\n # Create chunks based on boundaries\n chunks = []\n start_idx = 0\n\n for boundary_idx in boundaries:\n if boundary_idx > start_idx:\n chunk_sentences = sentences[start_idx:boundary_idx + 1]\n chunk_text = \" \".join(chunk_sentences)\n\n chunks.append({\n \"text\": chunk_text,\n \"sentence_count\": len(chunk_sentences),\n \"start_sentence\": start_idx,\n \"end_sentence\": boundary_idx,\n \"method\": \"semantic_boundary\"\n })\n\n start_idx = boundary_idx + 1\n\n # Add remaining sentences\n if start_idx \u003c len(sentences):\n chunk_sentences = sentences[start_idx:]\n chunk_text = \" \".join(chunk_sentences)\n\n chunks.append({\n \"text\": chunk_text,\n \"sentence_count\": len(chunk_sentences),\n \"start_sentence\": start_idx,\n \"end_sentence\": len(sentences) - 1,\n \"method\": \"semantic_boundary\"\n })\n\n return self._merge_small_chunks(chunks)\n\n def _find_semantic_boundaries(self, embeddings):\n \"\"\"Find semantic boundaries based on similarity thresholds\"\"\"\n boundaries = []\n\n for i in range(len(embeddings) - 1):\n # Calculate similarity between consecutive sentences\n similarity = cosine_similarity(\n embeddings[i].reshape(1, -1),\n embeddings[i + 1].reshape(1, -1)\n )[0][0]\n\n # If similarity is below threshold, create boundary\n if similarity \u003c self.similarity_threshold:\n boundaries.append(i)\n\n return boundaries\n\n def _split_into_sentences(self, text):\n \"\"\"Split text into sentences\"\"\"\n import re\n # Enhanced sentence splitting\n sentences = re.split(r'(?\u003c=[.!?])\\s+(?=[A-Z])', text)\n return [s.strip() for s in sentences if s.strip()]\n\n def _merge_small_chunks(self, chunks):\n \"\"\"Merge chunks that are too small\"\"\"\n if not chunks:\n return chunks\n\n merged_chunks = []\n current_chunk = chunks[0].copy()\n\n for next_chunk in chunks[1:]:\n if (current_chunk[\"sentence_count\"] \u003c self.min_chunk_size and\n current_chunk[\"sentence_count\"] + next_chunk[\"sentence_count\"] \u003c= self.max_chunk_size):\n\n # Merge chunks\n current_chunk[\"text\"] += \" \" + next_chunk[\"text\"]\n current_chunk[\"sentence_count\"] += next_chunk[\"sentence_count\"]\n current_chunk[\"end_sentence\"] = next_chunk[\"end_sentence\"]\n else:\n merged_chunks.append(current_chunk)\n current_chunk = next_chunk.copy()\n\n merged_chunks.append(current_chunk)\n return merged_chunks\n\n def adaptive_semantic_chunking(self, text, content_analyzer=None):\n \"\"\"Semantic chunking with adaptive threshold\"\"\"\n sentences = self._split_into_sentences(text)\n if len(sentences) \u003c= 2:\n return [{\"text\": text, \"method\": \"too_short\"}]\n\n # Generate embeddings\n embeddings = self.model.encode(sentences)\n\n # Analyze content complexity\n if content_analyzer:\n complexity = content_analyzer.analyze_complexity(text)\n # Adjust threshold based on complexity\n adaptive_threshold = self.similarity_threshold * (1.0 + complexity * 0.2)\n else:\n adaptive_threshold = self.similarity_threshold\n\n # Find boundaries with adaptive threshold\n boundaries = self._find_adaptive_boundaries(embeddings, adaptive_threshold)\n\n # Create chunks\n chunks = []\n start_idx = 0\n\n for boundary_idx in boundaries:\n if boundary_idx > start_idx:\n chunk_sentences = sentences[start_idx:boundary_idx + 1]\n chunk_text = \" \".join(chunk_sentences)\n\n chunks.append({\n \"text\": chunk_text,\n \"sentence_count\": len(chunk_sentences),\n \"start_sentence\": start_idx,\n \"end_sentence\": boundary_idx,\n \"method\": \"adaptive_semantic\",\n \"threshold_used\": adaptive_threshold\n })\n\n start_idx = boundary_idx + 1\n\n # Add remaining sentences\n if start_idx \u003c len(sentences):\n chunk_sentences = sentences[start_idx:]\n chunk_text = \" \".join(chunk_sentences)\n\n chunks.append({\n \"text\": chunk_text,\n \"sentence_count\": len(chunk_sentences),\n \"start_sentence\": start_idx,\n \"end_sentence\": len(sentences) - 1,\n \"method\": \"adaptive_semantic\",\n \"threshold_used\": adaptive_threshold\n })\n\n return chunks\n\n def _find_adaptive_boundaries(self, embeddings, threshold):\n \"\"\"Find boundaries with adaptive threshold based on local context\"\"\"\n boundaries = []\n\n for i in range(len(embeddings) - 1):\n # Calculate local similarity\n local_similarities = []\n\n # Look at local window of similarities\n window_size = min(3, i)\n for j in range(max(0, i - window_size), i + 1):\n if j \u003c len(embeddings) - 1:\n similarity = cosine_similarity(\n embeddings[j].reshape(1, -1),\n embeddings[j + 1].reshape(1, -1)\n )[0][0]\n local_similarities.append(similarity)\n\n # Use local average for comparison\n if local_similarities:\n local_avg = np.mean(local_similarities)\n current_similarity = local_similarities[-1]\n\n # Create boundary if current similarity is significantly lower than local average\n if current_similarity \u003c local_avg * threshold:\n boundaries.append(i)\n else:\n # Fallback to global threshold\n similarity = cosine_similarity(\n embeddings[i].reshape(1, -1),\n embeddings[i + 1].reshape(1, -1)\n )[0][0]\n\n if similarity \u003c threshold:\n boundaries.append(i)\n\n return boundaries\n\n def hierarchical_semantic_chunking(self, text, max_levels=3):\n \"\"\"Multi-level semantic chunking\"\"\"\n sentences = self._split_into_sentences(text)\n\n if len(sentences) \u003c= 4:\n return [{\n \"text\": text,\n \"level\": 0,\n \"sentence_count\": len(sentences),\n \"method\": \"hierarchical_semantic\"\n }]\n\n # Level 0: Original text\n chunks = [{\n \"text\": text,\n \"level\": 0,\n \"sentence_count\": len(sentences),\n \"method\": \"hierarchical_semantic\"\n }]\n\n # Generate embeddings once\n embeddings = self.model.encode(sentences)\n\n # Create hierarchical chunks\n current_level_sentences = sentences\n current_level_embeddings = embeddings\n\n for level in range(1, max_levels + 1):\n if len(current_level_sentences) \u003c= 2:\n break\n\n # Find boundaries at this level\n boundaries = self._find_semantic_boundaries(current_level_embeddings)\n\n # Create chunks at this level\n level_chunks = []\n start_idx = 0\n\n for boundary_idx in boundaries:\n if boundary_idx > start_idx:\n chunk_sentences = current_level_sentences[start_idx:boundary_idx + 1]\n chunk_text = \" \".join(chunk_sentences)\n\n level_chunks.append({\n \"text\": chunk_text,\n \"level\": level,\n \"sentence_count\": len(chunk_sentences),\n \"start_sentence\": start_idx,\n \"end_sentence\": boundary_idx,\n \"method\": \"hierarchical_semantic\"\n })\n\n start_idx = boundary_idx + 1\n\n # Add remaining sentences\n if start_idx \u003c len(current_level_sentences):\n chunk_sentences = current_level_sentences[start_idx:]\n chunk_text = \" \".join(chunk_sentences)\n\n level_chunks.append({\n \"text\": chunk_text,\n \"level\": level,\n \"sentence_count\": len(chunk_sentences),\n \"start_sentence\": start_idx,\n \"end_sentence\": len(current_level_sentences) - 1,\n \"method\": \"hierarchical_semantic\"\n })\n\n chunks.extend(level_chunks)\n\n # Prepare for next level\n if len(level_chunks) > 1:\n current_level_sentences = [chunk[\"text\"] for chunk in level_chunks]\n current_level_embeddings = self.model.encode(current_level_sentences)\n else:\n break\n\n return chunks\n```\n\n## 6. Recursive Chunking\n\n### Overview\nHierarchical splitting using ordered separators to preserve document structure.\n\n### Implementation\n```python\nfrom typing import List, Dict, Optional\nimport re\n\nclass RecursiveChunker:\n def __init__(self, chunk_size=1000, chunk_overlap=200,\n separators=None, length_function=len):\n self.chunk_size = chunk_size\n self.chunk_overlap = chunk_overlap\n self.length_function = length_function\n\n # Default separators in order of preference\n self.separators = separators or [\n \"\\n\\n\\n\", # Triple newlines (section breaks)\n \"\\n\\n\", # Double newlines (paragraph breaks)\n \"\\n\", # Single newlines (line breaks)\n \" \", # Spaces (word breaks)\n \"\" # Character-level (last resort)\n ]\n\n def recursive_split(self, text, separators=None):\n \"\"\"Recursively split text using hierarchical separators\"\"\"\n separators = separators or self.separators\n final_chunks = []\n\n # Try each separator in order\n for separator in separators:\n if separator == \"\":\n # Last resort: split by characters\n return self._split_by_characters(text)\n\n # Split by current separator\n splits = text.split(separator)\n\n # Filter out empty splits\n splits = [split for split in splits if split.strip()]\n\n if len(splits) > 1:\n # Found a good separator\n for split in splits:\n if self.length_function(split) \u003c= self.chunk_size:\n final_chunks.append(split)\n else:\n # Recursively split this piece\n sub_chunks = self.recursive_split(split, separators[separators.index(separator) + 1:])\n final_chunks.extend(sub_chunks)\n\n return self._merge_chunks(final_chunks)\n\n # No separator worked, split by characters\n return self._split_by_characters(text)\n\n def _split_by_characters(self, text):\n \"\"\"Split text by characters as last resort\"\"\"\n chunks = []\n start = 0\n\n while start \u003c len(text):\n end = min(start + self.chunk_size, len(text))\n chunk = text[start:end]\n chunks.append(chunk)\n\n # Calculate next start with overlap\n start = max(0, end - self.chunk_overlap)\n\n if end >= len(text):\n break\n\n return chunks\n\n def _merge_chunks(self, chunks):\n \"\"\"Merge chunks that are too small\"\"\"\n if not chunks:\n return chunks\n\n merged_chunks = []\n current_chunk = chunks[0]\n\n for next_chunk in chunks[1:]:\n combined_length = self.length_function(current_chunk + next_chunk)\n\n if combined_length \u003c= self.chunk_size:\n # Merge chunks\n current_chunk += \"\\n\\n\" + next_chunk\n else:\n # Add current chunk and start new one\n merged_chunks.append(current_chunk)\n current_chunk = next_chunk\n\n merged_chunks.append(current_chunk)\n return merged_chunks\n\n def recursive_split_with_metadata(self, text, separators=None):\n \"\"\"Recursive split with detailed metadata\"\"\"\n separators = separators or self.separators\n chunks = []\n\n def _recursive_split_with_context(text_chunk, parent_separator=\"\"):\n nonlocal chunks\n\n for separator in separators:\n if separator == \"\":\n sub_chunks = self._split_by_characters(text_chunk)\n for i, chunk in enumerate(sub_chunks):\n chunks.append({\n \"text\": chunk,\n \"separator\": \"character\",\n \"parent_separator\": parent_separator,\n \"level\": len(separators) - separators.index(separator),\n \"chunk_index\": len(chunks),\n \"size\": self.length_function(chunk)\n })\n return\n\n splits = text_chunk.split(separator)\n splits = [split for split in splits if split.strip()]\n\n if len(splits) > 1:\n for i, split in enumerate(splits):\n if self.length_function(split) \u003c= self.chunk_size:\n chunks.append({\n \"text\": split,\n \"separator\": separator,\n \"parent_separator\": parent_separator,\n \"level\": len(separators) - separators.index(separator),\n \"chunk_index\": len(chunks),\n \"size\": self.length_function(split)\n })\n else:\n # Recursively split this piece\n _recursive_split_with_context(split, separator)\n return\n\n # No separator worked\n sub_chunks = self._split_by_characters(text_chunk)\n for i, chunk in enumerate(sub_chunks):\n chunks.append({\n \"text\": chunk,\n \"separator\": \"character_fallback\",\n \"parent_separator\": parent_separator,\n \"level\": len(separators),\n \"chunk_index\": len(chunks),\n \"size\": self.length_function(chunk)\n })\n\n _recursive_split_with_context(text)\n return chunks\n\n def markdown_aware_recursive_split(self, text):\n \"\"\"Recursive splitting optimized for Markdown documents\"\"\"\n markdown_separators = [\n \"\\n# \", # H1 headers\n \"\\n## \", # H2 headers\n \"\\n### \", # H3 headers\n \"\\n#### \", # H4 headers\n \"\\n##### \", # H5 headers\n \"\\n###### \", # H6 headers\n \"\\n\\n\", # Paragraph breaks\n \"\\n\", # Line breaks\n \" \", # Spaces\n \"\" # Characters\n ]\n\n chunks = []\n\n def _split_markdown(text_chunk, separator_idx=0):\n if separator_idx >= len(markdown_separators):\n return self._split_by_characters(text_chunk)\n\n separator = markdown_separators[separator_idx]\n\n if separator.startswith(\"\\n#\"):\n # Markdown headers\n pattern = re.escape(separator)\n splits = re.split(pattern, text_chunk)\n\n if len(splits) > 1:\n # Re-add separator to splits (except first)\n for i in range(1, len(splits)):\n splits[i] = separator + splits[i]\n\n result_chunks = []\n for split in splits:\n if self.length_function(split) \u003c= self.chunk_size:\n result_chunks.append(split)\n else:\n # Try next level separator\n sub_chunks = _split_markdown(split, separator_idx + 1)\n result_chunks.extend(sub_chunks)\n\n return result_chunks\n else:\n # Regular separators\n splits = text_chunk.split(separator)\n splits = [split for split in splits if split.strip()]\n\n if len(splits) > 1:\n result_chunks = []\n for split in splits:\n if self.length_function(split) \u003c= self.chunk_size:\n result_chunks.append(split)\n else:\n # Try next level separator\n sub_chunks = _split_markdown(split, separator_idx + 1)\n result_chunks.extend(sub_chunks)\n\n return result_chunks\n\n # Try next separator\n return _split_markdown(text_chunk, separator_idx + 1)\n\n raw_chunks = _split_markdown(text)\n\n # Add metadata\n for i, chunk in enumerate(raw_chunks):\n chunks.append({\n \"text\": chunk,\n \"chunk_index\": i,\n \"size\": self.length_function(chunk),\n \"format\": \"markdown\",\n \"contains_header\": bool(re.search(r'^#+\\s', chunk, re.MULTILINE)),\n \"contains_code\": bool(re.search(r'```', chunk)),\n \"contains_list\": bool(re.search(r'^\\s*[-*+]\\s', chunk, re.MULTILINE))\n })\n\n return chunks\n```\n\n## 7-11. Additional Advanced Strategies\n\n### 7. Context-Enriched Chunking\n\n```python\nclass ContextEnrichedChunker:\n def __init__(self, base_chunker, context_generator=None):\n self.base_chunker = base_chunker\n self.context_generator = context_generator\n\n def enrich_chunks(self, text, query_context=None):\n \"\"\"Add contextual information to chunks\"\"\"\n base_chunks = self.base_chunker.chunk(text)\n enriched_chunks = []\n\n for i, chunk in enumerate(base_chunks):\n # Generate context for this chunk\n context = self._generate_context(chunk, text, i, query_context)\n\n enriched_chunk = {\n \"original_text\": chunk,\n \"context\": context,\n \"enriched_text\": f\"Context: {context}\\n\\nContent: {chunk}\",\n \"chunk_index\": i,\n \"method\": \"context_enriched\"\n }\n enriched_chunks.append(enriched_chunk)\n\n return enriched_chunks\n\n def _generate_context(self, chunk, full_text, chunk_index, query_context):\n \"\"\"Generate contextual information for a chunk\"\"\"\n # Simple context generation\n sentences = full_text.split('.')\n\n # Find sentences before and after\n chunk_start = full_text.find(chunk)\n chunk_end = chunk_start + len(chunk)\n\n # Get preceding and following context\n pre_context = full_text[max(0, chunk_start - 200):chunk_start]\n post_context = full_text[chunk_end:chunk_end + 200]\n\n context_parts = []\n if pre_context.strip():\n context_parts.append(f\"Preceding: {pre_context.strip()}\")\n if post_context.strip():\n context_parts.append(f\"Following: {post_context.strip()}\")\n\n return \" | \".join(context_parts)\n```\n\n### 8. Modality-Specific Chunking\n\n```python\nclass ModalitySpecificChunker:\n def __init__(self):\n self.chunkers = {\n \"text\": RecursiveChunker(),\n \"code\": CodeChunker(),\n \"table\": TableChunker(),\n \"image\": ImageChunker()\n }\n\n def chunk_mixed_content(self, document):\n \"\"\"Chunk document with multiple content types\"\"\"\n chunks = []\n\n # Detect content types\n sections = self._detect_content_types(document)\n\n for section in sections:\n content_type = section[\"type\"]\n content = section[\"content\"]\n\n if content_type in self.chunkers:\n section_chunks = self.chunkers[content_type].chunk(content)\n\n for chunk in section_chunks:\n chunks.append({\n \"content\": chunk,\n \"type\": content_type,\n \"metadata\": section.get(\"metadata\", {}),\n \"method\": f\"modality_specific_{content_type}\"\n })\n\n return chunks\n\n def _detect_content_types(self, document):\n \"\"\"Detect different content types in document\"\"\"\n sections = []\n\n # Simple detection logic\n if \"```\" in document:\n # Code blocks detected\n code_blocks = re.findall(r'```(\\w+)?\\n(.*?)\\n```', document, re.DOTALL)\n for lang, code in code_blocks:\n sections.append({\n \"type\": \"code\",\n \"content\": code,\n \"metadata\": {\"language\": lang}\n })\n\n if \"|\" in document and \"\\n\" in document:\n # Potential table detected\n sections.append({\n \"type\": \"table\",\n \"content\": document, # Simplified\n \"metadata\": {}\n })\n\n # Default to text\n sections.append({\n \"type\": \"text\",\n \"content\": document,\n \"metadata\": {}\n })\n\n return sections\n```\n\n### 9. Agentic Chunking\n\n```python\nclass AgenticChunker:\n def __init__(self, chunking_agents):\n self.agents = chunking_agents\n\n def adaptive_chunking(self, text, requirements):\n \"\"\"Use agents to determine optimal chunking strategy\"\"\"\n # Analyze text characteristics\n text_analysis = self._analyze_text(text)\n\n # Select appropriate agent based on requirements and text\n selected_agent = self._select_agent(text_analysis, requirements)\n\n # Use selected agent for chunking\n chunks = selected_agent.chunk(text, requirements)\n\n return {\n \"chunks\": chunks,\n \"selected_agent\": selected_agent.name,\n \"reasoning\": selected_agent.reasoning,\n \"text_analysis\": text_analysis\n }\n\n def _analyze_text(self, text):\n \"\"\"Analyze text characteristics\"\"\"\n return {\n \"length\": len(text),\n \"complexity\": self._calculate_complexity(text),\n \"structure\": self._detect_structure(text),\n \"content_type\": self._detect_content_type(text)\n }\n\n def _select_agent(self, analysis, requirements):\n \"\"\"Select best chunking agent\"\"\"\n for agent in self.agents:\n if agent.can_handle(analysis, requirements):\n return agent\n\n # Fallback to first agent\n return self.agents[0]\n```\n\n### 10. Subdocument Chunking\n\n```python\nclass SubdocumentChunker:\n def __init__(self, max_size=5000):\n self.max_size = max_size\n\n def chunk_by_logical_sections(self, document):\n \"\"\"Chunk document by logical sections\"\"\"\n sections = self._identify_logical_sections(document)\n chunks = []\n\n for section in sections:\n if len(section[\"content\"]) \u003c= self.max_size:\n chunks.append({\n \"content\": section[\"content\"],\n \"title\": section[\"title\"],\n \"level\": section[\"level\"],\n \"method\": \"subdocument_section\"\n })\n else:\n # Further split large sections\n sub_chunks = self._split_large_section(section)\n chunks.extend(sub_chunks)\n\n return chunks\n\n def _identify_logical_sections(self, document):\n \"\"\"Identify logical sections in document\"\"\"\n sections = []\n\n # Simple heading detection\n heading_pattern = r'^(#{1,6})\\s+(.+)

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.

\n lines = document.split('\\n')\n\n current_section = {\"title\": \"Introduction\", \"content\": \"\", \"level\": 0}\n\n for line in lines:\n match = re.match(heading_pattern, line)\n if match:\n # Save current section\n if current_section[\"content\"].strip():\n sections.append(current_section)\n\n # Start new section\n level = len(match.group(1))\n title = match.group(2)\n current_section = {\n \"title\": title,\n \"content\": \"\",\n \"level\": level\n }\n else:\n current_section[\"content\"] += line + \"\\n\"\n\n # Add final section\n if current_section[\"content\"].strip():\n sections.append(current_section)\n\n return sections\n```\n\n### 11. Hybrid Chunking\n\n```python\nclass HybridChunker:\n def __init__(self, strategies, weights=None):\n self.strategies = strategies\n self.weights = weights or [1.0 / len(strategies)] * len(strategies)\n\n def hybrid_chunk(self, text, evaluation_criteria=None):\n \"\"\"Combine multiple chunking strategies\"\"\"\n all_chunks = []\n\n # Apply all strategies\n for i, strategy in enumerate(self.strategies):\n strategy_chunks = strategy.chunk(text)\n\n for chunk in strategy_chunks:\n all_chunks.append({\n \"content\": chunk,\n \"strategy\": strategy.name,\n \"strategy_weight\": self.weights[i],\n \"method\": \"hybrid\"\n })\n\n # Evaluate and select best chunks\n if evaluation_criteria:\n evaluated_chunks = self._evaluate_chunks(all_chunks, evaluation_criteria)\n else:\n evaluated_chunks = all_chunks\n\n # Merge overlapping chunks from different strategies\n merged_chunks = self._merge_overlapping_chunks(evaluated_chunks)\n\n return merged_chunks\n\n def _evaluate_chunks(self, chunks, criteria):\n \"\"\"Evaluate chunks based on criteria\"\"\"\n for chunk in chunks:\n score = 0.0\n for criterion, weight in criteria.items():\n criterion_score = self._evaluate_criterion(chunk, criterion)\n score += criterion_score * weight\n\n chunk[\"evaluation_score\"] = score\n\n # Sort by evaluation score\n chunks.sort(key=lambda x: x[\"evaluation_score\"], reverse=True)\n return chunks\n\n def _merge_overlapping_chunks(self, chunks):\n \"\"\"Merge chunks that overlap significantly\"\"\"\n # Simple implementation - could be more sophisticated\n merged = []\n used_indices = set()\n\n for i, chunk1 in enumerate(chunks):\n if i in used_indices:\n continue\n\n best_chunk = chunk1.copy()\n\n for j, chunk2 in enumerate(chunks[i+1:], i+1):\n if j in used_indices:\n continue\n\n # Check overlap\n overlap = self._calculate_overlap(chunk1[\"content\"], chunk2[\"content\"])\n\n if overlap > 0.7: # High overlap\n # Merge chunks\n best_chunk[\"content\"] = max(\n chunk1[\"content\"],\n chunk2[\"content\"],\n key=len\n )\n best_chunk[\"merged_strategies\"] = [\n chunk1[\"strategy\"],\n chunk2[\"strategy\"]\n ]\n used_indices.add(j)\n\n merged.append(best_chunk)\n used_indices.add(i)\n\n return merged\n\n def _calculate_overlap(self, text1, text2):\n \"\"\"Calculate text overlap ratio\"\"\"\n words1 = set(text1.lower().split())\n words2 = set(text2.lower().split())\n\n intersection = words1 & words2\n union = words1 | words2\n\n return len(intersection) / len(union) if union else 0\n```\n\n## Usage Examples\n\n### Basic Usage\n```python\n# Initialize different chunkers\nfixed_chunker = FixedLengthChunker(chunk_size=1000, chunk_overlap=200)\nsemantic_chunker = SemanticChunker(similarity_threshold=0.8)\nhybrid_chunker = HybridChunker([fixed_chunker, semantic_chunker])\n\n# Apply chunking\ntext = \"Your long document text here...\"\n\nfixed_chunks = fixed_chunker.chunk_optimized(text, strategy=\"balanced\")\nsemantic_chunks = semantic_chunker.semantic_chunk_sentences(text)\nhybrid_chunks = hybrid_chunker.hybrid_chunk(text)\n\nprint(f\"Fixed chunks: {len(fixed_chunks)}\")\nprint(f\"Semantic chunks: {len(semantic_chunks)}\")\nprint(f\"Hybrid chunks: {len(hybrid_chunks)}\")\n```\n\n### Advanced Usage with Evaluation\n```python\n# Create evaluation criteria\nevaluation_criteria = {\n \"coherence\": 0.4,\n \"size_appropriateness\": 0.3,\n \"content_completeness\": 0.3\n}\n\n# Apply hybrid chunking with evaluation\nresults = hybrid_chunker.hybrid_chunk(text, evaluation_criteria)\n\n# Analyze results\nfor chunk in results[:5]:\n print(f\"Strategy: {chunk['strategy']}\")\n print(f\"Score: {chunk.get('evaluation_score', 'N/A')}\")\n print(f\"Content preview: {chunk['content'][:100]}...\")\n print(\"-\" * 50)\n```\n\nThese 11 advanced chunking strategies provide comprehensive coverage of different approaches for various document types and use cases, from simple fixed-size chunking to sophisticated hybrid methods that combine multiple strategies.","content_type":"text/markdown; charset=utf-8","language":"markdown","size":47732,"content_sha256":"3d9b45e91fd3f91a90116430f3cc5b9c4bf68dd1b401a7d957700966b559f5ea"},{"filename":"references/evaluation.md","content":"# Performance Evaluation Framework\n\nThis document provides comprehensive methodologies for evaluating chunking strategy performance and effectiveness.\n\n## Evaluation Metrics\n\n### Core Retrieval Metrics\n\n#### Retrieval Precision\nMeasures the fraction of retrieved chunks that are relevant to the query.\n\n```python\ndef calculate_precision(retrieved_chunks: List[Dict], relevant_chunks: List[Dict]) -> float:\n \"\"\"\n Calculate retrieval precision\n Precision = |Relevant ∩ Retrieved| / |Retrieved|\n \"\"\"\n retrieved_ids = {chunk.get('id') for chunk in retrieved_chunks}\n relevant_ids = {chunk.get('id') for chunk in relevant_chunks}\n\n intersection = retrieved_ids & relevant_ids\n\n if not retrieved_ids:\n return 0.0\n\n return len(intersection) / len(retrieved_ids)\n```\n\n#### Retrieval Recall\nMeasures the fraction of relevant chunks that are successfully retrieved.\n\n```python\ndef calculate_recall(retrieved_chunks: List[Dict], relevant_chunks: List[Dict]) -> float:\n \"\"\"\n Calculate retrieval recall\n Recall = |Relevant ∩ Retrieved| / |Relevant|\n \"\"\"\n retrieved_ids = {chunk.get('id') for chunk in retrieved_chunks}\n relevant_ids = {chunk.get('id') for chunk in relevant_chunks}\n\n intersection = retrieved_ids & relevant_ids\n\n if not relevant_ids:\n return 0.0\n\n return len(intersection) / len(relevant_ids)\n```\n\n#### F1-Score\nHarmonic mean of precision and recall.\n\n```python\ndef calculate_f1_score(precision: float, recall: float) -> float:\n \"\"\"\n Calculate F1-score\n F1 = 2 * (Precision * Recall) / (Precision + Recall)\n \"\"\"\n if precision + recall == 0:\n return 0.0\n\n return 2 * (precision * recall) / (precision + recall)\n```\n\n### Mean Reciprocal Rank (MRR)\nMeasures the rank of the first relevant result.\n\n```python\ndef calculate_mrr(queries: List[Dict], results: List[List[Dict]]) -> float:\n \"\"\"\n Calculate Mean Reciprocal Rank\n \"\"\"\n reciprocal_ranks = []\n\n for query, query_results in zip(queries, results):\n relevant_found = False\n\n for rank, result in enumerate(query_results, 1):\n if result.get('is_relevant', False):\n reciprocal_ranks.append(1.0 / rank)\n relevant_found = True\n break\n\n if not relevant_found:\n reciprocal_ranks.append(0.0)\n\n return sum(reciprocal_ranks) / len(reciprocal_ranks)\n```\n\n### Mean Average Precision (MAP)\nConsiders both precision and the ranking of relevant documents.\n\n```python\ndef calculate_average_precision(retrieved_chunks: List[Dict], relevant_chunks: List[Dict]) -> float:\n \"\"\"\n Calculate Average Precision for a single query\n \"\"\"\n retrieved_ids = {chunk.get('id') for chunk in retrieved_chunks}\n relevant_ids = {chunk.get('id') for chunk in relevant_chunks}\n\n if not relevant_ids:\n return 0.0\n\n precisions = []\n relevant_count = 0\n\n for rank, chunk in enumerate(retrieved_chunks, 1):\n if chunk.get('id') in relevant_ids:\n relevant_count += 1\n precision_at_rank = relevant_count / rank\n precisions.append(precision_at_rank)\n\n return sum(precisions) / len(relevant_ids) if relevant_ids else 0.0\n\ndef calculate_map(queries: List[Dict], results: List[List[Dict]]) -> float:\n \"\"\"\n Calculate Mean Average Precision across multiple queries\n \"\"\"\n average_precisions = []\n\n for query, query_results in zip(queries, results):\n ap = calculate_average_precision(query_results, query.get('relevant_chunks', []))\n average_precisions.append(ap)\n\n return sum(average_precisions) / len(average_precisions) if average_precisions else 0.0\n```\n\n### Normalized Discounted Cumulative Gain (NDCG)\nMeasures ranking quality with emphasis on highly relevant results.\n\n```python\ndef calculate_dcg(retrieved_chunks: List[Dict]) -> float:\n \"\"\"\n Calculate Discounted Cumulative Gain\n \"\"\"\n dcg = 0.0\n\n for rank, chunk in enumerate(retrieved_chunks, 1):\n relevance = chunk.get('relevance_score', 0)\n dcg += relevance / np.log2(rank + 1)\n\n return dcg\n\ndef calculate_ndcg(retrieved_chunks: List[Dict], ideal_chunks: List[Dict]) -> float:\n \"\"\"\n Calculate Normalized Discounted Cumulative Gain\n \"\"\"\n dcg = calculate_dcg(retrieved_chunks)\n idcg = calculate_dcg(ideal_chunks)\n\n if idcg == 0:\n return 0.0\n\n return dcg / idcg\n```\n\n## End-to-End RAG Evaluation\n\n### Answer Quality Metrics\n\n#### Factual Consistency\nMeasures how well the generated answer aligns with retrieved chunks.\n\n```python\nimport spacy\nfrom transformers import pipeline\n\nclass FactualConsistencyEvaluator:\n def __init__(self):\n self.nlp = spacy.load(\"en_core_web_sm\")\n self.nli_pipeline = pipeline(\"text-classification\",\n model=\"roberta-large-mnli\")\n\n def evaluate_consistency(self, answer: str, retrieved_chunks: List[str]) -> float:\n \"\"\"\n Evaluate factual consistency between answer and retrieved context\n \"\"\"\n if not retrieved_chunks:\n return 0.0\n\n # Combine retrieved chunks as context\n context = \" \".join(retrieved_chunks[:3]) # Use top 3 chunks\n\n # Use Natural Language Inference to check consistency\n result = self.nli_pipeline(f\"premise: {context} hypothesis: {answer}\")\n\n # Extract consistency score (entailment probability)\n for item in result:\n if item['label'] == 'ENTAILMENT':\n return item['score']\n elif item['label'] == 'CONTRADICTION':\n return 1.0 - item['score']\n\n return 0.5 # Neutral if NLI is inconclusive\n```\n\n#### Answer Completeness\nMeasures how completely the answer addresses the user's query.\n\n```python\ndef evaluate_completeness(answer: str, query: str, reference_answer: str = None) -> float:\n \"\"\"\n Evaluate answer completeness\n \"\"\"\n # Extract key entities from query\n query_entities = extract_entities(query)\n answer_entities = extract_entities(answer)\n\n # Calculate entity coverage\n if not query_entities:\n return 0.5 # Neutral if no entities in query\n\n covered_entities = query_entities & answer_entities\n entity_coverage = len(covered_entities) / len(query_entities)\n\n # If reference answer is available, compare against it\n if reference_answer:\n reference_entities = extract_entities(reference_answer)\n answer_reference_overlap = len(answer_entities & reference_entities) / max(len(reference_entities), 1)\n return (entity_coverage + answer_reference_overlap) / 2\n\n return entity_coverage\n\ndef extract_entities(text: str) -> set:\n \"\"\"\n Extract named entities from text (simplified)\n \"\"\"\n # This would use a proper NER model in practice\n import re\n\n # Simple noun phrase extraction as placeholder\n noun_phrases = re.findall(r'\\b[A-Z][a-z]+(?:\\s+[A-Z][a-z]+)*\\b', text)\n return set(noun_phrases)\n```\n\n#### Response Relevance\nMeasures how relevant the answer is to the original query.\n\n```python\nfrom sentence_transformers import SentenceTransformer\nfrom sklearn.metrics.pairwise import cosine_similarity\n\nclass RelevanceEvaluator:\n def __init__(self, model_name=\"all-MiniLM-L6-v2\"):\n self.model = SentenceTransformer(model_name)\n\n def evaluate_relevance(self, query: str, answer: str) -> float:\n \"\"\"\n Evaluate semantic relevance between query and answer\n \"\"\"\n # Generate embeddings\n query_embedding = self.model.encode([query])\n answer_embedding = self.model.encode([answer])\n\n # Calculate cosine similarity\n similarity = cosine_similarity(query_embedding, answer_embedding)[0][0]\n\n return float(similarity)\n```\n\n## Performance Metrics\n\n### Processing Time\n\n```python\nimport time\nfrom dataclasses import dataclass\nfrom typing import List, Dict\n\n@dataclass\nclass PerformanceMetrics:\n total_time: float\n chunking_time: float\n embedding_time: float\n search_time: float\n generation_time: float\n throughput: float # documents per second\n\nclass PerformanceProfiler:\n def __init__(self):\n self.timings = {}\n self.start_times = {}\n\n def start_timer(self, operation: str):\n self.start_times[operation] = time.time()\n\n def end_timer(self, operation: str):\n if operation in self.start_times:\n duration = time.time() - self.start_times[operation]\n if operation not in self.timings:\n self.timings[operation] = []\n self.timings[operation].append(duration)\n return duration\n return 0.0\n\n def get_performance_metrics(self, document_count: int) -> PerformanceMetrics:\n total_time = sum(sum(times) for times in self.timings.values())\n\n return PerformanceMetrics(\n total_time=total_time,\n chunking_time=sum(self.timings.get('chunking', [0])),\n embedding_time=sum(self.timings.get('embedding', [0])),\n search_time=sum(self.timings.get('search', [0])),\n generation_time=sum(self.timings.get('generation', [0])),\n throughput=document_count / total_time if total_time > 0 else 0\n )\n```\n\n### Memory Usage\n\n```python\nimport psutil\nimport os\nfrom typing import Dict, List\n\nclass MemoryProfiler:\n def __init__(self):\n self.process = psutil.Process(os.getpid())\n self.memory_snapshots = []\n\n def take_memory_snapshot(self, label: str):\n \"\"\"Take a snapshot of current memory usage\"\"\"\n memory_info = self.process.memory_info()\n memory_mb = memory_info.rss / 1024 / 1024 # Convert to MB\n\n self.memory_snapshots.append({\n 'label': label,\n 'memory_mb': memory_mb,\n 'timestamp': time.time()\n })\n\n def get_peak_memory_usage(self) -> float:\n \"\"\"Get peak memory usage in MB\"\"\"\n if not self.memory_snapshots:\n return 0.0\n return max(snapshot['memory_mb'] for snapshot in self.memory_snapshots)\n\n def get_memory_usage_by_operation(self) -> Dict[str, float]:\n \"\"\"Get memory usage breakdown by operation\"\"\"\n if not self.memory_snapshots:\n return {}\n\n memory_by_op = {}\n for i in range(1, len(self.memory_snapshots)):\n prev_snapshot = self.memory_snapshots[i-1]\n curr_snapshot = self.memory_snapshots[i]\n\n operation = curr_snapshot['label']\n memory_delta = curr_snapshot['memory_mb'] - prev_snapshot['memory_mb']\n\n if operation not in memory_by_op:\n memory_by_op[operation] = []\n memory_by_op[operation].append(memory_delta)\n\n return {op: sum(deltas) for op, deltas in memory_by_op.items()}\n```\n\n## Evaluation Datasets\n\n### Standardized Test Sets\n\n#### Question-Answer Pairs\n\n```python\nfrom dataclasses import dataclass\nfrom typing import List, Optional\nimport json\n\n@dataclass\nclass EvaluationQuery:\n id: str\n question: str\n reference_answer: Optional[str]\n relevant_chunk_ids: List[str]\n query_type: str # factoid, analytical, comparative\n difficulty: str # easy, medium, hard\n domain: str # finance, medical, legal, technical\n\nclass EvaluationDataset:\n def __init__(self, name: str):\n self.name = name\n self.queries: List[EvaluationQuery] = []\n self.documents: Dict[str, str] = {}\n self.chunks: Dict[str, Dict] = {}\n\n def add_query(self, query: EvaluationQuery):\n self.queries.append(query)\n\n def add_document(self, doc_id: str, content: str):\n self.documents[doc_id] = content\n\n def add_chunk(self, chunk_id: str, content: str, doc_id: str, metadata: Dict):\n self.chunks[chunk_id] = {\n 'id': chunk_id,\n 'content': content,\n 'doc_id': doc_id,\n 'metadata': metadata\n }\n\n def save_to_file(self, filepath: str):\n data = {\n 'name': self.name,\n 'queries': [\n {\n 'id': q.id,\n 'question': q.question,\n 'reference_answer': q.reference_answer,\n 'relevant_chunk_ids': q.relevant_chunk_ids,\n 'query_type': q.query_type,\n 'difficulty': q.difficulty,\n 'domain': q.domain\n }\n for q in self.queries\n ],\n 'documents': self.documents,\n 'chunks': self.chunks\n }\n\n with open(filepath, 'w') as f:\n json.dump(data, f, indent=2)\n\n @classmethod\n def load_from_file(cls, filepath: str):\n with open(filepath, 'r') as f:\n data = json.load(f)\n\n dataset = cls(data['name'])\n dataset.documents = data['documents']\n dataset.chunks = data['chunks']\n\n for q_data in data['queries']:\n query = EvaluationQuery(\n id=q_data['id'],\n question=q_data['question'],\n reference_answer=q_data.get('reference_answer'),\n relevant_chunk_ids=q_data['relevant_chunk_ids'],\n query_type=q_data['query_type'],\n difficulty=q_data['difficulty'],\n domain=q_data['domain']\n )\n dataset.add_query(query)\n\n return dataset\n```\n\n### Dataset Generation\n\n#### Synthetic Query Generation\n\n```python\nimport random\nfrom typing import List, Dict\n\nclass SyntheticQueryGenerator:\n def __init__(self):\n self.query_templates = {\n 'factoid': [\n \"What is {concept}?\",\n \"When did {event} occur?\",\n \"Who developed {technology}?\",\n \"How many {items} are mentioned?\",\n \"What is the value of {metric}?\"\n ],\n 'analytical': [\n \"Compare and contrast {concept1} and {concept2}.\",\n \"Analyze the impact of {concept} on {domain}.\",\n \"What are the advantages and disadvantages of {technology}?\",\n \"Explain the relationship between {concept1} and {concept2}.\",\n \"Evaluate the effectiveness of {approach} for {problem}.\"\n ],\n 'comparative': [\n \"Which is better: {option1} or {option2}?\",\n \"How does {method1} differ from {method2}?\",\n \"Compare the performance of {system1} and {system2}.\",\n \"What are the key differences between {approach1} and {approach2}?\"\n ]\n }\n\n def generate_queries_from_chunks(self, chunks: List[Dict], num_queries: int = 100) -> List[EvaluationQuery]:\n \"\"\"Generate synthetic queries from document chunks\"\"\"\n queries = []\n\n # Extract entities and concepts from chunks\n entities = self._extract_entities_from_chunks(chunks)\n\n for i in range(num_queries):\n query_type = random.choice(['factoid', 'analytical', 'comparative'])\n template = random.choice(self.query_templates[query_type])\n\n # Fill template with extracted entities\n query_text = self._fill_template(template, entities)\n\n # Find relevant chunks for this query\n relevant_chunks = self._find_relevant_chunks(query_text, chunks)\n\n query = EvaluationQuery(\n id=f\"synthetic_{i}\",\n question=query_text,\n reference_answer=None, # Would need generation model\n relevant_chunk_ids=[chunk['id'] for chunk in relevant_chunks],\n query_type=query_type,\n difficulty=random.choice(['easy', 'medium', 'hard']),\n domain='synthetic'\n )\n\n queries.append(query)\n\n return queries\n\n def _extract_entities_from_chunks(self, chunks: List[Dict]) -> Dict[str, List[str]]:\n \"\"\"Extract entities, concepts, and relationships from chunks\"\"\"\n # This would use proper NER in practice\n entities = {\n 'concepts': [],\n 'technologies': [],\n 'methods': [],\n 'metrics': [],\n 'events': []\n }\n\n for chunk in chunks:\n content = chunk['content']\n # Simplified entity extraction\n words = content.split()\n entities['concepts'].extend([word for word in words if len(word) > 6])\n entities['technologies'].extend([word for word in words if 'technology' in word.lower()])\n entities['methods'].extend([word for word in words if 'method' in word.lower()])\n entities['metrics'].extend([word for word in words if '%' in word or '

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.

in word])\n\n # Remove duplicates and limit\n for key in entities:\n entities[key] = list(set(entities[key]))[:50]\n\n return entities\n\n def _fill_template(self, template: str, entities: Dict[str, List[str]]) -> str:\n \"\"\"Fill query template with random entities\"\"\"\n import re\n\n def replace_placeholder(match):\n placeholder = match.group(1)\n\n # Map placeholders to entity types\n entity_mapping = {\n 'concept': 'concepts',\n 'concept1': 'concepts',\n 'concept2': 'concepts',\n 'technology': 'technologies',\n 'method': 'methods',\n 'method1': 'methods',\n 'method2': 'methods',\n 'metric': 'metrics',\n 'event': 'events',\n 'items': 'concepts',\n 'option1': 'concepts',\n 'option2': 'concepts',\n 'approach': 'methods',\n 'problem': 'concepts',\n 'domain': 'concepts',\n 'system1': 'concepts',\n 'system2': 'concepts'\n }\n\n entity_type = entity_mapping.get(placeholder, 'concepts')\n available_entities = entities.get(entity_type, ['something'])\n\n if available_entities:\n return random.choice(available_entities)\n else:\n return 'something'\n\n return re.sub(r'\\{(\\w+)\\}', replace_placeholder, template)\n\n def _find_relevant_chunks(self, query: str, chunks: List[Dict], k: int = 3) -> List[Dict]:\n \"\"\"Find chunks most relevant to the query\"\"\"\n # Simple keyword matching for synthetic generation\n query_words = set(query.lower().split())\n\n chunk_scores = []\n for chunk in chunks:\n chunk_words = set(chunk['content'].lower().split())\n overlap = len(query_words & chunk_words)\n chunk_scores.append((overlap, chunk))\n\n # Sort by overlap and return top k\n chunk_scores.sort(key=lambda x: x[0], reverse=True)\n return [chunk for _, chunk in chunk_scores[:k]]\n```\n\n## A/B Testing Framework\n\n### Statistical Significance Testing\n\n```python\nimport numpy as np\nfrom scipy import stats\nfrom typing import List, Dict, Tuple\n\nclass ABTestAnalyzer:\n def __init__(self):\n self.significance_level = 0.05\n\n def compare_metrics(self, control_metrics: List[float],\n treatment_metrics: List[float],\n metric_name: str) -> Dict:\n \"\"\"\n Compare metrics between control and treatment groups\n \"\"\"\n control_mean = np.mean(control_metrics)\n treatment_mean = np.mean(treatment_metrics)\n\n control_std = np.std(control_metrics)\n treatment_std = np.std(treatment_metrics)\n\n # Perform t-test\n t_statistic, p_value = stats.ttest_ind(control_metrics, treatment_metrics)\n\n # Calculate effect size (Cohen's d)\n pooled_std = np.sqrt(((len(control_metrics) - 1) * control_std**2 +\n (len(treatment_metrics) - 1) * treatment_std**2) /\n (len(control_metrics) + len(treatment_metrics) - 2))\n\n cohens_d = (treatment_mean - control_mean) / pooled_std if pooled_std > 0 else 0\n\n # Determine significance\n is_significant = p_value \u003c self.significance_level\n\n return {\n 'metric_name': metric_name,\n 'control_mean': control_mean,\n 'treatment_mean': treatment_mean,\n 'absolute_difference': treatment_mean - control_mean,\n 'relative_difference': ((treatment_mean - control_mean) / control_mean * 100) if control_mean != 0 else 0,\n 'control_std': control_std,\n 'treatment_std': treatment_std,\n 't_statistic': t_statistic,\n 'p_value': p_value,\n 'is_significant': is_significant,\n 'effect_size': cohens_d,\n 'significance_level': self.significance_level\n }\n\n def analyze_ab_test_results(self,\n control_results: Dict[str, List[float]],\n treatment_results: Dict[str, List[float]]) -> Dict:\n \"\"\"\n Analyze A/B test results across multiple metrics\n \"\"\"\n analysis_results = {}\n\n # Ensure both dictionaries have the same keys\n all_metrics = set(control_results.keys()) & set(treatment_results.keys())\n\n for metric in all_metrics:\n if metric in control_results and metric in treatment_results:\n analysis_results[metric] = self.compare_metrics(\n control_results[metric],\n treatment_results[metric],\n metric\n )\n\n # Calculate overall summary\n significant_improvements = sum(1 for result in analysis_results.values()\n if result['is_significant'] and result['relative_difference'] > 0)\n significant_degradations = sum(1 for result in analysis_results.values()\n if result['is_significant'] and result['relative_difference'] \u003c 0)\n\n analysis_results['summary'] = {\n 'total_metrics_compared': len(analysis_results),\n 'significant_improvements': significant_improvements,\n 'significant_degradations': significant_degradations,\n 'no_significant_change': len(analysis_results) - significant_improvements - significant_degradations\n }\n\n return analysis_results\n```\n\n## Automated Evaluation Pipeline\n\n### End-to-End Evaluation\n\n```python\nclass ChunkingEvaluationPipeline:\n def __init__(self, strategies: Dict[str, Any], dataset: EvaluationDataset):\n self.strategies = strategies\n self.dataset = dataset\n self.results = {}\n self.profiler = PerformanceProfiler()\n self.memory_profiler = MemoryProfiler()\n\n def run_evaluation(self) -> Dict:\n \"\"\"Run comprehensive evaluation of all strategies\"\"\"\n evaluation_results = {}\n\n for strategy_name, strategy in self.strategies.items():\n print(f\"Evaluating strategy: {strategy_name}\")\n\n # Reset profilers for each strategy\n self.profiler = PerformanceProfiler()\n self.memory_profiler = MemoryProfiler()\n\n # Evaluate strategy\n strategy_results = self._evaluate_strategy(strategy, strategy_name)\n evaluation_results[strategy_name] = strategy_results\n\n # Compare strategies\n comparison_results = self._compare_strategies(evaluation_results)\n\n return {\n 'individual_results': evaluation_results,\n 'comparison': comparison_results,\n 'recommendations': self._generate_recommendations(comparison_results)\n }\n\n def _evaluate_strategy(self, strategy: Any, strategy_name: str) -> Dict:\n \"\"\"Evaluate a single chunking strategy\"\"\"\n results = {\n 'strategy_name': strategy_name,\n 'retrieval_metrics': {},\n 'quality_metrics': {},\n 'performance_metrics': {}\n }\n\n # Track memory usage\n self.memory_profiler.take_memory_snapshot(f\"{strategy_name}_start\")\n\n # Process all documents\n self.profiler.start_timer('total_processing')\n\n all_chunks = {}\n for doc_id, content in self.dataset.documents.items():\n self.profiler.start_timer('chunking')\n chunks = strategy.chunk(content)\n self.profiler.end_timer('chunking')\n\n all_chunks[doc_id] = chunks\n\n self.memory_profiler.take_memory_snapshot(f\"{strategy_name}_after_chunking\")\n\n # Generate embeddings for chunks\n self.profiler.start_timer('embedding')\n chunk_embeddings = self._generate_embeddings(all_chunks)\n self.profiler.end_timer('embedding')\n\n self.memory_profiler.take_memory_snapshot(f\"{strategy_name}_after_embedding\")\n\n # Evaluate retrieval performance\n retrieval_results = self._evaluate_retrieval(all_chunks, chunk_embeddings)\n results['retrieval_metrics'] = retrieval_results\n\n # Evaluate chunk quality\n quality_results = self._evaluate_chunk_quality(all_chunks)\n results['quality_metrics'] = quality_results\n\n # Get performance metrics\n self.profiler.end_timer('total_processing')\n performance_metrics = self.profiler.get_performance_metrics(len(self.dataset.documents))\n results['performance_metrics'] = performance_metrics.__dict__\n\n # Get memory metrics\n self.memory_profiler.take_memory_snapshot(f\"{strategy_name}_end\")\n results['memory_metrics'] = {\n 'peak_memory_mb': self.memory_profiler.get_peak_memory_usage(),\n 'memory_by_operation': self.memory_profiler.get_memory_usage_by_operation()\n }\n\n return results\n\n def _evaluate_retrieval(self, all_chunks: Dict, chunk_embeddings: Dict) -> Dict:\n \"\"\"Evaluate retrieval performance\"\"\"\n retrieval_metrics = {\n 'precision': [],\n 'recall': [],\n 'f1_score': [],\n 'mrr': [],\n 'map': []\n }\n\n for query in self.dataset.queries:\n # Perform retrieval\n self.profiler.start_timer('search')\n retrieved_chunks = self._retrieve_chunks(query.question, chunk_embeddings, k=10)\n self.profiler.end_timer('search')\n\n # Get relevant chunks for this query\n relevant_chunk_ids = set(query.relevant_chunk_ids)\n relevant_chunks = [chunk for chunk in retrieved_chunks\n if chunk.get('id') in relevant_chunk_ids]\n\n # Calculate metrics\n precision = calculate_precision(retrieved_chunks, relevant_chunks)\n recall = calculate_recall(retrieved_chunks, relevant_chunks)\n f1 = calculate_f1_score(precision, recall)\n\n retrieval_metrics['precision'].append(precision)\n retrieval_metrics['recall'].append(recall)\n retrieval_metrics['f1_score'].append(f1)\n\n # Calculate averages\n return {metric: np.mean(values) for metric, values in retrieval_metrics.items()}\n\n def _evaluate_chunk_quality(self, all_chunks: Dict) -> Dict:\n \"\"\"Evaluate quality of generated chunks\"\"\"\n quality_assessor = ChunkQualityAssessor()\n quality_scores = []\n\n for doc_id, chunks in all_chunks.items():\n # Analyze document\n content = self.dataset.documents[doc_id]\n analyzer = DocumentAnalyzer()\n analysis = analyzer.analyze(content)\n\n # Assess chunk quality\n scores = quality_assessor.assess_chunks(chunks, analysis)\n quality_scores.append(scores)\n\n # Aggregate quality scores\n if quality_scores:\n avg_scores = {}\n for metric in quality_scores[0].keys():\n avg_scores[metric] = np.mean([scores[metric] for scores in quality_scores])\n return avg_scores\n\n return {}\n\n def _compare_strategies(self, evaluation_results: Dict) -> Dict:\n \"\"\"Compare performance across strategies\"\"\"\n ab_analyzer = ABTestAnalyzer()\n\n comparison = {}\n\n # Compare each metric across strategies\n strategy_names = list(evaluation_results.keys())\n\n for i in range(len(strategy_names)):\n for j in range(i + 1, len(strategy_names)):\n strategy1 = strategy_names[i]\n strategy2 = strategy_names[j]\n\n comparison_key = f\"{strategy1}_vs_{strategy2}\"\n comparison[comparison_key] = {}\n\n # Compare retrieval metrics\n for metric in ['precision', 'recall', 'f1_score']:\n if (metric in evaluation_results[strategy1]['retrieval_metrics'] and\n metric in evaluation_results[strategy2]['retrieval_metrics']):\n\n comparison[comparison_key][f\"retrieval_{metric}\"] = ab_analyzer.compare_metrics(\n [evaluation_results[strategy1]['retrieval_metrics'][metric]],\n [evaluation_results[strategy2]['retrieval_metrics'][metric]],\n f\"retrieval_{metric}\"\n )\n\n return comparison\n\n def _generate_recommendations(self, comparison_results: Dict) -> Dict:\n \"\"\"Generate recommendations based on evaluation results\"\"\"\n recommendations = {\n 'best_overall': None,\n 'best_for_precision': None,\n 'best_for_recall': None,\n 'best_for_performance': None,\n 'trade_offs': []\n }\n\n # This would analyze the comparison results and generate specific recommendations\n # Implementation depends on specific use case requirements\n\n return recommendations\n\n def _generate_embeddings(self, all_chunks: Dict) -> Dict:\n \"\"\"Generate embeddings for all chunks\"\"\"\n # This would use the actual embedding model\n # Placeholder implementation\n embeddings = {}\n\n for doc_id, chunks in all_chunks.items():\n embeddings[doc_id] = []\n for chunk in chunks:\n # Generate embedding for chunk content\n embedding = np.random.rand(384) # Placeholder\n embeddings[doc_id].append({\n 'chunk': chunk,\n 'embedding': embedding\n })\n\n return embeddings\n\n def _retrieve_chunks(self, query: str, chunk_embeddings: Dict, k: int = 10) -> List[Dict]:\n \"\"\"Retrieve most relevant chunks for a query\"\"\"\n # This would use actual similarity search\n # Placeholder implementation\n all_chunks = []\n\n for doc_embeddings in chunk_embeddings.values():\n for chunk_data in doc_embeddings:\n all_chunks.append(chunk_data['chunk'])\n\n # Simple random selection as placeholder\n selected = random.sample(all_chunks, min(k, len(all_chunks)))\n\n return selected\n```\n\nThis comprehensive evaluation framework provides the tools needed to thoroughly assess chunking strategies across multiple dimensions: retrieval effectiveness, answer quality, system performance, and statistical significance. The modular design allows for easy extension and customization based on specific requirements and use cases.","content_type":"text/markdown; charset=utf-8","language":"markdown","size":31173,"content_sha256":"0c14e0eacbac69c81e55a994c33b854c4b9b8cd76e74d22ba93d8b5d858f7f8f"},{"filename":"references/implementation.md","content":"# Complete Implementation Guidelines\n\nThis document provides comprehensive implementation guidance for building effective chunking systems.\n\n## System Architecture\n\n### Core Components\n\n```\nDocument Processor\n├── Ingestion Layer\n│ ├── Document Type Detection\n│ ├── Format Parsing (PDF, HTML, Markdown, etc.)\n│ └── Content Extraction\n├── Analysis Layer\n│ ├── Structure Analysis\n│ ├── Content Type Identification\n│ └── Complexity Assessment\n├── Strategy Selection Layer\n│ ├── Rule-based Selection\n│ ├── ML-based Prediction\n│ └── Adaptive Configuration\n├── Chunking Layer\n│ ├── Strategy Implementation\n│ ├── Parameter Optimization\n│ └── Quality Validation\n└── Output Layer\n ├── Chunk Metadata Generation\n ├── Embedding Integration\n └── Storage Preparation\n```\n\n## Pre-processing Pipeline\n\n### Document Analysis Framework\n\n```python\nfrom dataclasses import dataclass\nfrom typing import List, Dict, Any\nimport re\n\n@dataclass\nclass DocumentAnalysis:\n doc_type: str\n structure_score: float # 0-1, higher means more structured\n complexity_score: float # 0-1, higher means more complex\n content_types: List[str]\n language: str\n estimated_tokens: int\n has_multimodal: bool\n\nclass DocumentAnalyzer:\n def __init__(self):\n self.structure_patterns = {\n 'markdown': [r'^#+\\s', r'^\\*\\*.*\\*\\*

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.

, r'^\\* ', r'^\\d+\\. '],\n 'html': [r'\u003ch[1-6]>', r'\u003cp>', r'\u003cdiv>', r'\u003ctable>'],\n 'latex': [r'\\\\section', r'\\\\subsection', r'\\\\begin\\{', r'\\\\end\\{'],\n 'academic': [r'^\\d+\\.', r'^\\d+\\.\\d+', r'^[A-Z]\\.', r'^Figure \\d+']\n }\n\n def analyze(self, content: str) -> DocumentAnalysis:\n doc_type = self.detect_document_type(content)\n structure_score = self.calculate_structure_score(content, doc_type)\n complexity_score = self.calculate_complexity_score(content)\n content_types = self.identify_content_types(content)\n language = self.detect_language(content)\n estimated_tokens = self.estimate_tokens(content)\n has_multimodal = self.detect_multimodal_content(content)\n\n return DocumentAnalysis(\n doc_type=doc_type,\n structure_score=structure_score,\n complexity_score=complexity_score,\n content_types=content_types,\n language=language,\n estimated_tokens=estimated_tokens,\n has_multimodal=has_multimodal\n )\n\n def detect_document_type(self, content: str) -> str:\n content_lower = content.lower()\n\n if '\u003chtml' in content_lower or '\u003cbody' in content_lower:\n return 'html'\n elif '#' in content and '##' in content:\n return 'markdown'\n elif '\\\\documentclass' in content_lower or '\\\\begin{' in content_lower:\n return 'latex'\n elif any(keyword in content_lower for keyword in ['abstract', 'introduction', 'conclusion', 'references']):\n return 'academic'\n elif 'def ' in content or 'class ' in content or 'function ' in content_lower:\n return 'code'\n else:\n return 'plain'\n\n def calculate_structure_score(self, content: str, doc_type: str) -> float:\n patterns = self.structure_patterns.get(doc_type, [])\n if not patterns:\n return 0.5 # Default for unstructured content\n\n line_count = len(content.split('\\n'))\n structured_lines = 0\n\n for line in content.split('\\n'):\n for pattern in patterns:\n if re.search(pattern, line.strip()):\n structured_lines += 1\n break\n\n return min(structured_lines / max(line_count, 1), 1.0)\n\n def calculate_complexity_score(self, content: str) -> float:\n # Factors that increase complexity\n avg_sentence_length = self.calculate_avg_sentence_length(content)\n vocabulary_richness = self.calculate_vocabulary_richness(content)\n nested_structure = self.detect_nested_structure(content)\n\n # Normalize and combine\n complexity = (\n min(avg_sentence_length / 30, 1.0) * 0.3 +\n vocabulary_richness * 0.4 +\n nested_structure * 0.3\n )\n\n return min(complexity, 1.0)\n\n def identify_content_types(self, content: str) -> List[str]:\n types = []\n\n if '```' in content or 'def ' in content or 'function ' in content.lower():\n types.append('code')\n if '|' in content and '\\n' in content:\n types.append('tables')\n if re.search(r'\\!\\[.*\\]\$.*\$', content):\n types.append('images')\n if re.search(r'http[s]?://', content):\n types.append('links')\n if re.search(r'\\d+\\.\\d+', content) or re.search(r'\\$\\d', content):\n types.append('numbers')\n\n return types if types else ['text']\n\n def detect_language(self, content: str) -> str:\n # Simple language detection - can be enhanced with proper language detection libraries\n if re.search(r'[\\u4e00-\\u9fff]', content):\n return 'chinese'\n elif re.search(r'[u0600-\\u06ff]', content):\n return 'arabic'\n elif re.search(r'[u0400-\\u04ff]', content):\n return 'russian'\n else:\n return 'english' # Default assumption\n\n def estimate_tokens(self, content: str) -> int:\n # Rough estimation - actual tokenization varies by model\n word_count = len(content.split())\n return int(word_count * 1.3) # Average tokens per word\n\n def detect_multimodal_content(self, content: str) -> bool:\n multimodal_indicators = [\n r'\\!\\[.*\\]\$.*\$', # Images\n r'\u003ciframe', # Embedded content\n r'\u003cobject', # Embedded objects\n r'\u003cembed', # Embedded media\n ]\n\n return any(re.search(pattern, content) for pattern in multimodal_indicators)\n\n def calculate_avg_sentence_length(self, content: str) -> float:\n sentences = re.split(r'[.!?]+', content)\n sentences = [s.strip() for s in sentences if s.strip()]\n if not sentences:\n return 0\n return sum(len(s.split()) for s in sentences) / len(sentences)\n\n def calculate_vocabulary_richness(self, content: str) -> float:\n words = content.lower().split()\n if not words:\n return 0\n unique_words = set(words)\n return len(unique_words) / len(words)\n\n def detect_nested_structure(self, content: str) -> float:\n # Detect nested lists, indented content, etc.\n lines = content.split('\\n')\n indented_lines = 0\n\n for line in lines:\n if line.strip() and line.startswith(' '):\n indented_lines += 1\n\n return indented_lines / max(len(lines), 1)\n```\n\n### Strategy Selection Engine\n\n```python\nfrom abc import ABC, abstractmethod\nfrom typing import Dict, Any\n\nclass ChunkingStrategy(ABC):\n @abstractmethod\n def chunk(self, content: str, analysis: DocumentAnalysis) -> List[Dict[str, Any]]:\n pass\n\nclass StrategySelector:\n def __init__(self):\n self.strategies = {\n 'fixed_size': FixedSizeStrategy(),\n 'recursive': RecursiveStrategy(),\n 'structure_aware': StructureAwareStrategy(),\n 'semantic': SemanticStrategy(),\n 'adaptive': AdaptiveStrategy()\n }\n\n def select_strategy(self, analysis: DocumentAnalysis) -> str:\n # Rule-based selection logic\n if analysis.structure_score > 0.8 and analysis.doc_type in ['markdown', 'html', 'latex']:\n return 'structure_aware'\n elif analysis.complexity_score > 0.7 and analysis.estimated_tokens \u003c 10000:\n return 'semantic'\n elif analysis.doc_type == 'code':\n return 'structure_aware'\n elif analysis.structure_score \u003c 0.3:\n return 'fixed_size'\n elif analysis.complexity_score > 0.5:\n return 'recursive'\n else:\n return 'adaptive'\n\n def get_strategy(self, analysis: DocumentAnalysis) -> ChunkingStrategy:\n strategy_name = self.select_strategy(analysis)\n return self.strategies[strategy_name]\n\n# Example strategy implementations\nclass FixedSizeStrategy(ChunkingStrategy):\n def __init__(self, default_size=512, default_overlap=50):\n self.default_size = default_size\n self.default_overlap = default_overlap\n\n def chunk(self, content: str, analysis: DocumentAnalysis) -> List[Dict[str, Any]]:\n # Adjust parameters based on analysis\n if analysis.complexity_score > 0.7:\n chunk_size = 1024\n elif analysis.complexity_score \u003c 0.3:\n chunk_size = 256\n else:\n chunk_size = self.default_size\n\n overlap = int(chunk_size * 0.1) # 10% overlap\n\n # Implementation here...\n return self._fixed_size_chunk(content, chunk_size, overlap)\n\n def _fixed_size_chunk(self, content: str, chunk_size: int, overlap: int) -> List[Dict[str, Any]]:\n # Implementation using RecursiveCharacterTextSplitter or custom logic\n pass\n\nclass AdaptiveStrategy(ChunkingStrategy):\n def chunk(self, content: str, analysis: DocumentAnalysis) -> List[Dict[str, Any]]:\n # Combine multiple strategies based on content characteristics\n if analysis.structure_score > 0.6:\n # Use structure-aware for structured parts\n structured_chunks = self._chunk_structured_parts(content, analysis)\n else:\n # Use fixed-size for unstructured parts\n unstructured_chunks = self._chunk_unstructured_parts(content, analysis)\n\n # Merge and optimize\n return self._merge_chunks(structured_chunks + unstructured_chunks)\n\n def _chunk_structured_parts(self, content: str, analysis: DocumentAnalysis) -> List[Dict[str, Any]]:\n # Implementation for structured content\n pass\n\n def _chunk_unstructured_parts(self, content: str, analysis: DocumentAnalysis) -> List[Dict[str, Any]]:\n # Implementation for unstructured content\n pass\n\n def _merge_chunks(self, chunks: List[Dict[str, Any]]) -> List[Dict[str, Any]]:\n # Implementation for merging and optimizing chunks\n pass\n```\n\n## Quality Assurance Framework\n\n### Chunk Quality Metrics\n\n```python\nfrom typing import List, Dict, Any\nimport numpy as np\nfrom sklearn.metrics.pairwise import cosine_similarity\n\nclass ChunkQualityAssessor:\n def __init__(self):\n self.quality_weights = {\n 'coherence': 0.3,\n 'completeness': 0.25,\n 'size_appropriateness': 0.2,\n 'semantic_similarity': 0.15,\n 'boundary_quality': 0.1\n }\n\n def assess_chunks(self, chunks: List[Dict[str, Any]], analysis: DocumentAnalysis) -> Dict[str, float]:\n scores = {}\n\n # Coherence: Do chunks make sense on their own?\n scores['coherence'] = self._assess_coherence(chunks)\n\n # Completeness: Do chunks preserve important information?\n scores['completeness'] = self._assess_completeness(chunks, analysis)\n\n # Size appropriateness: Are chunks within optimal size range?\n scores['size_appropriateness'] = self._assess_size(chunks)\n\n # Semantic similarity: Are chunks thematically consistent?\n scores['semantic_similarity'] = self._assess_semantic_consistency(chunks)\n\n # Boundary quality: Are chunk boundaries placed well?\n scores['boundary_quality'] = self._assess_boundary_quality(chunks)\n\n # Calculate overall quality score\n overall_score = sum(\n score * self.quality_weights[metric]\n for metric, score in scores.items()\n )\n\n scores['overall'] = overall_score\n return scores\n\n def _assess_coherence(self, chunks: List[Dict[str, Any]]) -> float:\n # Simple heuristic-based coherence assessment\n coherence_scores = []\n\n for chunk in chunks:\n content = chunk['content']\n\n # Check for complete sentences\n sentences = re.split(r'[.!?]+', content)\n complete_sentences = sum(1 for s in sentences if s.strip())\n coherence = complete_sentences / max(len(sentences), 1)\n\n coherence_scores.append(coherence)\n\n return np.mean(coherence_scores)\n\n def _assess_completeness(self, chunks: List[Dict[str, Any]], analysis: DocumentAnalysis) -> float:\n # Check if important structural elements are preserved\n if analysis.doc_type in ['markdown', 'html']:\n return self._assess_structure_preservation(chunks, analysis)\n else:\n return self._assess_content_preservation(chunks)\n\n def _assess_structure_preservation(self, chunks: List[Dict[str, Any]], analysis: DocumentAnalysis) -> float:\n # Check if headings, lists, and other structural elements are preserved\n preserved_elements = 0\n total_elements = 0\n\n for chunk in chunks:\n content = chunk['content']\n\n # Count preserved structural elements\n headings = len(re.findall(r'^#+\\s', content, re.MULTILINE))\n lists = len(re.findall(r'^\\s*[-*+]\\s', content, re.MULTILINE))\n\n preserved_elements += headings + lists\n total_elements += 1 # Simplified count\n\n return preserved_elements / max(total_elements, 1)\n\n def _assess_content_preservation(self, chunks: List[Dict[str, Any]]) -> float:\n # Simple check based on content ratio\n total_content = ''.join(chunk['content'] for chunk in chunks)\n # This would need comparison with original content\n return 0.8 # Placeholder\n\n def _assess_size(self, chunks: List[Dict[str, Any]]) -> float:\n optimal_min = 100 # tokens\n optimal_max = 1000 # tokens\n\n size_scores = []\n for chunk in chunks:\n token_count = self._estimate_tokens(chunk['content'])\n if optimal_min \u003c= token_count \u003c= optimal_max:\n score = 1.0\n elif token_count \u003c optimal_min:\n score = token_count / optimal_min\n else:\n score = max(0, 1 - (token_count - optimal_max) / optimal_max)\n\n size_scores.append(score)\n\n return np.mean(size_scores)\n\n def _assess_semantic_consistency(self, chunks: List[Dict[str, Any]]) -> float:\n # This would require embedding models for actual implementation\n # Placeholder implementation\n return 0.7\n\n def _assess_boundary_quality(self, chunks: List[Dict[str, Any]]) -> float:\n # Check if boundaries don't split important content\n boundary_scores = []\n\n for i, chunk in enumerate(chunks):\n content = chunk['content']\n\n # Check for incomplete sentences at boundaries\n if not content.strip().endswith(('.', '!', '?', '>', '}')):\n boundary_scores.append(0.5)\n else:\n boundary_scores.append(1.0)\n\n return np.mean(boundary_scores)\n\n def _estimate_tokens(self, content: str) -> int:\n # Simple token estimation\n return len(content.split()) * 4 // 3 # Rough approximation\n```\n\n## Error Handling and Edge Cases\n\n### Robust Error Handling\n\n```python\nimport logging\nfrom typing import Optional, List\nfrom dataclasses import dataclass\n\n@dataclass\nclass ChunkingError:\n error_type: str\n message: str\n chunk_index: Optional[int] = None\n recovery_action: Optional[str] = None\n\nclass ChunkingErrorHandler:\n def __init__(self):\n self.logger = logging.getLogger(__name__)\n self.error_handlers = {\n 'empty_content': self._handle_empty_content,\n 'oversized_chunk': self._handle_oversized_chunk,\n 'encoding_error': self._handle_encoding_error,\n 'memory_error': self._handle_memory_error,\n 'structure_parsing_error': self._handle_structure_parsing_error\n }\n\n def handle_error(self, error: Exception, context: Dict[str, Any]) -> ChunkingError:\n error_type = self._classify_error(error)\n handler = self.error_handlers.get(error_type, self._handle_generic_error)\n return handler(error, context)\n\n def _classify_error(self, error: Exception) -> str:\n if isinstance(error, ValueError) and 'empty' in str(error).lower():\n return 'empty_content'\n elif isinstance(error, MemoryError):\n return 'memory_error'\n elif isinstance(error, UnicodeError):\n return 'encoding_error'\n elif 'too large' in str(error).lower():\n return 'oversized_chunk'\n elif 'parsing' in str(error).lower():\n return 'structure_parsing_error'\n else:\n return 'generic_error'\n\n def _handle_empty_content(self, error: Exception, context: Dict[str, Any]) -> ChunkingError:\n self.logger.warning(f\"Empty content encountered: {error}\")\n return ChunkingError(\n error_type='empty_content',\n message=str(error),\n recovery_action='skip_empty_content'\n )\n\n def _handle_oversized_chunk(self, error: Exception, context: Dict[str, Any]) -> ChunkingError:\n self.logger.warning(f\"Oversized chunk detected: {error}\")\n return ChunkingError(\n error_type='oversized_chunk',\n message=str(error),\n chunk_index=context.get('chunk_index'),\n recovery_action='reduce_chunk_size'\n )\n\n def _handle_encoding_error(self, error: Exception, context: Dict[str, Any]) -> ChunkingError:\n self.logger.error(f\"Encoding error: {error}\")\n return ChunkingError(\n error_type='encoding_error',\n message=str(error),\n recovery_action='fallback_encoding'\n )\n\n def _handle_memory_error(self, error: Exception, context: Dict[str, Any]) -> ChunkingError:\n self.logger.error(f\"Memory error during chunking: {error}\")\n return ChunkingError(\n error_type='memory_error',\n message=str(error),\n recovery_action='process_in_batches'\n )\n\n def _handle_structure_parsing_error(self, error: Exception, context: Dict[str, Any]) -> ChunkingError:\n self.logger.warning(f\"Structure parsing failed: {error}\")\n return ChunkingError(\n error_type='structure_parsing_error',\n message=str(error),\n recovery_action='fallback_to_fixed_size'\n )\n\n def _handle_generic_error(self, error: Exception, context: Dict[str, Any]) -> ChunkingError:\n self.logger.error(f\"Unexpected error during chunking: {error}\")\n return ChunkingError(\n error_type='generic_error',\n message=str(error),\n recovery_action='skip_and_continue'\n )\n```\n\n## Performance Optimization\n\n### Caching and Memoization\n\n```python\nimport hashlib\nimport pickle\nfrom functools import lru_cache\nfrom typing import Dict, Any, Optional\nimport redis\nimport json\n\nclass ChunkingCache:\n def __init__(self, redis_url: Optional[str] = None):\n if redis_url:\n self.redis_client = redis.from_url(redis_url)\n else:\n self.redis_client = None\n self.local_cache = {}\n\n def _generate_cache_key(self, content: str, strategy: str, params: Dict[str, Any]) -> str:\n content_hash = hashlib.md5(content.encode()).hexdigest()\n params_str = json.dumps(params, sort_keys=True)\n params_hash = hashlib.md5(params_str.encode()).hexdigest()\n return f\"chunking:{strategy}:{content_hash}:{params_hash}\"\n\n def get(self, content: str, strategy: str, params: Dict[str, Any]) -> Optional[List[Dict[str, Any]]]:\n cache_key = self._generate_cache_key(content, strategy, params)\n\n # Try local cache first\n if cache_key in self.local_cache:\n return self.local_cache[cache_key]\n\n # Try Redis cache\n if self.redis_client:\n try:\n cached_data = self.redis_client.get(cache_key)\n if cached_data:\n chunks = pickle.loads(cached_data)\n self.local_cache[cache_key] = chunks # Cache locally too\n return chunks\n except Exception as e:\n logging.warning(f\"Redis cache error: {e}\")\n\n return None\n\n def set(self, content: str, strategy: str, params: Dict[str, Any], chunks: List[Dict[str, Any]]) -> None:\n cache_key = self._generate_cache_key(content, strategy, params)\n\n # Store in local cache\n self.local_cache[cache_key] = chunks\n\n # Store in Redis cache\n if self.redis_client:\n try:\n cached_data = pickle.dumps(chunks)\n self.redis_client.setex(cache_key, 3600, cached_data) # 1 hour TTL\n except Exception as e:\n logging.warning(f\"Redis cache set error: {e}\")\n\n def clear_local_cache(self):\n self.local_cache.clear()\n\n def clear_redis_cache(self):\n if self.redis_client:\n pattern = \"chunking:*\"\n keys = self.redis_client.keys(pattern)\n if keys:\n self.redis_client.delete(*keys)\n```\n\n### Batch Processing\n\n```python\nimport asyncio\nimport concurrent.futures\nfrom typing import List, Callable, Any\n\nclass BatchChunkingProcessor:\n def __init__(self, max_workers: int = 4, batch_size: int = 10):\n self.max_workers = max_workers\n self.batch_size = batch_size\n\n def process_documents_batch(self, documents: List[str],\n chunking_function: Callable[[str], List[Dict[str, Any]]]) -> List[List[Dict[str, Any]]]:\n \"\"\"Process multiple documents in parallel\"\"\"\n results = []\n\n # Process in batches to avoid memory issues\n for i in range(0, len(documents), self.batch_size):\n batch = documents[i:i + self.batch_size]\n\n with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:\n future_to_doc = {\n executor.submit(chunking_function, doc): doc\n for doc in batch\n }\n\n batch_results = []\n for future in concurrent.futures.as_completed(future_to_doc):\n try:\n chunks = future.result()\n batch_results.append(chunks)\n except Exception as e:\n logging.error(f\"Error processing document: {e}\")\n batch_results.append([]) # Empty result for failed processing\n\n results.extend(batch_results)\n\n return results\n\n async def process_documents_async(self, documents: List[str],\n chunking_function: Callable[[str], List[Dict[str, Any]]]) -> List[List[Dict[str, Any]]]:\n \"\"\"Process documents asynchronously\"\"\"\n semaphore = asyncio.Semaphore(self.max_workers)\n\n async def process_single_document(doc: str) -> List[Dict[str, Any]]:\n async with semaphore:\n # Run the synchronous chunking function in an executor\n loop = asyncio.get_event_loop()\n return await loop.run_in_executor(None, chunking_function, doc)\n\n tasks = [process_single_document(doc) for doc in documents]\n return await asyncio.gather(*tasks, return_exceptions=True)\n```\n\n## Monitoring and Observability\n\n### Metrics Collection\n\n```python\nimport time\nfrom dataclasses import dataclass\nfrom typing import Dict, Any, List\nfrom collections import defaultdict\n\n@dataclass\nclass ChunkingMetrics:\n total_documents: int\n total_chunks: int\n avg_chunk_size: float\n processing_time: float\n memory_usage: float\n error_count: int\n strategy_distribution: Dict[str, int]\n\nclass MetricsCollector:\n def __init__(self):\n self.metrics = defaultdict(list)\n self.start_time = None\n\n def start_timing(self):\n self.start_time = time.time()\n\n def end_timing(self) -> float:\n if self.start_time:\n duration = time.time() - self.start_time\n self.metrics['processing_time'].append(duration)\n self.start_time = None\n return duration\n return 0.0\n\n def record_chunk_count(self, count: int):\n self.metrics['chunk_count'].append(count)\n\n def record_chunk_size(self, size: int):\n self.metrics['chunk_size'].append(size)\n\n def record_strategy_usage(self, strategy: str):\n self.metrics['strategy'][strategy] = self.metrics['strategy'].get(strategy, 0) + 1\n\n def record_error(self, error_type: str):\n self.metrics['errors'].append(error_type)\n\n def record_memory_usage(self, memory_mb: float):\n self.metrics['memory_usage'].append(memory_mb)\n\n def get_summary(self) -> ChunkingMetrics:\n return ChunkingMetrics(\n total_documents=len(self.metrics['processing_time']),\n total_chunks=sum(self.metrics['chunk_count']),\n avg_chunk_size=sum(self.metrics['chunk_size']) / max(len(self.metrics['chunk_size']), 1),\n processing_time=sum(self.metrics['processing_time']),\n memory_usage=sum(self.metrics['memory_usage']) / max(len(self.metrics['memory_usage']), 1),\n error_count=len(self.metrics['errors']),\n strategy_distribution=dict(self.metrics['strategy'])\n )\n\n def reset(self):\n self.metrics.clear()\n self.start_time = None\n```\n\nThis implementation guide provides a comprehensive foundation for building robust, scalable chunking systems that can handle various document types and use cases while maintaining high quality and performance.","content_type":"text/markdown; charset=utf-8","language":"markdown","size":25813,"content_sha256":"b7064f3b36dd1792341d6574079582856da2f20f6e4eb46b19ad5bc9bfa3a4f9"},{"filename":"references/research.md","content":"# Key Research Papers and Findings\n\nThis document summarizes important research papers and findings related to chunking strategies for RAG systems.\n\n## Seminal Papers\n\n### \"Reconstructing Context: Evaluating Advanced Chunking Strategies for RAG\" (arXiv:2504.19754)\n\n**Key Findings**:\n- Page-level chunking achieved highest average accuracy (0.648) with lowest variance across different query types\n- Optimal chunk size varies significantly by document type and query complexity\n- Factoid queries perform better with smaller chunks (256-512 tokens)\n- Complex analytical queries benefit from larger chunks (1024+ tokens)\n\n**Methodology**:\n- Evaluated 7 different chunking strategies across multiple document types\n- Tested with both factoid and analytical queries\n- Measured end-to-end RAG performance\n\n**Practical Implications**:\n- Start with page-level chunking for general-purpose RAG systems\n- Adapt chunk size based on expected query patterns\n- Consider hybrid approaches for mixed query types\n\n### \"Lost in the Middle: How Language Models Use Long Contexts\"\n\n**Key Findings**:\n- Language models tend to pay more attention to information at the beginning and end of context\n- Information in the middle of long contexts is often ignored\n- Performance degradation is most severe for centrally located information\n\n**Practical Implications**:\n- Place most important information at chunk boundaries\n- Consider chunk overlap to ensure important context appears multiple times\n- Use ranking to prioritize relevant chunks for inclusion in context\n\n### \"Grounded Language Learning in a Simulated 3D World\"\n\n**Related Concepts**:\n- Importance of grounding text in visual/contextual information\n- Multi-modal learning approaches for better understanding\n\n**Relevance to Chunking**:\n- Supports contextual chunking approaches that preserve visual/contextual relationships\n- Validates importance of maintaining document structure and relationships\n\n## Industry Research\n\n### NVIDIA Research: \"Finding the Best Chunking Strategy for Accurate AI Responses\"\n\n**Key Findings**:\n- Page-level chunking outperformed sentence and paragraph-level approaches\n- Fixed-size chunking showed consistent but suboptimal performance\n- Semantic chunking provided improvements for complex documents\n\n**Technical Details**:\n- Tested chunk sizes from 128 to 2048 tokens\n- Evaluated across financial, technical, and legal documents\n- Measured both retrieval accuracy and generation quality\n\n**Recommendations**:\n- Use 512-1024 token chunks as starting point\n- Implement adaptive chunking based on document complexity\n- Consider page boundaries as natural chunk separators\n\n### Cohere Research: \"Effective Chunking Strategies for RAG\"\n\n**Key Findings**:\n- Recursive character splitting provides good balance of performance and simplicity\n- Document structure awareness improves retrieval by 15-20%\n- Overlap of 10-20% provides optimal context preservation\n\n**Methodology**:\n- Compared 12 chunking strategies across 6 document types\n- Measured retrieval precision, recall, and F1-score\n- Tested with both dense and sparse retrieval\n\n**Best Practices Identified**:\n- Start with recursive character splitting with 10-20% overlap\n- Preserve document structure (headings, lists, tables)\n- Customize chunk size based on embedding model context window\n\n### Anthropic: \"Contextual Retrieval\"\n\n**Key Innovation**:\n- Enhance each chunk with LLM-generated contextual information before embedding\n- Improves retrieval precision by 25-30% for complex documents\n- Particularly effective for technical and academic content\n\n**Implementation Approach**:\n1. Split document using traditional methods\n2. For each chunk, generate contextual information using LLM\n3. Prepend context to chunk before embedding\n4. Use hybrid search (dense + sparse) with weighted ranking\n\n**Trade-offs**:\n- Significant computational overhead (2-3x processing time)\n- Higher embedding storage requirements\n- Improved retrieval precision justifies cost for high-value applications\n\n## Algorithmic Advances\n\n### Semantic Chunking Algorithms\n\n#### \"Semantic Segmentation of Text Documents\"\n\n**Core Idea**: Use cosine similarity between consecutive sentence embeddings to identify natural boundaries.\n\n**Algorithm**:\n1. Split document into sentences\n2. Generate embeddings for each sentence\n3. Calculate similarity between consecutive sentences\n4. Create boundaries where similarity drops below threshold\n5. Merge short segments with neighbors\n\n**Performance**: 20-30% improvement in retrieval relevance over fixed-size chunking for technical documents.\n\n#### \"Hierarchical Semantic Chunking\"\n\n**Core Idea**: Multi-level semantic segmentation for document organization.\n\n**Algorithm**:\n1. Document-level semantic analysis\n2. Section-level boundary detection\n3. Paragraph-level segmentation\n4. Sentence-level refinement\n\n**Benefits**: Maintains document hierarchy while adapting to semantic structure.\n\n### Advanced Embedding Techniques\n\n#### \"Late Chunking: Contextual Chunk Embeddings\"\n\n**Core Innovation**: Generate embeddings for entire document first, then create chunk embeddings from token-level embeddings.\n\n**Advantages**:\n- Preserves global document context\n- Reduces context fragmentation\n- Better for documents with complex inter-relationships\n\n**Requirements**:\n- Long-context embedding models (8k+ tokens)\n- Significant computational resources\n- Specialized implementation\n\n#### \"Hierarchical Embedding Retrieval\"\n\n**Approach**: Create embeddings at multiple granularities (document, section, paragraph, sentence).\n\n**Implementation**:\n1. Generate embeddings at each level\n2. Store in hierarchical vector database\n3. Query at appropriate granularity based on information needs\n\n**Performance**: 15-25% improvement in precision for complex queries.\n\n## Evaluation Methodologies\n\n### Retrieval-Augmented Generation Assessment Frameworks\n\n#### RAGAS Framework\n\n**Metrics**:\n- **Faithfulness**: Consistency between generated answer and retrieved context\n- **Answer Relevancy**: Relevance of generated answer to the question\n- **Context Relevancy**: Relevance of retrieved context to the question\n- **Context Recall**: Coverage of relevant information in retrieved context\n\n**Evaluation Process**:\n1. Generate questions from document corpus\n2. Retrieve relevant chunks using different strategies\n3. Generate answers using retrieved chunks\n4. Evaluate using automated metrics and human judgment\n\n#### ARES Framework\n\n**Innovation**: Automated evaluation using synthetic questions and LLM-based assessment.\n\n**Key Features**:\n- Generates diverse question types (factoid, analytical, comparative)\n- Uses LLMs to evaluate answer quality\n- Provides scalable evaluation without human annotation\n\n### Benchmark Datasets\n\n#### Natural Questions (NQ)\n\n**Description**: Real user questions from Google Search with relevant Wikipedia passages.\n\n**Relevance**: Natural language queries with authentic relevance judgments.\n\n#### MS MARCO\n\n**Description**: Large-scale passage ranking dataset with real search queries.\n\n**Relevance**: High-quality relevance judgments for passage retrieval.\n\n#### HotpotQA\n\n**Description**: Multi-hop question answering requiring information from multiple documents.\n\n**Relevance**: Tests ability to retrieve and synthesize information from multiple chunks.\n\n## Domain-Specific Research\n\n### Medical Documents\n\n#### \"Optimal Chunking for Medical Question Answering\"\n\n**Key Findings**:\n- Medical terminology requires specialized handling\n- Section-based chunking (History, Diagnosis, Treatment) most effective\n- Preserving doctor-patient dialogue context crucial\n\n**Recommendations**:\n- Use medical-specific tokenizers\n- Preserve section headers and structure\n- Maintain temporal relationships in medical histories\n\n### Legal Documents\n\n#### \"Chunking Strategies for Legal Document Analysis\"\n\n**Key Findings**:\n- Legal citations and cross-references require special handling\n- Contract clause boundaries serve as natural chunk separators\n- Case law benefits from hierarchical chunking\n\n**Best Practices**:\n- Preserve legal citation structure\n- Use clause and section boundaries\n- Maintain context for legal definitions and references\n\n### Financial Documents\n\n#### \"SEC Filing Chunking for Financial Analysis\"\n\n**Key Findings**:\n- Table preservation critical for financial data\n- XBRL tagging provides natural segmentation\n- Risk factors sections benefit from specialized treatment\n\n**Approach**:\n- Preserve complete tables when possible\n- Use XBRL tags for structured data\n- Create specialized chunks for risk sections\n\n## Emerging Trends\n\n### Multi-Modal Chunking\n\n#### \"Integrating Text, Tables, and Images in RAG Systems\"\n\n**Innovation**: Unified chunking approach for mixed-modal content.\n\n**Approach**:\n- Extract and describe images using vision models\n- Preserve table structure and relationships\n- Create unified embeddings for mixed content\n\n**Results**: 35% improvement in complex document understanding.\n\n### Adaptive Chunking\n\n#### \"Machine Learning-Based Chunk Size Optimization\"\n\n**Core Idea**: Use ML models to predict optimal chunking parameters.\n\n**Features**:\n- Document length and complexity\n- Query type distribution\n- Embedding model characteristics\n- Performance requirements\n\n**Benefits**: Dynamic optimization based on use case and content.\n\n### Real-time Chunking\n\n#### \"Streaming Chunking for Live Document Processing\"\n\n**Innovation**: Process documents as they become available.\n\n**Techniques**:\n- Incremental boundary detection\n- Dynamic chunk size adjustment\n- Context preservation across chunks\n\n**Applications**: Live news feeds, social media analysis, meeting transcripts.\n\n## Implementation Challenges\n\n### Computational Efficiency\n\n#### \"Scalable Chunking for Large Document Collections\"\n\n**Challenges**:\n- Processing millions of documents efficiently\n- Memory usage optimization\n- Distributed processing requirements\n\n**Solutions**:\n- Batch processing with parallel execution\n- Streaming approaches for large documents\n- Distributed chunking with load balancing\n\n### Quality Assurance\n\n#### \"Evaluating Chunk Quality at Scale\"\n\n**Challenges**:\n- Automated quality assessment\n- Detecting poor chunk boundaries\n- Maintaining consistency across document types\n\n**Approaches**:\n- Heuristic-based quality metrics\n- LLM-based evaluation\n- Human-in-the-loop validation\n\n## Future Research Directions\n\n### Context-Aware Chunking\n\n**Open Questions**:\n- How to optimally preserve cross-chunk relationships?\n- Can we predict chunk quality without human evaluation?\n- What is the optimal balance between size and context?\n\n### Domain Adaptation\n\n**Research Areas**:\n- Automatic domain detection and adaptation\n- Transfer learning across domains\n- Zero-shot chunking for new document types\n\n### Evaluation Standards\n\n**Needs**:\n- Standardized evaluation benchmarks\n- Cross-paper comparison methodologies\n- Real-world performance metrics\n\n## Practical Recommendations Based on Research\n\n### Starting Points\n\n1. **For General RAG Systems**: Page-level or recursive character chunking with 512-1024 tokens and 10-20% overlap\n2. **For Technical Documents**: Structure-aware chunking with semantic boundary detection\n3. **For High-Value Applications**: Contextual retrieval with LLM-generated context\n\n### Evolution Strategy\n\n1. **Begin**: Simple fixed-size chunking (512 tokens)\n2. **Improve**: Add document structure awareness\n3. **Optimize**: Implement semantic boundaries\n4. **Advanced**: Consider contextual retrieval for critical use cases\n\n### Key Success Factors\n\n1. **Match strategy to document type and query patterns**\n2. **Preserve document structure when beneficial**\n3. **Use overlap to maintain context across boundaries**\n4. **Monitor both accuracy and computational costs**\n5. **Iterate based on specific use case requirements**\n\nThis research foundation provides evidence-based guidance for implementing effective chunking strategies across various domains and use cases.","content_type":"text/markdown; charset=utf-8","language":"markdown","size":11954,"content_sha256":"2b1fcac71a30a054b4afd93bb70842ee442ced4e0329924009629ec99ba153d9"},{"filename":"references/semantic-methods.md","content":"# Semantic and Contextual Chunking Methods\n\nThis document provides comprehensive coverage of semantic and contextual chunking approaches for advanced RAG systems.\n\n## Overview of Semantic Methods\n\nSemantic chunking methods use understanding of text meaning and relationships to create more meaningful chunks than simple size-based approaches.\n\n| Method | Approach | Best For | Complexity |\n|--------|----------|----------|------------|\n| Embedding-Based Similarity | Sentence embeddings to find boundaries | Thematic documents | High |\n| Topic Modeling | LDA/NMF to identify topic segments | Mixed-topic documents | Medium |\n| Named Entity Recognition | Entity-aware boundaries | Technical/medical docs | Medium |\n| Dependency Parsing | Syntactic relationships | Structured content | High |\n| Cross-Encoder Scoring | BERT-based boundary detection | High-precision needs | Very High |\n\n## 1. Embedding-Based Semantic Chunking\n\n### Core Concept\nUse sentence embeddings to identify semantic boundaries where topic shifts occur.\n\n### Advanced Implementation\n\n```python\nimport numpy as np\nfrom sentence_transformers import SentenceTransformer\nfrom sklearn.metrics.pairwise import cosine_similarity\nfrom sklearn.cluster import KMeans\nfrom typing import List, Dict, Tuple\nimport re\n\nclass AdvancedSemanticChunker:\n def __init__(self, model_name=\"all-MiniLM-L6-v2\",\n boundary_threshold=0.7,\n min_chunk_size=3,\n max_chunk_size=15,\n clustering_method=\"kmeans\"):\n self.model = SentenceTransformer(model_name)\n self.boundary_threshold = boundary_threshold\n self.min_chunk_size = min_chunk_size\n self.max_chunk_size = max_chunk_size\n self.clustering_method = clustering_method\n\n def multi_level_semantic_chunking(self, text):\n \"\"\"Multi-level semantic analysis for optimal chunking\"\"\"\n # Extract sentences\n sentences = self._extract_sentences(text)\n if len(sentences) \u003c= self.min_chunk_size:\n return [{\"text\": text, \"method\": \"too_short\", \"level\": 0}]\n\n # Generate embeddings\n embeddings = self.model.encode(sentences)\n\n # Level 1: Similarity-based boundaries\n similarity_boundaries = self._find_similarity_boundaries(embeddings)\n similarity_chunks = self._create_chunks_from_boundaries(\n sentences, similarity_boundaries, \"similarity\"\n )\n\n # Level 2: Clustering-based boundaries\n clustering_boundaries = self._find_clustering_boundaries(embeddings)\n clustering_chunks = self._create_chunks_from_boundaries(\n sentences, clustering_boundaries, \"clustering\"\n )\n\n # Level 3: Hybrid approach\n hybrid_boundaries = self._find_hybrid_boundaries(\n similarity_boundaries, clustering_boundaries, embeddings\n )\n hybrid_chunks = self._create_chunks_from_boundaries(\n sentences, hybrid_boundaries, \"hybrid\"\n )\n\n # Evaluate and select best approach\n approaches = [\n {\"chunks\": similarity_chunks, \"method\": \"similarity\"},\n {\"chunks\": clustering_chunks, \"method\": \"clustering\"},\n {\"chunks\": hybrid_chunks, \"method\": \"hybrid\"}\n ]\n\n best_approach = self._evaluate_approaches(approaches, sentences, embeddings)\n\n return best_approach[\"chunks\"]\n\n def _extract_sentences(self, text):\n \"\"\"Enhanced sentence extraction\"\"\"\n # Multiple sentence splitting approaches\n patterns = [\n r'(?\u003c=[.!?])\\s+(?=[A-Z])', # Standard sentence boundaries\n r'(?\u003c=[.!?]\\s)\\s*(?=[A-Z])', # Handle multiple spaces\n r'(?\u003c=[.!?])\\s+(?=[a-z])', # Handle lowercase starts\n ]\n\n sentences = []\n for pattern in patterns:\n potential_sentences = re.split(pattern, text)\n potential_sentences = [s.strip() for s in potential_sentences if s.strip()]\n\n if len(potential_sentences) > len(sentences):\n sentences = potential_sentences\n\n return sentences if sentences else [text]\n\n def _find_similarity_boundaries(self, embeddings):\n \"\"\"Find boundaries based on similarity drops\"\"\"\n boundaries = []\n\n for i in range(len(embeddings) - 1):\n # Calculate similarity between consecutive sentences\n similarity = cosine_similarity(\n embeddings[i].reshape(1, -1),\n embeddings[i + 1].reshape(1, -1)\n )[0][0]\n\n # Dynamic threshold based on local context\n local_threshold = self._calculate_local_threshold(embeddings, i)\n\n if similarity \u003c local_threshold:\n boundaries.append(i)\n\n return self._filter_boundaries(boundaries, len(embeddings))\n\n def _calculate_local_threshold(self, embeddings, index):\n \"\"\"Calculate dynamic threshold based on local similarity patterns\"\"\"\n window_size = min(5, index, len(embeddings) - index - 1)\n start_idx = max(0, index - window_size)\n end_idx = min(len(embeddings), index + window_size + 1)\n\n local_embeddings = embeddings[start_idx:end_idx]\n\n if len(local_embeddings) \u003c 2:\n return self.boundary_threshold\n\n # Calculate local similarity statistics\n similarities = []\n for i in range(len(local_embeddings) - 1):\n sim = cosine_similarity(\n local_embeddings[i].reshape(1, -1),\n local_embeddings[i + 1].reshape(1, -1)\n )[0][0]\n similarities.append(sim)\n\n mean_sim = np.mean(similarities)\n std_sim = np.std(similarities)\n\n # Threshold based on local statistics\n return max(0.3, mean_sim - std_sim * 0.5)\n\n def _find_clustering_boundaries(self, embeddings):\n \"\"\"Find boundaries using clustering approaches\"\"\"\n if self.clustering_method == \"kmeans\":\n return self._kmeans_boundaries(embeddings)\n elif self.clustering_method == \"hierarchical\":\n return self._hierarchical_boundaries(embeddings)\n else:\n return self._dbscan_boundaries(embeddings)\n\n def _kmeans_boundaries(self, embeddings):\n \"\"\"K-means clustering for boundary detection\"\"\"\n n_clusters = min(max(len(embeddings) // 5, 2), 10) # Adaptive cluster count\n\n kmeans = KMeans(n_clusters=n_clusters, random_state=42)\n labels = kmeans.fit_predict(embeddings)\n\n boundaries = []\n for i in range(len(labels) - 1):\n if labels[i] != labels[i + 1]:\n boundaries.append(i)\n\n return self._filter_boundaries(boundaries, len(embeddings))\n\n def _hierarchical_boundaries(self, embeddings):\n \"\"\"Hierarchical clustering for boundary detection\"\"\"\n from sklearn.cluster import AgglomerativeClustering\n\n n_clusters = min(max(len(embeddings) // 4, 2), 8)\n clustering = AgglomerativeClustering(n_clusters=n_clusters)\n labels = clustering.fit_predict(embeddings)\n\n boundaries = []\n for i in range(len(labels) - 1):\n if labels[i] != labels[i + 1]:\n boundaries.append(i)\n\n return self._filter_boundaries(boundaries, len(embeddings))\n\n def _dbscan_boundaries(self, embeddings):\n \"\"\"DBSCAN clustering for boundary detection\"\"\"\n from sklearn.cluster import DBSCAN\n\n # Adaptive eps based on data characteristics\n distances = []\n for i in range(min(10, len(embeddings))):\n for j in range(i + 1, min(i + 10, len(embeddings))):\n dist = cosine_similarity(\n embeddings[i].reshape(1, -1),\n embeddings[j].reshape(1, -1)\n )[0][0]\n distances.append(dist)\n\n eps = np.percentile(distances, 25) if distances else 0.5\n\n clustering = DBSCAN(eps=eps, min_samples=2, metric=\"cosine\")\n labels = clustering.fit_predict(embeddings)\n\n boundaries = []\n for i in range(len(labels) - 1):\n if labels[i] != labels[i + 1] and labels[i] != -1 and labels[i + 1] != -1:\n boundaries.append(i)\n\n return self._filter_boundaries(boundaries, len(embeddings))\n\n def _find_hybrid_boundaries(self, similarity_boundaries, clustering_boundaries, embeddings):\n \"\"\"Combine similarity and clustering approaches\"\"\"\n # Combine boundaries from both methods\n all_boundaries = set(similarity_boundaries) | set(clustering_boundaries)\n combined_boundaries = sorted(list(all_boundaries))\n\n # Refine using consensus scoring\n refined_boundaries = []\n for boundary in combined_boundaries:\n score = self._calculate_boundary_score(boundary, embeddings)\n if score > 0.5: # Threshold for hybrid approach\n refined_boundaries.append(boundary)\n\n return self._filter_boundaries(refined_boundaries, len(embeddings))\n\n def _calculate_boundary_score(self, boundary_idx, embeddings):\n \"\"\"Calculate confidence score for a boundary\"\"\"\n if boundary_idx >= len(embeddings) - 1:\n return 0.0\n\n # Similarity drop score\n similarity = cosine_similarity(\n embeddings[boundary_idx].reshape(1, -1),\n embeddings[boundary_idx + 1].reshape(1, -1)\n )[0][0]\n\n similarity_score = 1.0 - similarity # Lower similarity = higher boundary score\n\n # Local variance score\n window_size = min(3, boundary_idx, len(embeddings) - boundary_idx - 1)\n start_idx = max(0, boundary_idx - window_size)\n end_idx = min(len(embeddings), boundary_idx + window_size + 1)\n\n window_embeddings = embeddings[start_idx:end_idx]\n if len(window_embeddings) > 1:\n similarities = []\n for i in range(len(window_embeddings) - 1):\n sim = cosine_similarity(\n window_embeddings[i].reshape(1, -1),\n window_embeddings[i + 1].reshape(1, -1)\n )[0][0]\n similarities.append(sim)\n\n variance_score = np.var(similarities)\n else:\n variance_score = 0.0\n\n # Combined score\n return (similarity_score * 0.7 + variance_score * 0.3)\n\n def _filter_boundaries(self, boundaries, total_sentences):\n \"\"\"Filter boundaries to meet size constraints\"\"\"\n if not boundaries:\n return boundaries\n\n filtered_boundaries = []\n last_boundary = -1\n\n for boundary in boundaries:\n chunk_size = boundary - last_boundary\n\n # Check if chunk meets minimum size\n if chunk_size >= self.min_chunk_size:\n # Check if next chunk won't be too small\n remaining_sentences = total_sentences - boundary - 1\n if (remaining_sentences >= self.min_chunk_size or\n boundary == total_sentences - 1):\n filtered_boundaries.append(boundary)\n last_boundary = boundary\n else:\n # Chunk too small, try to extend it\n next_boundary = boundary + (self.min_chunk_size - chunk_size)\n if next_boundary \u003c total_sentences:\n filtered_boundaries.append(next_boundary)\n last_boundary = next_boundary\n\n return filtered_boundaries\n\n def _create_chunks_from_boundaries(self, sentences, boundaries, method):\n \"\"\"Create chunks using identified boundaries\"\"\"\n chunks = []\n start_idx = 0\n\n for boundary in boundaries:\n if boundary > start_idx:\n chunk_sentences = sentences[start_idx:boundary + 1]\n chunk_text = \" \".join(chunk_sentences)\n\n chunks.append({\n \"text\": chunk_text,\n \"sentence_count\": len(chunk_sentences),\n \"start_sentence\": start_idx,\n \"end_sentence\": boundary,\n \"method\": method\n })\n\n start_idx = boundary + 1\n\n # Add remaining sentences\n if start_idx \u003c len(sentences):\n chunk_sentences = sentences[start_idx:]\n chunk_text = \" \".join(chunk_sentences)\n\n chunks.append({\n \"text\": chunk_text,\n \"sentence_count\": len(chunk_sentences),\n \"start_sentence\": start_idx,\n \"end_sentence\": len(sentences) - 1,\n \"method\": method\n })\n\n return chunks\n\n def _evaluate_approaches(self, approaches, sentences, embeddings):\n \"\"\"Evaluate different chunking approaches\"\"\"\n best_approach = approaches[0]\n best_score = 0.0\n\n for approach in approaches:\n score = self._calculate_approach_score(\n approach[\"chunks\"], sentences, embeddings\n )\n approach[\"score\"] = score\n\n if score > best_score:\n best_score = score\n best_approach = approach\n\n return best_approach\n\n def _calculate_approach_score(self, chunks, sentences, embeddings):\n \"\"\"Calculate quality score for a chunking approach\"\"\"\n if not chunks:\n return 0.0\n\n # Coherence score (average intra-chunk similarity)\n coherence_scores = []\n for chunk in chunks:\n start_idx = chunk[\"start_sentence\"]\n end_idx = chunk[\"end_sentence\"]\n\n if end_idx > start_idx:\n chunk_embeddings = embeddings[start_idx:end_idx + 1]\n similarities = []\n\n for i in range(len(chunk_embeddings) - 1):\n sim = cosine_similarity(\n chunk_embeddings[i].reshape(1, -1),\n chunk_embeddings[i + 1].reshape(1, -1)\n )[0][0]\n similarities.append(sim)\n\n coherence_scores.append(np.mean(similarities) if similarities else 0.0)\n\n coherence_score = np.mean(coherence_scores) if coherence_scores else 0.0\n\n # Boundary score (low inter-chunk similarity)\n boundary_scores = []\n for i in range(len(chunks) - 1):\n current_end = chunks[i][\"end_sentence\"]\n next_start = chunks[i + 1][\"start_sentence\"]\n\n if current_end \u003c len(embeddings) and next_start \u003c len(embeddings):\n similarity = cosine_similarity(\n embeddings[current_end].reshape(1, -1),\n embeddings[next_start].reshape(1, -1)\n )[0][0]\n boundary_scores.append(1.0 - similarity) # Lower similarity = better boundary\n\n boundary_score = np.mean(boundary_scores) if boundary_scores else 0.0\n\n # Size appropriateness score\n size_scores = []\n for chunk in chunks:\n sentence_count = chunk[\"sentence_count\"]\n if self.min_chunk_size \u003c= sentence_count \u003c= self.max_chunk_size:\n size_scores.append(1.0)\n else:\n # Penalty for size violations\n deviation = min(\n abs(sentence_count - self.min_chunk_size),\n abs(sentence_count - self.max_chunk_size)\n )\n size_scores.append(max(0.0, 1.0 - deviation / 10.0))\n\n size_score = np.mean(size_scores) if size_scores else 0.0\n\n # Combined score\n return (\n coherence_score * 0.4 +\n boundary_score * 0.4 +\n size_score * 0.2\n )\n```\n\n## 2. Contextual Retrieval\n\n### Core Concept\nEnhance each chunk with LLM-generated contextual information to improve retrieval and understanding.\n\n### Implementation\n\n```python\nimport openai\nfrom typing import List, Dict, Optional\nimport json\nimport asyncio\n\nclass ContextualRetriever:\n def __init__(self, api_key, model=\"gpt-3.5-turbo\", max_context_length=200):\n self.client = openai.OpenAI(api_key=api_key)\n self.model = model\n self.max_context_length = max_context_length\n\n def contextual_chunking(self, text, base_chunker):\n \"\"\"Add contextual information to chunks\"\"\"\n # First, create base chunks\n base_chunks = base_chunker.chunk(text)\n\n # Generate context for each chunk\n contextualized_chunks = []\n total_chunks = len(base_chunks)\n\n for i, chunk in enumerate(base_chunks):\n print(f\"Processing chunk {i + 1}/{total_chunks}\")\n\n # Generate contextual information\n context = self._generate_context(chunk, text, i, total_chunks)\n\n # Create contextualized chunk\n contextualized_chunk = {\n \"original_text\": chunk,\n \"context\": context,\n \"contextualized_text\": f\"Context: {context}\\n\\nContent: {chunk}\",\n \"chunk_index\": i,\n \"total_chunks\": total_chunks,\n \"method\": \"contextual_retrieval\"\n }\n\n contextualized_chunks.append(contextualized_chunk)\n\n return contextualized_chunks\n\n def _generate_context(self, chunk, full_document, chunk_index, total_chunks):\n \"\"\"Generate contextual information for a chunk\"\"\"\n # Create a comprehensive prompt for context generation\n prompt = self._create_context_prompt(\n chunk, full_document, chunk_index, total_chunks\n )\n\n try:\n response = self.client.chat.completions.create(\n model=self.model,\n messages=[\n {\"role\": \"system\", \"content\": \"You are an expert at providing contextual information for document chunks. Generate concise, relevant context that helps understand the chunk's place in the larger document.\"},\n {\"role\": \"user\", \"content\": prompt}\n ],\n max_tokens=300,\n temperature=0.3\n )\n\n return response.choices[0].message.content.strip()\n\n except Exception as e:\n print(f\"Error generating context: {e}\")\n return self._generate_fallback_context(chunk, chunk_index, total_chunks)\n\n def _create_context_prompt(self, chunk, full_document, chunk_index, total_chunks):\n \"\"\"Create prompt for context generation\"\"\"\n # Get surrounding context\n surrounding_context = self._get_surrounding_context(\n full_document, chunk, chunk_index, total_chunks\n )\n\n prompt = f\"\"\"\nI need you to provide brief contextual information for a chunk from a larger document.\n\nDocument position: Chunk {chunk_index + 1} of {total_chunks}\n\nSurrounding context:\n{surrounding_context}\n\nCurrent chunk:\n{chunk}\n\nPlease provide a concise context (maximum {self.max_context_length} characters) that:\n1. Explains where this chunk fits in the overall document\n2. Mentions key topics or themes immediately before/after\n3. Helps understand the chunk's purpose and relevance\n4. Is written in a clear, informative style\n\nContext:\n\"\"\"\n return prompt\n\n def _get_surrounding_context(self, full_document, chunk, chunk_index, total_chunks):\n \"\"\"Extract context from surrounding parts of the document\"\"\"\n # Find chunk position in document\n chunk_start = full_document.find(chunk)\n if chunk_start == -1:\n return \"Context not available\"\n\n # Extract surrounding text\n context_window = 500 # characters\n start_pos = max(0, chunk_start - context_window)\n end_pos = min(len(full_document), chunk_start + len(chunk) + context_window)\n\n surrounding_text = full_document[start_pos:end_pos]\n\n # Highlight the chunk position\n relative_start = chunk_start - start_pos\n relative_end = relative_start + len(chunk)\n\n before_chunk = surrounding_text[:relative_start]\n after_chunk = surrounding_text[relative_end:]\n\n context_parts = []\n if before_chunk.strip():\n context_parts.append(f\"Before: {before_chunk.strip()[-100:]}...\")\n if after_chunk.strip():\n context_parts.append(f\"After: ...{after_chunk.strip()[:100]}\")\n\n return \" | \".join(context_parts)\n\n def _generate_fallback_context(self, chunk, chunk_index, total_chunks):\n \"\"\"Generate simple fallback context\"\"\"\n return f\"This is chunk {chunk_index + 1} of {total_chunks} in the document.\"\n\n async def async_contextual_chunking(self, text, base_chunker, max_concurrent=5):\n \"\"\"Asynchronous contextual chunking for better performance\"\"\"\n # Create base chunks first\n base_chunks = base_chunker.chunk(text)\n total_chunks = len(base_chunks)\n\n # Create semaphore to limit concurrent API calls\n semaphore = asyncio.Semaphore(max_concurrent)\n\n async def process_chunk(chunk, index):\n async with semaphore:\n return await self._async_generate_context(chunk, text, index, total_chunks)\n\n # Process all chunks concurrently\n tasks = [\n process_chunk(chunk, i) for i, chunk in enumerate(base_chunks)\n ]\n\n contextualized_chunks = await asyncio.gather(*tasks)\n\n # Add chunk information\n for i, chunk in enumerate(base_chunks):\n contextualized_chunks[i][\"original_text\"] = chunk\n contextualized_chunks[i][\"chunk_index\"] = i\n contextualized_chunks[i][\"total_chunks\"] = total_chunks\n contextualized_chunks[i][\"method\"] = \"async_contextual_retrieval\"\n\n return contextualized_chunks\n\n async def _async_generate_context(self, chunk, full_document, chunk_index, total_chunks):\n \"\"\"Asynchronous context generation\"\"\"\n surrounding_context = self._get_surrounding_context(\n full_document, chunk, chunk_index, total_chunks\n )\n\n prompt = self._create_context_prompt(\n chunk, full_document, chunk_index, total_chunks\n )\n\n try:\n # Note: This would require an async OpenAI client\n # For now, we'll use the synchronous version\n context = self._generate_context(chunk, full_document, chunk_index, total_chunks)\n\n return {\n \"context\": context,\n \"contextualized_text\": f\"Context: {context}\\n\\nContent: {chunk}\"\n }\n\n except Exception as e:\n print(f\"Error in async context generation: {e}\")\n fallback_context = self._generate_fallback_context(chunk, chunk_index, total_chunks)\n return {\n \"context\": fallback_context,\n \"contextualized_text\": f\"Context: {fallback_context}\\n\\nContent: {chunk}\"\n }\n\n def hierarchical_contextual_chunking(self, text, base_chunker):\n \"\"\"Hierarchical contextual chunking with multiple context levels\"\"\"\n # Create base chunks\n base_chunks = base_chunker.chunk(text)\n\n # Level 1: Document-level context\n document_summary = self._generate_document_summary(text)\n\n # Level 2: Section-level context\n section_contexts = self._generate_section_contexts(text, base_chunks)\n\n # Level 3: Local context for each chunk\n contextualized_chunks = []\n for i, chunk in enumerate(base_chunks):\n local_context = self._generate_context(chunk, text, i, len(base_chunks))\n\n # Combine all context levels\n combined_context = f\"\"\"\nDocument Overview: {document_summary}\n\nSection Context: {section_contexts.get(i, \"Section context not available\")}\n\nLocal Context: {local_context}\n\"\"\"\n\n contextualized_chunk = {\n \"original_text\": chunk,\n \"document_context\": document_summary,\n \"section_context\": section_contexts.get(i, \"\"),\n \"local_context\": local_context,\n \"combined_context\": combined_context.strip(),\n \"contextualized_text\": f\"Context: {combined_context.strip()}\\n\\nContent: {chunk}\",\n \"chunk_index\": i,\n \"method\": \"hierarchical_contextual\"\n }\n\n contextualized_chunks.append(contextualized_chunk)\n\n return contextualized_chunks\n\n def _generate_document_summary(self, text):\n \"\"\"Generate a summary of the entire document\"\"\"\n try:\n prompt = f\"\"\"\nPlease provide a brief summary (maximum 100 words) of this document:\n\n{text[:1000]}...\n\"\"\"\n response = self.client.chat.completions.create(\n model=self.model,\n messages=[\n {\"role\": \"system\", \"content\": \"You are an expert at summarizing documents concisely.\"},\n {\"role\": \"user\", \"content\": prompt}\n ],\n max_tokens=150,\n temperature=0.3\n )\n\n return response.choices[0].message.content.strip()\n\n except Exception as e:\n print(f\"Error generating document summary: {e}\")\n return \"Document summary not available\"\n\n def _generate_section_contexts(self, text, chunks):\n \"\"\"Generate context for different sections of the document\"\"\"\n section_contexts = {}\n\n # Simple section detection based on position\n total_chunks = len(chunks)\n sections = 3 # Divide document into 3 sections\n\n for i, chunk in enumerate(chunks):\n section_number = min(i // (total_chunks // sections), sections - 1)\n\n if section_number not in section_contexts:\n section_contexts[section_number] = self._generate_section_context(\n text, section_number, sections\n )\n\n return {i: section_contexts.get(i // (total_chunks // sections), \"\")\n for i in range(total_chunks)}\n\n def _generate_section_context(self, text, section_number, total_sections):\n \"\"\"Generate context for a specific section\"\"\"\n try:\n section_size = len(text) // total_sections\n start_pos = section_number * section_size\n end_pos = min((section_number + 1) * section_size, len(text))\n\n section_text = text[start_pos:end_pos]\n\n prompt = f\"\"\"\nProvide brief context for section {section_number + 1} of {total_sections} in this document:\n\nSection text:\n{section_text[:500]}...\n\nContext (maximum 50 words):\n\"\"\"\n response = self.client.chat.completions.create(\n model=self.model,\n messages=[\n {\"role\": \"system\", \"content\": \"You are an expert at providing document section context.\"},\n {\"role\": \"user\", \"content\": prompt}\n ],\n max_tokens=100,\n temperature=0.3\n )\n\n return response.choices[0].message.content.strip()\n\n except Exception as e:\n print(f\"Error generating section context: {e}\")\n return f\"This is section {section_number + 1} of the document\"\n```\n\n## 3. Late Chunking\n\n### Core Concept\nGenerate embeddings for the entire document first, then create chunk embeddings from token-level embeddings.\n\n### Implementation\n\n```python\nimport torch\nimport numpy as np\nfrom transformers import AutoTokenizer, AutoModel\nfrom typing import List, Dict, Tuple\nimport logging\n\nclass LateChunker:\n def __init__(self, model_name=\"sentence-transformers/all-MiniLM-L6-v2\",\n device=\"cpu\", chunk_size=512):\n self.device = device\n self.chunk_size = chunk_size\n\n # Load tokenizer and model\n self.tokenizer = AutoTokenizer.from_pretrained(model_name)\n self.model = AutoModel.from_pretrained(model_name)\n self.model.to(device)\n self.model.eval()\n\n def late_chunk_embedding(self, text, chunk_sizes=None):\n \"\"\"Generate late chunk embeddings\"\"\"\n if chunk_sizes is None:\n chunk_sizes = [256, 512, 1024] # Multiple chunk sizes to try\n\n results = {}\n\n for chunk_size in chunk_sizes:\n # Tokenize entire document\n encoded = self.tokenizer(\n text,\n return_tensors=\"pt\",\n truncation=False,\n padding=False\n ).to(self.device)\n\n input_ids = encoded[\"input_ids\"][0]\n attention_mask = encoded[\"attention_mask\"][0]\n\n # Generate embeddings for all tokens\n with torch.no_grad():\n outputs = self.model(**encoded, output_hidden_states=True)\n # Use last hidden state\n token_embeddings = outputs.last_hidden_state[0] # Shape: [seq_len, hidden_dim]\n\n # Create chunks from token embeddings\n chunk_embeddings = self._create_chunk_embeddings(\n token_embeddings, attention_mask, chunk_size, input_ids\n )\n\n # Create chunk texts\n chunk_texts = self._create_chunk_texts(input_ids, chunk_size)\n\n results[chunk_size] = {\n \"chunk_embeddings\": chunk_embeddings,\n \"chunk_texts\": chunk_texts,\n \"chunk_size\": chunk_size,\n \"method\": \"late_chunking\"\n }\n\n return results\n\n def _create_chunk_embeddings(self, token_embeddings, attention_mask, chunk_size, input_ids):\n \"\"\"Create chunk embeddings from token embeddings\"\"\"\n chunk_embeddings = []\n valid_token_indices = torch.where(attention_mask == 1)[0]\n\n for i in range(0, len(valid_token_indices), chunk_size):\n end_idx = min(i + chunk_size, len(valid_token_indices))\n chunk_token_indices = valid_token_indices[i:end_idx]\n\n # Get embeddings for tokens in this chunk\n chunk_token_embeddings = token_embeddings[chunk_token_indices]\n\n # Pool token embeddings to create chunk embedding\n # Using mean pooling\n chunk_embedding = torch.mean(chunk_token_embeddings, dim=0)\n chunk_embeddings.append(chunk_embedding.cpu().numpy())\n\n return chunk_embeddings\n\n def _create_chunk_texts(self, input_ids, chunk_size):\n \"\"\"Create text for each chunk\"\"\"\n chunk_texts = []\n total_tokens = input_ids.shape[0]\n\n for i in range(0, total_tokens, chunk_size):\n end_idx = min(i + chunk_size, total_tokens)\n chunk_ids = input_ids[i:end_idx]\n chunk_text = self.tokenizer.decode(chunk_ids, skip_special_tokens=True)\n chunk_texts.append(chunk_text)\n\n return chunk_texts\n\n def adaptive_late_chunking(self, text, complexity_analyzer=None):\n \"\"\"Adaptive late chunking based on document complexity\"\"\"\n # Analyze document complexity\n if complexity_analyzer:\n complexity = complexity_analyzer.analyze_complexity(text)\n else:\n complexity = self._simple_complexity_analysis(text)\n\n # Adjust chunk size based on complexity\n if complexity \u003c 0.3: # Simple document\n chunk_sizes = [512, 1024]\n elif complexity \u003c 0.7: # Medium complexity\n chunk_sizes = [256, 512, 1024]\n else: # Complex document\n chunk_sizes = [128, 256, 512]\n\n results = self.late_chunk_embedding(text, chunk_sizes)\n\n # Evaluate and select best chunk size\n best_chunk_size = self._evaluate_chunk_sizes(results, complexity)\n\n return {\n \"best_results\": results[best_chunk_size],\n \"all_results\": results,\n \"selected_chunk_size\": best_chunk_size,\n \"complexity\": complexity,\n \"method\": \"adaptive_late_chunking\"\n }\n\n def _simple_complexity_analysis(self, text):\n \"\"\"Simple complexity analysis\"\"\"\n # Factors: average sentence length, vocabulary diversity, punctuation complexity\n sentences = text.split('.')\n if not sentences:\n return 0.0\n\n avg_sentence_length = sum(len(s.split()) for s in sentences) / len(sentences)\n unique_words = len(set(text.lower().split()))\n total_words = len(text.split())\n vocab_diversity = unique_words / total_words if total_words > 0 else 0\n\n # Normalize and combine\n length_score = min(avg_sentence_length / 20, 1.0) # Normalize to 0-1\n diversity_score = vocab_diversity\n\n complexity = (length_score + diversity_score) / 2\n return complexity\n\n def _evaluate_chunk_sizes(self, results, complexity):\n \"\"\"Evaluate different chunk sizes and select the best one\"\"\"\n best_chunk_size = list(results.keys())[0]\n best_score = 0.0\n\n for chunk_size, result in results.items():\n score = self._calculate_chunking_score(result, complexity)\n result[\"score\"] = score\n\n if score > best_score:\n best_score = score\n best_chunk_size = chunk_size\n\n return best_chunk_size\n\n def _calculate_chunking_score(self, result, complexity):\n \"\"\"Calculate quality score for a chunking result\"\"\"\n chunk_texts = result[\"chunk_texts\"]\n chunk_embeddings = result[\"chunk_embeddings\"]\n\n if not chunk_texts or not chunk_embeddings:\n return 0.0\n\n # Factors to consider:\n # 1. Number of chunks (moderate number is better)\n # 2. Chunk size consistency\n # 3. Content preservation (estimated)\n\n num_chunks = len(chunk_texts)\n chunk_lengths = [len(text.split()) for text in chunk_texts]\n\n # Score based on optimal number of chunks (5-15 is ideal)\n if 5 \u003c= num_chunks \u003c= 15:\n chunk_count_score = 1.0\n elif num_chunks \u003c 5:\n chunk_count_score = num_chunks / 5.0\n else:\n chunk_count_score = max(0.0, 1.0 - (num_chunks - 15) / 20.0)\n\n # Score based on size consistency\n if chunk_lengths:\n mean_length = np.mean(chunk_lengths)\n std_length = np.std(chunk_lengths)\n consistency_score = max(0.0, 1.0 - (std_length / mean_length))\n else:\n consistency_score = 0.0\n\n # Adjust score based on document complexity\n complexity_adjustment = 0.5 + complexity * 0.5\n\n total_score = (\n chunk_count_score * 0.4 +\n consistency_score * 0.3 +\n complexity_adjustment * 0.3\n )\n\n return total_score\n\n def contextual_late_chunking(self, text, context_generator=None):\n \"\"\"Combine late chunking with contextual information\"\"\"\n # Generate late chunk embeddings\n late_results = self.late_chunk_embedding(text)\n\n # Use best chunk size (default to 512)\n best_chunk_size = 512\n if best_chunk_size in late_results:\n result = late_results[best_chunk_size]\n else:\n result = list(late_results.values())[0]\n\n chunk_texts = result[\"chunk_texts\"]\n chunk_embeddings = result[\"chunk_embeddings\"]\n\n # Add contextual information if available\n if context_generator:\n contextualized_chunks = []\n for i, chunk_text in enumerate(chunk_texts):\n context = context_generator.generate_context(\n chunk_text, text, i, len(chunk_texts)\n )\n\n contextualized_chunk = {\n \"text\": chunk_text,\n \"embedding\": chunk_embeddings[i],\n \"context\": context,\n \"contextualized_text\": f\"Context: {context}\\n\\nContent: {chunk_text}\",\n \"chunk_index\": i,\n \"method\": \"contextual_late_chunking\"\n }\n contextualized_chunks.append(contextualized_chunk)\n\n return {\n \"chunks\": contextualized_chunks,\n \"chunk_size\": best_chunk_size,\n \"method\": \"contextual_late_chunking\"\n }\n\n # Return without context\n chunks = []\n for i, (chunk_text, embedding) in enumerate(zip(chunk_texts, chunk_embeddings)):\n chunks.append({\n \"text\": chunk_text,\n \"embedding\": embedding,\n \"chunk_index\": i,\n \"method\": \"late_chunking\"\n })\n\n return {\n \"chunks\": chunks,\n \"chunk_size\": best_chunk_size,\n \"method\": \"late_chunking\"\n }\n\n def semantic_late_chunking(self, text, semantic_model=None):\n \"\"\"Combine late chunking with semantic boundary detection\"\"\"\n # Generate late chunks\n late_results = self.late_chunk_embedding(text)\n best_chunk_size = 512\n\n if best_chunk_size in late_results:\n result = late_results[best_chunk_size]\n else:\n result = list(late_results.values())[0]\n\n chunk_texts = result[\"chunk_texts\"]\n chunk_embeddings = result[\"chunk_embeddings\"]\n\n # Use semantic model to analyze chunk boundaries\n if semantic_model:\n # Analyze semantic coherence between chunks\n semantic_chunks = self._analyze_semantic_boundaries(\n chunk_texts, chunk_embeddings, semantic_model\n )\n else:\n # Simple semantic analysis using the embeddings we already have\n semantic_chunks = self._simple_semantic_analysis(\n chunk_texts, chunk_embeddings\n )\n\n return {\n \"chunks\": semantic_chunks,\n \"chunk_size\": best_chunk_size,\n \"method\": \"semantic_late_chunking\"\n }\n\n def _analyze_semantic_boundaries(self, chunk_texts, chunk_embeddings, semantic_model):\n \"\"\"Analyze semantic boundaries using external semantic model\"\"\"\n # This would use a semantic model to analyze boundaries\n # For now, return chunks with similarity information\n chunks = []\n\n for i, (text, embedding) in enumerate(zip(chunk_texts, chunk_embeddings)):\n chunk_data = {\n \"text\": text,\n \"embedding\": embedding,\n \"chunk_index\": i,\n \"method\": \"semantic_late_chunking\"\n }\n\n # Calculate similarity with adjacent chunks\n if i > 0:\n prev_embedding = chunk_embeddings[i - 1]\n similarity = np.dot(embedding, prev_embedding) / (\n np.linalg.norm(embedding) * np.linalg.norm(prev_embedding)\n )\n chunk_data[\"similarity_with_previous\"] = similarity\n\n if i \u003c len(chunk_texts) - 1:\n next_embedding = chunk_embeddings[i + 1]\n similarity = np.dot(embedding, next_embedding) / (\n np.linalg.norm(embedding) * np.linalg.norm(next_embedding)\n )\n chunk_data[\"similarity_with_next\"] = similarity\n\n chunks.append(chunk_data)\n\n return chunks\n\n def _simple_semantic_analysis(self, chunk_texts, chunk_embeddings):\n \"\"\"Simple semantic analysis using available embeddings\"\"\"\n chunks = []\n\n for i, (text, embedding) in enumerate(zip(chunk_texts, chunk_embeddings)):\n chunk_data = {\n \"text\": text,\n \"embedding\": embedding,\n \"chunk_index\": i,\n \"method\": \"simple_semantic_late_chunking\"\n }\n\n # Calculate similarity with adjacent chunks\n if i > 0:\n prev_embedding = chunk_embeddings[i - 1]\n similarity = np.dot(embedding, prev_embedding) / (\n np.linalg.norm(embedding) * np.linalg.norm(prev_embedding)\n )\n chunk_data[\"similarity_with_previous\"] = similarity\n\n if i \u003c len(chunk_texts) - 1:\n next_embedding = chunk_embeddings[i + 1]\n similarity = np.dot(embedding, next_embedding) / (\n np.linalg.norm(embedding) * np.linalg.norm(next_embedding)\n )\n chunk_data[\"similarity_with_next\"] = similarity\n\n chunks.append(chunk_data)\n\n return chunks\n```\n\n## 4. Usage Examples and Integration\n\n### Complete Semantic Chunking Pipeline\n\n```python\nclass SemanticChunkingPipeline:\n def __init__(self, config):\n self.config = config\n self.semantic_chunker = AdvancedSemanticChunker(\n model_name=config.get(\"semantic_model\", \"all-MiniLM-L6-v2\"),\n boundary_threshold=config.get(\"boundary_threshold\", 0.7)\n )\n self.contextual_retriever = ContextualRetriever(\n api_key=config.get(\"openai_api_key\"),\n model=config.get(\"context_model\", \"gpt-3.5-turbo\")\n )\n self.late_chunker = LateChunker(\n model_name=config.get(\"embedding_model\", \"sentence-transformers/all-MiniLM-L6-v2\")\n )\n\n def process_document(self, text, method=\"hybrid\"):\n \"\"\"Process document using specified semantic method\"\"\"\n if method == \"semantic\":\n return self.semantic_chunker.multi_level_semantic_chunking(text)\n\n elif method == \"contextual\":\n base_chunker = FixedSizeChunker(chunk_size=512, chunk_overlap=50)\n return self.contextual_retriever.contextual_chunking(text, base_chunker)\n\n elif method == \"late\":\n return self.late_chunker.adaptive_late_chunking(text)\n\n elif method == \"hybrid\":\n # Combine multiple approaches\n semantic_chunks = self.semantic_chunker.multi_level_semantic_chunking(text)\n\n # Add contextual information to semantic chunks\n contextualized_chunks = []\n for i, chunk in enumerate(semantic_chunks):\n context = self._generate_simple_context(chunk[\"text\"], text, i)\n\n contextualized_chunk = chunk.copy()\n contextualized_chunk.update({\n \"context\": context,\n \"contextualized_text\": f\"Context: {context}\\n\\nContent: {chunk['text']}\",\n \"method\": \"hybrid_semantic_contextual\"\n })\n contextualized_chunks.append(contextualized_chunk)\n\n return contextualized_chunks\n\n else:\n raise ValueError(f\"Unknown method: {method}\")\n\n def _generate_simple_context(self, chunk, full_text, chunk_index):\n \"\"\"Generate simple context without API calls\"\"\"\n # Find chunk position\n chunk_start = full_text.find(chunk)\n if chunk_start == -1:\n return f\"Chunk {chunk_index + 1}\"\n\n # Extract surrounding text\n context_window = 200\n start_pos = max(0, chunk_start - context_window)\n end_pos = min(len(full_text), chunk_start + len(chunk) + context_window)\n\n surrounding = full_text[start_pos:end_pos]\n return f\"Part of larger document (position {chunk_start}-{chunk_start + len(chunk)})\"\n\n def evaluate_chunking(self, chunks, text):\n \"\"\"Evaluate chunking quality\"\"\"\n if not chunks:\n return {\"overall_score\": 0.0, \"metrics\": {}}\n\n # Generate embeddings for evaluation\n sentences = self._extract_sentences(text)\n sentence_embeddings = self.semantic_chunker.model.encode(sentences)\n\n # Calculate metrics\n metrics = {\n \"coherence\": self._calculate_coherence(chunks, sentence_embeddings),\n \"coverage\": self._calculate_coverage(chunks, text),\n \"size_distribution\": self._analyze_size_distribution(chunks),\n \"boundary_quality\": self._assess_boundary_quality(chunks)\n }\n\n # Calculate overall score\n weights = {\"coherence\": 0.3, \"coverage\": 0.3, \"size_distribution\": 0.2, \"boundary_quality\": 0.2}\n overall_score = sum(metrics[metric] * weights[metric] for metric in weights)\n\n return {\n \"overall_score\": overall_score,\n \"metrics\": metrics,\n \"chunks_count\": len(chunks)\n }\n\n def _extract_sentences(self, text):\n \"\"\"Extract sentences for evaluation\"\"\"\n import re\n sentences = re.split(r'[.!?]+', text)\n return [s.strip() for s in sentences if s.strip()]\n\n def _calculate_coherence(self, chunks, sentence_embeddings):\n \"\"\"Calculate semantic coherence of chunks\"\"\"\n coherence_scores = []\n\n for chunk in chunks:\n chunk_text = chunk.get(\"text\", chunk.get(\"original_text\", \"\"))\n chunk_sentences = self._extract_sentences(chunk_text)\n\n if len(chunk_sentences) > 1:\n # Find sentence indices\n sentence_indices = []\n for sentence in chunk_sentences:\n # Simple matching - could be improved\n for i, sent in enumerate(self._extract_sentences(\" \".join([s for s in chunk_sentences]))):\n if sentence.strip() == sent.strip():\n sentence_indices.append(i)\n break\n\n if len(sentence_indices) > 1:\n # Calculate average similarity between consecutive sentences\n similarities = []\n for i in range(len(sentence_indices) - 1):\n idx1, idx2 = sentence_indices[i], sentence_indices[i + 1]\n if idx1 \u003c len(sentence_embeddings) and idx2 \u003c len(sentence_embeddings):\n sim = cosine_similarity(\n sentence_embeddings[idx1].reshape(1, -1),\n sentence_embeddings[idx2].reshape(1, -1)\n )[0][0]\n similarities.append(sim)\n\n if similarities:\n coherence_scores.append(np.mean(similarities))\n\n return np.mean(coherence_scores) if coherence_scores else 0.0\n\n def _calculate_coverage(self, chunks, original_text):\n \"\"\"Calculate how well chunks cover the original text\"\"\"\n combined_chunk_text = \" \".join([\n chunk.get(\"text\", chunk.get(\"original_text\", \"\")) for chunk in chunks\n ])\n\n # Simple coverage based on text overlap\n original_words = set(original_text.lower().split())\n chunk_words = set(combined_chunk_text.lower().split())\n\n if not original_words:\n return 0.0\n\n coverage = len(original_words & chunk_words) / len(original_words)\n return coverage\n\n def _analyze_size_distribution(self, chunks):\n \"\"\"Analyze the distribution of chunk sizes\"\"\"\n sizes = [len(chunk.get(\"text\", chunk.get(\"original_text\", \"\")).split())\n for chunk in chunks]\n\n if not sizes:\n return 0.0\n\n mean_size = np.mean(sizes)\n std_size = np.std(sizes)\n\n # Ideal distribution has low variance\n consistency_score = max(0.0, 1.0 - (std_size / mean_size))\n return consistency_score\n\n def _assess_boundary_quality(self, chunks):\n \"\"\"Assess the quality of chunk boundaries\"\"\"\n if len(chunks) \u003c 2:\n return 1.0\n\n boundary_scores = []\n\n for i in range(len(chunks) - 1):\n current_chunk = chunks[i].get(\"text\", chunks[i].get(\"original_text\", \"\"))\n next_chunk = chunks[i + 1].get(\"text\", chunks[i + 1].get(\"original_text\", \"\"))\n\n # Check if chunks end/begin naturally\n ends_naturally = current_chunk.strip().endswith(('.', '!', '?', ':', ';'))\n begins_naturally = next_chunk.strip()[0].isupper() if next_chunk.strip() else False\n\n boundary_score = 0.0\n if ends_naturally:\n boundary_score += 0.5\n if begins_naturally:\n boundary_score += 0.5\n\n boundary_scores.append(boundary_score)\n\n return np.mean(boundary_scores) if boundary_scores else 0.0\n\n\n# Usage Example\nif __name__ == \"__main__\":\n config = {\n \"semantic_model\": \"all-MiniLM-L6-v2\",\n \"context_model\": \"gpt-3.5-turbo\",\n \"embedding_model\": \"sentence-transformers/all-MiniLM-L6-v2\",\n \"openai_api_key\": \"your-api-key-here\"\n }\n\n pipeline = SemanticChunkingPipeline(config)\n\n # Sample document\n sample_text = \"\"\"\n Natural language processing has evolved significantly over the past decade.\n Modern transformer models have revolutionized how we approach text understanding and generation.\n These models use attention mechanisms to process input text in parallel.\n The attention mechanism allows the model to focus on different parts of the input when producing each part of the output.\n\n Retrieval-Augmented Generation (RAG) combines the power of large language models with external knowledge retrieval.\n This approach enables models to access up-to-date information beyond their training data.\n RAG systems typically consist of three main components: a retriever, a knowledge base, and a generator.\n The retriever finds relevant documents from the knowledge base based on the user's query.\n The generator then uses these retrieved documents to produce a more informed response.\n \"\"\"\n\n # Process with different methods\n semantic_chunks = pipeline.process_document(sample_text, method=\"semantic\")\n contextual_chunks = pipeline.process_document(sample_text, method=\"contextual\")\n late_chunks = pipeline.process_document(sample_text, method=\"late\")\n hybrid_chunks = pipeline.process_document(sample_text, method=\"hybrid\")\n\n # Evaluate results\n methods = {\n \"semantic\": semantic_chunks,\n \"contextual\": contextual_chunks,\n \"late\": late_chunks,\n \"hybrid\": hybrid_chunks\n }\n\n for method_name, chunks in methods.items():\n evaluation = pipeline.evaluate_chunking(chunks, sample_text)\n print(f\"\\n{method_name.upper()} Method:\")\n print(f\" Chunks: {len(chunks)}\")\n print(f\" Overall Score: {evaluation['overall_score']:.3f}\")\n print(f\" Coherence: {evaluation['metrics']['coherence']:.3f}\")\n print(f\" Coverage: {evaluation['metrics']['coverage']:.3f}\")\n```\n\nThese semantic and contextual chunking methods provide advanced approaches for creating meaningful, context-aware chunks that significantly improve RAG system performance compared to traditional size-based chunking methods.","content_type":"text/markdown; charset=utf-8","language":"markdown","size":50188,"content_sha256":"0304faf69890b188fb307472e98b3c89b8968111b7ea83a8495559f3d33a974f"},{"filename":"references/strategies.md","content":"# Detailed Chunking Strategies\n\nThis document provides comprehensive implementation details for all chunking strategies mentioned in the main skill.\n\n## Level 1: Fixed-Size Chunking\n\n### Implementation\n\n```python\nfrom langchain.text_splitter import RecursiveCharacterTextSplitter\n\nclass FixedSizeChunker:\n def __init__(self, chunk_size=512, chunk_overlap=50):\n self.chunk_size = chunk_size\n self.chunk_overlap = chunk_overlap\n self.splitter = RecursiveCharacterTextSplitter(\n chunk_size=chunk_size,\n chunk_overlap=chunk_overlap,\n length_function=len,\n separators=[\"\\n\\n\", \"\\n\", \" \", \"\"]\n )\n\n def chunk(self, documents):\n return self.splitter.split_documents(documents)\n```\n\n### Parameter Recommendations\n\n| Use Case | Chunk Size | Overlap | Rationale |\n|----------|------------|---------|-----------|\n| Factoid Queries | 256 | 25 | Small chunks for precise answers |\n| General Q&A | 512 | 50 | Balanced approach for most cases |\n| Analytical Queries | 1024 | 100 | Larger context for complex analysis |\n| Code Documentation | 300 | 30 | Preserve code context while maintaining focus |\n\n### Best Practices\n\n- Start with 512 tokens and 10-20% overlap\n- Adjust based on embedding model context window\n- Use overlap for queries where context might span boundaries\n- Monitor token count vs. character count based on model\n\n## Level 2: Recursive Character Chunking\n\n### Implementation\n\n```python\nfrom langchain.text_splitter import RecursiveCharacterTextSplitter\n\nclass RecursiveChunker:\n def __init__(self, chunk_size=512, separators=None):\n self.chunk_size = chunk_size\n self.separators = separators or [\"\\n\\n\", \"\\n\", \" \", \"\"]\n self.splitter = RecursiveCharacterTextSplitter(\n chunk_size=chunk_size,\n chunk_overlap=0,\n length_function=len,\n separators=self.separators\n )\n\n def chunk(self, text):\n return self.splitter.create_documents([text])\n\n# Document-specific configurations\ndef get_chunker_for_document_type(doc_type):\n configurations = {\n \"markdown\": [\"\\n## \", \"\\n### \", \"\\n\\n\", \"\\n\", \" \", \"\"],\n \"html\": [\"\u003c/div>\", \"\u003c/p>\", \"\\n\\n\", \"\\n\", \" \", \"\"],\n \"code\": [\"\\n\\n\", \"\\n\", \" \", \"\"],\n \"plain\": [\"\\n\\n\", \"\\n\", \" \", \"\"]\n }\n return RecursiveChunker(separators=configurations.get(doc_type, [\"\\n\\n\", \"\\n\", \" \", \"\"]))\n```\n\n### Customization Guidelines\n\n- **Markdown**: Use headings as primary separators\n- **HTML**: Use block-level tags as separators\n- **Code**: Preserve function and class boundaries\n- **Academic papers**: Prioritize paragraph and section breaks\n\n## Level 3: Structure-Aware Chunking\n\n### Markdown Documents\n\n```python\nimport markdown\nfrom bs4 import BeautifulSoup\n\nclass MarkdownChunker:\n def __init__(self, max_chunk_size=512):\n self.max_chunk_size = max_chunk_size\n\n def chunk(self, markdown_text):\n html = markdown.markdown(markdown_text)\n soup = BeautifulSoup(html, 'html.parser')\n\n chunks = []\n current_chunk = \"\"\n current_heading = \"Introduction\"\n\n for element in soup.find_all(['h1', 'h2', 'h3', 'p', 'pre', 'table']):\n if element.name.startswith('h'):\n if current_chunk.strip():\n chunks.append({\n \"content\": current_chunk.strip(),\n \"heading\": current_heading\n })\n current_heading = element.get_text().strip()\n current_chunk = f\"{element}\\n\"\n elif element.name in ['pre', 'table']:\n # Preserve code blocks and tables intact\n if len(current_chunk) + len(str(element)) > self.max_chunk_size:\n if current_chunk.strip():\n chunks.append({\n \"content\": current_chunk.strip(),\n \"heading\": current_heading\n })\n current_chunk = f\"{element}\\n\"\n else:\n current_chunk += f\"{element}\\n\"\n else:\n current_chunk += str(element)\n\n if current_chunk.strip():\n chunks.append({\n \"content\": current_chunk.strip(),\n \"heading\": current_heading\n })\n\n return chunks\n```\n\n### Code Documents\n\n```python\nimport ast\nimport re\n\nclass CodeChunker:\n def __init__(self, language='python'):\n self.language = language\n\n def chunk_python(self, code):\n tree = ast.parse(code)\n chunks = []\n\n for node in ast.walk(tree):\n if isinstance(node, (ast.FunctionDef, ast.ClassDef)):\n start_line = node.lineno - 1\n end_line = node.end_lineno if hasattr(node, 'end_lineno') else start_line + 10\n lines = code.split('\\n')\n chunk_lines = lines[start_line:end_line]\n chunks.append('\\n'.join(chunk_lines))\n\n return chunks\n\n def chunk_javascript(self, code):\n # Use regex for languages without AST parsers\n function_pattern = r'(function\\s+\\w+\\s*\$[^)]*\$\\s*\\{[^}]*\\})'\n class_pattern = r'(class\\s+\\w+\\s*\\{[^}]*\\})'\n\n patterns = [function_pattern, class_pattern]\n chunks = []\n\n for pattern in patterns:\n matches = re.finditer(pattern, code, re.MULTILINE | re.DOTALL)\n for match in matches:\n chunks.append(match.group(1))\n\n return chunks\n\n def chunk(self, code):\n if self.language == 'python':\n return self.chunk_python(code)\n elif self.language == 'javascript':\n return self.chunk_javascript(code)\n else:\n # Fallback to line-based chunking\n return self.chunk_by_lines(code)\n\n def chunk_by_lines(self, code, max_lines=50):\n lines = code.split('\\n')\n chunks = []\n\n for i in range(0, len(lines), max_lines):\n chunk = '\\n'.join(lines[i:i+max_lines])\n chunks.append(chunk)\n\n return chunks\n```\n\n### Tabular Data\n\n```python\nimport pandas as pd\n\nclass TableChunker:\n def __init__(self, max_rows=100, summary_rows=5):\n self.max_rows = max_rows\n self.summary_rows = summary_rows\n\n def chunk(self, table_data):\n if isinstance(table_data, str):\n df = pd.read_csv(StringIO(table_data))\n else:\n df = table_data\n\n chunks = []\n\n if len(df) \u003c= self.max_rows:\n # Small table - keep intact\n chunks.append({\n \"type\": \"full_table\",\n \"content\": df.to_string(),\n \"metadata\": {\n \"rows\": len(df),\n \"columns\": len(df.columns)\n }\n })\n else:\n # Large table - create summary + chunks\n summary = df.head(self.summary_rows)\n chunks.append({\n \"type\": \"table_summary\",\n \"content\": f\"Table Summary ({len(df)} rows, {len(df.columns)} columns):\\n{summary.to_string()}\",\n \"metadata\": {\n \"total_rows\": len(df),\n \"summary_rows\": self.summary_rows,\n \"columns\": list(df.columns)\n }\n })\n\n # Chunk the remaining data\n for i in range(self.summary_rows, len(df), self.max_rows):\n chunk_df = df.iloc[i:i+self.max_rows]\n chunks.append({\n \"type\": \"table_chunk\",\n \"content\": f\"Rows {i+1}-{min(i+self.max_rows, len(df))}:\\n{chunk_df.to_string()}\",\n \"metadata\": {\n \"start_row\": i + 1,\n \"end_row\": min(i + self.max_rows, len(df)),\n \"columns\": list(df.columns)\n }\n })\n\n return chunks\n```\n\n## Level 4: Semantic Chunking\n\n### Implementation\n\n```python\nimport numpy as np\nfrom sentence_transformers import SentenceTransformer\nfrom sklearn.metrics.pairwise import cosine_similarity\n\nclass SemanticChunker:\n def __init__(self, model_name=\"all-MiniLM-L6-v2\", similarity_threshold=0.8, buffer_size=3):\n self.model = SentenceTransformer(model_name)\n self.similarity_threshold = similarity_threshold\n self.buffer_size = buffer_size\n\n def split_into_sentences(self, text):\n # Simple sentence splitting - can be enhanced with nltk/spacy\n sentences = re.split(r'[.!?]+', text)\n return [s.strip() for s in sentences if s.strip()]\n\n def chunk(self, text):\n sentences = self.split_into_sentences(text)\n\n if len(sentences) \u003c= self.buffer_size:\n return [text]\n\n # Create embeddings\n embeddings = self.model.encode(sentences)\n\n chunks = []\n current_chunk_sentences = []\n\n for i in range(len(sentences)):\n current_chunk_sentences.append(sentences[i])\n\n # Check if we should create a boundary\n if i \u003c len(sentences) - 1:\n similarity = cosine_similarity(\n [embeddings[i]],\n [embeddings[i + 1]]\n )[0][0]\n\n if similarity \u003c self.similarity_threshold and len(current_chunk_sentences) >= 2:\n chunks.append(' '.join(current_chunk_sentences))\n current_chunk_sentences = []\n\n # Add remaining sentences\n if current_chunk_sentences:\n chunks.append(' '.join(current_chunk_sentences))\n\n return chunks\n```\n\n### Parameter Tuning\n\n| Parameter | Range | Effect |\n|-----------|-------|--------|\n| similarity_threshold | 0.5-0.9 | Higher values create more chunks |\n| buffer_size | 1-10 | Larger buffers provide more context |\n| model_name | Various | Different models for different domains |\n\n### Optimization Tips\n\n- Use domain-specific models for specialized content\n- Adjust threshold based on content complexity\n- Cache embeddings for repeated processing\n- Consider batch processing for large documents\n\n## Level 5: Advanced Contextual Methods\n\n### Late Chunking\n\n```python\nimport torch\nfrom transformers import AutoTokenizer, AutoModel\n\nclass LateChunker:\n def __init__(self, model_name=\"microsoft/DialoGPT-medium\"):\n self.tokenizer = AutoTokenizer.from_pretrained(model_name)\n self.model = AutoModel.from_pretrained(model_name)\n\n def chunk(self, text, chunk_size=512):\n # Tokenize entire document\n tokens = self.tokenizer(text, return_tensors=\"pt\", truncation=False)\n\n # Get token-level embeddings\n with torch.no_grad():\n outputs = self.model(**tokens, output_hidden_states=True)\n token_embeddings = outputs.last_hidden_state[0]\n\n # Create chunk embeddings from token embeddings\n chunks = []\n for i in range(0, len(token_embeddings), chunk_size):\n chunk_tokens = token_embeddings[i:i+chunk_size]\n chunk_embedding = torch.mean(chunk_tokens, dim=0)\n chunks.append({\n \"content\": self.tokenizer.decode(tokens[\"input_ids\"][0][i:i+chunk_size]),\n \"embedding\": chunk_embedding.numpy()\n })\n\n return chunks\n```\n\n### Contextual Retrieval\n\n```python\nimport openai\n\nclass ContextualChunker:\n def __init__(self, api_key):\n self.client = openai.OpenAI(api_key=api_key)\n\n def generate_context(self, chunk, full_document):\n prompt = f\"\"\"\n Given the following document and a chunk from it, provide a brief context\n that helps understand the chunk's meaning within the full document.\n\n Document:\n {full_document[:2000]}...\n\n Chunk:\n {chunk}\n\n Context (max 50 words):\n \"\"\"\n\n response = self.client.chat.completions.create(\n model=\"gpt-3.5-turbo\",\n messages=[{\"role\": \"user\", \"content\": prompt}],\n max_tokens=100,\n temperature=0\n )\n\n return response.choices[0].message.content.strip()\n\n def chunk_with_context(self, text, base_chunker):\n # First create base chunks\n base_chunks = base_chunker.chunk(text)\n\n # Then add context to each chunk\n contextualized_chunks = []\n for chunk in base_chunks:\n context = self.generate_context(chunk.page_content, text)\n contextualized_content = f\"Context: {context}\\n\\nContent: {chunk.page_content}\"\n\n contextualized_chunks.append({\n \"content\": contextualized_content,\n \"original_content\": chunk.page_content,\n \"context\": context\n })\n\n return contextualized_chunks\n```\n\n## Performance Considerations\n\n### Computational Cost Analysis\n\n| Strategy | Time Complexity | Space Complexity | Relative Cost |\n|----------|-----------------|------------------|---------------|\n| Fixed-Size | O(n) | O(n) | Low |\n| Recursive | O(n) | O(n) | Low |\n| Structure-Aware | O(n log n) | O(n) | Medium |\n| Semantic | O(n²) | O(n²) | High |\n| Late Chunking | O(n) | O(n) | Very High |\n| Contextual | O(n²) | O(n²) | Very High |\n\n### Optimization Strategies\n\n1. **Parallel Processing**: Process chunks concurrently when possible\n2. **Caching**: Store embeddings and intermediate results\n3. **Batch Operations**: Group similar operations together\n4. **Progressive Loading**: Process large documents in streaming fashion\n5. **Model Selection**: Choose appropriate models for task complexity","content_type":"text/markdown; charset=utf-8","language":"markdown","size":13540,"content_sha256":"533ef3342b95f8440fbadd3d0b78a134fceaf90cf16b7ecd7f53d165b90320f5"},{"filename":"references/tools.md","content":"# Recommended Libraries and Frameworks\n\nThis document provides a comprehensive guide to tools, libraries, and frameworks for implementing chunking strategies.\n\n## Core Chunking Libraries\n\n### LangChain\n\n**Overview**: Comprehensive framework for building applications with large language models, includes robust text splitting utilities.\n\n**Installation**:\n```bash\npip install langchain langchain-text-splitters\n```\n\n**Key Features**:\n- Multiple text splitting strategies\n- Integration with various document loaders\n- Support for different content types (code, markdown, etc.)\n- Customizable separators and parameters\n\n**Example Usage**:\n\n```python\nfrom langchain.text_splitter import (\n RecursiveCharacterTextSplitter,\n CharacterTextSplitter,\n TokenTextSplitter,\n MarkdownTextSplitter,\n PythonCodeTextSplitter\n)\n\n# Basic recursive splitting\nsplitter = RecursiveCharacterTextSplitter(\n chunk_size=1000,\n chunk_overlap=200,\n length_function=len,\n separators=[\"\\n\\n\", \"\\n\", \" \", \"\"]\n)\n\nchunks = splitter.split_text(large_text)\n\n# Markdown-specific splitting\nmarkdown_splitter = MarkdownTextSplitter(\n chunk_size=1000,\n chunk_overlap=100\n)\n\n# Code-specific splitting\ncode_splitter = PythonCodeTextSplitter(\n chunk_size=1000,\n chunk_overlap=100\n)\n```\n\n**Pros**:\n- Well-maintained and actively developed\n- Extensive documentation and examples\n- Integrates well with other LangChain components\n- Supports multiple document types\n\n**Cons**:\n- Can be heavy dependency for simple use cases\n- Some advanced features require LangChain ecosystem\n\n### LlamaIndex\n\n**Overview**: Data framework for LLM applications with advanced indexing and retrieval capabilities.\n\n**Installation**:\n```bash\npip install llama-index\n```\n\n**Key Features**:\n- Advanced semantic chunking\n- Hierarchical indexing\n- Context-aware retrieval\n- Integration with vector databases\n\n**Example Usage**:\n\n```python\nfrom llama_index.core.node_parser import (\n SentenceSplitter,\n SemanticSplitterNodeParser\n)\nfrom llama_index.core import SimpleDirectoryReader\nfrom llama_index.embeddings.openai import OpenAIEmbedding\n\n# Basic sentence splitting\nsplitter = SentenceSplitter(\n chunk_size=1024,\n chunk_overlap=20\n)\n\n# Semantic chunking with embeddings\nembed_model = OpenAIEmbedding()\nsemantic_splitter = SemanticSplitterNodeParser(\n buffer_size=1,\n breakpoint_percentile_threshold=95,\n embed_model=embed_model\n)\n\n# Load and process documents\ndocuments = SimpleDirectoryReader(\"./data\").load_data()\nnodes = semantic_splitter.get_nodes_from_documents(documents)\n```\n\n**Pros**:\n- Excellent semantic chunking capabilities\n- Built for production RAG systems\n- Strong vector database integration\n- Active community support\n\n**Cons**:\n- More complex setup for basic use cases\n- Semantic chunking requires embedding model setup\n\n### Unstructured\n\n**Overview**: Open-source library for processing unstructured documents, especially strong with multi-modal content.\n\n**Installation**:\n```bash\npip install \"unstructured[pdf,png,jpg]\"\n```\n\n**Key Features**:\n- Multi-modal document processing\n- Support for PDFs, images, and various formats\n- Structure preservation\n- Table extraction and processing\n\n**Example Usage**:\n\n```python\nfrom unstructured.partition.auto import partition\nfrom unstructured.chunking.title import chunk_by_title\n\n# Partition document by type\nelements = partition(filename=\"document.pdf\")\n\n# Chunk by title/heading structure\nchunks = chunk_by_title(\n elements,\n combine_text_under_n_chars=2000,\n max_characters=10000,\n new_after_n_chars=1500,\n multipage_sections=True\n)\n\n# Access chunked content\nfor chunk in chunks:\n print(f\"Category: {chunk.category}\")\n print(f\"Content: {chunk.text[:200]}...\")\n```\n\n**Pros**:\n- Excellent for PDF and image processing\n- Preserves document structure\n- Handles tables and figures well\n- Strong multi-modal capabilities\n\n**Cons**:\n- Can be slower for large documents\n- Requires additional dependencies for some formats\n\n## Text Processing Libraries\n\n### NLTK (Natural Language Toolkit)\n\n**Installation**:\n```bash\npip install nltk\n```\n\n**Key Features**:\n- Sentence tokenization\n- Language detection\n- Text preprocessing\n- Linguistic analysis\n\n**Example Usage**:\n\n```python\nimport nltk\nfrom nltk.tokenize import sent_tokenize, word_tokenize\nfrom nltk.corpus import stopwords\n\n# Download required data\nnltk.download('punkt')\nnltk.download('stopwords')\n\n# Sentence and word tokenization\ntext = \"This is a sample sentence. This is another sentence.\"\nsentences = sent_tokenize(text)\nwords = word_tokenize(text)\n\n# Stop words removal\nstop_words = set(stopwords.words('english'))\nfiltered_words = [word for word in words if word.lower() not in stop_words]\n```\n\n### spaCy\n\n**Installation**:\n```bash\npip install spacy\npython -m spacy download en_core_web_sm\n```\n\n**Key Features**:\n- Industrial-strength NLP\n- Named entity recognition\n- Dependency parsing\n- Sentence boundary detection\n\n**Example Usage**:\n\n```python\nimport spacy\n\n# Load language model\nnlp = spacy.load(\"en_core_web_sm\")\n\n# Process text\ndoc = nlp(\"This is a sample sentence. This is another sentence.\")\n\n# Extract sentences\nsentences = [sent.text for sent in doc.sents]\n\n# Named entities\nentities = [(ent.text, ent.label_) for ent in doc.ents]\n\n# Dependency parsing for better chunking\nfor token in doc:\n print(f\"{token.text}: {token.dep_} (head: {token.head.text})\")\n```\n\n### Sentence Transformers\n\n**Installation**:\n```bash\npip install sentence-transformers\n```\n\n**Key Features**:\n- Pre-trained sentence embeddings\n- Semantic similarity calculation\n- Multi-lingual support\n- Custom model training\n\n**Example Usage**:\n\n```python\nfrom sentence_transformers import SentenceTransformer, util\nimport numpy as np\n\n# Load pre-trained model\nmodel = SentenceTransformer('all-MiniLM-L6-v2')\n\n# Generate embeddings\nsentences = [\"This is a sentence.\", \"This is another sentence.\"]\nembeddings = model.encode(sentences)\n\n# Calculate semantic similarity\nsimilarity = util.cos_sim(embeddings[0], embeddings[1])\n\n# Find semantic boundaries for chunking\ndef find_semantic_boundaries(text, model, threshold=0.8):\n sentences = [s.strip() for s in text.split('.') if s.strip()]\n embeddings = model.encode(sentences)\n\n boundaries = [0]\n for i in range(1, len(sentences)):\n similarity = util.cos_sim(embeddings[i-1], embeddings[i])\n if similarity \u003c threshold:\n boundaries.append(i)\n\n return boundaries\n```\n\n## Vector Databases and Search\n\n### ChromaDB\n\n**Installation**:\n```bash\npip install chromadb\n```\n\n**Key Features**:\n- In-memory and persistent storage\n- Built-in embedding functions\n- Similarity search\n- Metadata filtering\n\n**Example Usage**:\n\n```python\nimport chromadb\nfrom chromadb.utils import embedding_functions\n\n# Initialize client\nclient = chromadb.Client()\n\n# Create collection\ncollection = client.create_collection(\n name=\"document_chunks\",\n embedding_function=embedding_functions.DefaultEmbeddingFunction()\n)\n\n# Add chunks\ncollection.add(\n documents=[chunk[\"content\"] for chunk in chunks],\n metadatas=[chunk.get(\"metadata\", {}) for chunk in chunks],\n ids=[chunk[\"id\"] for chunk in chunks]\n)\n\n# Search\nresults = collection.query(\n query_texts=[\"What is chunking?\"],\n n_results=5\n)\n```\n\n### Pinecone\n\n**Installation**:\n```bash\npip install pinecone-client\n```\n\n**Key Features**:\n- Managed vector database service\n- High-performance similarity search\n- Metadata filtering\n- Scalable infrastructure\n\n**Example Usage**:\n\n```python\nimport pinecone\nfrom sentence_transformers import SentenceTransformer\n\n# Initialize\npinecone.init(api_key=\"your-api-key\", environment=\"your-environment\")\nindex_name = \"document-chunks\"\n\n# Create index if it doesn't exist\nif index_name not in pinecone.list_indexes():\n pinecone.create_index(\n name=index_name,\n dimension=384, # Match embedding model\n metric=\"cosine\"\n )\n\nindex = pinecone.Index(index_name)\n\n# Generate embeddings and upsert\nmodel = SentenceTransformer('all-MiniLM-L6-v2')\nfor chunk in chunks:\n embedding = model.encode(chunk[\"content\"])\n index.upsert(\n vectors=[{\n \"id\": chunk[\"id\"],\n \"values\": embedding.tolist(),\n \"metadata\": chunk.get(\"metadata\", {})\n }]\n )\n\n# Search\nquery_embedding = model.encode(\"search query\")\nresults = index.query(\n vector=query_embedding.tolist(),\n top_k=5,\n include_metadata=True\n)\n```\n\n### Weaviate\n\n**Installation**:\n```bash\npip install weaviate-client\n```\n\n**Key Features**:\n- GraphQL API\n- Hybrid search (dense + sparse)\n- Real-time updates\n- Schema validation\n\n**Example Usage**:\n\n```python\nimport weaviate\n\n# Connect to Weaviate\nclient = weaviate.Client(\"http://localhost:8080\")\n\n# Define schema\nclient.schema.create_class({\n \"class\": \"DocumentChunk\",\n \"description\": \"A chunk of document content\",\n \"properties\": [\n {\n \"name\": \"content\",\n \"dataType\": [\"text\"]\n },\n {\n \"name\": \"source\",\n \"dataType\": [\"string\"]\n }\n ]\n})\n\n# Add data\nfor chunk in chunks:\n client.data_object.create(\n data_object={\n \"content\": chunk[\"content\"],\n \"source\": chunk.get(\"source\", \"unknown\")\n },\n class_name=\"DocumentChunk\"\n )\n\n# Search\nresults = client.query.get(\n \"DocumentChunk\",\n [\"content\", \"source\"]\n).with_near_text({\n \"concepts\": [\"search query\"]\n}).with_limit(5).do()\n```\n\n## Evaluation and Testing\n\n### RAGAS\n\n**Installation**:\n```bash\npip install ragas\n```\n\n**Key Features**:\n- RAG evaluation metrics\n- Answer quality assessment\n- Context relevance measurement\n- Faithfulness evaluation\n\n**Example Usage**:\n\n```python\nfrom ragas import evaluate\nfrom ragas.metrics import (\n faithfulness,\n answer_relevancy,\n context_relevancy,\n context_recall\n)\nfrom datasets import Dataset\n\n# Prepare evaluation data\ndataset = Dataset.from_dict({\n \"question\": [\"What is chunking?\"],\n \"answer\": [\"Chunking is the process of breaking large documents into smaller segments\"],\n \"contexts\": [[\"Chunking involves dividing text into manageable pieces for better processing\"]],\n \"ground_truth\": [\"Chunking is a document processing technique\"]\n})\n\n# Evaluate\nresult = evaluate(\n dataset=dataset,\n metrics=[\n faithfulness,\n answer_relevancy,\n context_relevancy,\n context_recall\n ]\n)\n\nprint(result)\n```\n\n### TruEra (TruLens)\n\n**Installation**:\n```bash\npip install trulens trulens-apps\n```\n\n**Key Features**:\n- LLM application evaluation\n- Feedback functions\n- Hallucination detection\n- Performance monitoring\n\n**Example Usage**:\n\n```python\nfrom trulens.core import TruSession\nfrom trulens.apps.custom import instrument\nfrom trulens.feedback import GroundTruthAgreement\n\n# Initialize session\nsession = TruSession()\n\n# Define feedback functions\nf_groundedness = GroundTruthAgreement(ground_truth)\n\n# Evaluate chunks\n@instrument\ndef chunk_and_query(text, query):\n chunks = chunk_function(text)\n relevant_chunks = search_function(chunks, query)\n answer = generate_function(relevant_chunks, query)\n return answer\n\n# Record evaluation\nwith session:\n chunk_and_query(\"large document text\", \"what is the main topic?\")\n```\n\n## Document Processing\n\n### PyPDF2\n\n**Installation**:\n```bash\npip install PyPDF2\n```\n\n**Key Features**:\n- PDF text extraction\n- Page manipulation\n- Metadata extraction\n- Form field processing\n\n**Example Usage**:\n\n```python\nimport PyPDF2\n\ndef extract_text_from_pdf(pdf_path):\n text = \"\"\n with open(pdf_path, 'rb') as file:\n reader = PyPDF2.PdfReader(file)\n for page in reader.pages:\n text += page.extract_text()\n return text\n\n# Extract text by page for better chunking\ndef extract_pages(pdf_path):\n pages = []\n with open(pdf_path, 'rb') as file:\n reader = PyPDF2.PdfReader(file)\n for i, page in enumerate(reader.pages):\n pages.append({\n \"page_number\": i + 1,\n \"content\": page.extract_text()\n })\n return pages\n```\n\n### python-docx\n\n**Installation**:\n```bash\npip install python-docx\n```\n\n**Key Features**:\n- Microsoft Word document processing\n- Paragraph and table extraction\n- Style preservation\n- Metadata access\n\n**Example Usage**:\n\n```python\nfrom docx import Document\n\ndef extract_from_docx(docx_path):\n doc = Document(docx_path)\n content = []\n\n for paragraph in doc.paragraphs:\n if paragraph.text.strip():\n content.append({\n \"type\": \"paragraph\",\n \"text\": paragraph.text,\n \"style\": paragraph.style.name\n })\n\n for table in doc.tables:\n table_text = []\n for row in table.rows:\n row_text = [cell.text for cell in row.cells]\n table_text.append(\" | \".join(row_text))\n\n content.append({\n \"type\": \"table\",\n \"text\": \"\\n\".join(table_text)\n })\n\n return content\n```\n\n## Specialized Libraries\n\n### tiktoken (OpenAI)\n\n**Installation**:\n```bash\npip install tiktoken\n```\n\n**Key Features**:\n- Accurate token counting for OpenAI models\n- Fast encoding/decoding\n- Multiple model support\n- Language model specific tokenization\n\n**Example Usage**:\n\n```python\nimport tiktoken\n\n# Get encoding for specific model\nencoding = tiktoken.encoding_for_model(\"gpt-3.5-turbo\")\n\n# Encode text\ntokens = encoding.encode(\"This is a sample text\")\nprint(f\"Token count: {len(tokens)}\")\n\n# Decode tokens\ntext = encoding.decode(tokens)\n\n# Count tokens without full encoding\ndef count_tokens(text, model=\"gpt-3.5-turbo\"):\n encoding = tiktoken.encoding_for_model(model)\n return len(encoding.encode(text))\n\n# Use in chunking\ndef chunk_by_tokens(text, max_tokens=1000):\n encoding = tiktoken.encoding_for_model(\"gpt-3.5-turbo\")\n tokens = encoding.encode(text)\n\n chunks = []\n for i in range(0, len(tokens), max_tokens):\n chunk_tokens = tokens[i:i + max_tokens]\n chunk_text = encoding.decode(chunk_tokens)\n chunks.append(chunk_text)\n\n return chunks\n```\n\n### PDFMiner\n\n**Installation**:\n```bash\npip install pdfminer.six\n```\n\n**Key Features**:\n- Detailed PDF analysis\n- Layout preservation\n- Font and style information\n- High-precision text extraction\n\n**Example Usage**:\n\n```python\nfrom pdfminer.high_level import extract_pages\nfrom pdfminer.layout import LTTextContainer\n\ndef extract_structured_text(pdf_path):\n structured_content = []\n\n for page_layout in extract_pages(pdf_path):\n page_content = []\n\n for element in page_layout:\n if isinstance(element, LTTextContainer):\n text = element.get_text()\n font_info = {\n \"font_size\": element.height,\n \"is_bold\": \"Bold\" in element.fontname,\n \"x0\": element.x0,\n \"y0\": element.y0\n }\n page_content.append({\n \"text\": text.strip(),\n \"font_info\": font_info\n })\n\n structured_content.append({\n \"page_number\": page_layout.pageid,\n \"content\": page_content\n })\n\n return structured_content\n```\n\n## Performance and Optimization\n\n### Dask\n\n**Installation**:\n```bash\npip install dask[complete]\n```\n\n**Key Features**:\n- Parallel processing\n- Out-of-core computation\n- Distributed computing\n- Integration with pandas\n\n**Example Usage**:\n\n```python\nimport dask.bag as db\nfrom dask.distributed import Client\n\n# Setup distributed client\nclient = Client(n_workers=4)\n\n# Parallel chunking of multiple documents\ndef chunk_document(document):\n # Your chunking logic here\n return chunk_function(document)\n\n# Process documents in parallel\ndocuments = [\"doc1\", \"doc2\", \"doc3\", ...] # List of document contents\ndocument_bag = db.from_sequence(documents)\n\n# Apply chunking function in parallel\nchunked_documents = document_bag.map(chunk_document)\n\n# Compute results\nresults = chunked_documents.compute()\n```\n\n### Ray\n\n**Installation**:\n```bash\npip install ray\n```\n\n**Key Features**:\n- Distributed computing\n- Actor model\n- Autoscaling\n- ML pipeline integration\n\n**Example Usage**:\n\n```python\nimport ray\n\n# Initialize Ray\nray.init()\n\[email protected]\nclass ChunkingWorker:\n def __init__(self, strategy):\n self.strategy = strategy\n\n def chunk_documents(self, documents):\n results = []\n for doc in documents:\n chunks = self.strategy.chunk(doc)\n results.append(chunks)\n return results\n\n# Create workers\nworkers = [ChunkingWorker.remote(strategy) for _ in range(4)]\n\n# Distribute work\ndocuments_batch = [documents[i::4] for i in range(4)]\nfutures = [worker.chunk_documents.remote(batch)\n for worker, batch in zip(workers, documents_batch)]\n\n# Get results\nresults = ray.get(futures)\n```\n\n## Development and Testing\n\n### pytest\n\n**Installation**:\n```bash\npip install pytest pytest-asyncio\n```\n\n**Example Tests**:\n\n```python\nimport pytest\nfrom your_chunking_module import FixedSizeChunker, SemanticChunker\n\nclass TestFixedSizeChunker:\n def test_chunk_size_respect(self):\n chunker = FixedSizeChunker(chunk_size=100, chunk_overlap=10)\n text = \"word \" * 50 # 50 words\n\n chunks = chunker.chunk(text)\n\n for chunk in chunks:\n assert len(chunk.split()) \u003c= 100 # Account for word boundaries\n\n def test_overlap_consistency(self):\n chunker = FixedSizeChunker(chunk_size=50, chunk_overlap=10)\n text = \"word \" * 30\n\n chunks = chunker.chunk(text)\n\n # Check overlap between consecutive chunks\n for i in range(1, len(chunks)):\n chunk1_words = set(chunks[i-1].split()[-10:])\n chunk2_words = set(chunks[i].split()[:10])\n overlap = len(chunk1_words & chunk2_words)\n assert overlap >= 5 # Allow some tolerance\n\[email protected]\nasync def test_semantic_chunker():\n chunker = SemanticChunker()\n text = \"First topic sentence. Another sentence about first topic. \" \\\n \"Now switching to second topic. More about second topic.\"\n\n chunks = await chunker.chunk_async(text)\n\n # Should detect topic change and create boundary\n assert len(chunks) >= 2\n assert \"first topic\" in chunks[0].lower()\n assert \"second topic\" in chunks[1].lower()\n```\n\n### Memory Profiler\n\n**Installation**:\n```bash\npip install memory-profiler\n```\n\n**Example Usage**:\n\n```python\nfrom memory_profiler import profile\n\n@profile\ndef chunk_large_document():\n chunker = FixedSizeChunker(chunk_size=1000)\n large_text = \"word \" * 100000 # Large document\n\n chunks = chunker.chunk(large_text)\n return chunks\n\n# Run with: python -m memory_profiler your_script.py\n```\n\nThis comprehensive toolset provides everything needed to implement, test, and optimize chunking strategies for various use cases, from simple text processing to production-grade RAG systems.","content_type":"text/markdown; charset=utf-8","language":"markdown","size":18918,"content_sha256":"813d24ede6f7654387947de55d882939913b92bc259d533caf9d87438654f1a5"},{"filename":"references/visualization-tools.md","content":"# Visualization and Evaluation Tools\n\nThis document covers tools and methodologies for visualizing chunking strategies and evaluating their effectiveness.\n\n## Overview of Visualization Tools\n\n| Tool | Purpose | Complexity | Integration |\n|------|---------|------------|-------------|\n| ChunkViz | Visualize chunk boundaries and overlaps | Low | Standalone |\n| Plotly | Interactive visualizations | Medium | Library |\n| TensorBoard | ML experiment tracking | High | Framework |\n| Streamlit | Web-based dashboards | Medium | Web App |\n| D3.js | Custom visualizations | High | Web Library |\n\n## 1. ChunkViz - Chunking Visualization Tool\n\n### Overview\nChunkViz is a specialized tool for visualizing how different chunking strategies behave with various parameters.\n\n### Installation and Setup\n\n```bash\n# Clone ChunkViz repository\ngit clone https://github.com/gkamradt/ChunkViz.git\ncd ChunkViz\n\n# Install dependencies\npip install -r requirements.txt\n```\n\n### Basic Usage\n\n```python\nfrom chunkviz import visualize_chunking\nimport matplotlib.pyplot as plt\n\n# Sample text\nsample_text = \"\"\"\nNatural language processing has evolved significantly over the past decade.\nModern transformer models have revolutionized how we approach text understanding.\nThese models use attention mechanisms to process input text efficiently.\nThe attention mechanism allows models to focus on relevant input parts.\n\nRetrieval-Augmented Generation (RAG) combines LLMs with external knowledge.\nThis approach enables models to access current information beyond training data.\nRAG systems typically have three main components: retriever, knowledge base, generator.\nThe retriever finds relevant documents based on user queries.\nThe generator uses retrieved documents to produce informed responses.\n\"\"\"\n\n# Visualize different chunking strategies\nstrategies = [\n {\"name\": \"Fixed Size 100\", \"chunk_size\": 100, \"overlap\": 0},\n {\"name\": \"Fixed Size 200\", \"chunk_size\": 200, \"overlap\": 0},\n {\"name\": \"Fixed Size 100 + 20% Overlap\", \"chunk_size\": 100, \"overlap\": 20},\n {\"name\": \"Semantic\", \"method\": \"semantic\", \"threshold\": 0.7}\n]\n\nfig, axes = plt.subplots(2, 2, figsize=(15, 10))\naxes = axes.flatten()\n\nfor i, strategy in enumerate(strategies):\n if i \u003c len(axes):\n visualize_chunking(\n text=sample_text,\n strategy=strategy,\n ax=axes[i],\n title=strategy[\"name\"]\n )\n\nplt.tight_layout()\nplt.show()\n```\n\n### Advanced ChunkViz Implementation\n\n```python\nimport matplotlib.pyplot as plt\nimport matplotlib.patches as patches\nimport numpy as np\nfrom typing import List, Dict, Tuple\nimport re\n\nclass AdvancedChunkViz:\n def __init__(self, figsize=(12, 8)):\n self.figsize = figsize\n self.colors = plt.cm.Set3(np.linspace(0, 1, 12))\n\n def visualize_multiple_strategies(self, text, strategies, save_path=None):\n \"\"\"Visualize multiple chunking strategies side by side\"\"\"\n n_strategies = len(strategies)\n n_cols = min(3, n_strategies)\n n_rows = (n_strategies + n_cols - 1) // n_cols\n\n fig, axes = plt.subplots(n_rows, n_cols, figsize=(n_cols * 4, n_rows * 3))\n if n_strategies == 1:\n axes = [axes]\n elif n_rows == 1:\n axes = axes.reshape(1, -1)\n\n for i, strategy in enumerate(strategies):\n row, col = i // n_cols, i % n_cols\n ax = axes[row, col] if n_rows > 1 else axes[col]\n\n self._visualize_single_strategy(text, strategy, ax)\n\n # Hide unused subplots\n for i in range(n_strategies, n_rows * n_cols):\n row, col = i // n_cols, i % n_cols\n ax = axes[row, col] if n_rows > 1 else axes[col]\n ax.set_visible(False)\n\n plt.tight_layout()\n if save_path:\n plt.savefig(save_path, dpi=300, bbox_inches='tight')\n plt.show()\n\n def _visualize_single_strategy(self, text, strategy, ax):\n \"\"\"Visualize a single chunking strategy\"\"\"\n chunks = self._apply_strategy(text, strategy)\n\n # Set up the plot\n ax.set_xlim(0, len(text))\n ax.set_ylim(0, len(chunks))\n ax.set_xlabel('Character Position')\n ax.set_ylabel('Chunk Index')\n ax.set_title(self._get_strategy_title(strategy))\n\n # Color code chunks\n for i, chunk in enumerate(chunks):\n start_pos = chunk['start']\n end_pos = chunk['end']\n color = self.colors[i % len(self.colors)]\n\n # Draw chunk rectangle\n rect = patches.Rectangle(\n (start_pos, i), end_pos - start_pos, 0.8,\n linewidth=1, edgecolor='black', facecolor=color, alpha=0.7\n )\n ax.add_patch(rect)\n\n # Add chunk text (truncated)\n chunk_text = chunk['text'][:20] + \"...\" if len(chunk['text']) > 20 else chunk['text']\n ax.text(\n start_pos + (end_pos - start_pos) / 2, i + 0.4,\n chunk_text, ha='center', va='center', fontsize=8\n )\n\n # Add chunk size info\n ax.text(\n start_pos, i - 0.1,\n f\"{len(chunk['text'])} chars\", ha='left', va='top', fontsize=6\n )\n\n # Add overlap indicators\n for i in range(1, len(chunks)):\n prev_end = chunks[i-1]['end']\n curr_start = chunks[i]['start']\n if curr_start \u003c prev_end: # Overlap exists\n overlap_len = prev_end - curr_start\n ax.axvspan(\n curr_start, prev_end, alpha=0.3, color='red',\n ymin=(i-0.5)/len(chunks), ymax=(i+0.5)/len(chunks)\n )\n ax.text(\n (curr_start + prev_end) / 2, i + 0.9,\n f\"Overlap: {overlap_len}\", ha='center', va='bottom',\n fontsize=6, color='red', weight='bold'\n )\n\n def _apply_strategy(self, text, strategy):\n \"\"\"Apply chunking strategy to text\"\"\"\n if strategy.get('method') == 'semantic':\n return self._semantic_chunking(text, strategy)\n else:\n return self._fixed_size_chunking(text, strategy)\n\n def _fixed_size_chunking(self, text, strategy):\n \"\"\"Fixed-size chunking\"\"\"\n chunk_size = strategy.get('chunk_size', 100)\n overlap = strategy.get('overlap', 0)\n overlap_chars = int(chunk_size * overlap / 100) if isinstance(overlap, int) else overlap\n\n chunks = []\n start = 0\n\n while start \u003c len(text):\n end = min(start + chunk_size, len(text))\n chunks.append({\n 'text': text[start:end],\n 'start': start,\n 'end': end\n })\n\n # Calculate next start with overlap\n start = max(0, end - overlap_chars)\n\n if end >= len(text):\n break\n\n return chunks\n\n def _semantic_chunking(self, text, strategy):\n \"\"\"Simple semantic chunking simulation\"\"\"\n # For visualization purposes, split by sentences\n sentences = re.split(r'[.!?]+', text)\n sentences = [s.strip() for s in sentences if s.strip()]\n\n threshold = strategy.get('threshold', 0.7)\n chunks = []\n current_chunk = []\n current_start = 0\n\n for i, sentence in enumerate(sentences):\n current_chunk.append(sentence)\n\n # Simple heuristic: create new chunk after certain number of sentences\n if len(current_chunk) >= 3:\n chunk_text = '. '.join(current_chunk) + '.'\n end_pos = current_start + len(chunk_text)\n chunks.append({\n 'text': chunk_text,\n 'start': current_start,\n 'end': end_pos\n })\n\n # Start new chunk\n current_start = end_pos\n current_chunk = []\n\n # Add remaining sentences\n if current_chunk:\n chunk_text = '. '.join(current_chunk) + '.'\n chunks.append({\n 'text': chunk_text,\n 'start': current_start,\n 'end': current_start + len(chunk_text)\n })\n\n return chunks\n\n def _get_strategy_title(self, strategy):\n \"\"\"Get descriptive title for strategy\"\"\"\n if strategy.get('method') == 'semantic':\n return f\"Semantic (threshold={strategy.get('threshold', 0.7)})\"\n else:\n chunk_size = strategy.get('chunk_size', 100)\n overlap = strategy.get('overlap', 0)\n return f\"Fixed Size ({chunk_size} chars, {overlap}% overlap)\"\n\n def create_interactive_visualization(self, text, strategies):\n \"\"\"Create interactive visualization using Plotly\"\"\"\n import plotly.graph_objects as go\n from plotly.subplots import make_subplots\n\n n_strategies = len(strategies)\n fig = make_subplots(\n rows=n_strategies, cols=1,\n subplot_titles=[self._get_strategy_title(s) for s in strategies],\n vertical_spacing=0.1\n )\n\n for i, strategy in enumerate(strategies):\n chunks = self._apply_strategy(text, strategy)\n\n for j, chunk in enumerate(chunks):\n fig.add_trace(\n go.Scatter(\n x=[chunk['start'], chunk['end']],\n y=[j, j],\n mode='lines+markers',\n line=dict(width=20),\n hovertemplate='\u003cb>Chunk %{text}\u003c/b>\u003cbr>' +\n 'Start: %{x}\u003cbr>' +\n 'End: %{customdata}\u003cbr>' +\n 'Length: %{marker.size}\u003cextra>\u003c/extra>',\n text=[f\"Chunk {j+1}\"],\n customdata=[chunk['end']],\n marker=dict(size=len(chunk['text'])),\n name=f\"Chunk {j+1}\"\n ),\n row=i+1, col=1\n )\n\n fig.update_layout(\n height=300 * n_strategies,\n title_text=\"Interactive Chunking Visualization\",\n showlegend=False\n )\n\n return fig\n\n def compare_chunk_size_distributions(self, text, chunk_sizes):\n \"\"\"Compare chunk size distributions\"\"\"\n fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 6))\n\n # Collect chunk sizes for each strategy\n all_chunk_sizes = []\n labels = []\n\n for size in chunk_sizes:\n strategy = {\"chunk_size\": size, \"overlap\": 0}\n chunks = self._apply_strategy(text, strategy)\n chunk_sizes_list = [len(chunk['text']) for chunk in chunks]\n all_chunk_sizes.append(chunk_sizes_list)\n labels.append(f\"Size {size}\")\n\n # Box plot\n ax1.boxplot(all_chunk_sizes, labels=labels)\n ax1.set_title('Chunk Size Distribution')\n ax1.set_ylabel('Chunk Size (characters)')\n ax1.grid(True, alpha=0.3)\n\n # Histogram\n for i, (sizes, label) in enumerate(zip(all_chunk_sizes, labels)):\n ax2.hist(sizes, alpha=0.7, label=label, bins=20)\n\n ax2.set_title('Chunk Size Histograms')\n ax2.set_xlabel('Chunk Size (characters)')\n ax2.set_ylabel('Frequency')\n ax2.legend()\n ax2.grid(True, alpha=0.3)\n\n plt.tight_layout()\n return fig\n\n def visualize_overlap_effects(self, text, base_chunk_size, overlap_percentages):\n \"\"\"Visualize effects of different overlap percentages\"\"\"\n fig, axes = plt.subplots(len(overlap_percentages), 1,\n figsize=(12, 3 * len(overlap_percentages)))\n\n if len(overlap_percentages) == 1:\n axes = [axes]\n\n for i, overlap in enumerate(overlap_percentages):\n strategy = {\"chunk_size\": base_chunk_size, \"overlap\": overlap}\n chunks = self._apply_strategy(text, strategy)\n\n self._visualize_single_strategy(text, strategy, axes[i])\n\n # Add overlap statistics\n total_overlap = sum(\n max(0, chunks[j-1]['end'] - chunks[j]['start'])\n for j in range(1, len(chunks))\n )\n total_text = len(text)\n overlap_percentage = (total_overlap / total_text) * 100 if total_text > 0 else 0\n\n axes[i].text(\n 0.02, 0.98, f\"Total Overlap: {overlap_percentage:.1f}%\",\n transform=axes[i].transAxes, ha='left', va='top',\n bbox=dict(boxstyle='round', facecolor='wheat', alpha=0.8)\n )\n\n plt.tight_layout()\n return fig\n```\n\n## 2. Plotly Interactive Visualizations\n\n### Advanced Interactive Dashboard\n\n```python\nimport plotly.graph_objects as go\nfrom plotly.subplots import make_subplots\nimport plotly.express as px\nimport pandas as pd\nimport numpy as np\nfrom typing import List, Dict\n\nclass InteractiveChunkingDashboard:\n def __init__(self):\n self.figures = {}\n\n def create_strategy_comparison_dashboard(self, text, strategies):\n \"\"\"Create comprehensive dashboard for strategy comparison\"\"\"\n # Create subplots\n fig = make_subplots(\n rows=3, cols=2,\n subplot_titles=[\n 'Chunk Boundaries', 'Chunk Size Distribution',\n 'Overlap Analysis', 'Coverage Analysis',\n 'Performance Metrics', 'Interactive Chunk Explorer'\n ],\n specs=[\n [{\"type\": \"scatter\"}, {\"type\": \"histogram\"}],\n [{\"type\": \"bar\"}, {\"type\": \"scatter\"}],\n [{\"type\": \"bar\"}, {\"type\": \"table\"}]\n ]\n )\n\n # Process all strategies\n strategy_data = {}\n for strategy in strategies:\n chunks = self._apply_chunking_strategy(text, strategy)\n strategy_data[strategy['name']] = {\n 'chunks': chunks,\n 'strategy': strategy\n }\n\n # 1. Chunk Boundaries Visualization\n self._add_chunk_boundaries(fig, strategy_data, row=1, col=1)\n\n # 2. Chunk Size Distribution\n self._add_size_distribution(fig, strategy_data, row=1, col=2)\n\n # 3. Overlap Analysis\n self._add_overlap_analysis(fig, strategy_data, row=2, col=1)\n\n # 4. Coverage Analysis\n self._add_coverage_analysis(fig, strategy_data, text, row=2, col=2)\n\n # 5. Performance Metrics\n self._add_performance_metrics(fig, strategy_data, row=3, col=1)\n\n # 6. Interactive Chunk Explorer\n self._add_chunk_explorer(fig, strategy_data, row=3, col=2)\n\n # Update layout\n fig.update_layout(\n height=1200,\n title_text=\"Comprehensive Chunking Strategy Analysis\",\n showlegend=True\n )\n\n return fig\n\n def _apply_chunking_strategy(self, text, strategy):\n \"\"\"Apply chunking strategy and return chunks\"\"\"\n # This would use actual chunking implementations\n # For now, simulate fixed-size chunking\n chunk_size = strategy.get('chunk_size', 100)\n overlap = strategy.get('overlap', 0)\n\n chunks = []\n start = 0\n chunk_id = 0\n\n while start \u003c len(text):\n end = min(start + chunk_size, len(text))\n chunks.append({\n 'id': chunk_id,\n 'text': text[start:end],\n 'start': start,\n 'end': end,\n 'size': end - start\n })\n chunk_id += 1\n\n # Calculate next start with overlap\n start = max(0, end - int(chunk_size * overlap / 100))\n\n if end >= len(text):\n break\n\n return chunks\n\n def _add_chunk_boundaries(self, fig, strategy_data, row, col):\n \"\"\"Add chunk boundaries visualization\"\"\"\n colors = px.colors.qualitative.Set3\n\n for i, (strategy_name, data) in enumerate(strategy_data.items()):\n chunks = data['chunks']\n\n for j, chunk in enumerate(chunks):\n fig.add_trace(\n go.Scatter(\n x=[chunk['start'], chunk['end']],\n y=[i, i],\n mode='lines+markers',\n line=dict(width=15, color=colors[i % len(colors)]),\n hovertemplate=f'\u003cb>{strategy_name}\u003c/b>\u003cbr>' +\n f'Chunk {j+1}\u003cbr>' +\n f'Start: {chunk[\"start\"]}\u003cbr>' +\n f'End: {chunk[\"end\"]}\u003cbr>' +\n f'Size: {chunk[\"size\"]}\u003cextra>\u003c/extra>',\n name=f'{strategy_name} - Chunk {j+1}' if j \u003c 3 else None,\n showlegend=j \u003c 3\n ),\n row=row, col=col\n )\n\n fig.update_xaxes(title_text=\"Character Position\", row=row, col=col)\n fig.update_yaxes(title_text=\"Strategy\", row=row, col=col)\n\n def _add_size_distribution(self, fig, strategy_data, row, col):\n \"\"\"Add chunk size distribution histogram\"\"\"\n for i, (strategy_name, data) in enumerate(strategy_data.items()):\n chunks = data['chunks']\n sizes = [chunk['size'] for chunk in chunks]\n\n fig.add_trace(\n go.Histogram(\n x=sizes,\n name=strategy_name,\n opacity=0.7,\n nbinsx=20\n ),\n row=row, col=col\n )\n\n fig.update_xaxes(title_text=\"Chunk Size (characters)\", row=row, col=col)\n fig.update_yaxes(title_text=\"Frequency\", row=row, col=col)\n\n def _add_overlap_analysis(self, fig, strategy_data, row, col):\n \"\"\"Add overlap analysis bar chart\"\"\"\n strategy_names = []\n overlap_totals = []\n overlap_percentages = []\n\n for strategy_name, data in strategy_data.items():\n chunks = data['chunks']\n strategy = data['strategy']\n overlap = strategy.get('overlap', 0)\n\n if overlap > 0:\n total_overlap = sum(\n max(0, chunks[j-1]['end'] - chunks[j]['start'])\n for j in range(1, len(chunks))\n )\n overlap_pct = (total_overlap / len(chunks[0]['text'])) * 100 if chunks else 0\n else:\n total_overlap = 0\n overlap_pct = 0\n\n strategy_names.append(strategy_name)\n overlap_totals.append(total_overlap)\n overlap_percentages.append(overlap_pct)\n\n fig.add_trace(\n go.Bar(\n x=strategy_names,\n y=overlap_percentages,\n name=\"Overlap %\",\n marker_color='lightblue'\n ),\n row=row, col=col\n )\n\n fig.update_xaxes(title_text=\"Strategy\", row=row, col=col)\n fig.update_yaxes(title_text=\"Overlap Percentage\", row=row, col=col)\n\n def _add_coverage_analysis(self, fig, strategy_data, text, row, col):\n \"\"\"Add coverage analysis scatter plot\"\"\"\n strategy_names = []\n chunk_counts = []\n avg_sizes = []\n coverage_scores = []\n\n for strategy_name, data in strategy_data.items():\n chunks = data['chunks']\n\n # Calculate coverage (simplified)\n total_covered = sum(chunk['size'] for chunk in chunks)\n coverage = (total_covered / len(text)) * 100 if text else 0\n\n strategy_names.append(strategy_name)\n chunk_counts.append(len(chunks))\n avg_sizes.append(np.mean([chunk['size'] for chunk in chunks]) if chunks else 0)\n coverage_scores.append(coverage)\n\n fig.add_trace(\n go.Scatter(\n x=chunk_counts,\n y=avg_sizes,\n mode='markers+text',\n text=strategy_names,\n textposition=\"top center\",\n marker=dict(\n size=coverage_scores,\n color=coverage_scores,\n colorscale='Viridis',\n showscale=True,\n colorbar=dict(title=\"Coverage %\", x=1.02)\n ),\n name=\"Strategies\"\n ),\n row=row, col=col\n )\n\n fig.update_xaxes(title_text=\"Number of Chunks\", row=row, col=col)\n fig.update_yaxes(title_text=\"Average Chunk Size\", row=row, col=col)\n\n def _add_performance_metrics(self, fig, strategy_data, row, col):\n \"\"\"Add performance metrics comparison\"\"\"\n metrics = ['Coherence', 'Speed', 'Memory', 'Accuracy']\n strategies = list(strategy_data.keys())\n\n # Simulated metrics (in real implementation, these would be calculated)\n metric_values = {\n 'Coherence': [0.8, 0.7, 0.9, 0.6],\n 'Speed': [0.9, 0.95, 0.7, 0.85],\n 'Memory': [0.7, 0.8, 0.6, 0.9],\n 'Accuracy': [0.75, 0.8, 0.85, 0.7]\n }\n\n for i, metric in enumerate(metrics):\n fig.add_trace(\n go.Bar(\n x=strategies,\n y=metric_values[metric][:len(strategies)],\n name=metric,\n opacity=0.8\n ),\n row=row, col=col\n )\n\n fig.update_xaxes(title_text=\"Strategy\", row=row, col=col)\n fig.update_yaxes(title_text=\"Score (0-1)\", row=row, col=col)\n\n def _add_chunk_explorer(self, fig, strategy_data, row, col):\n \"\"\"Add interactive chunk explorer table\"\"\"\n # Select first strategy for table display\n first_strategy = list(strategy_data.keys())[0]\n chunks = strategy_data[first_strategy]['chunks']\n\n # Create table data\n table_data = [\n [\n f\"Chunk {i+1}\",\n f\"{chunk['size']}\",\n f\"{chunk['start']}-{chunk['end']}\",\n chunk['text'][:50] + \"...\" if len(chunk['text']) > 50 else chunk['text']\n ]\n for i, chunk in enumerate(chunks[:10]) # Limit to 10 chunks\n ]\n\n fig.add_trace(\n go.Table(\n header=dict(\n values=['Chunk ID', 'Size', 'Position', 'Preview'],\n fill_color='lightblue',\n align='left'\n ),\n cells=dict(\n values=list(zip(*table_data)),\n fill_color='white',\n align='left'\n ),\n name=\"Chunk Details\"\n ),\n row=row, col=col\n )\n\n def create_real_time_analysis(self, text_input_callback):\n \"\"\"Create real-time analysis dashboard\"\"\"\n fig = go.Figure()\n\n # This would be connected to real-time data\n fig.add_trace(\n go.Scatter(\n x=[1, 2, 3, 4],\n y=[10, 11, 12, 13],\n mode='lines+markers',\n name='Real-time Performance'\n )\n )\n\n fig.update_layout(\n title=\"Real-time Chunking Analysis\",\n xaxis_title=\"Time\",\n yaxis_title=\"Performance Metric\",\n updatemenus=[{\n 'buttons': [\n {\n 'label': 'Play',\n 'method': 'animate',\n 'args': [None]\n },\n {\n 'label': 'Pause',\n 'method': 'animate',\n 'args': [[None], {'frame': {'duration': 0, 'redraw': False}}]\n }\n ],\n 'direction': 'left',\n 'pad': {'r': 10, 't': 87},\n 'showactive': False,\n 'x': 0.011,\n 'xanchor': 'right',\n 'y': 0,\n 'yanchor': 'top'\n }]\n )\n\n return fig\n```\n\n## 3. Streamlit Web Dashboard\n\n### Complete Web Application\n\n```python\nimport streamlit as st\nimport plotly.graph_objects as go\nimport plotly.express as px\nimport pandas as pd\nimport numpy as np\nfrom typing import List, Dict\nimport time\n\nclass StreamlitChunkingDashboard:\n def __init__(self):\n self.chunking_strategies = {\n \"Fixed Size - Small\": {\"chunk_size\": 100, \"overlap\": 0},\n \"Fixed Size - Medium\": {\"chunk_size\": 250, \"overlap\": 0},\n \"Fixed Size - Large\": {\"chunk_size\": 500, \"overlap\": 0},\n \"With Overlap - 10%\": {\"chunk_size\": 250, \"overlap\": 10},\n \"With Overlap - 25%\": {\"chunk_size\": 250, \"overlap\": 25},\n \"Semantic\": {\"method\": \"semantic\", \"threshold\": 0.7}\n }\n\n def run(self):\n \"\"\"Run the Streamlit dashboard\"\"\"\n st.set_page_config(\n page_title=\"Chunking Strategy Analyzer\",\n page_icon=\"📊\",\n layout=\"wide\"\n )\n\n st.title(\"🔪 Chunking Strategy Analyzer\")\n st.markdown(\"\"\"\n Analyze and visualize different chunking strategies for RAG systems.\n Compare performance metrics and find the optimal approach for your use case.\n \"\"\")\n\n # Sidebar for configuration\n self._render_sidebar()\n\n # Main content area\n tab1, tab2, tab3, tab4 = st.tabs([\"📊 Analysis\", \"🔍 Comparison\", \"📈 Metrics\", \"⚙️ Settings\"])\n\n with tab1:\n self._render_analysis_tab()\n\n with tab2:\n self._render_comparison_tab()\n\n with tab3:\n self._render_metrics_tab()\n\n with tab4:\n self._render_settings_tab()\n\n def _render_sidebar(self):\n \"\"\"Render sidebar configuration\"\"\"\n st.sidebar.header(\"Configuration\")\n\n # Text input\n input_method = st.sidebar.radio(\n \"Input Method\",\n [\"Sample Text\", \"Upload File\", \"Enter Text\"]\n )\n\n if input_method == \"Sample Text\":\n self._load_sample_text()\n elif input_method == \"Upload File\":\n self._handle_file_upload()\n else:\n self._handle_text_input()\n\n # Strategy selection\n st.sidebar.subheader(\"Select Strategies\")\n selected_strategies = []\n for strategy_name in self.chunking_strategies.keys():\n if st.sidebar.checkbox(strategy_name, value=True):\n selected_strategies.append(strategy_name)\n\n st.session_state.selected_strategies = selected_strategies\n\n # Analysis parameters\n st.sidebar.subheader(\"Analysis Parameters\")\n st.session_state.min_chunk_size = st.sidebar.slider(\n \"Min Chunk Size\", 50, 500, 100\n )\n st.session_state.max_chunk_size = st.sidebar.slider(\n \"Max Chunk Size\", 100, 2000, 500\n )\n\n def _load_sample_text(self):\n \"\"\"Load sample text for demonstration\"\"\"\n sample_texts = {\n \"Technical Document\": \"\"\"\n Machine learning is a subset of artificial intelligence that enables systems to learn and improve from experience without being explicitly programmed.\n Deep learning, a subset of machine learning, uses neural networks with multiple layers to progressively extract higher-level features from raw input.\n Natural Language Processing (NLP) combines computational linguistics with statistical and machine learning models to enable computers to process human language.\n Transformer models have revolutionized NLP by using attention mechanisms to weigh the importance of different words in the input text.\n Retrieval-Augmented Generation (RAG) enhances language models by retrieving relevant information from external knowledge bases.\n \"\"\",\n \"Legal Document\": \"\"\"\n This agreement is entered into on this date between the parties herein mentioned.\n The terms and conditions outlined herein shall govern the relationship between the parties.\n Any breach of contract shall result in immediate termination and possible legal action.\n The jurisdiction for any disputes shall be the courts of the specified region.\n All amendments must be made in writing and signed by both parties.\n \"\"\",\n \"Scientific Paper\": \"\"\"\n Our research demonstrates significant improvements in chunking strategies for RAG systems.\n We evaluated multiple approaches across different document types and query patterns.\n Experimental results show that semantic chunking outperforms fixed-size methods by 23% on average.\n Contextual retrieval provides additional benefits but requires increased computational resources.\n Future work will explore hybrid approaches that combine multiple strategies.\n \"\"\"\n }\n\n selected_sample = st.sidebar.selectbox(\n \"Select Sample Text\",\n list(sample_texts.keys())\n )\n\n st.session_state.input_text = sample_texts[selected_sample]\n\n def _handle_file_upload(self):\n \"\"\"Handle file upload\"\"\"\n uploaded_file = st.sidebar.file_uploader(\n \"Choose a file\",\n type=['txt', 'md', 'csv']\n )\n\n if uploaded_file is not None:\n try:\n content = uploaded_file.read().decode('utf-8')\n st.session_state.input_text = content\n st.sidebar.success(f\"Loaded {uploaded_file.name}\")\n except Exception as e:\n st.sidebar.error(f\"Error reading file: {e}\")\n\n def _handle_text_input(self):\n \"\"\"Handle manual text input\"\"\"\n text = st.sidebar.text_area(\n \"Enter your text:\",\n height=200\n )\n\n if text:\n st.session_state.input_text = text\n\n def _render_analysis_tab(self):\n \"\"\"Render analysis tab\"\"\"\n if 'input_text' not in st.session_state or not st.session_state.input_text:\n st.warning(\"Please input text using the sidebar configuration.\")\n return\n\n st.header(\"📊 Chunk Analysis\")\n\n # Display text statistics\n self._display_text_statistics()\n\n # Apply selected strategies\n if 'selected_strategies' in st.session_state and st.session_state.selected_strategies:\n strategy_results = self._apply_strategies()\n\n # Create visualizations\n col1, col2 = st.columns(2)\n\n with col1:\n self._create_chunk_boundary_chart(strategy_results)\n\n with col2:\n self._create_size_distribution_chart(strategy_results)\n\n # Detailed analysis\n st.subheader(\"Detailed Strategy Analysis\")\n self._create_detailed_analysis_table(strategy_results)\n\n def _render_comparison_tab(self):\n \"\"\"Render comparison tab\"\"\"\n if 'selected_strategies' not in st.session_state or not st.session_state.selected_strategies:\n st.warning(\"Please select strategies in the sidebar.\")\n return\n\n st.header(\"🔍 Strategy Comparison\")\n\n strategy_results = self._apply_strategies()\n\n # Comparison metrics\n col1, col2 = st.columns(2)\n\n with col1:\n self._create_performance_radar_chart(strategy_results)\n\n with col2:\n self._create_efficiency_comparison(strategy_results)\n\n # Detailed comparison table\n st.subheader(\"Detailed Comparison\")\n self._create_comparison_table(strategy_results)\n\n def _render_metrics_tab(self):\n \"\"\"Render metrics tab\"\"\"\n st.header(\"📈 Performance Metrics\")\n\n if 'input_text' not in st.session_state:\n st.warning(\"Please input text first.\")\n return\n\n # Metric selection\n selected_metrics = st.multiselect(\n \"Select Metrics to Display\",\n [\"Coverage\", \"Overlap\", \"Coherence\", \"Speed\", \"Memory Usage\"],\n default=[\"Coverage\", \"Overlap\", \"Coherence\"]\n )\n\n if selected_metrics:\n strategy_results = self._apply_strategies()\n self._create_metrics_dashboard(strategy_results, selected_metrics)\n\n def _render_settings_tab(self):\n \"\"\"Render settings tab\"\"\"\n st.header(\"⚙️ Advanced Settings\")\n\n # Chunking parameters\n st.subheader(\"Chunking Parameters\")\n col1, col2 = st.columns(2)\n\n with col1:\n st.number_input(\"Default Chunk Size\", value=250, min_value=50, max_value=2000)\n st.number_input(\"Default Overlap %\", value=10, min_value=0, max_value=50)\n\n with col2:\n st.selectbox(\"Default Strategy\", [\"Fixed Size\", \"Semantic\", \"Hybrid\"])\n st.slider(\"Similarity Threshold\", 0.0, 1.0, 0.7)\n\n # Visualization settings\n st.subheader(\"Visualization Settings\")\n st.checkbox(\"Show Overlap Regions\", value=True)\n st.checkbox(\"Show Chunk Statistics\", value=True)\n st.checkbox(\"Color Code by Strategy\", value=True)\n\n # Export settings\n st.subheader(\"Export Options\")\n export_format = st.selectbox(\"Export Format\", [\"PNG\", \"SVG\", \"PDF\", \"CSV\"])\n\n if st.button(\"Export Analysis\"):\n st.success(f\"Analysis exported as {export_format}\")\n\n def _display_text_statistics(self):\n \"\"\"Display input text statistics\"\"\"\n text = st.session_state.input_text\n word_count = len(text.split())\n char_count = len(text)\n sentence_count = text.count('.') + text.count('!') + text.count('?')\n\n col1, col2, col3, col4 = st.columns(4)\n\n with col1:\n st.metric(\"Characters\", f\"{char_count:,}\")\n with col2:\n st.metric(\"Words\", f\"{word_count:,}\")\n with col3:\n st.metric(\"Sentences\", f\"{sentence_count:,}\")\n with col4:\n st.metric(\"Avg Word Length\", f\"{char_count/word_count:.1f}\")\n\n def _apply_strategies(self):\n \"\"\"Apply selected chunking strategies\"\"\"\n text = st.session_state.input_text\n results = {}\n\n for strategy_name in st.session_state.selected_strategies:\n strategy = self.chunking_strategies[strategy_name]\n chunks = self._chunk_text(text, strategy)\n results[strategy_name] = {\n 'chunks': chunks,\n 'strategy': strategy,\n 'metrics': self._calculate_metrics(chunks, text)\n }\n\n return results\n\n def _chunk_text(self, text, strategy):\n \"\"\"Apply chunking strategy to text\"\"\"\n # Simplified chunking implementation\n if strategy.get('method') == 'semantic':\n # Simulate semantic chunking\n sentences = text.split('. ')\n chunks = []\n current_chunk = []\n\n for sentence in sentences:\n current_chunk.append(sentence)\n if len(' '.join(current_chunk)) > strategy.get('chunk_size', 250):\n chunks.append({\n 'text': '. '.join(current_chunk),\n 'size': len('. '.join(current_chunk))\n })\n current_chunk = []\n\n if current_chunk:\n chunks.append({\n 'text': '. '.join(current_chunk),\n 'size': len('. '.join(current_chunk))\n })\n\n else:\n # Fixed-size chunking\n chunk_size = strategy.get('chunk_size', 250)\n overlap = strategy.get('overlap', 0)\n overlap_chars = int(chunk_size * overlap / 100)\n\n chunks = []\n start = 0\n\n while start \u003c len(text):\n end = min(start + chunk_size, len(text))\n chunks.append({\n 'text': text[start:end],\n 'size': end - start\n })\n\n start = max(0, end - overlap_chars)\n if end >= len(text):\n break\n\n return chunks\n\n def _calculate_metrics(self, chunks, original_text):\n \"\"\"Calculate metrics for chunks\"\"\"\n if not chunks:\n return {}\n\n total_chars = sum(chunk['size'] for chunk in chunks)\n coverage = (total_chars / len(original_text)) * 100 if original_text else 0\n\n chunk_sizes = [chunk['size'] for chunk in chunks]\n avg_size = np.mean(chunk_sizes) if chunk_sizes else 0\n size_variance = np.var(chunk_sizes) if chunk_sizes else 0\n\n # Simulated metrics\n coherence_score = np.random.uniform(0.6, 0.9)\n speed_score = 1.0 - (len(chunks) / 100) # More chunks = slower\n memory_score = 1.0 - (total_chars / 10000) # More text = more memory\n\n return {\n 'num_chunks': len(chunks),\n 'avg_chunk_size': avg_size,\n 'coverage': coverage,\n 'size_variance': size_variance,\n 'coherence': coherence_score,\n 'speed': speed_score,\n 'memory': memory_score\n }\n\n def _create_chunk_boundary_chart(self, strategy_results):\n \"\"\"Create chunk boundary visualization\"\"\"\n fig = go.Figure()\n\n colors = px.colors.qualitative.Set3\n\n for i, (strategy_name, data) in enumerate(strategy_results.items()):\n chunks = data['chunks']\n\n for j, chunk in enumerate(chunks):\n # Create chunk visualization\n fig.add_trace(\n go.Scatter(\n x=[j, j + 1],\n y=[i, i],\n mode='lines',\n line=dict(width=chunk['size']/10, color=colors[i % len(colors)]),\n hovertemplate=f'\u003cb>{strategy_name}\u003c/b>\u003cbr>' +\n f'Chunk {j+1}\u003cbr>' +\n f'Size: {chunk[\"size\"]}\u003cextra>\u003c/extra>',\n name=f'{strategy_name}' if j == 0 else None,\n showlegend=j == 0\n )\n )\n\n fig.update_layout(\n title=\"Chunk Boundaries Visualization\",\n xaxis_title=\"Chunk Index\",\n yaxis_title=\"Strategy\",\n height=400\n )\n\n st.plotly_chart(fig, use_container_width=True)\n\n def _create_size_distribution_chart(self, strategy_results):\n \"\"\"Create size distribution chart\"\"\"\n fig = go.Figure()\n\n for strategy_name, data in strategy_results.items():\n chunk_sizes = [chunk['size'] for chunk in data['chunks']]\n\n fig.add_trace(\n go.Histogram(\n x=chunk_sizes,\n name=strategy_name,\n opacity=0.7,\n nbinsx=20\n )\n )\n\n fig.update_layout(\n title=\"Chunk Size Distribution\",\n xaxis_title=\"Chunk Size (characters)\",\n yaxis_title=\"Frequency\",\n height=400\n )\n\n st.plotly_chart(fig, use_container_width=True)\n\n def _create_performance_radar_chart(self, strategy_results):\n \"\"\"Create performance radar chart\"\"\"\n metrics = ['Coherence', 'Speed', 'Memory', 'Coverage', 'Consistency']\n\n fig = go.Figure()\n\n for strategy_name, data in strategy_results.items():\n metrics_data = data['metrics']\n values = [\n metrics_data.get('coherence', 0.5),\n metrics_data.get('speed', 0.5),\n metrics_data.get('memory', 0.5),\n metrics_data.get('coverage', 0.5) / 100,\n 1.0 - (metrics_data.get('size_variance', 0) / 10000) # Inverse variance\n ]\n\n fig.add_trace(\n go.Scatterpolar(\n r=values,\n theta=metrics,\n fill='toself',\n name=strategy_name\n )\n )\n\n fig.update_layout(\n polar=dict(\n radialaxis=dict(\n visible=True,\n range=[0, 1]\n )\n ),\n title=\"Performance Comparison\",\n height=400\n )\n\n st.plotly_chart(fig, use_container_width=True)\n\n def _create_efficiency_comparison(self, strategy_results):\n \"\"\"Create efficiency comparison chart\"\"\"\n strategies = list(strategy_results.keys())\n processing_times = [np.random.uniform(0.5, 2.0) for _ in strategies] # Simulated\n memory_usage = [np.random.uniform(10, 50) for _ in strategies] # Simulated\n\n fig = go.Figure()\n\n # Add processing time\n fig.add_trace(\n go.Scatter(\n x=strategies,\n y=processing_times,\n mode='markers+lines',\n name='Processing Time (s)',\n marker=dict(size=10),\n line=dict(width=2)\n )\n )\n\n # Add memory usage\n fig.add_trace(\n go.Scatter(\n x=strategies,\n y=memory_usage,\n mode='markers+lines',\n name='Memory Usage (MB)',\n marker=dict(size=10),\n line=dict(width=2),\n yaxis='y2'\n )\n )\n\n fig.update_layout(\n title=\"Efficiency Comparison\",\n xaxis_title=\"Strategy\",\n yaxis=dict(title=\"Processing Time (s)\"),\n yaxis2=dict(title=\"Memory Usage (MB)\", overlaying=\"y\", side=\"right\"),\n height=400\n )\n\n st.plotly_chart(fig, use_container_width=True)\n\n def _create_detailed_analysis_table(self, strategy_results):\n \"\"\"Create detailed analysis table\"\"\"\n table_data = []\n\n for strategy_name, data in strategy_results.items():\n metrics = data['metrics']\n table_data.append({\n 'Strategy': strategy_name,\n 'Chunks': metrics['num_chunks'],\n 'Avg Size': f\"{metrics['avg_chunk_size']:.0f}\",\n 'Coverage (%)': f\"{metrics['coverage']:.1f}\",\n 'Coherence': f\"{metrics['coherence']:.2f}\",\n 'Speed': f\"{metrics['speed']:.2f}\",\n 'Memory': f\"{metrics['memory']:.2f}\"\n })\n\n df = pd.DataFrame(table_data)\n st.dataframe(df, use_container_width=True)\n\n def _create_comparison_table(self, strategy_results):\n \"\"\"Create detailed comparison table\"\"\"\n comparison_data = []\n\n metrics = ['num_chunks', 'avg_chunk_size', 'coverage', 'coherence', 'speed', 'memory']\n metric_names = ['Chunks', 'Avg Size', 'Coverage %', 'Coherence', 'Speed', 'Memory']\n\n for i, metric in enumerate(metrics):\n row = {'Metric': metric_names[i]}\n for strategy_name, data in strategy_results.items():\n value = data['metrics'].get(metric, 0)\n if metric == 'coverage':\n row[strategy_name] = f\"{value:.1f}%\"\n else:\n row[strategy_name] = f\"{value:.3f}\"\n comparison_data.append(row)\n\n df = pd.DataFrame(comparison_data)\n st.dataframe(df, use_container_width=True)\n\n def _create_metrics_dashboard(self, strategy_results, selected_metrics):\n \"\"\"Create metrics dashboard\"\"\"\n for metric in selected_metrics:\n st.subheader(f\"{metric} Analysis\")\n\n if metric == \"Coverage\":\n self._create_coverage_chart(strategy_results)\n elif metric == \"Overlap\":\n self._create_overlap_chart(strategy_results)\n elif metric == \"Coherence\":\n self._create_coherence_chart(strategy_results)\n elif metric == \"Speed\":\n self._create_speed_chart(strategy_results)\n elif metric == \"Memory Usage\":\n self._create_memory_chart(strategy_results)\n\n def _create_coverage_chart(self, strategy_results):\n \"\"\"Create coverage analysis chart\"\"\"\n strategies = list(strategy_results.keys())\n coverage_values = [data['metrics']['coverage'] for data in strategy_results.values()]\n\n fig = go.Figure(data=[\n go.Bar(x=strategies, y=coverage_values, marker_color='lightblue')\n ])\n\n fig.update_layout(\n title=\"Text Coverage by Strategy\",\n xaxis_title=\"Strategy\",\n yaxis_title=\"Coverage (%)\",\n height=300\n )\n\n st.plotly_chart(fig, use_container_width=True)\n\n def _create_overlap_chart(self, strategy_results):\n \"\"\"Create overlap analysis chart\"\"\"\n # Simulated overlap data\n strategies = list(strategy_results.keys())\n overlap_data = []\n\n for strategy_name, data in strategy_results.items():\n strategy = data['strategy']\n overlap = strategy.get('overlap', 0)\n overlap_data.append(overlap)\n\n fig = go.Figure(data=[\n go.Bar(x=strategies, y=overlap_data, marker_color='lightcoral')\n ])\n\n fig.update_layout(\n title=\"Overlap Percentage by Strategy\",\n xaxis_title=\"Strategy\",\n yaxis_title=\"Overlap (%)\",\n height=300\n )\n\n st.plotly_chart(fig, use_container_width=True)\n\n def _create_coherence_chart(self, strategy_results):\n \"\"\"Create coherence analysis chart\"\"\"\n strategies = list(strategy_results.keys())\n coherence_values = [data['metrics']['coherence'] for data in strategy_results.values()]\n\n fig = go.Figure(data=[\n go.Bar(x=strategies, y=coherence_values, marker_color='lightgreen')\n ])\n\n fig.update_layout(\n title=\"Coherence Score by Strategy\",\n xaxis_title=\"Strategy\",\n yaxis_title=\"Coherence Score\",\n yaxis=dict(range=[0, 1]),\n height=300\n )\n\n st.plotly_chart(fig, use_container_width=True)\n\n def _create_speed_chart(self, strategy_results):\n \"\"\"Create speed analysis chart\"\"\"\n strategies = list(strategy_results.keys())\n speed_values = [data['metrics']['speed'] for data in strategy_results.values()]\n\n fig = go.Figure(data=[\n go.Bar(x=strategies, y=speed_values, marker_color='gold')\n ])\n\n fig.update_layout(\n title=\"Processing Speed by Strategy\",\n xaxis_title=\"Strategy\",\n yaxis_title=\"Speed Score\",\n height=300\n )\n\n st.plotly_chart(fig, use_container_width=True)\n\n def _create_memory_chart(self, strategy_results):\n \"\"\"Create memory usage analysis chart\"\"\"\n strategies = list(strategy_results.keys())\n memory_values = [data['metrics']['memory'] for data in strategy_results.values()]\n\n fig = go.Figure(data=[\n go.Bar(x=strategies, y=memory_values, marker_color='plum')\n ])\n\n fig.update_layout(\n title=\"Memory Usage by Strategy\",\n xaxis_title=\"Strategy\",\n yaxis_title=\"Memory Score\",\n height=300\n )\n\n st.plotly_chart(fig, use_container_width=True)\n\n\n# Run the dashboard\nif __name__ == \"__main__\":\n dashboard = StreamlitChunkingDashboard()\n dashboard.run()\n```\n\n## 4. Command Line Usage Examples\n\n### Basic Visualization\n\n```python\n# Example 1: Basic ChunkViz usage\nfrom chunkviz.advanced import AdvancedChunkViz\n\ntext = \"Your long document text here...\"\nviz = AdvancedChunkViz()\n\n# Compare multiple strategies\nstrategies = [\n {\"name\": \"Small Chunks\", \"chunk_size\": 100, \"overlap\": 0},\n {\"name\": \"Medium Chunks\", \"chunk_size\": 250, \"overlap\": 0},\n {\"name\": \"Large Chunks\", \"chunk_size\": 500, \"overlap\": 0},\n {\"name\": \"With Overlap\", \"chunk_size\": 250, \"overlap\": 20}\n]\n\nviz.visualize_multiple_strategies(text, strategies, save_path=\"chunking_comparison.png\")\n\n# Analyze overlap effects\nviz.visualize_overlap_effects(text, 250, [0, 10, 20, 30], save_path=\"overlap_analysis.png\")\n```\n\n### Interactive Dashboard\n\n```python\n# Example 2: Interactive dashboard\nfrom chunkviz.dashboard import InteractiveChunkingDashboard\n\ndashboard = InteractiveChunkingDashboard()\n\n# Create comprehensive analysis\nstrategies = [\n {\"name\": \"Fixed-100\", \"chunk_size\": 100, \"overlap\": 0},\n {\"name\": \"Fixed-250\", \"chunk_size\": 250, \"overlap\": 0},\n {\"name\": \"Semantic\", \"method\": \"semantic\"}\n]\n\nfig = dashboard.create_strategy_comparison_dashboard(text, strategies)\nfig.show()\n\n# Create size distribution comparison\nfig = dashboard.compare_chunk_size_distributions(text, [100, 200, 300, 400, 500])\nfig.show()\n```\n\n### Streamlit Web Application\n\n```python\n# Example 3: Run Streamlit dashboard\n# Save as app.py and run: streamlit run app.py\n\nfrom chunkviz.streamlit_app import StreamlitChunkingDashboard\n\ndashboard = StreamlitChunkingDashboard()\ndashboard.run()\n```\n\nThese visualization and evaluation tools provide comprehensive capabilities for analyzing, comparing, and optimizing chunking strategies across different use cases and requirements.","content_type":"text/markdown; charset=utf-8","language":"markdown","size":48928,"content_sha256":"3cbd984b0a6820e250915e5017f2c11c82d4d69f4455882925e1e83c43b5a55a"}],"content_json":{"type":"doc","content":[{"type":"heading","attrs":{"level":1},"content":[{"text":"Chunking Strategy for RAG Systems","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Overview","type":"text"}]},{"type":"paragraph","content":[{"text":"Provides chunking strategies for RAG systems, vector databases, and document processing. Recommends chunk sizes, overlap percentages, and boundary detection methods; validates semantic coherence; evaluates retrieval metrics.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"When to Use","type":"text"}]},{"type":"paragraph","content":[{"text":"Use when building or optimizing RAG systems, vector search pipelines, document chunking workflows, or performance-tuning existing systems with poor retrieval quality.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Instructions","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Choose Chunking Strategy","type":"text"}]},{"type":"paragraph","content":[{"text":"Select based on document type and use case:","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Fixed-Size Chunking","type":"text","marks":[{"type":"strong"}]},{"text":" (Level 1)","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Use for simple documents without clear structure","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Start with 512 tokens and 10-20% overlap","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Adjust: 256 for factoid queries, 1024 for analytical","type":"text"}]}]}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Recursive Character Chunking","type":"text","marks":[{"type":"strong"}]},{"text":" (Level 2)","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Use for documents with structural boundaries","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Hierarchical separators: paragraphs → sentences → words","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Customize for document types (HTML, Markdown, JSON)","type":"text"}]}]}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Structure-Aware Chunking","type":"text","marks":[{"type":"strong"}]},{"text":" (Level 3)","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Use for structured content (Markdown, code, tables, PDFs)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Preserve semantic units: functions, sections, table blocks","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Validate structure preservation post-split","type":"text"}]}]}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Semantic Chunking","type":"text","marks":[{"type":"strong"}]},{"text":" (Level 4)","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Use for complex documents with thematic shifts","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Embedding-based boundary detection with 0.8 similarity threshold","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Buffer size: 3-5 sentences","type":"text"}]}]}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Advanced Methods","type":"text","marks":[{"type":"strong"}]},{"text":" (Level 5)","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Late Chunking for long-context models","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Contextual Retrieval for high-precision requirements","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Monitor computational cost vs. retrieval gain","type":"text"}]}]}]}]}]},{"type":"paragraph","content":[{"text":"Reference: ","type":"text"},{"text":"references/strategies.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/strategies.md","title":null}}]},{"text":".","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Implement Chunking Pipeline","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Pre-process documents","type":"text","marks":[{"type":"strong"}]}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Analyze structure, content types, information density","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Identify multi-modal content (tables, images, code)","type":"text"}]}]}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Select parameters","type":"text","marks":[{"type":"strong"}]}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Chunk size: embedding model context window / 4","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Overlap: 10-20% for most cases","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Strategy-specific settings","type":"text"}]}]}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Process and validate","type":"text","marks":[{"type":"strong"}]}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Apply chunking strategy","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Validate coherence: run ","type":"text"},{"text":"evaluate_chunks.py --coherence","type":"text","marks":[{"type":"code_inline"}]},{"text":" (see below)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Test with representative documents","type":"text"}]}]}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Evaluate and iterate","type":"text","marks":[{"type":"strong"}]}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Measure precision and recall","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"If precision \u003c 0.7: reduce chunk_size by 25% and re-evaluate","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"If recall \u003c 0.6: increase overlap by 10% and re-evaluate","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Monitor latency and memory usage","type":"text"}]}]}]}]}]},{"type":"paragraph","content":[{"text":"Reference: ","type":"text"},{"text":"references/implementation.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/implementation.md","title":null}}]},{"text":".","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Validate Chunk Quality","type":"text"}]},{"type":"paragraph","content":[{"text":"Run validation commands to assess chunk quality:","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"# Check semantic coherence (requires sentence-transformers)\npython -c \"\nfrom sentence_transformers import SentenceTransformer\nmodel = SentenceTransformer('all-MiniLM-L6-v2')\nchunks = [...] # your chunks\nembeddings = model.encode(chunks)\nsimilarity = (embeddings @ embeddings.T).mean()\nprint(f'Cohesion: {similarity:.3f}') # target: 0.3-0.7\n\"\n\n# Measure retrieval precision\npython -c \"\nrelevant = sum(1 for c in retrieved if c in relevant_chunks)\nprecision = relevant / len(retrieved)\nprint(f'Precision: {precision:.2f}') # target: >= 0.7\n\"\n\n# Check chunk size distribution\npython -c \"\nimport numpy as np\nsizes = [len(c.split()) for c in chunks]\nprint(f'Mean: {np.mean(sizes):.0f}, Std: {np.std(sizes):.0f}')\nprint(f'Min: {min(sizes)}, Max: {max(sizes)}')\n\"","type":"text"}]},{"type":"paragraph","content":[{"text":"Reference: ","type":"text"},{"text":"references/evaluation.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/evaluation.md","title":null}}]},{"text":".","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Examples","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Fixed-Size Chunking","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"from langchain.text_splitter import RecursiveCharacterTextSplitter\n\nsplitter = RecursiveCharacterTextSplitter(\n chunk_size=256,\n chunk_overlap=25,\n length_function=len\n)\nchunks = splitter.split_documents(documents)","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Structure-Aware Code Chunking","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"import ast\n\ndef chunk_python_code(code):\n tree = ast.parse(code)\n chunks = []\n for node in ast.walk(tree):\n if isinstance(node, (ast.FunctionDef, ast.ClassDef)):\n chunks.append(ast.get_source_segment(code, node))\n return chunks","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Semantic Chunking","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"def semantic_chunk(text, similarity_threshold=0.8):\n sentences = split_into_sentences(text)\n embeddings = generate_embeddings(sentences)\n chunks, current = [], [sentences[0]]\n for i in range(1, len(sentences)):\n sim = cosine_similarity(embeddings[i-1], embeddings[i])\n if sim \u003c similarity_threshold:\n chunks.append(\" \".join(current))\n current = [sentences[i]]\n else:\n current.append(sentences[i])\n chunks.append(\" \".join(current))\n return chunks","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Best Practices","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Core Principles","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Balance context preservation with retrieval precision","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Maintain semantic coherence within chunks","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Optimize for embedding model context window constraints","type":"text"}]}]}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Implementation","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Start with fixed-size (512 tokens, 15% overlap)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Iterate based on document characteristics","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Test with domain-specific documents before deployment","type":"text"}]}]}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Pitfalls to Avoid","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Over-chunking: context-poor small chunks","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Under-chunking: missing information in oversized chunks","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Ignoring semantic boundaries and document structure","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"One-size-fits-all for diverse content types","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Constraints and Warnings","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Resource Considerations","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Semantic methods require significant compute resources","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Late chunking needs long-context embedding models","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Complex strategies increase processing latency","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Monitor memory for large document batches","type":"text"}]}]}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Quality Requirements","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Validate semantic coherence post-processing","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Test with representative documents before deployment","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Ensure chunks maintain standalone meaning","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Implement error handling for malformed content","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"References","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"strategies.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/strategies.md","title":null}}]},{"text":" - Detailed strategies","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"implementation.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/implementation.md","title":null}}]},{"text":" - Implementation guidelines","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"evaluation.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/evaluation.md","title":null}}]},{"text":" - Performance metrics","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"tools.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/tools.md","title":null}}]},{"text":" - Libraries and frameworks","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"research.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/research.md","title":null}}]},{"text":" - Research papers","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"advanced-strategies.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/advanced-strategies.md","title":null}}]},{"text":" - 11 advanced methods","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"semantic-methods.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/semantic-methods.md","title":null}}]},{"text":" - Semantic approaches","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"visualization-tools.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/visualization-tools.md","title":null}}]},{"text":" - Visualization tools","type":"text"}]}]}]}]},"metadata":{"date":"2026-11-02","name":"chunking-strategy","author":"@skillopedia","source":{"stars":366,"repo_name":"developer-kit","origin_url":"https://github.com/giuseppe-trisciuoglio/developer-kit/blob/HEAD/plugins/developer-kit-ai/skills/chunking-strategy/SKILL.md","repo_owner":"giuseppe-trisciuoglio","body_sha256":"50e95b88a2e015994fab6e6431a98254713ffe3067146a772d9cd6be3e15c643","cluster_key":"28068e06cff69f17a401e3f7d40ce2085ef534c3abb7a3db74a2ae9403f22a58","clean_bundle":{"format":"clean-skill-bundle-v1","source":"giuseppe-trisciuoglio/developer-kit/plugins/developer-kit-ai/skills/chunking-strategy/SKILL.md","text_bytes":248246,"source_commit":"fe73fb33a09ee6562eeec7415a3f9a3bcf795177","skills_sh_rank":5296,"attachment_count":8,"text_attachments":8,"attachment_storage":"skill_attachments","binary_attachments":0},"cluster_size":1,"skill_md_path":"plugins/developer-kit-ai/skills/chunking-strategy/SKILL.md","import_metadata":{"date":"2026-11-02","author":"@skillopedia","version":"v1","category":"ai-agent-development","category_label":"AI"},"exact_dupes_collapsed_into_this":0},"version":"v1","category":"ai-agent-development","import_tag":"skills-sh-top20k-2026-11","description":"Provides chunking strategies for RAG systems. Generates chunk size recommendations (256-1024 tokens), overlap percentages (10-20%), and semantic boundary detection methods. Validates semantic coherence and evaluates retrieval precision/recall metrics. Use when building retrieval-augmented generation systems, vector databases, or processing large documents.","allowed-tools":"Read, Write, Bash"}},"renderedAt":1793666233614}

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.