legacy_compat module

Overview

These five symbols implement an older API generation that operated on plain (chunk, score) tuples rather than ScoredChunk objects. They are not part of context/__init__.py's __all__, and — as documented in api/overview.md — are actually shadowed with None at the package root due to a bug in __init__.py. Import them directly from this submodule:

from context.legacy_compat import  (
    validate_chunk, PrefixDeduplication, HashDeduplication, GreedyCompression, ScoreRanking,
)

validate_chunk(chunk)

def validate_chunk(chunk: object) -> None: ...

Overview

Why it exists: to give callers of the legacy tuple-based API an explicit, early TypeError for malformed chunk objects, rather than a confusing failure deep inside ranking or composition.

Parameters

Returns

None. This function only validates; it has no return value to act on.

Raises

Examples

from zerda.context.legacy_compat import validate_chunk

class GoodChunk:
    doc_id = "doc-1"
    text = "some text"

validate_chunk(GoodChunk())  # passes silently

class BadChunk:
    doc_id = "doc-1"
    text = 123  # wrong type

try:
    validate_chunk(BadChunk())
except TypeError as e:
    print(e)  # "chunk.text must be str, got int."

PrefixDeduplication

class PrefixDeduplication:
    def __init__(self, prefix_length: int = 100): ...
    def deduplicate(self, chunks: List[Tuple]) -> List[Tuple]: ...

Implements strategy.DeduplicationStrategy (structurally — not declared with explicit inheritance).

Algorithm

For each (chunk, score) tuple, takes the lowercased first prefix_length characters of chunk.text as a signature; keeps only the first tuple seen per unique signature.

Raises

ValueError if prefix_length > len(chunk.text) for any chunk in the input — i.e. this class cannot safely process a corpus containing any chunk shorter than prefix_length, and will raise mid-iteration (chunks processed before the short one are simply lost, since the method doesn't return partial results on failure).

dedup = PrefixDeduplication(prefix_length=100)
short_chunk = SimpleNamespace(text="short")  # 5 chars
try:
    dedup.deduplicate([(short_chunk, 0.9)])
except ValueError as e:
    print(e)  # "prefix_length 100 exceeds chunk text length 5."

Complexity

O(n) in number of chunks; O(p) per chunk for the prefix slice/lowercasing, where p = prefix_length.

HashDeduplication

class HashDeduplication:
    def deduplicate(self, chunks: List[Tuple]) -> List[Tuple]: ...

Implements DeduplicationStrategy. Same approach as ScoredChunk._content_hash (whitespace-normalize, lowercase, SHA-256), but operating on tuples instead of ScoredChunk objects, and with no prefix_length restriction — does not raise on short chunks, unlike PrefixDeduplication.

ScoreRanking

class ScoreRanking:
    def rank(self, chunks: List[Tuple], max_chunks: Optional[int]) -> List[Tuple]: ...

Implements strategy.RankingStrategy. Pure sorted(..., key=lambda x: x[1], reverse=True), truncated to max_chunks if given (passing None returns the full sorted list, not an empty one — max_chunks is only applied via slicing if it's not None).

GreedyCompression

class GreedyCompression:
    MIN_CONTENT_LENGTH = 50
    def compress(self, context: str, target: int, counter, separator: str) -> str: ...

Implements strategy.CompressionStrategy. This is a standalone alternative to SmartContextComposer's budget-fitting logic, operating on an already-composed context string rather than a list of chunks.

Algorithm

1. If counter(context) <= target, returns unchanged.
2. If target <= 0, returns "".
3. Splits context by separator. If 2 or fewer parts result, falls back to a plain word-boundary truncation (_truncate) of the whole string.
4. Otherwise treats the first part as a header and the last as a footer, computes overhead (header + footer + 2×separator), and greedily keeps middle parts until the budget (target - overhead) runs out — truncating (not dropping) the part that doesn't fully fit, then stopping.
5. If available < MIN_CONTENT_LENGTH (50 chars) after subtracting overhead, falls back to truncating just the header.
6. Final safety clamp: if the assembled result still exceeds target, truncates the whole thing.

This logic is structurally very similar to SmartContextComposer._assemble, but implemented independently — they are not refactored to share code, so changes to one will not automatically apply to the other.

_truncate(text, max_len, counter) (static method)

Same binary-search-to-last-space-plus-ellipsis approach used elsewhere in the package (SmartContextComposer._hard_truncate, ContextEngine._hard_truncate) — implemented a third time here, independently, with a slightly different space-backtrack threshold (last_space > lo * 0.8 here, vs. 0.7 in SmartContextComposer).

Examples

1. Basic:

from zerda.context.legacy_compat import HashDeduplication
dedup = HashDeduplication()
unique = dedup.deduplicate([(chunk1, 0.9), (chunk2, 0.8), (chunk1, 0.95)])  # chunk1 kept once

2. Intermediate — full legacy pipeline:

from zerda.context.legacy_compat import HashDeduplication, ScoreRanking, GreedyCompression

chunks = [(c, s) for c, s in retrieved_tuples]
chunks = HashDeduplication().deduplicate(chunks)
chunks = ScoreRanking().rank(chunks, max_chunks=10)
context = "\n---\n".join(c.text for c, s in chunks)
context = GreedyCompression().compress(context, target=1500, counter=len, separator="\n---\n")

3. Advanced — validating chunks before legacy processing:

from zerda.context.legacy_compat import validate_chunk
for chunk, score in raw_chunks:
    validate_chunk(chunk)  # raises TypeError early if malformed

4. Error handling — prefix dedup on short chunks:

from zerda.context.legacy_compat import PrefixDeduplication
try:
    PrefixDeduplication(prefix_length=200).deduplicate(short_chunks)
except ValueError:
    # Fall back to HashDeduplication, which has no length constraint
    from context.legacy_compat import HashDeduplication
    unique = HashDeduplication().deduplicate(short_chunks)

5. Production usage — migrating off the legacy API:

# Old:
from zerda.context.legacy_compat import HashDeduplication, ScoreRanking
# New (preferred):
from zerda.context import ChunkFilter, CompositeRanker, ContextEngineConfig
cfg = ContextEngineConfig()
chunk_filter = ChunkFilter(cfg)   # handles hash dedup + noise + length filtering
ranker = CompositeRanker(cfg)     # handles multi-factor ranking, not just raw score sort

Best practices

• Prefer HashDeduplication over PrefixDeduplication unless you specifically need prefix-based matching — it has no minimum-length failure mode.
• For new code, prefer the modern ChunkFilter/CompositeRanker/SmartContextComposer trio over this module; it offers richer scoring (keyword boost, source quality, recency, position) that the legacy classes don't have.

Common mistakes

• Importing these five names from the package root (from context import validate_chunk) — due to the __init__.py quirk described in api/overview.md, this silently gives you None instead of the function/class.
• Using PrefixDeduplication on a corpus with variable-length chunks without first checking minimum length.

Related

• strategy — the four legacy ABCs these classes implement
• engine.ContextManager — accepts (but, per its docs, does not actually use) legacy strategy instances
• API Overview → the __init__.py quirk

← Back to API overview