Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-prompt/src/lexigram/ai/prompt/service/service.py: 38%
112 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""PromptService — centralised template management, rendering, and escaping.
3Design decisions
4----------------
5* Templates are loaded **at construction time** (boot phase). Hot-reload is
6 out of scope; the service is read-only at runtime.
7* Templates are addressed by ``(name, version)``. ``version="latest"`` resolves
8 to the most recently registered version for that name.
9* Variable substitution delegates to :class:`~lexigram.ai.prompt.rendering.engine.PromptRenderer`
10 so every :class:`~lexigram.ai.prompt.rendering.engine.RenderFormat` is
11 supported through the service.
12* Provider-specific escaping is applied to variable *values* **before**
13 substitution so that special characters in values do not break the template.
14 Full escaping rules are stubbed — see ``_apply_provider_escaping``.
15* The observability hook (``PromptObserverProtocol``) is called synchronously
16 after every successful render. Exceptions from the observer are caught and
17 logged as warnings so they never break the caller.
18"""
20from __future__ import annotations
22import asyncio
23from typing import Any, Protocol, runtime_checkable
25from lexigram.ai.prompt.constants import DEFAULT_RENDER_FORMAT
26from lexigram.ai.prompt.exceptions import (
27 PromptConfigError,
28 PromptNotFoundError,
29 PromptRenderError,
30)
31from lexigram.ai.prompt.hooks import (
32 PromptInputSanitizedHook,
33 PromptRenderedHook,
34 PromptTemplateResolvedHook,
35)
36from lexigram.ai.prompt.rendering.engine import PromptRenderer, RenderFormat
37from lexigram.ai.prompt.rendering.sanitizer import InputSanitizer
38from lexigram.ai.prompt.service.models import (
39 LLMProvider,
40 PromptRenderRequest,
41 PromptRenderResult,
42 PromptTemplate,
43)
44from lexigram.ai.prompt.service.observer import (
45 NoOpPromptObserver,
46 PromptObserverProtocol,
47)
48from lexigram.contracts.core.hooks import HookRegistryProtocol
49from lexigram.logging import get_logger
51logger = get_logger(__name__)
53# Sentinel used internally when resolving "latest".
54_LATEST = "latest"
57@runtime_checkable
58class PromptServiceProtocol(Protocol):
59 """Structural protocol for PromptService.
61 Inject this type in DI containers so consumers don't depend on the
62 concrete class.
63 """
65 def render(self, request: PromptRenderRequest) -> PromptRenderResult:
66 """Render a named prompt template.
68 Args:
69 request: Lookup key + variable values.
71 Returns:
72 The rendered result.
74 Raises:
75 :class:`~lexigram.ai.prompt.exceptions.PromptNotFoundError`:
76 No template with the given ``(name, version)`` is registered.
77 :class:`~lexigram.ai.prompt.exceptions.PromptRenderError`:
78 A required variable is missing from the request.
79 """
80 ...
82 def get_template(self, name: str, version: str = _LATEST) -> PromptTemplate:
83 """Return the :class:`~lexigram.ai.prompt.service.models.PromptTemplate`
84 for ``(name, version)`` without rendering it."""
85 ...
87 def list_templates(self) -> list[tuple[str, str]]:
88 """Return all registered ``(name, version)`` pairs."""
89 ...
92class PromptService:
93 """Injectable service for loading, versioning, and rendering prompt templates.
95 Instantiate at boot time (via the DI provider) with one or more
96 :class:`~lexigram.ai.prompt.service.loader.PromptLoaderProtocol` sources.
98 Args:
99 templates: Pre-built list of :class:`~lexigram.ai.prompt.service.models.PromptTemplate`
100 objects. Pass an empty list and call :meth:`_register` if
101 you are building the service manually.
102 observer: Optional observability hook. Defaults to
103 :class:`~lexigram.ai.prompt.service.observer.NoOpPromptObserver`.
104 sanitizer: Optional :class:`~lexigram.ai.prompt.rendering.sanitizer.InputSanitizer`
105 applied to resolved variables before substitution. When
106 attached, the ``prompt.input_sanitized`` hook fires after
107 every successful scan.
109 Example::
111 from lexigram.ai.prompt.service import (
112 DictPromptLoader, PromptService, PromptRenderRequest
113 )
115 loader = DictPromptLoader([
116 {
117 "name": "greeting",
118 "version": "v1",
119 "content": "Hello, {name}!",
120 "required_variables": ["name"],
121 "provider": "anthropic",
122 }
123 ])
124 service = PromptService(loader.load())
125 result = service.render(PromptRenderRequest(name="greeting", variables={"name": "Alice"}))
126 print(result.rendered) # Hello, Alice!
127 """
129 def __init__(
130 self,
131 templates: list[PromptTemplate],
132 observer: PromptObserverProtocol | None = None,
133 hook_registry: HookRegistryProtocol | None = None,
134 sanitizer: InputSanitizer | None = None,
135 ) -> None:
136 self._observer: PromptObserverProtocol = observer or NoOpPromptObserver()
137 self._hooks: HookRegistryProtocol | None = hook_registry
138 self._sanitizer = sanitizer
139 self._hook_tasks: set[asyncio.Task[None]] = set()
140 # _store: name → {version → PromptTemplate}
141 self._store: dict[str, dict[str, PromptTemplate]] = {}
142 # _latest: name → most-recently-registered version string
143 self._latest: dict[str, str] = {}
145 for tmpl in templates:
146 self._register(tmpl)
148 logger.debug(
149 "prompt_service_initialised",
150 template_count=sum(len(v) for v in self._store.values()),
151 )
153 def attach_hook_registry(self, hook_registry: HookRegistryProtocol | None) -> None:
154 """Attach (or detach, when ``None``) the framework hook registry.
156 Hook actions are emitted fire-and-forget (see
157 :meth:`~lexigram.ai.prompt.service.service.PromptService._emit_action`).
158 """
159 self._hooks = hook_registry
161 # ------------------------------------------------------------------
162 # Public API
163 # ------------------------------------------------------------------
165 def render(self, request: PromptRenderRequest) -> PromptRenderResult:
166 """Render a named prompt template.
168 Args:
169 request: Lookup key, variables, and optional pinned version.
171 Returns:
172 :class:`~lexigram.ai.prompt.service.models.PromptRenderResult`
174 Raises:
175 :class:`~lexigram.ai.prompt.exceptions.PromptNotFoundError`:
176 No template with the requested ``(name, version)``.
177 :class:`~lexigram.ai.prompt.exceptions.PromptRenderError`:
178 A required variable is missing or substitution fails.
179 """
180 tmpl = self.get_template(request.name, request.version)
181 resolved = self._resolve_variables(tmpl, request.variables)
182 self._sanitize_inputs(resolved)
183 escaped = self._apply_provider_escaping(tmpl.provider, resolved)
184 rendered = self._substitute(tmpl.content, escaped, tmpl.name, tmpl.format)
186 result = PromptRenderResult(
187 name=tmpl.name,
188 version=tmpl.version,
189 rendered=rendered,
190 provider=tmpl.provider,
191 )
193 self._notify_observer(tmpl.name, tmpl.version, resolved, rendered, tmpl.format)
194 return result
196 def get_template(self, name: str, version: str = _LATEST) -> PromptTemplate:
197 """Look up a template by name and optional version.
199 Args:
200 name: Template name.
201 version: Version string or ``"latest"`` (default).
203 Returns:
204 The matching :class:`~lexigram.ai.prompt.service.models.PromptTemplate`.
206 Raises:
207 :class:`~lexigram.ai.prompt.exceptions.PromptNotFoundError`:
208 Name or version is not registered.
209 """
210 versions = self._store.get(name)
211 if versions is None:
212 available = sorted(self._store.keys())
213 raise PromptNotFoundError(
214 f"No prompt template registered under '{name}'. "
215 f"Available names: {available!r}"
216 )
218 resolved_version = self._latest[name] if version == _LATEST else version
219 tmpl = versions.get(resolved_version)
220 if tmpl is None:
221 raise PromptNotFoundError(
222 f"Template '{name}' version '{resolved_version}' not found. "
223 f"Available versions: {sorted(versions.keys())!r}"
224 )
225 self._emit_action(
226 "prompt.template_resolved",
227 PromptTemplateResolvedHook(template_name=name),
228 )
229 return tmpl
231 def list_templates(self) -> list[tuple[str, str]]:
232 """Return all ``(name, version)`` pairs sorted by name then version."""
233 pairs: list[tuple[str, str]] = []
234 for name, versions in sorted(self._store.items()):
235 for version in sorted(versions.keys()):
236 pairs.append((name, version))
237 return pairs
239 # ------------------------------------------------------------------
240 # Internal helpers
241 # ------------------------------------------------------------------
243 def _register(self, tmpl: PromptTemplate) -> None:
244 """Store a template and update the 'latest' pointer."""
245 versions = self._store.setdefault(tmpl.name, {})
246 versions[tmpl.version] = tmpl
247 # Last-registered wins for "latest".
248 self._latest[tmpl.name] = tmpl.version
250 def _resolve_variables(
251 self,
252 tmpl: PromptTemplate,
253 supplied: dict[str, Any],
254 ) -> dict[str, Any]:
255 """Merge supplied variables with optional defaults and enforce required vars.
257 Raises:
258 :class:`~lexigram.ai.prompt.exceptions.PromptRenderError`:
259 A required variable was not supplied.
260 """
261 resolved: dict[str, Any] = dict(tmpl.optional_defaults)
262 resolved.update(supplied)
264 missing = [v for v in tmpl.required_variables if v not in resolved]
265 if missing:
266 raise PromptRenderError(
267 f"Template '{tmpl.name}' v{tmpl.version}: "
268 f"missing required variable(s): {missing!r}. "
269 f"Supplied: {sorted(supplied.keys())!r}"
270 )
272 return resolved
274 def _substitute(
275 self,
276 content: str,
277 variables: dict[str, Any],
278 template_name: str,
279 render_format: RenderFormat = DEFAULT_RENDER_FORMAT,
280 ) -> str:
281 """Perform variable substitution using the template's declared format.
283 Delegates to :class:`~lexigram.ai.prompt.rendering.engine.PromptRenderer`
284 so the service supports every :class:`RenderFormat` value:
285 ``f_string`` (``{variable}``), ``jinja2`` (``{{ variable }}``,
286 loops/filters/conditionals), ``dollar`` (``$variable``), and
287 ``simple`` (literal, no substitution).
289 Raises:
290 :class:`~lexigram.ai.prompt.exceptions.PromptRenderError`:
291 A placeholder in the template has no corresponding variable, or
292 the Jinja2 template contains a syntax/undefined-variable error.
293 :class:`~lexigram.ai.prompt.exceptions.PromptConfigError`:
294 ``render_format`` is not a valid
295 :class:`~lexigram.ai.prompt.rendering.engine.RenderFormat`.
296 """
297 if not isinstance(render_format, RenderFormat):
298 raise PromptConfigError(
299 f"Template '{template_name}': unknown render format {render_format!r}. "
300 f"Valid values: {[f.value for f in RenderFormat]!r}"
301 )
303 renderer = PromptRenderer(render_format)
304 try:
305 return renderer.render(content, variables)
306 except PromptRenderError:
307 raise
308 except Exception as exc:
309 raise PromptRenderError(
310 f"Template '{template_name}': {render_format.value} rendering error — {exc}"
311 ) from exc
313 @staticmethod
314 def _apply_provider_escaping(
315 provider: LLMProvider,
316 variables: dict[str, Any],
317 ) -> dict[str, Any]:
318 """Apply provider-specific escaping to variable *values* before substitution.
320 Status: scaffolded seam — minimal real escaping is in place; full
321 provider-specific rules are a TODO.
323 Anthropic:
324 Values are passed through as-is. A future implementation would
325 wrap values that contain ``{`` / ``}`` or XML-special characters in
326 ``<parameter name="…">…</parameter>`` tags to prevent prompt
327 injection. The seam is here; the full rule set is out of scope.
329 OpenAI / Azure:
330 Literal ``{`` and ``}`` characters inside values are escaped to
331 ``{{`` and ``}}`` to prevent them being interpreted as format
332 placeholders by downstream callers that re-template the output.
334 Generic:
335 No escaping.
336 """
337 if provider in (LLMProvider.GENERIC, LLMProvider.ANTHROPIC):
338 # TODO(LEX-009): Anthropic — wrap values in <parameter> XML tags
339 # when the value contains characters that could break the Claude
340 # system prompt (e.g. angle brackets or unbalanced braces).
341 return variables
343 if provider in (LLMProvider.OPENAI, LLMProvider.AZURE):
344 escaped: dict[str, Any] = {}
345 for key, val in variables.items():
346 if isinstance(val, str):
347 # Escape literal braces so they survive any subsequent
348 # str.format_map calls on the rendered output.
349 escaped[key] = val.replace("{", "{{").replace("}", "}}")
350 else:
351 escaped[key] = val
352 return escaped
354 # Unreachable with current enum, but safe fallback.
355 return variables # pragma: no cover
357 def _sanitize_inputs(self, variables: dict[str, Any]) -> None:
358 """Scan resolved variables for prompt-injection patterns when a sanitizer is attached.
360 Raises:
361 :class:`~lexigram.ai.prompt.exceptions.PromptValidationError`:
362 An injection pattern is detected and the sanitizer is strict.
363 """
364 if self._sanitizer is None:
365 return
366 self._sanitizer.sanitize_all(variables)
367 self._emit_action("prompt.input_sanitized", PromptInputSanitizedHook())
369 def _notify_observer(
370 self,
371 name: str,
372 version: str,
373 variables: dict[str, Any],
374 rendered_output: str,
375 render_format: RenderFormat,
376 ) -> None:
377 """Call the observer and emit the ``prompt.rendered`` hook, swallowing errors."""
378 try:
379 self._observer.on_render(
380 name,
381 version,
382 variables,
383 rendered_output,
384 render_format,
385 )
386 except Exception as exc: # noqa: BLE001
387 logger.warning(
388 "prompt_observer_error",
389 template=name,
390 version=version,
391 error=str(exc),
392 )
393 self._emit_action(
394 "prompt.rendered",
395 PromptRenderedHook(render_format=render_format),
396 )
398 def _emit_action(self, hook_name: str, payload: object) -> None:
399 """Fire a hook action fire-and-forget when a registry and running event loop exist.
401 Action hooks are fire-and-forget by design
402 (:class:`~lexigram.hooks.registry.HookRegistry`). When no registry is
403 attached, or no event loop is running (e.g. sync CLI code), the hook
404 is skipped — the observer remains the always-synchronous
405 observability surface.
406 """
407 if self._hooks is None:
408 return
409 try:
410 asyncio.get_running_loop()
411 except RuntimeError:
412 return
413 task = asyncio.create_task(self._hooks.call_action(hook_name, payload=payload))
414 self._hook_tasks.add(task)
415 task.add_done_callback(self._hook_tasks.discard)
418__all__ = ["PromptService", "PromptServiceProtocol"]