Beyond Prompt Engineering: Systematic Prompting for Production Agentic Systems

import json
import hashlib
from pathlib import Path
from datetime import datetime
from typing import Optional
from jinja2 import Environment, FileSystemLoader, StrictUndefined
from pydantic import BaseModel
import logging

logger = logging.getLogger(__name__)


class PromptVersion(BaseModel):
    """Metadata for a versioned prompt template."""
    slug: str
    version: str  # semver: "1.2.0"
    description: str
    author: str
    date_modified: str
    template_hash: str  # SHA256 of template content for integrity checking
    variables: list[str]  # required Jinja2 variables
    model_compatibility: list[str]  # e.g., ["gpt-4o", "claude-3-5-sonnet"]


class PromptRegistry:
    """
    Production prompt registry that loads versioned Jinja2 templates from disk,
    validates variable injection, and logs render calls for observability.

    Directory structure:
    prompts/
        customer_support_agent/
            v1.2.0.j2          # Jinja2 template
            v1.2.0.meta.json   # PromptVersion metadata
        contract_analysis/
            v2.0.0.j2
            v2.0.0.meta.json
    """

    def __init__(self, prompts_dir: str = "./prompts"):
        self.prompts_dir = Path(prompts_dir)
        self._cache: dict[str, tuple[PromptVersion, str]] = {}
        self.env = Environment(
            loader=FileSystemLoader(str(self.prompts_dir)),
            undefined=StrictUndefined,  # raises on missing variables — fail fast
            trim_blocks=True,
            lstrip_blocks=True
        )

    def load(self, slug: str, version: Optional[str] = None) -> tuple[PromptVersion, str]:
        """
        Load a prompt template by slug and optional version.
        If version is None, loads the latest version (highest semver).
        Returns (metadata, template_string).
        """
        prompt_dir = self.prompts_dir / slug
        if not prompt_dir.exists():
            raise FileNotFoundError(f"No prompt found with slug '{slug}' at {prompt_dir}")

        if version is None:
            version = self._resolve_latest_version(prompt_dir)

        cache_key = f"{slug}:{version}"
        if cache_key in self._cache:
            return self._cache[cache_key]

        template_path = prompt_dir / f"{version}.j2"
        meta_path = prompt_dir / f"{version}.meta.json"

        if not template_path.exists():
            raise FileNotFoundError(f"Template not found: {template_path}")
        if not meta_path.exists():
            raise FileNotFoundError(f"Metadata not found: {meta_path}")

        template_content = template_path.read_text(encoding="utf-8")
        metadata_raw = json.loads(meta_path.read_text(encoding="utf-8"))
        metadata = PromptVersion(**metadata_raw)

        # Integrity check: verify template content matches recorded hash
        computed_hash = hashlib.sha256(template_content.encode()).hexdigest()
        if computed_hash != metadata.template_hash:
            raise ValueError(
                f"Template integrity check failed for {slug} v{version}. "
                f"Expected hash {metadata.template_hash}, got {computed_hash}. "
                f"Template file may have been modified outside of version control."
            )

        self._cache[cache_key] = (metadata, template_content)
        logger.info(f"Loaded prompt: {slug} v{version} (hash: {computed_hash[:8]}...)")
        return metadata, template_content

    def render(self, slug: str, variables: dict, version: Optional[str] = None) -> str:
        """
        Render a prompt template with the provided variables.
        Raises if any required variable is missing (StrictUndefined).
        Logs render call with variable keys (not values) for observability.
        """
        metadata, _ = self.load(slug, version)

        # Validate all required variables are present
        missing = [v for v in metadata.variables if v not in variables]
        if missing:
            raise ValueError(
                f"Missing required variables for prompt '{slug}' v{metadata.version}: {missing}"
            )

        template = self.env.get_template(f"{slug}/{metadata.version}.j2")

        try:
            rendered = template.render(**variables)
        except Exception as e:
            logger.error(f"Template render failed for {slug} v{metadata.version}: {e}")
            raise

        logger.info(
            f"Rendered prompt: {slug} v{metadata.version} | "
            f"variables: {list(variables.keys())} | "
            f"output_chars: {len(rendered)}"
        )
        return rendered

    def _resolve_latest_version(self, prompt_dir: Path) -> str:
        """Find the highest semver among available template files."""
        from packaging.version import Version  # pip install packaging
        versions = [
            f.stem for f in prompt_dir.glob("*.j2")
            if not f.stem.startswith(".")
        ]
        if not versions:
            raise FileNotFoundError(f"No template versions found in {prompt_dir}")
        return str(max(versions, key=lambda v: Version(v)))

    def list_versions(self, slug: str) -> list[str]:
        """List all available versions for a prompt slug."""
        prompt_dir = self.prompts_dir / slug
        if not prompt_dir.exists():
            return []
        return [f.stem for f in sorted(prompt_dir.glob("*.j2"))]


# --- Usage Example ---
if __name__ == "__main__":
    registry = PromptRegistry("./prompts")

    # Render the latest version of the contract analysis prompt
    rendered_prompt = registry.render(
        slug="contract_analysis",
        variables={
            "contract_text": "This Agreement is entered into as of January 1, 2025...",
            "analysis_focus": "liability clauses and indemnification terms",
            "output_format": "structured JSON with risk scores"
        }
    )
    print(rendered_prompt[:200] + "...")

    # Pin to a specific version for A/B testing
    rendered_v1 = registry.render(
        slug="contract_analysis",
        variables={"contract_text": "...", "analysis_focus": "...", "output_format": "..."},
        version="1.0.0"
    )
    rendered_v2 = registry.render(
        slug="contract_analysis",
        variables={"contract_text": "...", "analysis_focus": "...", "output_format": "..."},
        version="2.0.0"
    )