Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-prompt/src/lexigram/ai/prompt/service/loader.py: 26%
86 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""Template loaders for PromptService.
3Three loaders are provided:
5- :class:`DictPromptLoader` — load from an in-process Python dict (inline
6 config or app-registered templates).
7- :class:`DirectoryPromptLoader` — load all YAML/JSON files from a directory.
8 Each file may contain one or more templates.
9- The :class:`PromptLoaderProtocol` defines the interface for custom loaders.
11YAML template format
12--------------------
13A single file may contain multiple templates under a top-level ``templates``
14key, or a single template as the top-level document::
16 # multi-template file
17 templates:
18 - name: wellness_check
19 version: v1
20 provider: anthropic
21 required_variables: [pet_name, species]
22 optional_defaults:
23 breed: Unknown
24 content: |
25 Assess {pet_name} ({species}, {breed}).
27 # single-template file (flat)
28 name: greeting
29 version: v1
30 content: Hello, {name}!
32Dict format
33-----------
34Pass a list of dicts, each matching the :class:`~lexigram.ai.prompt.service.models.PromptTemplate`
35fields::
37 loader = DictPromptLoader([
38 {
39 "name": "greeting",
40 "version": "v1",
41 "content": "Hello, {name}!",
42 "required_variables": ["name"],
43 },
44 ])
45"""
47from __future__ import annotations
49from pathlib import Path
50from typing import Any, Protocol, runtime_checkable
52from lexigram.ai.prompt.constants import DEFAULT_RENDER_FORMAT
53from lexigram.ai.prompt.rendering.engine import RenderFormat
54from lexigram.ai.prompt.service.models import LLMProvider, PromptTemplate
55from lexigram.logging import get_logger
56from lexigram.serialization import JSONDecodeError, loads
58logger = get_logger(__name__)
60_SUPPORTED_EXTENSIONS = frozenset({".yaml", ".yml", ".json"})
63@runtime_checkable
64class PromptLoaderProtocol(Protocol):
65 """Interface for objects that supply :class:`~lexigram.ai.prompt.service.models.PromptTemplate` instances."""
67 def load(self) -> list[PromptTemplate]:
68 """Return all templates from this source.
70 Raises:
71 OSError: If a file cannot be read.
72 ValueError: If a template document is malformed.
73 """
74 ...
77def _dict_to_template(
78 data: dict[str, Any], default_format: RenderFormat = DEFAULT_RENDER_FORMAT
79) -> PromptTemplate:
80 """Convert a raw dict (from YAML or inline config) to a :class:`PromptTemplate`.
82 Raises:
83 ValueError: If ``name``, ``version``, or ``content`` are missing,
84 or if ``format`` is not a valid
85 :class:`~lexigram.ai.prompt.rendering.engine.RenderFormat`
86 value.
87 """
88 missing = [k for k in ("name", "version", "content") if not data.get(k)]
89 if missing:
90 raise ValueError(
91 f"PromptTemplate dict is missing required keys: {missing!r}. Got: {list(data.keys())!r}"
92 )
94 raw_provider = data.get("provider", LLMProvider.GENERIC)
95 try:
96 provider = LLMProvider(raw_provider)
97 except ValueError:
98 raise ValueError(
99 f"Unknown provider {raw_provider!r}. "
100 f"Valid values: {[p.value for p in LLMProvider]!r}"
101 )
103 raw_format = data.get("format", default_format)
104 try:
105 render_format = RenderFormat(raw_format)
106 except ValueError:
107 raise ValueError(
108 f"Unknown render format {raw_format!r}. "
109 f"Valid values: {[f.value for f in RenderFormat]!r}"
110 )
112 required_variables = tuple(data.get("required_variables") or [])
113 optional_defaults: dict[str, Any] = dict(data.get("optional_defaults") or {})
115 return PromptTemplate(
116 name=str(data["name"]),
117 version=str(data["version"]),
118 content=str(data["content"]),
119 format=render_format,
120 provider=provider,
121 required_variables=required_variables,
122 optional_defaults=optional_defaults,
123 description=str(data.get("description", "")),
124 metadata=dict(data.get("metadata") or {}),
125 )
128def _parse_file_content(path: Path, text: str) -> list[dict[str, Any]]:
129 """Parse YAML or JSON text into a list of raw template dicts."""
130 if path.suffix.lower() == ".json":
131 data = loads(text)
132 else:
133 try:
134 import yaml
135 except ImportError as exc:
136 raise ImportError(
137 f"PyYAML is required to load YAML prompt templates. "
138 f"Install it: pip install pyyaml. File: {path}"
139 ) from exc
140 data = yaml.safe_load(text)
142 if data is None:
143 return []
145 if isinstance(data, list):
146 return [d for d in data if isinstance(d, dict)]
148 if isinstance(data, dict):
149 if "templates" in data and isinstance(data["templates"], list):
150 return [d for d in data["templates"] if isinstance(d, dict)]
151 # Single flat template document
152 return [data]
154 raise ValueError(f"Unexpected YAML/JSON root type in {path}: {type(data).__name__}")
157class DictPromptLoader:
158 """Load templates from an in-process Python list of dicts.
160 Args:
161 templates: List of raw template dicts (see module docstring for schema).
162 default_format: Format applied to dicts that don't declare ``format``.
163 """
165 def __init__(
166 self,
167 templates: list[dict[str, Any]],
168 default_format: RenderFormat = DEFAULT_RENDER_FORMAT,
169 ) -> None:
170 self._templates = templates
171 self._default_format = default_format
173 def load(self) -> list[PromptTemplate]:
174 """Parse and return all templates.
176 Raises:
177 ValueError: If any dict is malformed.
178 """
179 result: list[PromptTemplate] = []
180 for i, raw in enumerate(self._templates):
181 try:
182 result.append(_dict_to_template(raw, self._default_format))
183 except ValueError as exc:
184 raise ValueError(f"Invalid template at index {i}: {exc}") from exc
185 return result
188class DirectoryPromptLoader:
189 """Load templates from all YAML/JSON files in a directory.
191 Files are processed in sorted order. Files starting with ``.`` or ``_``
192 are skipped. Files that fail to parse emit a warning and are skipped
193 (non-fatal), matching the behaviour of
194 :class:`~lexigram.config.lib.sources.DirectoryConfigSource`.
196 Args:
197 directory: Path to the directory containing template files.
198 default_format: Format applied to files that don't declare ``format``.
199 """
201 def __init__(
202 self,
203 directory: str | Path,
204 default_format: RenderFormat = DEFAULT_RENDER_FORMAT,
205 ) -> None:
206 self._directory = Path(directory)
207 self._default_format = default_format
209 def load(self) -> list[PromptTemplate]:
210 """Scan the directory and return all parsed templates.
212 Raises:
213 OSError: If the directory cannot be read.
214 """
215 if not self._directory.is_dir():
216 logger.warning(
217 "prompt_loader_directory_not_found",
218 directory=str(self._directory),
219 )
220 return []
222 result: list[PromptTemplate] = []
223 for path in sorted(self._directory.iterdir()):
224 if path.suffix.lower() not in _SUPPORTED_EXTENSIONS:
225 continue
226 if path.name.startswith((".", "_")):
227 continue
229 try:
230 text = path.read_text(encoding="utf-8")
231 raw_dicts = _parse_file_content(path, text)
232 for raw in raw_dicts:
233 try:
234 result.append(_dict_to_template(raw, self._default_format))
235 except ValueError as exc:
236 logger.warning(
237 "prompt_loader_template_parse_error",
238 file=str(path),
239 error=str(exc),
240 )
241 except (OSError, JSONDecodeError, ValueError) as exc:
242 logger.warning(
243 "prompt_loader_file_error",
244 file=str(path),
245 error=str(exc),
246 )
248 logger.debug(
249 "prompt_loader_directory_loaded",
250 directory=str(self._directory),
251 count=len(result),
252 )
253 return result
256__all__ = [
257 "DictPromptLoader",
258 "DirectoryPromptLoader",
259 "PromptLoaderProtocol",
260]