File size: 1,492 Bytes
d8bfe4a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 | #!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Load pipeline prompts from pipeline/prompt.jsonl."""
from __future__ import annotations
import json
import os
from functools import lru_cache
from pathlib import Path
from typing import Any, Dict, Mapping
def default_prompt_path() -> Path:
return Path(__file__).resolve().parent / "pipeline" / "prompt.jsonl"
@lru_cache(maxsize=4)
def load_prompts(path: str = "") -> Dict[str, str]:
prompt_path = Path(path or os.environ.get("PIPELINE_PROMPT_JSONL", "") or default_prompt_path())
prompts: Dict[str, str] = {}
if not prompt_path.exists():
return prompts
with prompt_path.open("r", encoding="utf-8") as handle:
for line_no, line in enumerate(handle, 1):
line = line.strip()
if not line:
continue
row = json.loads(line)
key = str(row.get("key", "")).strip()
text = row.get("text")
if not key or not isinstance(text, str):
raise ValueError(f"Invalid prompt row {prompt_path}:{line_no}")
prompts[key] = text
return prompts
def get_prompt(key: str, default: str = "") -> str:
return load_prompts().get(key, default)
def render_prompt(key: str, default: str = "", replacements: Mapping[str, Any] | None = None) -> str:
text = get_prompt(key, default)
for name, value in (replacements or {}).items():
text = text.replace(f"__{name}__", str(value))
return text
|