Celes Renata
a72f336ad1
Multi-stage evidence-grounded inference architecture replacing the monolithic 9B model extraction pipeline. CPU-first specialist services handle routine extraction while the 9B vLLM model is preserved for semantic adjudication of ambiguous cases. Key components: - Capability-aware inference gateway (OpenAI-compatible + Ollama) - Endpoint registry with DB migrations and REST API - Sentence-aware document segmenter (property tests) - Deterministic financial parsing with offset integrity - Symbol resolution with ambiguity detection - Specialist service (GLiNER2, dynamic batching, K8s deployment) - Company-specific sentiment (FinBERT, calibration) - Retrieval-based novelty and duplicate detection - Confidence calibration pipeline - Deterministic routing engine (property tests) - 9B adjudication layer with VRAM gating - Stock-specific impact model (features, labels, baseline, trained) - Pipeline orchestrator (state machine, queues, leases, feature flags) - Bounded parallelism (async workers, semaphore, load shedding) - Observability (tracing, metrics, alerts) - Compatibility adapter (v3→v2 golden mapping tests) - Shadow/canary promotion framework - Active learning and fine-tuning pipeline Test results: 1,161 tests pass, ruff lint clean. All 282 spec tasks completed.
120 lines
3.1 KiB
Python
120 lines
3.1 KiB
Python
"""Exact and near-duplicate fingerprinting for document deduplication.
|
|
|
|
Provides:
|
|
- SHA-256 exact fingerprint on normalized text
|
|
- SimHash for near-duplicate detection with configurable threshold
|
|
- Hamming distance comparison between SimHash values
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import re
|
|
import struct
|
|
|
|
|
|
def _normalize_text(text: str) -> str:
|
|
"""Normalize text for fingerprinting.
|
|
|
|
Lowercases, collapses whitespace, strips leading/trailing space.
|
|
This ensures minor formatting differences don't defeat deduplication.
|
|
"""
|
|
text = text.lower()
|
|
text = re.sub(r"\s+", " ", text)
|
|
return text.strip()
|
|
|
|
|
|
def compute_exact_fingerprint(text: str) -> str:
|
|
"""Compute SHA-256 fingerprint of normalized text.
|
|
|
|
Args:
|
|
text: Raw document text.
|
|
|
|
Returns:
|
|
Hex-encoded SHA-256 digest of normalized text.
|
|
"""
|
|
normalized = _normalize_text(text)
|
|
return hashlib.sha256(normalized.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def _tokenize(text: str) -> list[str]:
|
|
"""Split normalized text into tokens for SimHash computation."""
|
|
normalized = _normalize_text(text)
|
|
return normalized.split()
|
|
|
|
|
|
def _hash_token(token: str) -> int:
|
|
"""Hash a single token to a 64-bit integer using MD5 truncation."""
|
|
digest = hashlib.md5(token.encode("utf-8")).digest() # noqa: S324
|
|
return struct.unpack("<Q", digest[:8])[0]
|
|
|
|
|
|
def compute_simhash(text: str) -> int:
|
|
"""Compute 64-bit SimHash of text for near-duplicate detection.
|
|
|
|
SimHash produces locality-sensitive fingerprints: similar documents
|
|
will have SimHash values with low Hamming distance.
|
|
|
|
Args:
|
|
text: Raw document text.
|
|
|
|
Returns:
|
|
64-bit SimHash integer value.
|
|
"""
|
|
tokens = _tokenize(text)
|
|
if not tokens:
|
|
return 0
|
|
|
|
# Accumulator for each bit position
|
|
v = [0] * 64
|
|
|
|
for token in tokens:
|
|
token_hash = _hash_token(token)
|
|
for i in range(64):
|
|
if token_hash & (1 << i):
|
|
v[i] += 1
|
|
else:
|
|
v[i] -= 1
|
|
|
|
# Build final hash from accumulator signs
|
|
fingerprint = 0
|
|
for i in range(64):
|
|
if v[i] > 0:
|
|
fingerprint |= 1 << i
|
|
|
|
return fingerprint
|
|
|
|
|
|
def hamming_distance(a: int, b: int) -> int:
|
|
"""Compute Hamming distance between two 64-bit SimHash values.
|
|
|
|
Args:
|
|
a: First SimHash value.
|
|
b: Second SimHash value.
|
|
|
|
Returns:
|
|
Number of differing bits (0-64).
|
|
"""
|
|
xor = a ^ b
|
|
# Count set bits (Brian Kernighan's algorithm)
|
|
distance = 0
|
|
while xor:
|
|
xor &= xor - 1
|
|
distance += 1
|
|
return distance
|
|
|
|
|
|
def is_near_duplicate(fp1: int, fp2: int, threshold: int = 3) -> bool:
|
|
"""Determine if two SimHash fingerprints indicate near-duplicate content.
|
|
|
|
Args:
|
|
fp1: First SimHash value.
|
|
fp2: Second SimHash value.
|
|
threshold: Maximum Hamming distance to consider near-duplicate.
|
|
Default of 3 is conservative for 64-bit SimHash.
|
|
|
|
Returns:
|
|
True if the documents are near-duplicates.
|
|
"""
|
|
return hamming_distance(fp1, fp2) <= threshold
|