ZerdaZerda

PromptEngine

On this page

The single public entry point for the whole system.

Table of contents

Overview

PromptEngine coordinates PromptBuilder → ContextManager → GuardrailEngine → strategy → PromptOptimizer, and adds cross-cutting concerns the builder doesn't handle itself: input normalisation, type/strategy auto-detection, hash-based caching, metrics aggregation, a bounded trace log, a simple pub/sub hook, and an adaptive-feedback loop for strategy selection.

Python · 2 lines
engine = PromptEngine()
prompt = engine.build(query="...", documents=[...], memory=[...], strategy="multi_hop")

Constructor

Python · 12 lines
PromptEngine(
    context_manager:  Optional[ContextManager]  = None,
    guardrail_engine: Optional[GuardrailEngine] = None,
    extra_guardrails: Optional[List[Guardrail]]  = None,
    enable_cache:       bool  = True,
    cache_ttl_sec:      int   = 300,
    max_cache_size:     int   = 256,
    enable_auto_detect: bool  = True,
    memory_store:  Optional[Any] = None,
    cache_store:   Optional[Any] = None,
    router:        Optional[Any] = None,
) -> None
ParameterTypeDefaultEffect
context_managerContextManagernew ContextManager()Injected into the internal PromptBuilder
guardrail_engineGuardrailEnginenew GuardrailEngine(extra_guardrails=...)If you pass this explicitly, extra_guardrails is ignored (only used to build the default engine)
extra_guardrailsList[Guardrail]NoneAdditional always-considered guardrails, merged in by GuardrailEngine.build()
enable_cacheboolTrueTurns the SHA-256 prompt cache on/off
cache_ttl_secint300Cache entry lifetime in seconds
max_cache_sizeint256LRU-by-timestamp eviction ceiling (oldest entry evicted, not truly LRU-on-access — see Cache)
enable_auto_detectboolTrueEnables _auto_detect inference of prompt_type/strategy/complexity
memory_store, cache_store, routerAnyNoneStored as attributes; not read anywhere else in the provided source (reserved integration slots — see Architecture)

No exceptions are raised by the constructor under normal use; all subsystem construction is deterministic and offline.

build()

Python · 22 lines
def build(
    self,
    query: str,
    documents: Optional[List[Union[Document, Dict, str]]] = None,
    memory:    Optional[List[Union[Message, Dict]]]        = None,
    prompt_type:    Union[PromptType, str]     = PromptType.QA,
    strategy:       Union[PromptStrategy, str] = PromptStrategy.SIMPLE,
    output_format:  Union[OutputFormat, str]   = OutputFormat.TEXT,
    complexity:     Union[QueryComplexity, str] = QueryComplexity.SIMPLE,
    user_profile:   Union[UserProfile, str]     = UserProfile.GENERAL,
    max_context_tokens: int  = 3000,
    max_answer_tokens:  int  = 512,
    output_schema:      Optional[Dict] = None,
    language:           str  = "en",
    enable_guardrails:  bool = True,
    enable_citations:   bool = True,
    enable_uncertainty: bool = True,
    session_id: str = "",
    user_id:    str = "",
    trace_id:   str = "",
    extra:      Optional[Dict] = None,
) -> BuiltPrompt

Why it exists

It is the one call site an application needs: accept loosely-typed inputs (raw dicts/strings for documents and memory), route them through the full pipeline, and return a ready-to-send BuiltPrompt.

Parameters

See the table under PromptRequest — every keyword here maps 1:1 onto a PromptRequest field of the same name and type, except documents/memory, which accept looser input (below).

documents — each item may be:

  • a Document instance (used as-is)
  • a dict — mapped via item.get("content", item.get("text", str(item))), item.get("source", item.get("url", "")), item.get("score", 1.0) (coerced to float), item.get("metadata", {})
  • a plain str — wrapped as Document(content=item)
  • any other type is silently dropped (no error raised)

memory — each item may be:

  • a Message instance (used as-is)
  • a dict — mapped via item.get("role", "user"), item.get("content", "")
  • any other type is silently dropped

Returns

A BuiltPrompt. Never None.

Raises

  • ValueError — if prompt_type, strategy, output_format, complexity, or user_profile is passed as a string that doesn't match any enum value (e.g. strategy="not_real" raises ValueError: 'not_real' is not a valid PromptStrategy). This is standard Enum(value) behavior, not custom validation.
  • No other exceptions are explicitly raised by build() itself. Exceptions from hook callbacks registered via on() are caught and logged, not propagated (see _fire_hook).

Internal workflow

  1. Normalise documents/memory into typed lists (_normalise_documents, _normalise_messages).
  2. Coerce string enum args to their enum types.
  3. If enable_auto_detect, call _auto_detect() to possibly override prompt_type, strategy, complexity.
  4. Assemble a PromptRequest.
  5. If enable_cache, compute a cache key and return a cached BuiltPrompt on hit (_metrics.cache_hits incremented).
  6. On cache miss, delegate to self._builder.build(request) (see PromptBuilder).
  7. Store the result in cache (evicting the oldest entry if max_cache_size reached).
  8. Record metrics (_metrics.record), append a trace entry (_log_trace), and fire the "prompt.built" hook.
  9. Log an INFO line and return the BuiltPrompt.

Complexity

  • Time: dominated by ContextManager.build()'s deduplication pass, which is O(n² · k) in the number of documents n (pairwise Jaccard comparisons over k-shingle sets) — see context_manager.md. Everything else in build() is O(n) in documents/messages.
  • Space: O(n) for normalised documents/messages plus one cache entry per unique (query, doc snippets, type, strategy, format, language) combination, up to max_cache_size.

Thread safety

Not thread-safe. _cache, _metrics, _feedback, _trace_log, and _hooks are plain mutable instance state mutated without locks. Concurrent calls to build() on the same PromptEngine instance from multiple threads can race on cache writes/evictions and on PromptMetrics counters. Use one PromptEngine per thread/process, or add external locking, if you need concurrent access.

Side effects

  • Logging: one logger.info per successful build ([PromptEngine] Built prompt | ...), plus logger.debug on cache hits.
  • State mutation: _cache, _metrics, _trace_log are all mutated on every non-cached call.
  • No file, network, or database I/O anywhere in this method.

Examples

1. Basic

Python · 2 lines
engine = PromptEngine()
prompt = engine.build(query="What is the capital of France?")

2. Intermediate — with documents and a strategy

Python · 6 lines
prompt = engine.build(
    query="Summarize our Q3 refund policy changes.",
    documents=[{"content": "...", "source": "policy.pdf", "score": 0.95}],
    strategy="cot",
    output_format="bullet_list",
)

3. Advanced — bilingual, structured output, custom schema

Python · 8 lines
prompt = engine.build(
    query="ما هي أنواع التأشيرات المتاحة؟",
    documents=visa_docs,
    output_format="structured",
    output_schema={"type": "object", "properties": {"visa_types": {"type": "array"}}},
    language="ar",
    user_profile="technical",
)

4. Error handling — invalid strategy string

Python · 4 lines
try:
    engine.build(query="test", strategy="not_a_real_strategy")
except ValueError as e:
    print(f"Invalid strategy: {e}")

5. Production usage — with tracing and feedback

Python · 10 lines
prompt = engine.build(
    query=user_query,
    documents=retrieved_docs,
    session_id=session.id,
    user_id=user.id,
    trace_id=str(uuid.uuid4()),
)
response = llm.chat(prompt.to_messages())
# later, once you've scored the response quality:
engine.record_feedback(trace_id=prompt.trace_id, quality_score=0.9)

Best practices

  • Reuse a single PromptEngine instance across requests to benefit from caching and metrics aggregation — construction itself is cheap, but the cache/metrics value compounds across calls.
  • Pass a trace_id explicitly if you plan to call record_feedback() later; otherwise you must recover it from prompt.trace_id, which is only populated if you passed one in (it is not auto-generated — see record_feedback).
  • Set enable_cache=False for tests/benchmarks where you need every call to actually execute the pipeline.

Common mistakes

  • Assuming trace_id is auto-generated. It defaults to "" if not supplied, and record_feedback("") will simply fail to find a matching trace (falls back to "unknown" for prompt_type/strategy in the resulting FeedbackEntry).
  • Mutating a returned BuiltPrompt object and expecting it not to affect the cache: cached entries are the same object reference, so mutating a cached BuiltPrompt in place will corrupt future cache hits within the TTL window.

build_from_request()

Python
def build_from_request(self, request: PromptRequest) -> BuiltPrompt

Convenience wrapper that unpacks every field of an existing PromptRequest and calls build() with them as keyword arguments. Useful when you've already constructed/validated a PromptRequest elsewhere (e.g. deserialized from storage) and don't want to re-specify each field.

Python · 2 lines
req = PromptRequest(query="What is RAG?", strategy=PromptStrategy.CHAIN_OF_THOUGHT)
prompt = engine.build_from_request(req)

Adaptive feedback

record_feedback()

Python
def record_feedback(self, trace_id: str, quality_score: float, notes: str = "") -> None

Looks up trace_id in the internal trace log (next((t for t in self._trace_log if t["trace_id"] == trace_id), None)), builds a FeedbackEntry (using "unknown" for prompt_type/strategy if no matching trace is found), and appends it to self._feedback. The feedback list is capped at 1000 entries (oldest popped via .pop(0) — an O(n) list-shift on every overflow).

Parameters

NameTypeNotes
trace_idstrMust match a trace_id previously logged by build(); if it doesn't, prompt_type/strategy in the recorded feedback are "unknown" rather than an error
quality_scorefloatDocumented as 0.0 (bad) to 1.0 (perfect); not validated or clamped — out-of-range values are stored as-is
notesstrFree text, default ""

Side effects: mutates self._feedback; logs one INFO line.

adaptive_strategy_for()

Python
def adaptive_strategy_for(self, prompt_type: PromptType) -> Optional[PromptStrategy]

Returns the historically best-performing PromptStrategy for a given prompt_type, based on mean quality_score across recorded feedback — but only if at least 5 matching feedback entries exist; otherwise returns None.

Python · 3 lines
best = engine.adaptive_strategy_for(PromptType.QA)
if best:
    prompt = engine.build(query=q, strategy=best)

Observability

metrics (property)

Python · 2 lines
@property
def metrics(self) -> Dict[str, Any]

Returns PromptMetrics.to_dict() — see PromptMetrics below.

get_trace_log()

Python
def get_trace_log(self, last_n: int = 20) -> List[Dict[str, Any]]

Returns the last last_n entries from the internal trace log (a plain Python slice self._trace_log[-last_n:]). The log itself is capped at 500 entries (oldest popped on overflow).

reset_metrics()

Python
def reset_metrics(self) -> None

Replaces self._metrics with a fresh PromptMetrics(). Does not clear the trace log or feedback history — only the aggregate counters.

on()

Python
def on(self, event: str, callback: Callable) -> None

Registers callback to be invoked on event. The only event fired anywhere in the provided source is "prompt.built", invoked as callback(prompt=<BuiltPrompt>, request=<PromptRequest>) after every successful (non-cached-hit... actually including after cache population) build() call.

Exceptions raised inside a callback are caught and logged as a warning ([PromptEngine] Hook '%s' error: %s) — they never propagate to the caller of build().

Python · 4 lines
def log_prompt(prompt, request):
    print(f"Built {prompt.strategy.value} prompt, {prompt.estimated_tokens} tokens")

engine.on("prompt.built", log_prompt)

Internals

These are private (_-prefixed) but documented here since they determine observable behavior.

_auto_detect (internal)

Python
def _auto_detect(self, query, documents, ptype, strat, cplx)

Only runs when enable_auto_detect=True (constructor default). Logic, in order:

  1. Prompt-type inference — only overrides ptype if it's still PromptType.QA (i.e. explicit non-QA types are never overridden). Lowercases the query and checks substrings:
    • "summarize", "summarise", "summary", "overview" → SUMMARIZATION
    • "compare", "contrast", "difference", "versus", "vs" → COMPARISON
    • "extract", "list all", "find all", "enumerate" → EXTRACTION
    • "why", "how", "explain", "reason", "cause" → REASONING
    • Checked in this order — the first match wins.
  2. Complexity inference — based on len(query.split()) and len(documents): > 40 words or > 5 docs → COMPLEX; > 20 words or > 2 docs → MODERATE; otherwise the passed-in cplx is kept.
  3. Strategy inference — if the (lowercased) query contains "and", "also", "additionally", "furthermore", or "relationship between" and more than 1 document was supplied, strat is forced to MULTI_HOP (this can override an explicitly-passed strategy).

[!WARNING] Auto-detection can silently override an explicit strategy= Step 3 overrides strat regardless of whether the caller explicitly passed a non-default strategy. If your query happens to contain the word "and" and you supplied more than one document, MULTI_HOP will be used even if you asked for strategy="cot". Set enable_auto_detect=False if you need strict control.

Input normalisation

_normalise_documents and _normalise_messages — see build() parameters above; both are @staticmethod.

Cache

  • _cache_key(request) — sha256 of a json.dumps(..., sort_keys=True) payload containing query, the first 100 characters of each document's content (doc_hashes), prompt_type, strategy, output_format, and language. Note: documents' score/source/metadata and memory are not part of the cache key — two requests differing only in document metadata or conversation history but identical otherwise will collide and return a stale cached prompt.
  • _cache_get(request) — returns None on miss or on TTL expiry (deleting the expired entry as a side effect).
  • _cache_set(request, prompt) — if at max_cache_size, evicts the globally-oldest entry (by insertion timestamp, via min(self._cache, key=lambda k: self._cache[k][1])) before inserting.

Trace log

_log_trace(request, prompt, elapsed_ms) appends a dict with trace_id, session_id, an 80-character query_preview, prompt_type, strategy, output_format, doc counts, token counts, guardrails, elapsed_ms, and a Unix ts. Capped at 500 entries.

PromptMetrics

Python · 9 lines
@dataclass
class PromptMetrics:
    total_builds:       int = 0
    total_tokens:       int = 0
    total_tokens_saved: int = 0
    cache_hits:         int = 0
    builds_by_type:     Dict[str, int]
    builds_by_strategy: Dict[str, int]
    avg_build_ms:       float = 0.0

record(prompt, elapsed_ms)

Increments all counters, updates builds_by_type[prompt.prompt_type.value] and builds_by_strategy[prompt.strategy.value], and recomputes the running average build time.

to_dict()

Returns a plain dict including a derived cache_hit_rate_pct (cache_hits / total_builds * 100, rounded to 1 decimal; 0 if total_builds == 0 — no division-by-zero error).

Python · 10 lines
{
    "total_builds": 12,
    "total_tokens": 3480,
    "total_tokens_saved": 96,
    "cache_hits": 3,
    "avg_build_ms": 4.21,
    "cache_hit_rate_pct": 25.0,
    "builds_by_type": {"qa": 9, "summarization": 3},
    "builds_by_strategy": {"simple": 9, "cot": 3},
}

FeedbackEntry

Python · 7 lines
@dataclass
class FeedbackEntry:
    trace_id:      str
    prompt_type:   str
    strategy:      str
    quality_score: float
    notes:         str = ""

Plain data container appended to PromptEngine._feedback by record_feedback().

Thread safety

See the build() section above — applies to the whole class. No locks are used anywhere in prompt_engine.py.

Best practices

  • One engine per logical "tenant"/config if you need isolated caches and metrics (e.g. per API key), since PromptEngine holds all state per-instance.
  • Call reset_metrics() on a regular cadence (e.g. per reporting window) rather than creating new engines, to preserve the cache and feedback history.

Common mistakes

  • Expecting on("prompt.built", ...) hooks to fire on cache hits with fully-updated metrics — cache hits still increment cache_hits and return early before the hook-firing block, so "prompt.built" is not fired on a cache hit. Only cache misses fire the hook.

Related pages