🧠 CrumbVector - Certified Code in Vector Space
Semantic Code Search mit Trust Layer
🎯 Die Idee
┌─────────────────────────────────────────────────────────────────────────────┐
│ │
│ CERTIFIED REPO → CHUNKING → EMBEDDING → VECTOR DB → SEMANTIC │
│ │
│ ┌─────────┐ ┌───────┐ ┌───────┐ ┌───────┐ ┌────────┐ │
│ │ .git │ │ Code │ │ float │ │ChromaDB│ │ "Wie │ │
│ │ CKL │ ──► │ Chunks│ ──► │ [384] │ ──► │Qdrant │ ◄──│ mache │ │
│ │ ✓ Trust│ │ +Meta │ │ Vector│ │Pinecone│ │ OneDrop│ │
│ └─────────┘ └───────┘ └───────┘ └───────┘ └────────┘ │
│ │
│ Nur TRUSTED Code wird vektorisiert! │
│ Query → Semantische Suche → Verified Code Snippets │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
🔐 Trust Layer
Was macht ein Repo "zertifiziert"?
# .crumbvector/trust.yaml
certification:
status: verified
license: CKL-1.0 # oder MIT, Apache, etc.
verified_by: crumbforest
verified_at: 2026-01-13
trust_level: 3 # 1-5 scale
# 1 = unknown
# 2 = community reviewed
# 3 = maintainer verified
# 4 = foundation certified
# 5 = core/official
allowed_for:
- teaching
- code_completion
- pattern_search
- remix
excluded_patterns:
- "**/*.env"
- "**/secrets/**"
- "**/node_modules/**"
🔄 Der Pipeline Flow
┌─────────────────────────────────────────────────────────────────────────────┐
│ │
│ 1. INGEST │
│ ├── Git Clone (certified repo) │
│ ├── Verify Trust (.crumbvector/trust.yaml) │
│ ├── Check License (CKL, MIT, Apache...) │
│ └── Extract Metadata (author, date, tags) │
│ │
│ 2. CHUNK │
│ ├── Split by: function, class, file, pattern │
│ ├── Keep context: imports, comments, docstrings │
│ ├── Attach metadata: file, line, repo, license │
│ └── Smart boundaries: AST-aware splitting │
│ │
│ 3. EMBED │
│ ├── Code-optimized model (CodeBERT, StarCoder, etc.) │
│ ├── Or: Local model (Ollama + nomic-embed-text) │
│ ├── Generate vectors: [384] or [768] dimensions │
│ └── Batch processing for efficiency │
│ │
│ 4. STORE │
│ ├── Vector DB: ChromaDB (local), Qdrant, Pinecone │
│ ├── Metadata: license, trust_level, source, language │
│ ├── Collections: by repo, by language, by topic │
│ └── Index: for fast similarity search │
│ │
│ 5. QUERY │
│ ├── Natural language: "Wie mache ich einen One Drop Beat?" │
│ ├── Code snippet: "sound('bd').delay()" │
│ ├── Filter: trust_level >= 3, license = CKL │
│ └── Return: relevant chunks + metadata + source link │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
📁 Architektur
crumbvector/
├── ingest/
│ ├── git_loader.py # Clone & verify repos
│ ├── trust_checker.py # Validate certification
│ └── license_parser.py # Extract license info
│
├── chunk/
│ ├── code_chunker.py # AST-aware splitting
│ ├── pattern_chunker.py # For Strudel patterns
│ └── doc_chunker.py # Markdown/comments
│
├── embed/
│ ├── local_embedder.py # Ollama integration
│ ├── openai_embedder.py # OpenAI embeddings
│ └── code_embedder.py # CodeBERT/StarCoder
│
├── store/
│ ├── chroma_store.py # ChromaDB (local, free)
│ ├── qdrant_store.py # Qdrant (local/cloud)
│ └── collections.py # Collection management
│
├── query/
│ ├── semantic_search.py # Natural language queries
│ ├── code_search.py # Code similarity
│ └── filter_engine.py # Trust/license filtering
│
├── api/
│ ├── server.py # FastAPI server
│ └── cli.py # Command line interface
│
└── config/
├── trusted_repos.yaml # List of certified repos
└── settings.yaml # Configuration
🛠️ Implementation
1. Git Loader mit Trust Check
#!/usr/bin/env python3
"""
CrumbVector - Git Loader with Trust Verification
"""
import os
import yaml
import subprocess
from pathlib import Path
from dataclasses import dataclass
from typing import Optional, List
@dataclass
class TrustInfo:
status: str
license: str
trust_level: int
verified_by: str
allowed_for: List[str]
@dataclass
class CertifiedRepo:
url: str
local_path: Path
trust: TrustInfo
def clone_repo(url: str, target_dir: Path) -> Path:
"""Clone a git repository."""
repo_name = url.split('/')[-1].replace('.git', '')
local_path = target_dir / repo_name
if local_path.exists():
# Pull latest
subprocess.run(['git', '-C', str(local_path), 'pull'], check=True)
else:
# Clone
subprocess.run(['git', 'clone', url, str(local_path)], check=True)
return local_path
def verify_trust(repo_path: Path) -> Optional[TrustInfo]:
"""Check if repo has valid trust certification."""
trust_file = repo_path / '.crumbvector' / 'trust.yaml'
# Also check for CKL license
ckl_file = repo_path / 'LICENSE.CKL'
license_file = repo_path / 'LICENSE'
if trust_file.exists():
with open(trust_file) as f:
data = yaml.safe_load(f)
cert = data.get('certification', {})
return TrustInfo(
status=cert.get('status', 'unknown'),
license=cert.get('license', 'unknown'),
trust_level=data.get('trust_level', 1),
verified_by=cert.get('verified_by', 'unknown'),
allowed_for=data.get('allowed_for', [])
)
# Fallback: Check for known licenses
if ckl_file.exists():
return TrustInfo(
status='license_only',
license='CKL-1.0',
trust_level=2,
verified_by='auto',
allowed_for=['teaching', 'remix']
)
return None
def load_certified_repo(url: str, target_dir: Path,
min_trust_level: int = 2) -> Optional[CertifiedRepo]:
"""Load and verify a repository."""
local_path = clone_repo(url, target_dir)
trust = verify_trust(local_path)
if trust is None:
print(f"⚠️ No trust info found for {url}")
return None
if trust.trust_level < min_trust_level:
print(f"⚠️ Trust level {trust.trust_level} below minimum {min_trust_level}")
return None
if trust.status not in ['verified', 'license_only']:
print(f"⚠️ Status '{trust.status}' not acceptable")
return None
print(f"✅ Loaded {url} (trust: {trust.trust_level}, license: {trust.license})")
return CertifiedRepo(url=url, local_path=local_path, trust=trust)
2. Code Chunker (AST-aware)
"""
CrumbVector - Smart Code Chunker
"""
import ast
from pathlib import Path
from dataclasses import dataclass
from typing import List, Optional
import re
@dataclass
class CodeChunk:
content: str
language: str
file_path: str
start_line: int
end_line: int
chunk_type: str # function, class, pattern, comment
metadata: dict
def chunk_python(file_path: Path, repo_info: dict) -> List[CodeChunk]:
"""Chunk Python file using AST."""
chunks = []
content = file_path.read_text()
try:
tree = ast.parse(content)
except SyntaxError:
# Fallback to line-based chunking
return chunk_by_lines(file_path, content, 'python', repo_info)
for node in ast.walk(tree):
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
chunk = extract_node_chunk(content, node, 'function', file_path, repo_info)
chunks.append(chunk)
elif isinstance(node, ast.ClassDef):
chunk = extract_node_chunk(content, node, 'class', file_path, repo_info)
chunks.append(chunk)
return chunks
def chunk_javascript(file_path: Path, repo_info: dict) -> List[CodeChunk]:
"""Chunk JavaScript/Strudel patterns."""
chunks = []
content = file_path.read_text()
lines = content.split('\n')
# Pattern: let name = ...
pattern_regex = r'^(let|const|var)\s+(\w+)\s*='
current_chunk = []
current_start = 0
current_name = None
for i, line in enumerate(lines):
match = re.match(pattern_regex, line)
if match and current_chunk:
# Save previous chunk
chunks.append(CodeChunk(
content='\n'.join(current_chunk),
language='javascript',
file_path=str(file_path),
start_line=current_start,
end_line=i - 1,
chunk_type='pattern',
metadata={**repo_info, 'name': current_name}
))
current_chunk = []
current_start = i
current_name = match.group(2)
current_chunk.append(line)
if match and current_name is None:
current_name = match.group(2)
current_start = i
# Don't forget last chunk
if current_chunk:
chunks.append(CodeChunk(
content='\n'.join(current_chunk),
language='javascript',
file_path=str(file_path),
start_line=current_start,
end_line=len(lines) - 1,
chunk_type='pattern',
metadata={**repo_info, 'name': current_name}
))
return chunks
def chunk_strudel_pattern(file_path: Path, repo_info: dict) -> List[CodeChunk]:
"""Special chunker for Strudel .js patterns."""
chunks = []
content = file_path.read_text()
# Split by section comments
sections = re.split(r'// [═─]{10,}', content)
for i, section in enumerate(sections):
if section.strip():
# Extract section title if present
title_match = re.search(r'// ([A-Z][A-Z\s]+)', section)
title = title_match.group(1).strip() if title_match else f"Section {i}"
chunks.append(CodeChunk(
content=section.strip(),
language='strudel',
file_path=str(file_path),
start_line=0, # Would need proper line tracking
end_line=0,
chunk_type='section',
metadata={**repo_info, 'section': title}
))
return chunks
3. Embedder (Local with Ollama)
"""
CrumbVector - Local Embedding with Ollama
"""
import requests
import numpy as np
from typing import List
from dataclasses import dataclass
@dataclass
class EmbeddedChunk:
chunk: 'CodeChunk'
vector: np.ndarray
model: str
class OllamaEmbedder:
"""Generate embeddings using local Ollama."""
def __init__(self,
model: str = "nomic-embed-text",
base_url: str = "http://localhost:11434"):
self.model = model
self.base_url = base_url
def embed_text(self, text: str) -> np.ndarray:
"""Generate embedding for single text."""
response = requests.post(
f"{self.base_url}/api/embeddings",
json={
"model": self.model,
"prompt": text
}
)
response.raise_for_status()
return np.array(response.json()['embedding'])
def embed_chunks(self, chunks: List['CodeChunk']) -> List[EmbeddedChunk]:
"""Generate embeddings for multiple chunks."""
embedded = []
for chunk in chunks:
# Prepare text with context
text = self._prepare_chunk_text(chunk)
vector = self.embed_text(text)
embedded.append(EmbeddedChunk(
chunk=chunk,
vector=vector,
model=self.model
))
return embedded
def _prepare_chunk_text(self, chunk: 'CodeChunk') -> str:
"""Prepare chunk text for embedding with metadata context."""
meta = chunk.metadata
# Add context for better semantic matching
context_parts = [
f"Language: {chunk.language}",
f"Type: {chunk.chunk_type}",
]
if 'name' in meta:
context_parts.append(f"Name: {meta['name']}")
if 'section' in meta:
context_parts.append(f"Section: {meta['section']}")
if 'genre' in meta:
context_parts.append(f"Genre: {meta['genre']}")
context = " | ".join(context_parts)
return f"{context}\n\n{chunk.content}"
4. ChromaDB Store
"""
CrumbVector - ChromaDB Storage
"""
import chromadb
from chromadb.config import Settings
from typing import List, Optional, Dict
import json
class CrumbVectorStore:
"""Store and query code vectors with ChromaDB."""
def __init__(self, persist_dir: str = "./crumbvector_db"):
self.client = chromadb.Client(Settings(
chroma_db_impl="duckdb+parquet",
persist_directory=persist_dir,
anonymized_telemetry=False
))
# Main collection for code
self.code_collection = self.client.get_or_create_collection(
name="certified_code",
metadata={"description": "Certified code from trusted repos"}
)
# Separate collection for Strudel patterns
self.pattern_collection = self.client.get_or_create_collection(
name="strudel_patterns",
metadata={"description": "Music patterns for Strudel"}
)
def add_chunks(self, embedded_chunks: List['EmbeddedChunk'],
collection: str = "code"):
"""Add embedded chunks to the store."""
coll = (self.pattern_collection if collection == "patterns"
else self.code_collection)
ids = []
embeddings = []
documents = []
metadatas = []
for i, ec in enumerate(embedded_chunks):
chunk_id = f"{ec.chunk.file_path}:{ec.chunk.start_line}:{i}"
ids.append(chunk_id)
embeddings.append(ec.vector.tolist())
documents.append(ec.chunk.content)
metadatas.append({
"language": ec.chunk.language,
"chunk_type": ec.chunk.chunk_type,
"file_path": ec.chunk.file_path,
"start_line": ec.chunk.start_line,
"end_line": ec.chunk.end_line,
"model": ec.model,
**{k: str(v) for k, v in ec.chunk.metadata.items()}
})
coll.add(
ids=ids,
embeddings=embeddings,
documents=documents,
metadatas=metadatas
)
def query(self,
query_text: str,
query_embedding: List[float],
n_results: int = 5,
collection: str = "code",
filters: Optional[Dict] = None) -> List[Dict]:
"""Query for similar code chunks."""
coll = (self.pattern_collection if collection == "patterns"
else self.code_collection)
where_filter = None
if filters:
where_filter = {"$and": [
{k: {"$eq": v}} for k, v in filters.items()
]}
results = coll.query(
query_embeddings=[query_embedding],
n_results=n_results,
where=where_filter,
include=["documents", "metadatas", "distances"]
)
# Format results
formatted = []
for i in range(len(results['ids'][0])):
formatted.append({
"id": results['ids'][0][i],
"content": results['documents'][0][i],
"metadata": results['metadatas'][0][i],
"distance": results['distances'][0][i],
"query": query_text
})
return formatted
def persist(self):
"""Persist the database to disk."""
self.client.persist()
5. Query API
"""
CrumbVector - Query API
"""
from fastapi import FastAPI, Query
from pydantic import BaseModel
from typing import List, Optional
import uvicorn
app = FastAPI(title="CrumbVector API", version="0.1.0")
# Initialize components
embedder = OllamaEmbedder()
store = CrumbVectorStore()
class SearchQuery(BaseModel):
query: str
n_results: int = 5
collection: str = "patterns"
language: Optional[str] = None
min_trust_level: Optional[int] = None
genre: Optional[str] = None
class SearchResult(BaseModel):
content: str
file_path: str
language: str
distance: float
metadata: dict
@app.post("/search", response_model=List[SearchResult])
async def search_code(search: SearchQuery):
"""Semantic search for code snippets."""
# Generate query embedding
query_vector = embedder.embed_text(search.query)
# Build filters
filters = {}
if search.language:
filters["language"] = search.language
if search.min_trust_level:
filters["trust_level"] = {"$gte": search.min_trust_level}
if search.genre:
filters["genre"] = search.genre
# Query
results = store.query(
query_text=search.query,
query_embedding=query_vector.tolist(),
n_results=search.n_results,
collection=search.collection,
filters=filters if filters else None
)
return [SearchResult(
content=r["content"],
file_path=r["metadata"]["file_path"],
language=r["metadata"]["language"],
distance=r["distance"],
metadata=r["metadata"]
) for r in results]
@app.get("/health")
async def health():
return {"status": "ok", "service": "crumbvector"}
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8765)
🎵 Use Case: Strudel Pattern Search
# Query
curl -X POST http://localhost:8765/search \
-H "Content-Type: application/json" \
-d '{
"query": "Wie mache ich einen One Drop Beat?",
"collection": "patterns",
"genre": "reggae",
"n_results": 3
}'
# Response
[
{
"content": "let oneDrop = stack(\n sound(\"~ ~ bd ~\").gain(1.1).lpf(150),\n sound(\"~ ~ rim ~\").gain(0.7).hpf(800)\n)",
"file_path": "crumbmidi/patterns/REGGAE_DUB_75BPM.js",
"language": "strudel",
"distance": 0.23,
"metadata": {
"genre": "reggae",
"section": "ONE DROP",
"trust_level": "3",
"license": "CKL-1.0"
}
},
...
]
🔐 Trust Levels für CrumbMidi
# crumbmidi/.crumbvector/trust.yaml
certification:
status: verified
license: CKL-1.0
verified_by: crumbforest
verified_at: 2026-01-13
trust_level: 4 # Foundation certified
allowed_for:
- teaching
- code_completion
- pattern_search
- remix
- ai_training # Explicit opt-in!
genres:
- gfunk
- dnb
- house
- baobab
- reggae
languages:
- javascript
- strudel
- bash
🦊 FunkFox Vision
┌─────────────────────────────────────────────────────────────────────────────┐
│ │
│ "Wie mache ich Drums wie Armand Van Helden?" │
│ │ │
│ ▼ │
│ ┌───────────────┐ │
│ │ CrumbVector │ │
│ │ Query API │ │
│ └───────────────┘ │
│ │ │
│ ┌──────────────────┼──────────────────┐ │
│ ▼ ▼ ▼ │
│ ┌────────────────┐ ┌────────────────┐ ┌────────────────┐ │
│ │ HOUSE Pattern │ │ Filter House │ │ Four on Floor │ │
│ │ from CrumbMidi │ │ Tutorial │ │ Example │ │
│ │ ✓ CKL License │ │ ✓ Trust: 4 │ │ ✓ Verified │ │
│ └────────────────┘ └────────────────┘ └────────────────┘ │
│ │
│ Nur TRUSTED Code • Nur LICENSED Content • Keine Halluzination │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
🚀 Nächste Schritte
- [ ] ChromaDB Setup - Lokale Vector DB
- [ ] Ollama Embeddings - nomic-embed-text oder mxbai-embed-large
- [ ] Trust YAML - Für alle CrumbMidi Patterns
- [ ] Ingest Pipeline - Git → Chunk → Embed → Store
- [ ] Query API - FastAPI Server
- [ ] CLI Tool -
crumbvector search "one drop beat" - [ ] Integration - Mit Strudel / Terminal Dojo
🧠 CRUMBVECTOR 🧠
Code verstehen.
Code finden.
Code vertrauen.
Semantic Search für Certified Code.
#crumbvector #rag #vectordb #chromadb #embeddings #trust
🧠🔐📚