nrchd/
├── app.py # Streamlit entry point, tab routing
├── tabs/
│ ├── data_tab.py # Upload, preview, column mapping, row filter
│ ├── analyse_tab.py # Sentence transformer features
│ ├── knowledge_tab.py # RAG — ingest, index, retrieval settings
│ ├── task_tab.py # Prompt editor, schema builder, test mode
│ ├── run_tab.py # Progress display, batch status, pause/stop
│ └── results_tab.py # Output table, filters, download buttons
├── engine/
│ ├── runner.py # Async batch executor, rate limiter, semaphore
│ ├── clients.py # Unified LLM interface (Claude / OpenAI / Groq / Ollama)
│ ├── prompt_builder.py # Template + row data + context → final prompt
│ ├── schema_validator.py # Validate & coerce LLM response to output schema
│ ├── column_mapper.py # Auto-detect + fuzzy match CSV column roles
│ ├── semantic/
│ │ ├── embedder.py # Encode text, cache embeddings
│ │ ├── deduper.py # Similarity pairs, greedy removal, export pairs
│ │ ├── clusterer.py # K-means / agglomerative, auto-N option
│ │ ├── outlier.py # Cosine distance from cluster centroid
│ │ └── cross_match.py # Two-file pairwise similarity
│ └── rag/
│ ├── ingester.py # Parse PDF / DOCX / TXT / MD / CSV / folder
│ ├── chunker.py # Fixed / paragraph / sentence chunking strategies
│ ├── indexer.py # Build & persist ChromaDB vector store
│ └── retriever.py # Query → top-K chunks with similarity scores
├── templates/ # Built-in + user-saved task templates (JSON)
│ ├── question_bank_validator.json
│ ├── sentiment_analyser.json
│ └── ...
├── knowledge_bases/ # Persisted ChromaDB indexes (gitignored)
├── outputs/ # Generated CSVs (gitignored)
├── .env.example # Config template (committed)
├── .env # User's actual keys (gitignored)
├── pyproject.toml # Dependencies + uv config
├── uv.lock # Exact dep lockfile (committed)
├── requirements.txt # Fallback for pip users
├── Makefile # Convenience commands
├── Dockerfile
├── docker-compose.yml
├── .gitignore
└── README.md
# Concurrency + rate limiting
semaphore = asyncio.Semaphore(MAX_CONCURRENT) # max 10 in-flight
rate_limiter = SlidingWindowRateLimiter( # max 10 per 60s
max_calls=MAX_PER_MINUTE,
window_seconds=60
)
async def call_api(batch, semaphore, rate_limiter):
await rate_limiter.acquire() # wait if > 10 calls in last 60s
async with semaphore: # wait if > 10 concurrent
prompt = build_prompt(batch)
response = await client.complete(prompt)
return parse_response(response)
# Fire all batches
tasks = [call_api(batch, semaphore, rate_limiter) for batch in batches]
results = await asyncio.gather(*tasks, return_exceptions=True)class SlidingWindowRateLimiter:
def __init__(self, max_calls, window_seconds):
self.max_calls = max_calls
self.window = window_seconds
self.timestamps = []
async def acquire(self):
now = time.monotonic()
# Drop timestamps outside window
self.timestamps = [t for t in self.timestamps if now - t < self.window]
if len(self.timestamps) >= self.max_calls:
sleep_for = self.window - (now - self.timestamps[0])
await asyncio.sleep(sleep_for)
self.timestamps.append(time.monotonic())# clients.py — same interface regardless of backend
class LLMClient(Protocol):
async def complete(self, prompt: str, schema: OutputSchema) -> dict: ...
class ClaudeClient:
async def complete(self, prompt, schema): ...
class OpenAIClient:
async def complete(self, prompt, schema): ...
class GroqClient:
async def complete(self, prompt, schema): ...
class OllamaClient:
base_url: str = "http://localhost:11434"
async def complete(self, prompt, schema): ...
def get_client(config: Settings) -> LLMClient:
match config.model_source:
case "claude": return ClaudeClient(config.api_key, config.model)
case "openai": return OpenAIClient(config.api_key, config.model)
case "groq": return GroqClient(config.api_key, config.model)
case "ollama": return OllamaClient(config.ollama_url, config.model){
"name": "Question Bank Validator",
"description": "Validates MCQ answers and explanations, generates hints",
"version": "1.0",
"column_hints": {
"question": ["question_text", "question", "q"],
"option_a": ["option_a", "opt_a", "a"],
"answer": ["correct_option", "answer", "correct"]
},
"prompt_template": "You are an expert educator...\n\nQuestion: {question_text}\n...\n\nReturn JSON with the fields below.",
"output_schema": [
{"name": "answer_correct", "type": "bool", "description": "Is the marked answer correct?"},
{"name": "explanation_correct", "type": "bool", "description": "Is the explanation accurate?"},
{"name": "tightened_answer", "type": "string", "description": "Concise version of correct answer"},
{"name": "detailed_explanation","type": "string", "description": "Richer conceptual explanation"},
{"name": "hint", "type": "string", "description": "Nudge toward answer without giving it away"}
],
"recommended_settings": {
"batch_size": 100,
"max_concurrent": 10,
"max_per_minute": 10,
"model": "claude-haiku-4-5-20251001"
}
}def parse_response(raw: str, schema: OutputSchema) -> list[dict]:
# 1. Strip markdown fences if present
text = re.sub(r'^```json\s*', '', raw.strip())
text = re.sub(r'\s*```$', '', text)
# 2. Find JSON array boundaries (handles trailing commentary)
start = text.find('[')
for end in range(len(text), start, -1):
try:
result = json.loads(text[start:end])
if isinstance(result, list):
break
except json.JSONDecodeError:
continue
# 3. Validate + coerce against schema
return [validate_and_coerce(item, schema) for item in result]def get_processed_ids(output_path: Path) -> set[str]:
if not output_path.exists():
return set()
with open(output_path, newline='') as f:
return {row['id'] for row in csv.DictReader(f)}
# On startup
processed_ids = get_processed_ids(output_path)
pending_rows = [r for r in all_rows if r['id'] not in processed_ids]
batches = chunk(pending_rows, BATCH_SIZE)Checkpointing: append results to output CSV after each batch completes.
Failed batches: written to {output_stem}.failed.json with full batch payload for re-run.
- Encode text column with sentence transformer
- Normalise embeddings, compute cosine similarity matrix per group (sub_area etc.)
- Find all pairs >= threshold
- Greedy removal sorted by similarity desc: keep lower ID, remove higher
- Export pairs CSV for manual review before removal
- K-means (fast, user sets K) or Agglomerative (auto-determine K via silhouette score)
- 2D visualisation via UMAP or t-SNE (optional, heavy dep — lazy import)
- Output: cluster label per row, sample rows per cluster shown in UI
- Compute centroid per cluster
- Flag rows with cosine distance > threshold from their cluster centroid
- Sensitivity slider maps to percentile cutoff
- Embed column from file A, embed column from file B
- Compute pairwise similarity (A_embeddings @ B_embeddings.T)
- Return top matches per row in A with similarity score
File/Folder
↓
[ingester.py] — detect type, extract text
PDF → pymupdf (fitz)
DOCX → python-docx
TXT/MD → direct read
CSV → each row as document
↓
[chunker.py] — split into chunks
Fixed: every N chars with M overlap
Paragraph: split on \n\n
Sentence: split on . / ? / !
↓
[indexer.py] — embed + store
Embed chunks with sentence transformer
Store in ChromaDB collection (named KB)
Persist to disk at knowledge_bases/{name}/
def retrieve(row: dict, query_columns: list[str], k: int) -> str:
query = " ".join(row[col] for col in query_columns)
results = collection.query(query_texts=[query], n_results=k)
chunks = results['documents'][0]
return "\n\n---\n\n".join(chunks)Retrieved context injected as {context} in prompt template.
ANTHROPIC_API_KEY
OPENAI_API_KEY
GROQ_API_KEY
OLLAMA_BASE_URL default: http://localhost:11434
DEFAULT_BATCH_SIZE default: 100
DEFAULT_MAX_CONCURRENT default: 10
DEFAULT_MAX_PER_MINUTE default: 10
DEFAULT_MAX_RETRIES default: 3
Sidebar always shows current values and allows override for the session.