models module

Overview

Defines the core value objects passed between every pipeline stage: four enums, five dataclasses, and one standalone helper function. None of these classes hold any retrieval or LLM logic themselves — they are pure data containers with a handful of convenience methods (to_dict(), property accessors, history management).

Enumerations

QueryIntent

class QueryIntent(Enum):
    FACTUAL      = "factual"
    REASONING    = "reasoning"
    COMPARATIVE  = "comparative"
    PROCEDURAL   = "procedural"
    CODE         = "code"
    SUMMARIZE    = "summarize"
    UNKNOWN      = "unknown"

Assigned by RuleBasedQueryAnalyzer._extract_intent. FACTUAL is the default/fallback when no intent signal matches.

QueryComplexity

class QueryComplexity(Enum):
    LOW    = "low"
    MEDIUM = "medium"
    HIGH   = "high"

Drives the context budget via QueryAnalyzerConfig.budget_low/medium/high. MEDIUM is the dataclass-level default for QueryAnalysis.complexity (i.e. before analysis is run on an unconstructed instance), but real analysis always sets one of the three values explicitly.

RetrievalMethod

class RetrievalMethod(Enum):
    VECTOR   = "vector"
    KEYWORD  = "keyword"
    METADATA = "metadata"
    HYBRID   = "hybrid"

Tags which retriever produced a given ScoredChunk. Note: despite HYBRID being defined here, no code path inspected ever assigns RetrievalMethod.HYBRID to a chunk — individual chunks are always tagged VECTOR, KEYWORD, or METADATA by their originating retriever; HYBRID appears to be reserved for future use or external consumers.

SourceQuality

class SourceQuality(Enum):
    HIGH   = 3
    MEDIUM = 2
    LOW    = 1
    UNKNOWN = 0

Integer-valued (not string-valued, unlike the other three enums) so it can be divided by 3 directly in CompositeRanker._compute_composite to normalize to [0, 1].

QueryAnalysis

@dataclass
class QueryAnalysis:
    original_query:  str
    normalized:      str
    intent:          QueryIntent   = QueryIntent.UNKNOWN
    complexity:      QueryComplexity = QueryComplexity.MEDIUM
    keywords:        List[str]     = field(default_factory=list)
    entities:        List[str]     = field(default_factory=list)
    is_arabic:       bool          = False
    language:        str           = "en"
    question_type:   str           = ""
    context_budget:  int           = 2000
    timestamp:       float         = field(default_factory=time.time)

The output of any QueryAnalyzerStrategy. original_query and normalized are required positional/keyword arguments; everything else has a default.

Methods

to_dict()

def to_dict(self) -> Dict[str, Any]: ...

Returns a JSON-serializable dict (enums converted via .value). Does not include timestamp — it is the one field deliberately omitted from the dict representation.

Example:

analysis = QueryAnalysis(original_query="what is RAG?", normalized="what is rag?")
analysis.to_dict()
# {'original_query': 'what is RAG?', 'normalized': 'what is rag?', 'intent': 'unknown',
#  'complexity': 'medium', 'keywords': [], 'entities': [], 'is_arabic': False,
#  'language': 'en', 'question_type': '', 'context_budget': 2000}

chunk_metadata_dict(meta)

def chunk_metadata_dict(meta: Any) -> Dict[str, Any]: ...

Standalone module-level helper, added alongside the native zerda.chunks/zerda.vector_database integration. Normalizes a chunk's metadata attribute into a plain dict, regardless of its actual runtime shape:

This exists because DocumentChunk.metadata from zerda.chunks chunkers is a ChunkMetadata dataclass, not a dict — while some zerda.vector_database backends (e.g. ChromaVectorDatabase.search()) already unwrap it into a plain dict before returning results. Every place in context that reads chunk metadata as a dict (ScoredChunk.metadata, retriever.MetadataRetriever._matches_filters) routes through this function first, so both shapes — and anything from a fully custom chunk object — are safe to .get()/.items() without an AttributeError.

Not thread-sensitive — pure function, no shared state, no exceptions raised for unexpected input (falls through to {}).

ScoredChunk

@dataclass
class ScoredChunk:
    chunk:              Any
    vector_score:       float  = 0.0
    keyword_score:      float  = 0.0
    metadata_score:     float  = 0.0
    source_quality:     SourceQuality = SourceQuality.UNKNOWN
    recency_score:      float  = 0.0
    composite_score:    float  = 0.0
    retrieval_method:   RetrievalMethod = RetrievalMethod.VECTOR
    retrieval_rank:     int    = 0

The unified wrapper used throughout the pipeline. chunk (your original chunk object — any type) is the only required field.

__post_init__

Computes _content_hash automatically: sha256(" ".join(self.text.lower().split())). This runs every time a ScoredChunk is constructed — there is no way to skip it, and it depends on the .text property (see below), so the wrapped chunk object must already support .text/.page_content access at construction time.

Properties (duck-typed access to the wrapped chunk)

[!WARNING] All of these silently degrade to empty values If your chunk object has none of text/page_content, .text simply returns "". There is no exception raised for a malformed chunk passed to ScoredChunk directly (contrast this with the stricter legacy_compat.validate_chunk, which does raise TypeError, but is only used by the legacy code path, not by ScoredChunk itself). The metadata property specifically degrades to {} for None, or for anything that's neither a dict nor a dataclass — it does not raise even for a genuinely malformed metadata value.

__iter__

def __iter__(self): ...

Yields (self.chunk, self.composite_score) — this lets you do chunk, score = scored_chunk for compatibility with code written against the older tuple-based API.

to_dict()

def to_dict(self) -> Dict[str, Any]: ...

Returns a dict with doc_id, a 120-character text_preview, rounded scores, source_quality.name, and retrieval_method.value.

ContextResult

@dataclass
class ContextResult:
    context:             str
    query_analysis:      QueryAnalysis
    chunks_used:         List[ScoredChunk]
    total_retrieved:     int    = 0
    total_after_filter:  int    = 0
    total_after_rank:    int    = 0
    total_included:      int    = 0
    context_length:      int    = 0
    pipeline_ms:         float  = 0.0
    guard_passed:        bool   = True
    guard_warnings:      List[str] = field(default_factory=list)
    metadata:            Dict[str, Any] = field(default_factory=dict)

The return type of ContextEngine.run(). context, query_analysis, and chunks_used are required.

__post_init__

Overwrites context_length and total_included immediately after construction, regardless of what you passed in for those two fields:

self.context_length = len(self.context)
self.total_included  = len(self.chunks_used)

If you construct a ContextResult manually and pass explicit values for context_length or total_included, they will be silently discarded and recomputed.

get_stats()

def get_stats(self) -> Dict[str, Any]: ...

Returns a flat dict combining pipeline counters with the query's intent, complexity, and budget — used in the Quickstart example.

ConversationTurn

@dataclass
class ConversationTurn:
    turn_id:       int
    query:         str
    answer:        str                = ""
    context_used:  str                = ""
    analysis:      Optional[QueryAnalysis] = None
    timestamp:     float              = field(default_factory=time.time)

A plain record; created exclusively by ConversationHistory.add_turn() — there is no reason to construct this directly in normal usage, since turn_id is assigned automatically based on history length.

[!NOTE] answer is rarely populated by ContextEngine itself ContextEngine.run() calls self._history.add_turn(query=query, context=context, analysis=analysis) — it never passes answer. The answer field stays "" unless you call add_turn() yourself after generating an LLM response, or set turn.answer directly. This matters for get_recent_context() below, which only includes a turn if both query and answer are non-empty.

ConversationHistory

@dataclass
class ConversationHistory:
    session_id:  str
    turns:       List[ConversationTurn] = field(default_factory=list)
    max_turns:   int = 10

add_turn()

def add_turn(
    self,
    query:    str,
    answer:   str        = "",
    context:  str        = "",
    analysis: Optional[QueryAnalysis] = None,
) -> ConversationTurn: ...

Appends a new ConversationTurn (with auto-incremented turn_id = len(self.turns) computed before appending) and trims the list to the most recent max_turns entries. Returns the created turn.

Side effects: mutates self.turns in place.

get_recent_context(n=3)

def get_recent_context(self, n: int = 3) -> str: ...

Returns the last n turns formatted as "Q: ...\nA: ..." blocks joined by blank lines — but only for turns where both query and answer are truthy. Since ContextEngine.run() never sets answer, calling this after only using engine.run() (and never manually setting answer) returns "" for every turn, every time.

get_enriched_query(current_query, n=2)

def get_enriched_query(self, current_query: str, n: int = 2) -> str: ...

If there's no history, or get_recent_context(n) returns an empty string, returns current_query unchanged. Otherwise wraps it as:

[Previous conversation:
{recent_ctx}]

Current question: {current_query}

This is what ContextEngine.run() calls internally to build effective_query — see Architecture → Multi-turn conversation flow. Given the answer-population caveat above, in practice (using only ContextEngine.run()'s default flow) this enrichment is a no-op unless you separately record answers into the history.

Examples

Basic:

from zerda.context import ConversationHistory
hist = ConversationHistory(session_id="demo")
hist.add_turn(query="what is RAG?", answer="Retrieval-Augmented Generation.")
print(hist.get_enriched_query("how does it work?"))

Intermediate — manually recording answers to make enrichment work:

result = engine.run("ما هو RAG؟")
# ... call your LLM with result.context to get an answer ...
answer = "RAG هو..."
engine.get_history().turns[-1].answer = answer  # patch the just-recorded turn

Advanced — resuming a saved session:

from zerda.context import ConversationHistory
saved = ConversationHistory(session_id="user-42", max_turns=20)
engine.set_conversation(saved)

Related

• config · strategy · engine
• retriever.MetadataRetriever — the other consumer of chunk_metadata_dict()

← Back to API overview