# Matriz de Políticas de Indexación — Klaus Knowledge Base
**Fecha:** 2026-07-28
**Escala:** 9,368 archivos | 9.5GB | 11 proyectos
**Objetivo:** Validación en UNA PASADA de todos los tipos de archivos con políticas diferenciadas
---
## I. CATEGORIZACIÓN SEMÁNTICA DE ARCHIVOS
Klaus está indexando 5 categorías semánticas principales, no solo extensiones. Cada categoría tiene necesidades diferentes de chunking, embedding y validación.
### Matriz de Políticas Diferenciadas
| **Categoría** | **Extensiones** | **Archivos Est.** | **Chunks/Archivo** | **Chunking** | **Embedding** | **Indexar?** | **Razón** | **Prioridad Validación** |
|---|---|---|---|---|---|---|---|---|
| **CODE_PYTHON** | .py | ~219 | 5-50 | AST-aware (función/clase) | Code-aware + símbolo | ✓ SÍ | Máximo valor semántico | CRÍTICA |
| **CODE_TYPESCRIPT** | .ts, .tsx | ~54 | 3-30 | AST → functions/classes | Code-aware + símbolo | ✓ SÍ | High-value queries | CRÍTICA |
| **CODE_JAVASCRIPT** | .js, .jsx | ~54 | 3-30 | Fixed (500c) + symbol extract | Code-aware + símbolo | ✓ SÍ | High-value queries | CRÍTICA |
| **CODE_SHELL** | .sh, .bash | ~75 | 2-15 | Fixed (400c) + extract symbols | Code-aware | ✓ SÍ | Automation scripts | ALTA |
| **CODE_OTHER** | .go, .rs, .sql, .java, .hcl, .tf | ~60+ | 3-20 | Fixed (500c) + extract symbols | Code-aware | ✓ SÍ | Infrastructure/queries | ALTA |
| **CONFIG_STRUCTURED** | .json, .yaml, .yml, .toml | ~2,854 | 1-5 | Semantic sections + keys | Structured + metadata | ✓ SÍ (filtrado) | Config patterns | MEDIA |
| **CONFIG_UNSTRUCTURED** | .ini, .conf, .cfg, .env | ~50+ | 1-3 | Line-by-line + context | Minimal | ⚠️ SELECTIVO | Noise vs signal | BAJA |
| **DOCUMENTATION** | .md, .rst, .txt | ~742 | 2-10 | Heading-split (H2+H3) | General + context | ✓ SÍ | Critical for context | CRÍTICA |
| **MARKUP** | .html, .xml, .svg | ~52 | 1-3 | Structure-aware (DOM nodes) | Minimal | ⚠️ SELECTIVO | Noise vs semantics | MEDIA |
| **STYLES** | .css, .scss, .less, .sass | ~54+ | 2-8 | Rule-group split | Minimal | ⚠️ SELECTIVO | Limited semantics | BAJA |
| **DATA_TABULAR** | .csv, .tsv, .sql (data) | Varios | Variable | Row sampling (10-15 rows) | Metadata-only | ❌ NO | Too large, low value | BAJA |
| **BINARY_IMAGES** | .png, .jpg, .gif, .svg (binary) | ~130 | N/A | N/A | N/A | ❌ NO | Non-textual, can't embed | CRÍTICA SKIP |
| **BINARY_ARCHIVES** | .zip, .tar, .gz, .7z, .rar | Varios | N/A | N/A | N/A | ❌ NO | Extracted content only | CRÍTICA SKIP |
| **BINARY_MEDIA** | .mp4, .mp3, .wav, .mov, .pdf | Varios | N/A | N/A | N/A | ❌ NO | Non-textual or external OCR | CRÍTICA SKIP |
| **GENERATED_LOGS** | .log, .out, .txt (logs) | Varios | Sampling | Line sampling (últimas 50) | Timestamp + context | ❌ NO | Noise, volatile | CRÍTICA SKIP |
| **TEST_FIXTURES** | .json (fixtures), test data | Varios | Sampling | N/A | Metadata-only | ❌ NO | Noise, not real code | MEDIA SKIP |
| **VENDORED** | node_modules/, vendor/, dist/ | Millones | N/A | N/A | N/A | ❌ SKIP (dir) | Already excluded | CRÍTICA SKIP |
| **BUILD_ARTIFACTS** | .o, .a, .so, .dll, .exe, .class | Varios | N/A | N/A | N/A | ❌ SKIP (dir) | Compiled binaries | CRÍTICA SKIP |
---
## II. PARÁMETROS DE CHUNKING POR CATEGORÍA
### **Código Python (AST-aware)**
```
├─ Estrategia: AST parsing → top-level functions/classes (1 chunk por def)
├─ Module-level code: fixed-size (400 chars, 80 overlap)
├─ Tamaño mínimo: 40 chars
├─ Tamaño máximo: unbounded (preserve context)
├─ Overlap: 0 (AST boundaries don't overlap)
├─ Fallback: fixed-size si SyntaxError
└─ Symbol enrichment: function/class name + docstring detection
```
### **TypeScript/JavaScript (Adaptive)**
```
├─ Estrategia: Symbol extraction + fixed chunking
├─ Símbolos detectados: function, class, export default, const assignments
├─ Chunking: fixed-size (500 chars, 100 overlap) si no hay AST
├─ Fallback: simple token-based (para .jsx con JSX syntax)
├─ Symbol enrichment: extracted name + type (function/class/export)
└─ Bonus: +20% quality score si contiene type annotations (TypeScript)
```
### **Shell Scripts**
```
├─ Estrategia: function-aware + fixed chunking
├─ Símbolos: function name() { ... } patterns
├─ Chunking: fixed-size (400 chars, 80 overlap)
├─ Líneas especiales: preserve shebang (#!) in context
├─ Comment preservation: keep inline documentation
└─ Complexity penalty: -10% si >50% comments (likely over-documented)
```
### **Config Structured (JSON/YAML/TOML)**
```
├─ Estrategia: Schema-aware semantic sections
├─ Secciones: top-level keys = chunks (if value < 1000 chars)
├─ Valor grande (>1000): split por sub-keys o fixed-size (800 chars)
├─ Chunking: preserve structure (no split mid-object)
├─ Symbol enrichment: key path (e.g., "database.connection.timeout")
├─ Metadata: type hints from schema (string, int, bool, object, array)
└─ Filtering: exclude value if matches _LLM_RESPONSE_PATTERNS
```
### **Documentation (Markdown)**
```
├─ Estrategia: Heading-split (H1 → section, H2/H3 → subsections)
├─ Chunking:
│ ├─ Level 1 (H1): per-file section
│ ├─ Level 2+ (H2-H6): split here, fixed-size within section
│ └─ Fixed: 800 chars, 120 overlap per subsection
├─ Symbol enrichment: heading text as symbol_name
├─ Type annotation: symbol_type = "section"
├─ Metadata: toc_level, in_code_block (skip code fences in chunks)
└─ Filtering: skip pure frontmatter (YAML block), code blocks >2000 chars
```
### **Other Code (Go, Rust, SQL, HCL, TF)**
```
├─ Estrategia: Language-specific symbol extraction + fixed
├─ Símbolos Go: func name, type name struct/interface
├─ Símbolos Rust: fn name, struct, enum, trait, impl
├─ Símbolos SQL: CREATE TABLE/PROCEDURE/FUNCTION name, SELECT blocks
├─ Chunking: fixed-size (500 chars, 100 overlap) si no hay symbols
├─ Symbol enrichment: detected name + type (function/struct/query)
└─ Language metadata: para syntax highlighting contextual en RAG
```
### **Config Unstructured (.ini, .conf, .cfg)**
```
├─ Estrategia: Line-by-line + minimal chunking
├─ Secciones: [section] headers = context markers
├─ Chunking: key=value pairs, grupo por [section] (~5 lines per chunk)
├─ Tamaño: ~200 chars, 30 overlap
├─ Filtering: skip empty lines, pure comments
└─ Warn: if >50% lines are comments (likely low-value)
```
### **Markup (HTML, XML)**
```
├─ Estrategia: DOM node-aware (si es XML bien-formado)
├─ Secciones: <section>, <article>, major tags
├─ Chunking: fixed-size (600 chars, 100 overlap) si parsing falla
├─ Exclude:
│ ├─ Script tags (<script>...</script>)
│ ├─ Style tags (<style>...</style>)
│ └─ Comments (<!-- ... -->)
├─ Symbol extraction: id, class, role attributes → context
└─ Quality filter: -50% score si >80% whitespace
```
### **Styles (CSS, SCSS)**
```
├─ Estrategia: Rule-group split (selector + declarations)
├─ Secciones: @media queries, @keyframes como boundaries
├─ Chunking:
│ ├─ Rule sets: .selector { ... } = 1 chunk (if <300 chars)
│ └─ Large rules: split por property groups (~200 chars, 30 overlap)
├─ Symbol: selector name (e.g., ".btn-primary")
├─ Filter: -80% score si no contiene ninguna property (just selectors)
└─ Exclude: vendor prefixes from symbol extraction
```
---
## III. POLÍTICAS DE EMBEDDING
### **Embedding Model Strategy**
**Estrategia actual (validada):**
- Dense vector: `text-embedding-3-small` via Anthropic (1536 dims)
- Sparse vector: BM25-lite hash (30k vocab, TF log-scaling)
- Fusion: RRF (Reciprocal Rank Fusion) para hybrid search
**Prefijo semántico (enriquecido):**
```python
def embed_text(content, file_path, symbol_name="", symbol_type=""):
prefix = f"{file_path}: "
if symbol_name:
kind = symbol_type if symbol_type in ("function", "class", ...) else "function"
prefix += f"{kind} {symbol_name} — "
return f"{prefix}{content}"
```
### **Prefijos por Categoría**
| Categoría | Prefijo Template | Ejemplo |
|---|---|---|
| CODE_PYTHON | `{path}: function {name} —` | `proxy/knowledge.py: function score_chunk_quality —` |
| CODE_TS/JS | `{path}: {type} {name} —` | `src/api.ts: function authenticate —` |
| DOCUMENTATION | `{path}: section {heading} —` | `docs/README.md: section Installation —` |
| CONFIG_STRUCTURED | `{path}: config {key_path} —` | `config/app.yaml: config database.connection —` |
| Other | `{path}:` | `scripts/deploy.sh:` |
**Threshold de similitud:**
- KB general: 0.75 (permisivo para documentación)
- Symbol-specific queries: 0.85 * 0.75 = 0.638 (reduced for variants)
- Duplicate detection: 0.92 (estricto)
---
## IV. DECISIONES DE INDEXACIÓN
### **INDEXAR (✓ SÍ)**
- **Code** (Python, TypeScript, JavaScript, Shell, Go, Rust, SQL, HCL, Terraform)
- Razón: Máxima densidad semántica, alto ROI para Q&A técnico
- Estrategia: AST-aware cuando sea posible, symbol extraction cuando no
- **Documentation** (Markdown, reStructuredText)
- Razón: Context critical, heading-driven chunking minimiza ruido
- Estrategia: H2+ split, preserve examples
- **Config Structured** (JSON, YAML, TOML)
- Razón: Schema patterns are queryable ("what's the timeout config?")
- **PERO:** Filtrar archivos que:
- Contengan `_LLM_RESPONSE_PATTERNS` (responses indexadas)
- Sean >50MB (bloat, probablemente fixtures o datos)
- Sean `.json.sample`, `.example`, `.template` (templated, not real)
- Estrategia: Semantic section split, preserve keys
### **INDEXAR SELECTIVAMENTE (⚠️)**
- **Config Unstructured** (.ini, .conf, .cfg)
- Razón: Bajo SNR, pero a veces críticos (nginx.conf, Apache)
- Filtro: Solo si proyecto **menciona explícitamente en docs** que use este tipo
- Chunk size: Reducido (200c) para minimizar noise
- **Markup** (HTML, XML)
- Razón: Baja densidad semántica (mucho ruido de tags)
- Filtro: Solo si es HTML template crítico (no generated output)
- Exclusiones: Script tags, style tags, comments
- **Styles** (CSS, SCSS, LESS)
- Razón: Muy baja densidad semántica
- Filtro: Solo si hay design patterns documentadas que merezcan búsqueda
- Penalty: -80% quality score si es puro selector sin properties
### **NO INDEXAR (❌)**
- **Data** (.csv, .tsv, .sql data dumps)
- Razón: Demasiado grande, bajo value, contamina índice
- Exception: Sample of 10-15 rows si schema es crítico → metadata-only
- **Binary** (imágenes, videos, PDFs, archives)
- Razón: No textual, requiere OCR/external vision
- Handling: Skip silently, no warning
- **Logs** (.log, .out, rotation logs)
- Razón: Volatile, noise, temporal data
- Handling: Skip on ingest detection pattern
- Detection: nombre contiene `.log`, `.out`, `debug.txt`, o empieza con fecha
- **Test Fixtures** (.json fixtures, test data)
- Razón: Noise, not real code, contaminan búsquedas
- Detection: Directorio `/tests/`, `/fixtures/`, `*.test.*`, `*.spec.*`, `mock*`
- **Vendored Code** (node_modules/, vendor/, dist/)
- Razón: Already handled by SKIP_DIRS in ingest_all.py
- Confirmation: Maintain existing list
- **Build Artifacts** (.o, .a, .so, .class, .exe)
- Razón: Binary, no semantic value
- Detection: Compiled extensions, extension patterns
---
## V. ALGORITMO DE VALIDACIÓN AUTOMATIZADO
### **A. Detector de Categoría (Single-Pass)**
```python
FileCategory = enum.Enum('FileCategory', [
'CODE_PYTHON', 'CODE_TYPESCRIPT', 'CODE_JAVASCRIPT', 'CODE_SHELL',
'CODE_OTHER', 'DOCUMENTATION', 'CONFIG_STRUCTURED', 'CONFIG_UNSTRUCTURED',
'MARKUP', 'STYLES', 'DATA_TABULAR', 'BINARY_IMAGES', 'BINARY_ARCHIVES',
'BINARY_MEDIA', 'GENERATED_LOGS', 'TEST_FIXTURES', 'VENDORED', 'BUILD_ARTIFACTS'
])
def detect_file_category(file_path: str, content: Optional[str] = None) -> FileCategory:
"""Detect file category in ONE pass, using extension + optional content heuristics.
Phase 1: Extension-based detection (fast)
Phase 2: Content heuristics if ambiguous
Phase 3: Directory heuristics if still ambiguous
"""
ext = file_path.rsplit('.', 1)[-1].lower() if '.' in file_path else ''
path_lower = file_path.lower()
# === PHASE 1: EXTENSION-BASED (primary signal) ===
# Code
if ext == 'py':
return FileCategory.CODE_PYTHON
if ext in ('ts', 'tsx'):
return FileCategory.CODE_TYPESCRIPT
if ext in ('js', 'jsx'):
return FileCategory.CODE_JAVASCRIPT
if ext in ('sh', 'bash'):
return FileCategory.CODE_SHELL
if ext in ('go', 'rs', 'java', 'sql', 'hcl', 'tf', 'rb', 'php', 'cpp', 'c', 'h'):
return FileCategory.CODE_OTHER
# Documentation
if ext in ('md', 'rst', 'txt'):
return FileCategory.DOCUMENTATION
# Config structured
if ext in ('json', 'yaml', 'yml', 'toml'):
return FileCategory.CONFIG_STRUCTURED
# Config unstructured
if ext in ('ini', 'conf', 'cfg'):
return FileCategory.CONFIG_UNSTRUCTURED
# Markup
if ext in ('html', 'xml', 'svg'):
return FileCategory.MARKUP
# Styles
if ext in ('css', 'scss', 'less', 'sass'):
return FileCategory.STYLES
# Data
if ext in ('csv', 'tsv', 'parquet', 'avro'):
return FileCategory.DATA_TABULAR
# Binary media
if ext in ('png', 'jpg', 'jpeg', 'gif', 'bmp', 'tiff', 'webp'):
return FileCategory.BINARY_IMAGES
if ext in ('mp4', 'mov', 'avi', 'mkv', 'flv'):
return FileCategory.BINARY_MEDIA
if ext in ('mp3', 'wav', 'flac', 'aac', 'm4a'):
return FileCategory.BINARY_MEDIA
if ext in ('pdf', 'docx', 'pptx'):
return FileCategory.BINARY_MEDIA
# Archives
if ext in ('zip', 'tar', 'gz', '7z', 'rar', 'bz2', 'xz'):
return FileCategory.BINARY_ARCHIVES
# === PHASE 2: DIRECTORY HEURISTICS ===
# Vendored
if any(seg in path_lower for seg in ('node_modules', 'vendor', 'venv', '.venv', 'dist', 'build')):
return FileCategory.VENDORED
# Test fixtures
if any(seg in path_lower for seg in ('tests', 'fixtures', '__tests__', 'spec', 'specs')):
if ext in ('json', 'yaml', 'yml', 'sql'):
return FileCategory.TEST_FIXTURES
# Build artifacts
if ext in ('o', 'a', 'so', 'dll', 'exe', 'class', 'pyc', 'pyo'):
return FileCategory.BUILD_ARTIFACTS
# === PHASE 3: FILENAME HEURISTICS ===
# Logs
if 'log' in path_lower or path_lower.endswith('.out'):
return FileCategory.GENERATED_LOGS
# Test data
if any(seg in file_path for seg in ('.test.', '.spec.', 'mock', 'fixture')):
return FileCategory.TEST_FIXTURES
# Sample/template configs
if any(seg in file_path for seg in ('.sample', '.example', '.template', '.dist')):
return FileCategory.CONFIG_STRUCTURED # But will be filtered in policy check
# === FALLBACK: Content heuristics if extension is unknown ===
if content is not None and len(content) > 0:
content_sample = content[:500].lower()
# Try JSON-like
if content_sample.strip().startswith(('{', '[')):
try:
json.loads(content)
return FileCategory.CONFIG_STRUCTURED
except:
pass
# Try YAML-like
if ':' in content_sample and not content_sample.startswith('#!'):
return FileCategory.CONFIG_STRUCTURED
# Try code-like (shebangs, import statements)
if content_sample.startswith(('#!', 'import ', 'from ', 'function ', 'class ')):
return FileCategory.CODE_OTHER
# Default: unknown, will be skipped
return None
```
### **B. Policy Validator (Pre-Ingest)**
```python
def should_index_file(
file_path: str,
category: FileCategory,
content: Optional[str] = None,
max_size_mb: int = 50,
) -> tuple[bool, str]:
"""Decide if a file should be indexed based on category + content filters.
Returns (should_index, reason)
"""
# === SIZE FILTERS ===
size_bytes = len(content.encode('utf-8')) if content else 0
size_mb = size_bytes / (1024 * 1024)
if size_mb > max_size_mb:
return False, f"file_too_large:{size_mb:.1f}MB"
if category == FileCategory.DATA_TABULAR:
if size_mb > 1.0:
return False, f"tabular_data_too_large:{size_mb:.1f}MB"
# === CONTENT FILTERS ===
# Skip files that are LLM responses
if content:
if is_llm_response(content):
return False, "llm_response_contamination"
# === CATEGORY-BASED FILTERS ===
if category == FileCategory.CODE_PYTHON:
return True, "code_python_index"
if category == FileCategory.CODE_TYPESCRIPT:
return True, "code_typescript_index"
if category == FileCategory.CODE_JAVASCRIPT:
return True, "code_javascript_index"
if category == FileCategory.CODE_SHELL:
return True, "code_shell_index"
if category == FileCategory.CODE_OTHER:
return True, "code_other_index"
if category == FileCategory.DOCUMENTATION:
return True, "documentation_index"
if category == FileCategory.CONFIG_STRUCTURED:
# Filter out template/sample files
if any(seg in file_path for seg in ('.sample', '.example', '.template', '.dist')):
return False, "config_sample_skip"
# Filter out LLM responses
if content and is_llm_response(content):
return False, "config_llm_response"
return True, "config_structured_index"
if category == FileCategory.CONFIG_UNSTRUCTURED:
# Selective: only if explicitly in docs or appears critical
# For now: skip to reduce noise
return False, "config_unstructured_selective_skip"
if category == FileCategory.MARKUP:
# Selective: skip unless it's template critical
if 'template' in file_path.lower():
return True, "markup_template_index"
return False, "markup_noise_skip"
if category == FileCategory.STYLES:
# Selective: high noise, skip by default
return False, "styles_low_semantic_value_skip"
if category == FileCategory.DATA_TABULAR:
return False, "data_tabular_skip"
if category in (FileCategory.BINARY_IMAGES, FileCategory.BINARY_ARCHIVES,
FileCategory.BINARY_MEDIA):
return False, "binary_skip"
if category == FileCategory.GENERATED_LOGS:
return False, "logs_volatile_skip"
if category == FileCategory.TEST_FIXTURES:
return False, "test_fixtures_noise_skip"
if category == FileCategory.VENDORED:
return False, "vendored_skip"
if category == FileCategory.BUILD_ARTIFACTS:
return False, "build_artifacts_skip"
return False, "unknown_category"
```
### **C. Chunking Policy Resolver**
```python
@dataclass
class ChunkingPolicy:
strategy: str # 'ast', 'heading', 'semantic_section', 'fixed', 'line_aware'
chunk_size: int
overlap: int
min_chunk_len: int
extract_symbols: bool
preserve_structure: bool
@staticmethod
def for_category(category: FileCategory) -> ChunkingPolicy:
"""Get chunking policy for a given file category."""
POLICIES = {
FileCategory.CODE_PYTHON: ChunkingPolicy(
strategy='ast',
chunk_size=500, # max (unbounded for functions)
overlap=0,
min_chunk_len=40,
extract_symbols=True,
preserve_structure=True,
),
FileCategory.CODE_TYPESCRIPT: ChunkingPolicy(
strategy='fixed',
chunk_size=500,
overlap=100,
min_chunk_len=40,
extract_symbols=True,
preserve_structure=False,
),
FileCategory.CODE_JAVASCRIPT: ChunkingPolicy(
strategy='fixed',
chunk_size=500,
overlap=100,
min_chunk_len=40,
extract_symbols=True,
preserve_structure=False,
),
FileCategory.CODE_SHELL: ChunkingPolicy(
strategy='line_aware',
chunk_size=400,
overlap=80,
min_chunk_len=40,
extract_symbols=True,
preserve_structure=False,
),
FileCategory.CODE_OTHER: ChunkingPolicy(
strategy='fixed',
chunk_size=500,
overlap=100,
min_chunk_len=40,
extract_symbols=True,
preserve_structure=False,
),
FileCategory.DOCUMENTATION: ChunkingPolicy(
strategy='heading',
chunk_size=800,
overlap=120,
min_chunk_len=40,
extract_symbols=True,
preserve_structure=True,
),
FileCategory.CONFIG_STRUCTURED: ChunkingPolicy(
strategy='semantic_section',
chunk_size=1000,
overlap=0,
min_chunk_len=40,
extract_symbols=True,
preserve_structure=True,
),
FileCategory.CONFIG_UNSTRUCTURED: ChunkingPolicy(
strategy='line_aware',
chunk_size=200,
overlap=30,
min_chunk_len=40,
extract_symbols=False,
preserve_structure=False,
),
FileCategory.MARKUP: ChunkingPolicy(
strategy='fixed',
chunk_size=600,
overlap=100,
min_chunk_len=50,
extract_symbols=False,
preserve_structure=True,
),
FileCategory.STYLES: ChunkingPolicy(
strategy='rule_group',
chunk_size=200,
overlap=30,
min_chunk_len=50,
extract_symbols=True,
preserve_structure=True,
),
}
return POLICIES.get(category, ChunkingPolicy(
strategy='fixed',
chunk_size=500,
overlap=100,
min_chunk_len=40,
extract_symbols=False,
preserve_structure=False,
))
```
### **D. Full Validation Pipeline**
```python
class KlausFileValidator:
"""Single-pass file validator that applies policies automatically."""
def __init__(self, config: KlausConfig):
self.config = config
self.stats = {
'total': 0,
'indexed': 0,
'skipped': 0,
'by_category': defaultdict(int),
'by_skip_reason': defaultdict(int),
'quality_issues': defaultdict(int),
}
async def validate_and_chunk_file(
self,
file_path: str,
content: str,
mtime: float,
) -> tuple[bool, Optional[list[dict]], str]:
"""Validate and chunk a file in ONE pass.
Returns (success, chunks_or_none, reason)
"""
self.stats['total'] += 1
# === PHASE 1: DETECT CATEGORY ===
category = detect_file_category(file_path, content[:1000])
self.stats['by_category'][category.name if category else 'UNKNOWN'] += 1
if category is None:
self.stats['skipped'] += 1
self.stats['by_skip_reason']['unknown_category'] += 1
return False, None, "unknown_category"
# === PHASE 2: APPLY POLICY ===
should_index, policy_reason = should_index_file(
file_path, category, content, max_size_mb=50
)
if not should_index:
self.stats['skipped'] += 1
self.stats['by_skip_reason'][policy_reason] += 1
log.debug(f"SKIP {file_path} — {policy_reason}")
return False, None, policy_reason
# === PHASE 3: GET CHUNKING POLICY ===
chunking_policy = ChunkingPolicy.for_category(category)
# === PHASE 4: CHUNK FILE ===
try:
chunks = chunk_file_with_policy(
file_path, content, chunking_policy, category
)
# === PHASE 5: QUALITY CHECKS ===
quality_issues = []
for chunk in chunks:
# Check chunk quality, complexity, contamination
if is_llm_response(chunk['content']):
quality_issues.append(f"chunk_{chunk['chunk_index']}_llm_response")
chunk['skip'] = True
if len(chunk['content']) < chunking_policy.min_chunk_len:
quality_issues.append(f"chunk_{chunk['chunk_index']}_too_short")
chunk['skip'] = True
chunks = [c for c in chunks if not c.get('skip', False)]
if quality_issues:
self.stats['quality_issues'][file_path] = quality_issues
if chunks:
self.stats['indexed'] += 1
return True, chunks, "indexed_success"
else:
self.stats['skipped'] += 1
self.stats['by_skip_reason']['all_chunks_filtered'] += 1
return False, None, "all_chunks_filtered"
except Exception as e:
log.error(f"ERROR chunking {file_path}: {e}")
self.stats['skipped'] += 1
self.stats['by_skip_reason']['chunking_error'] += 1
return False, None, f"chunking_error:{str(e)[:30]}"
def report(self) -> dict:
"""Generate validation report."""
return {
'total_files': self.stats['total'],
'indexed': self.stats['indexed'],
'skipped': self.stats['skipped'],
'by_category': dict(self.stats['by_category']),
'by_skip_reason': dict(self.stats['by_skip_reason']),
'quality_issues': dict(self.stats['quality_issues']),
}
```
---
## VI. VALIDACIÓN EN UNA PASADA: FLUJO INTEGRADO
```python
# En scripts/ingest_all.py o proxy/knowledge.py
async def ingest_with_validated_policies(project: str, project_root: Path) -> dict:
"""Ingestion with single-pass validation and categorized policies."""
validator = KlausFileValidator(config=settings)
client = await get_embedding_client()
kb = KnowledgeBase()
stats = {
'total': 0,
'indexed': 0,
'skipped': 0,
'stored': 0,
'failed': 0,
}
# === PHASE 1: FILE DISCOVERY ===
files = discover_files(project_root)
# === PHASE 2: SINGLE-PASS VALIDATION & INGEST ===
for file_path in files:
stats['total'] += 1
# Read content
try:
content = file_path.read_text(encoding='utf-8', errors='replace')
mtime = file_path.stat().st_mtime
except Exception as e:
log.error(f"Cannot read {file_path}: {e}")
stats['failed'] += 1
continue
# Validate & chunk in ONE pass
success, chunks, reason = await validator.validate_and_chunk_file(
str(file_path.relative_to(project_root)),
content,
mtime,
)
if not success:
stats['skipped'] += 1
continue
# Embed & store chunks
for chunk in chunks:
try:
# Get semantic prefix
prefix_text = embed_text(
chunk['content'],
chunk['file_path'],
chunk.get('symbol_name', ''),
chunk.get('symbol_type', ''),
)
# Embed
vector = await client.embed(prefix_text)
# Check duplicate (with category-aware threshold)
category = detect_file_category(chunk['file_path'])
dup_threshold = 0.92 # Standard duplicate check
if await kb.check_semantic_duplicate(vector, chunk['file_path'], dup_threshold):
log.debug(f"DEDUP: {chunk['file_path']}:{chunk['chunk_index']}")
continue
# Store
stored = await kb.store_chunk(chunk, vector)
if stored:
stats['stored'] += 1
else:
stats['failed'] += 1
except Exception as e:
log.error(f"Error embedding {file_path}: {e}")
stats['failed'] += 1
# === PHASE 3: REPORT ===
validation_report = validator.report()
log.info(f"Ingest complete: {stats}")
log.info(f"Validation: {validation_report}")
return {
**stats,
'validation': validation_report,
}
```
---
## VII. MATRIZ RESUMIDA PARA INGESTER
Usar esta tabla en el ingestor para validación rápida:
```python
# En proxy/knowledge.py o ingest_all.py
INDEXATION_POLICY = {
# (extension, category) → (should_index, strategy, chunk_size, overlap)
# CODE: Always index
('.py', 'CODE_PYTHON'): (True, 'ast', 500, 0),
('.ts', 'CODE_TYPESCRIPT'): (True, 'fixed', 500, 100),
('.tsx', 'CODE_TYPESCRIPT'): (True, 'fixed', 500, 100),
('.js', 'CODE_JAVASCRIPT'): (True, 'fixed', 500, 100),
('.jsx', 'CODE_JAVASCRIPT'): (True, 'fixed', 500, 100),
('.sh', 'CODE_SHELL'): (True, 'line_aware', 400, 80),
('.bash', 'CODE_SHELL'): (True, 'line_aware', 400, 80),
('.go', 'CODE_OTHER'): (True, 'fixed', 500, 100),
('.rs', 'CODE_OTHER'): (True, 'fixed', 500, 100),
# DOCS: Always index
('.md', 'DOCUMENTATION'): (True, 'heading', 800, 120),
('.rst', 'DOCUMENTATION'): (True, 'heading', 800, 120),
# CONFIG: Index with filtering
('.json', 'CONFIG_STRUCTURED'): (True, 'semantic_section', 1000, 0, ['.sample', '.example']),
('.yaml', 'CONFIG_STRUCTURED'): (True, 'semantic_section', 1000, 0, ['.sample', '.example']),
('.yml', 'CONFIG_STRUCTURED'): (True, 'semantic_section', 1000, 0, ['.sample', '.example']),
('.toml', 'CONFIG_STRUCTURED'): (True, 'semantic_section', 1000, 0, ['.sample', '.example']),
# CONFIG Unstructured: Skip (noise)
('.ini', 'CONFIG_UNSTRUCTURED'): (False, None, None, None),
('.conf', 'CONFIG_UNSTRUCTURED'): (False, None, None, None),
('.cfg', 'CONFIG_UNSTRUCTURED'): (False, None, None, None),
# MARKUP: Skip (noise)
('.html', 'MARKUP'): (False, None, None, None),
('.xml', 'MARKUP'): (False, None, None, None),
# STYLES: Skip (noise)
('.css', 'STYLES'): (False, None, None, None),
('.scss', 'STYLES'): (False, None, None, None),
# DATA: Skip (bloat)
('.csv', 'DATA_TABULAR'): (False, None, None, None),
('.tsv', 'DATA_TABULAR'): (False, None, None, None),
# BINARY: Skip (always)
# ... (all binary extensions)
}
```
---
## VIII. MÉTRICAS DE ÉXITO
Después de aplicar estas políticas, validar:
| Métrica | Target | Razón |
|---|---|---|
| **Indexation rate** | > 90% de archivos válidos indexados | Detectar categorías perdidas o falsos negativos |
| **Skip ratio** | < 10% por políticas, > 0% por contamination | Validar filtering efectivo |
| **Avg chunks per file** | Code: 8-15, Docs: 3-6, Config: 1-3 | Detectar over/under-chunking |
| **Quality score distribution** | 70% entre 0.8-1.0 | Validar enrichment y scoring |
| **Duplicate rate** | < 2% | Validar threshold de dedup |
| **LLM contamination** | < 0.5% | Validar _is_llm_response filter |
| **Symbol extraction** | Code: > 80%, Docs: > 60% | Validar semantic enrichment |
| **Search latency (hybrid)** | < 200ms p99 | Validar fusion RRF overhead |
| **Threshold compliance** | 0.75 general, 0.92 dedup | Validar consistency |
---
## IX. ROADMAP: IMPLEMENTACIÓN FASE A FASE
### **Fase 1: Validación (Semana 1)**
- [ ] Implementar `detect_file_category()` en `knowledge.py`
- [ ] Implementar `should_index_file()` con políticas
- [ ] Ejecutar validación en muestra de 500 archivos
- [ ] Reportar categorización y skip reasons
### **Fase 2: Integración (Semana 2)**
- [ ] Integrar validator en `ingest_all.py`
- [ ] Ejecutar full ingest con validators activados
- [ ] Comparar antes/después: chunks count, quality scores
- [ ] Ajustar thresholds si es necesario
### **Fase 3: Refinamiento (Semana 3)**
- [ ] Análisis de quality issues reportados
- [ ] Ajustar chunk sizes por categoría si es necesario
- [ ] Validar search quality (latency, recall, precision)
- [ ] Documentar políticas finales en runbook
### **Fase 4: Automatización (Semana 4)**
- [ ] Agregar automated testing para cada categoría
- [ ] Implementar pre-ingest hook en proxy para validación en tiempo real
- [ ] Dashboard de métricas de KB (indexation health)
- [ ] Runbook: cómo agregar nuevas categorías
---
## Conclusión
Esta matriz proporciona a Klaus un **framework de validación UNA PASADA** que:
1. **Categoriza** automáticamente cada archivo (18 categorías semánticas)
2. **Aplica políticas diferenciadas** sin solapamiento (chunking, embedding, filtrado)
3. **Valida en ingest time** (contamination, duplicates, quality)
4. **Genera métricas** para ajuste posterior
5. **Es extensible** para nuevas categorías sin reaprendizaje
El resultado es un **Knowledge Base limpio, bien-estructurado y altamente queryable** que maximiza ROI para Q&A técnico mientras minimiza noise.