ContextManager

Transforms a raw list of retrieved Documents into an optimally-ordered, deduplicated, token-budget-aware context block.

Table of contents

• Overview
• Constructor
• build()
• format_memory()
• Algorithm details
• ContextResult
• Complexity
• Thread safety
• Examples

Overview

Key behaviors, straight from the module docstring and verified against the implementation:

• Relevance-first ordering — documents sorted by score descending before anything else.
• Deduplication — exact-hash + near-duplicate (shingle/Jaccard) removal.
• "Lost-in-the-middle" reordering — most-relevant docs placed at the start and end of the context, since LLM attention is empirically weakest for content in the middle of a long context.
• Token budget enforcement — a whitespace-based token heuristic (not a real tokenizer), with partial inclusion of the final doc that would overflow.
• Citation index — each included doc gets a [1], [2], ... label mapped back to its source.

Constructor

ContextManager(
    dedup_threshold:     float = 0.85,
    min_doc_tokens:      int   = 5,
    use_lost_in_middle:  bool  = True,
    summarize_overflow:  bool  = False,
    max_memory_messages: int   = 10,
) -> None

[!WARNING] summarize_overflow has no effect The module docstring lists "Adaptive summarization — docs that don't fit get one-sentence synopses" as a feature, and the constructor accepts summarize_overflow, but no method in the class branches on self.summarize_overflow. Overflow handling is always the truncate-and-append-"[...]" behavior in _apply_budget (see below), regardless of this setting. Treat this as an unimplemented/stubbed feature, not a bug in your usage.

build()

def build(self, request: PromptRequest) -> ContextResult

Internal workflow

1. Filter: drop documents where _count_tokens(d.content) < min_doc_tokens.
2. Sort: descending by d.score.
3. Deduplicate: _deduplicate() — see Algorithm details.
4. Reorder: _lost_in_middle_reorder() if use_lost_in_middle and len(docs) > 3.
5. Budget: _apply_budget(docs, request.max_context_tokens) → (included, excluded, truncated).
6. Format: _format(included) → (context_block: str, citation_map: Dict[int, str]).
7. Compute tokens_used = _count_tokens(context_block) (re-tokenizes the formatted block, including the --- Source [n] --- headers — so this is slightly higher than the sum of the raw document token counts).
8. Return a ContextResult.

Parameters

Returns

ContextResult.

Raises

No exceptions raised for empty document lists — _format([]) returns ("", {}) and the resulting ContextResult.context_block is simply "".

Complexity

• Time: dominated by deduplication, which compares every candidate document's shingle set against every previously-kept document's shingle set: O(n²) document-pairs in the worst case (no near-duplicates found, so nothing short-circuits), each comparison being O(k) in shingle-set size. Sorting is O(n log n); budget/format passes are O(n).
• Space: O(n · k) for the shingle sets kept in seen_shingles during dedup, plus O(n) for the output lists.

format_memory()

def format_memory(self, memory: List[Message], max_turns: Optional[int] = None) -> str

Formats the tail of a conversation as "User: ...\nAssistant: ..." lines.

• max_turns defaults to self.max_memory_messages (constructor value).
• Slicing is memory[-max_turns * 2:] — i.e. max_turns is interpreted as conversation turns (user+assistant pairs), so the actual number of Message objects kept is up to 2 * max_turns.
• Returns "" if memory is empty (after slicing).
• Role labeling: any role other than exactly "user" is labeled "Assistant" — so a stray "system"-role message inside memory would be mislabeled as "Assistant" rather than dropped or flagged.

cm = ContextManager(max_memory_messages=5)
block = cm.format_memory(chat_history)  # last 10 messages (5 turns)

Algorithm details

Deduplication

_deduplicate(docs):

1. For each doc, compute an exact content hash: hashlib.md5(re.sub(r"\s+", " ", text.lower()).encode()). If seen before, count as a duplicate and skip.
2. Otherwise compute 5-word shingles (_shingles, k=5): the set of all 5-word windows over the lowercased, whitespace-split content.
3. Compare this shingle set against every previously-kept document's shingle set via Jaccard similarity (_jaccard, intersection/union). If any comparison is >= dedup_threshold, treat as duplicate and skip.
4. Otherwise, keep the document and add its hash/shingles to the seen sets.

Returns (unique_docs, n_dupes).

Lost-in-the-middle reordering

@staticmethod
def _lost_in_middle_reorder(docs: List[Document]) -> List[Document]

Given docs already sorted by relevance descending, this walks the list and alternately places even-indexed docs (0, 2, 4, ...) at the front of a result array (advancing a left pointer) and odd-indexed docs (1, 3, 5, ...) at the back (retreating a right pointer). Net effect: the single most relevant doc goes first, the second-most-relevant goes last, the third-most goes second, the fourth-most goes second-to-last, and so on — concentrating high-relevance content at both ends and pushing lower-relevance content toward the middle.

Token budget enforcement

def _apply_budget(self, docs, budget) -> Tuple[included, excluded, truncated]

Greedily includes documents in order until the running token total would exceed budget. For the first document that doesn't fit, if the remaining budget is > 50 tokens, it creates a partial Document copy: takes the first int(remaining / 1.3) words, appends " [...]", sets metadata["truncated"] = True, and includes that partial version (the original is also recorded in excluded). All subsequent documents are excluded outright without a partial-inclusion attempt (the loop doesn't break, but once used has consumed the budget, later docs also fail the if remaining > 50 check unless used somehow decreases, which it doesn't — so in practice only the first overflowing doc ever gets a partial inclusion).

Formatting

@staticmethod
def _format(docs) -> Tuple[str, Dict[int, str]]

For each doc (1-indexed): citation_map[i] = doc.source or f"doc_{i}"; header is f"--- Source [{i}] {doc.source} ---" (or f"--- Source [{i}] ---" if no source) with an appended " (relevance: X.XX)" when score < 1.0. Documents are joined with "\n\n".

ContextResult

@dataclass
class ContextResult:
    context_block:      str
    included_docs:      List[Document]
    excluded_docs:      List[Document]
    citation_map:       Dict[int, str]
    tokens_used:        int
    tokens_budget:      int
    duplicates_removed: int
    truncated:          bool

utilization_pct (property)

round(self.tokens_used / max(self.tokens_budget, 1) * 100, 1)

Percentage of the token budget actually consumed; guards against division by zero via max(tokens_budget, 1).

Thread safety

ContextManager holds only configuration (set at construction) as instance state — .build() and .format_memory() don't mutate self. Concurrent calls from multiple threads on the same instance are safe with respect to self's own state, though this is not explicitly documented or tested in the source.

Examples

1. Basic

from zerda.prompt.context_manager import ContextManager
from zerda.prompt.types import PromptRequest, Document

cm = ContextManager()
req = PromptRequest(query="q", documents=[Document(content="Paris is the capital of France.", score=0.9)])
result = cm.build(req)
print(result.context_block)

2. Intermediate — inspecting truncation

result = cm.build(req)
if result.truncated:
    print(f"{len(result.excluded_docs)} docs excluded; utilization={result.utilization_pct}%")

3. Advanced — tighter dedup threshold

cm = ContextManager(dedup_threshold=0.6, min_doc_tokens=10)

4. Error handling — empty documents

req = PromptRequest(query="q", documents=[])
result = cm.build(req)
assert result.context_block == ""

5. Production — formatting memory alongside context

result = cm.build(request)
memory_block = cm.format_memory(request.memory, max_turns=8)