ContextBuilder

Purpose

Assembles retrieved RAG documents, selected memories, conversation history, and a user-profile snippet into one structured, token-budgeted string ready to inject into an LLM prompt — the final step of the memory pipeline, consumed by AIMemoryManager.build_context.

Document (dataclass)

@dataclass
class Document:
    page_content: str
    metadata: Dict[str, Any] = field(default_factory=dict)
    doc_id: Optional[str] = None

    def token_estimate(self) -> int:
        return max(1, len(self.page_content.split()) // 4)

A RAG-retrieved chunk. token_estimate() is a word-count heuristic (words // 4, minimum 1) — note this is not called anywhere by ContextBuilder.build() itself, which computes its own token estimates inline; Document.token_estimate() is provided as a standalone utility for callers who want a per-document estimate before building.

BuiltContext (dataclass)

@dataclass
class BuiltContext:
    full_text: str
    token_estimate: int
    sections: Dict[str, str]
    truncated: bool = False
    document_count: int = 0
    memory_count: int = 0

The result of ContextBuilder.build(). sections maps section name ("user_profile", "memories", "documents", "history") to its fully-formatted text (including the === Section Name === header), for callers who want to inspect or re-order sections rather than use the concatenated full_text.

ContextBuilder constructor

def __init__(
    self,
    total_token_budget: int = 4_000,
    *,
    profile_budget_fraction: float = 0.05,
    memory_budget_fraction: float = 0.30,
    document_budget_fraction: float = 0.45,
    history_budget_fraction: float = 0.20,
) -> None

The four fractions sum to 1.0 by default, but this is not validated or enforced — you can pass fractions that sum to more or less than 1.0 (e.g. all 0.5) with no error; the resulting per-section budgets are simply int(total_token_budget * fraction) each, independently, and the sum of actual section budgets can exceed or fall short of total_token_budget as a result. There is no cross-section rebalancing.

build() — main API

def build(
    self, *,
    documents: Optional[List[Document]] = None,
    memories: Optional[List[MemoryEntry]] = None,
    history: Optional[List[Dict[str, str]]] = None,
    user_profile_text: Optional[str] = None,
    query: Optional[str] = None,
) -> BuiltContext

All parameters are optional keyword-only. Sections are built and added in this fixed priority order:

1. user_profile — only if user_profile_text is truthy.
2. memories — only if memories is truthy (non-empty list). Each entry rendered via _entry_to_text (see below) and numbered [Memory N].
3. documents — only if documents is truthy. Each rendered [Doc N (source: ...)]\n{content} if metadata.get('source') is set, else [Doc N]\n{content}.
4. history — only if history is truthy. Each turn dict rendered as User: ... / Assistant: ... lines, only including a line if the corresponding key ("input"/"output") is truthy in that turn dict.

Per-section truncation

def _add_section(name, content):
    budget = self._budgets.get(name, 500)
    token_est = max(1, len(content.split()) // 4)
    if token_est > budget:
        content = content[: budget * 4] + "\n[...truncated]"
        truncated = True
    sections[name] = TEMPLATE.format(content=content.strip())

If a section's estimated token count exceeds its budget, the content is character-truncated to budget * 4 characters (reversing the words // 4 ≈ tokens heuristic back into a character count) and a "\n[...truncated]" marker is appended. This is a blunt, non-word-boundary-aware cut — truncation can happen mid-word. truncated=True is set on the returned BuiltContext if any section was truncated (there's no per-section truncation flag on BuiltContext itself, only a single aggregate boolean).

[!NOTE] No section-level ordering by priority once one section is over budget Even if the documents section vastly exceeds its budget and gets truncated, this doesn't "borrow" spare budget from an under-used section like history — each section's budget is fixed and independent, computed once at ContextBuilder.__init__ time.

_entry_to_text (internal, for memories)

if "summary" in c: return c["summary"]
if "merged" in c: return c["merged"]
# else format as "User: ...\nAssistant: ..."

Recognizes the special content shapes produced by MemoryCompressor ({"summary": ...} from summarise(), {"merged": ...} from _merge_pair()) and renders those directly, falling back to the standard input/output format otherwise.

Final assembly

header = f"# Context for: {query}\n\n" if query else ""
full_text = header + "\n".join(sections.values())

Sections appear in dict insertion order (Python 3.7+ dicts preserve insertion order), which matches the fixed priority order above (profile → memories → documents → history) since they're always added in that sequence when present.

build_prompt() — convenience wrapper

def build_prompt(
    self, query: str, *,
    documents=None, memories=None, history=None,
    user_profile_text=None,
    system_instruction: str = "You are a helpful, knowledgeable assistant.",
) -> str

Calls build() internally, then wraps the result with a system instruction and a === Current Query === footer, textwrap.dedent-ed and .strip()-ed. Convenient for direct LLM calls; AIMemoryManager.build_prompt() does not call this cleanly — see the caveat below.

Raises

No exceptions are raised by build() or build_prompt() under any documented input — malformed history dicts (missing keys) are simply skipped for that line, not errored.

Thread safety

ContextBuilder holds only its budget configuration (set once at __init__, read-only thereafter) — safe to share one instance across threads and call build()/build_prompt() concurrently.

Complexity

O(total content size) — linear in the combined length of all documents, memories, history turns, and profile text. No search or sorting is performed by ContextBuilder itself (its inputs are assumed pre-sorted by relevance/priority by the caller).

Examples

from zerda.memory.rag import ContextBuilder, Document

cb = ContextBuilder(total_token_budget=2000)

# 1. Basic
ctx = cb.build(
    documents=[Document(page_content="Paris is the capital of France.", metadata={"source": "wiki"})],
    query="What's the capital of France?",
)
print(ctx.full_text)

# 2. Intermediate — with memories and history
from zerda.memory import MemoryEntry, MemoryType
import time
mem = MemoryEntry(content={"input": "I live in Paris", "output": "Noted!"},
                   timestamp=time.time(), memory_type=MemoryType.SEMANTIC)
ctx2 = cb.build(memories=[mem], history=[{"input": "Hi", "output": "Hello!"}])

# 3. Advanced — inspecting individual sections
print(ctx2.sections.keys())      # dict_keys(['memories', 'history'])
print(ctx2.truncated)            # False, unless content exceeded a section budget

# 4. Error handling — malformed history entries are simply skipped, not errored
ctx3 = cb.build(history=[{"input": None, "output": "reply with no input"}])
# only the "Assistant: reply with no input" line is included

# 5. Production — direct prompt string for an LLM call
prompt = cb.build_prompt(
    query="Summarize what we know about the user",
    memories=[mem],
    system_instruction="You are a CRM assistant.",
)
response = my_llm.generate(prompt, max_tokens=300)

Best practices

• Verify your *_budget_fraction values actually sum to 1.0 if you override them — ContextBuilder won't validate or warn you otherwise.
• Check BuiltContext.truncated after build() if losing content silently would be a problem for your use case — there's no per-section detail, only the aggregate flag.
• Pre-sort documents/memories by relevance before passing them in — ContextBuilder preserves input order, it does not re-rank.

Common mistakes

• Assuming truncation happens at a word boundary — it's a raw character-count cut.
• Assuming spare budget in one section is available to an over-budget section — budgets are fixed and independent.

Related APIs

• MemorySelector — typically produces the memories list passed into build().
• MemoryCompressor — produces the {"summary": ...}/{"merged": ...} content shapes _entry_to_text specially recognizes.
• AIMemoryManager.build_context / build_prompt — the primary caller, with a caveat about how it uses build_prompt (see that page).