data_construction/prompt_loader.py

#!/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