Privacy-Aware Corpus Intelligence Pipeline
Privacy-safe intelligence from large text archives.
A local-first corpus analysis platform that separates personal, regulated, and review-worthy conversational records from public-safe knowledge candidates. The project focuses on cost-conscious privacy screening, multi-model validation, and auditable data products without sending the full archive through bulk LLM inference.
The system ingests a large exported conversation corpus, normalizes it into classification units, applies deterministic privacy policy gates, cross-checks the result with independent non-LLM detectors, and writes JSON plus Markdown evidence for review.
What inspired this build
We often hear that conversations may be recorded for audit, quality, or training purposes. I wanted to go behind the curtain and build the data engineering version of that problem: how do you remove identity, sensitive context, and private identifiers from a large text archive while still preserving useful knowledge? This project became a real-time privacy filtering pipeline that validates multiple non-LLM approaches and chooses an ensemble route that is practical, inspectable, and affordable in 2026.
Dataset Information
The source material is described only as a large exported conversational corpus. The page intentionally does not disclose personal topics, raw excerpts, names, account details, or source-file identifiers. All public examples below are synthetic but shaped like the real pipeline contracts.
Worked Examples of the Classification Passes
These are precomputed worked examples, not a live service: the page is static. The code below each example is the real implementation, extracted from the repository. The examples are harmless and synthetic: a public technical note, an identity-heavy private note, and a borderline review item.
Source Payload
Select a sample and run the preview.
Curated Output
Waiting for classifier output...
def policy_classifier(unit: CorpusUnit) -> ValidatorDecision:
result = classify(unit)
if result.decision.startswith("exclude"):
label = PRIVATE_LABEL
elif result.decision == "public_candidate":
label = PUBLIC_LABEL
else:
label = REVIEW_LABEL
reasons = result.exclusion_reasons + result.identifier_hits
if result.public_topics:
reasons.extend(topic for topic, _ in result.public_topics[:2])
confidence = min(0.99, max(0.50, result.score / 220))
return ValidatorDecision("policy_classifier", label, confidence, reasons[:8]) Verbatim from src/corpus_privacy_intelligence/validators.py · policy_classifier()
def strict_detector(unit: CorpusUnit) -> ValidatorDecision:
text = f"{unit.title}\n{unit.text}"
identifier_hits = detect_identifiers(text)
private_hits = [name for name, pattern in STRICT_PRIVATE_PATTERNS.items() if pattern.search(text)]
public_hits = [name for name, pattern in PUBLIC_CONTEXT_PATTERNS.items() if pattern.search(text)]
if identifier_hits:
return ValidatorDecision("strict_detector", PRIVATE_LABEL, 0.98, identifier_hits)
if len(private_hits) >= 2:
return ValidatorDecision("strict_detector", PRIVATE_LABEL, 0.92, private_hits)
if len(private_hits) == 1:
return ValidatorDecision("strict_detector", PRIVATE_LABEL, 0.82, private_hits)
if public_hits:
return ValidatorDecision("strict_detector", PUBLIC_LABEL, 0.76, public_hits)
return ValidatorDecision("strict_detector", REVIEW_LABEL, 0.50, []) Verbatim from src/corpus_privacy_intelligence/validators.py · strict_detector()
def semantic_score_classifier(unit: CorpusUnit) -> ValidatorDecision:
text = f"{unit.title}\n{unit.text}"
identifier_hits = detect_identifiers(text)
if identifier_hits:
return ValidatorDecision("semantic_score_classifier", PRIVATE_LABEL, 0.97, identifier_hits)
phrase_private_hits = [name for name, pattern in STRICT_PRIVATE_PATTERNS.items() if pattern.search(text)]
counts = term_counter(text)
private_scores = {
name: sum(counts.get(normalize_token(term), 0) for term in terms)
for name, terms in SEMANTIC_PRIVATE_TERMS.items()
}
public_scores = {
name: sum(counts.get(normalize_token(term), 0) for term in terms)
for name, terms in SEMANTIC_PUBLIC_TERMS.items()
}
private_total = sum(private_scores.values())
public_total = sum(public_scores.values())
top_private = [name for name, value in private_scores.items() if value]
top_public = [name for name, value in public_scores.items() if value]
if phrase_private_hits:
confidence = 0.86 if public_total else 0.92
return ValidatorDecision("semantic_score_classifier", PRIVATE_LABEL, confidence, phrase_private_hits)
if private_total >= 3 and private_total >= public_total * 0.25:
confidence = min(0.95, 0.60 + private_total / max(private_total + public_total, 1) * 0.35)
return ValidatorDecision("semantic_score_classifier", PRIVATE_LABEL, confidence, top_private)
if public_total >= 5 and private_total <= 2:
confidence = min(0.90, 0.55 + public_total / max(private_total + public_total, 1) * 0.30)
return ValidatorDecision("semantic_score_classifier", PUBLIC_LABEL, confidence, top_public)
if private_total > public_total:
return ValidatorDecision("semantic_score_classifier", REVIEW_LABEL, 0.62, top_private + top_public)
if public_total:
return ValidatorDecision("semantic_score_classifier", PUBLIC_LABEL, 0.64, top_public)
return ValidatorDecision("semantic_score_classifier", REVIEW_LABEL, 0.50, []) Verbatim from src/corpus_privacy_intelligence/validators.py · semantic_score_classifier()
def weighted_ensemble(results: list[DetectorResult]) -> tuple[str, float]:
weights = {
"policy": 1.0,
"strict_detector": 0.9,
"semantic_score_classifier": 0.8,
"presidio": 1.2,
"spacy": 0.8,
}
scores = Counter()
for result in results:
if result.confidence == 0:
continue
scores[result.label] += weights.get(result.name, 0.75) * result.confidence
if scores[PRIVATE_LABEL] >= 1.15:
total = sum(scores.values()) or 1
return PRIVATE_LABEL, round(scores[PRIVATE_LABEL] / total, 4)
if scores[PUBLIC_LABEL] > scores[PRIVATE_LABEL] and scores[PUBLIC_LABEL] >= 1.25:
total = sum(scores.values()) or 1
return PUBLIC_LABEL, round(scores[PUBLIC_LABEL] / total, 4)
return REVIEW_LABEL, round((scores[REVIEW_LABEL] + 0.25) / (sum(scores.values()) + 0.25), 4) Verbatim from src/corpus_privacy_intelligence/advanced_validation.py · weighted_ensemble()
Schema and Source Contract
Classification Unit Contract
Classifier Result Contract
Model Comparison and Ensemble Results
The system does not trust one classifier blindly. It compares multiple independent views of the same corpus: a production policy classifier, a strict privacy detector, a semantic scoring classifier, Presidio, spaCy NER, and a final ensemble route. The large full-corpus pass establishes the production baseline. The advanced NLP sample adds a free, state-of-the-art non-LLM validation layer.
Agreement Snapshot
Why the ensemble works
The policy classifier enforces the product rule: private domains and identifiers override usefulness. The strict detector challenges that decision with a narrower safety lens. The semantic classifier catches topic intent. Presidio and spaCy provide independent entity recognition. The ensemble protects the corpus by preserving clear public candidates, excluding high-confidence private records, and routing ambiguous items to review.
Quality and Privacy Gates
The pipeline treats privacy as a data quality problem. Every unit receives evidence, not just a label. That evidence can be audited, sampled, validated, and improved without exposing the raw corpus.
Engineering Toolchain
Every tool listed is in the running system. Forward-looking work lives in the repository README, not on this page.
Verify It Yourself
Every number on this page comes from a committed artifact, and every line of code shown above is extracted verbatim from the repository. CI fails the build if either drifts.
# the suite that gates every push git clone https://github.com/svaddhiparthy/Privacy-Aware-Corpus-Intelligence-Pipeline.git cd Privacy-Aware-Corpus-Intelligence-Pipeline pip install -e ".[dev]" && python -m pytest -q # the checks that keep this page honest python scripts/check_published_numbers.py python scripts/export_page_snippets.py