Prompt Engineering Production Guide

/**
 * Production prompt registry.
 *
 * Manages versioned prompts as first-class artifacts
 * with evaluation metadata and rollback support.
 */

interface PromptVersion {
  id: string;            // e.g., "product-search-v3.2.1"
  name: string;          // Human-readable name
  version: string;       // Semantic version
  template: string;      // The prompt template with {{variables}}
  model: string;         // Target model (e.g., "claude-sonnet-4-20250514")
  maxTokens: number;
  temperature: number;
  evaluationScore: number;  // Score from automated evaluation
  evaluationDate: string;
  status: "draft" | "testing" | "canary" | "production" | "deprecated";
  changelog: string;
}

class PromptRegistry {
  private prompts: Map<string, PromptVersion[]> = new Map();

  register(prompt: PromptVersion): void {
    const versions = this.prompts.get(prompt.name) || [];
    versions.push(prompt);
    this.prompts.set(prompt.name, versions);
  }

  getProduction(name: string): PromptVersion | undefined {
    const versions = this.prompts.get(name) || [];
    return versions.find((p) => p.status === "production");
  }

  getCanary(name: string): PromptVersion | undefined {
    const versions = this.prompts.get(name) || [];
    return versions.find((p) => p.status === "canary");
  }

  promote(
    name: string, version: string, to: PromptVersion["status"],
  ): void {
    const versions = this.prompts.get(name) || [];
    const target = versions.find((p) => p.version === version);
    if (!target) throw new Error(`Version ${version} not found`);

    // If promoting to production, demote current production
    if (to === "production") {
      const current = this.getProduction(name);
      if (current) current.status = "deprecated";
    }

    target.status = to;
  }

  rollback(name: string): PromptVersion | undefined {
    const versions = this.prompts.get(name) || [];
    // Find the most recent deprecated version
    const previous = versions
      .filter((p) => p.status === "deprecated")
      .sort((a, b) => b.version.localeCompare(a.version))[0];

    if (previous) {
      this.promote(name, previous.version, "production");
    }
    return previous;
  }
}

"""Automated prompt evaluation pipeline.

Runs a prompt variant against a golden dataset
and produces a quality report that gates deployment.
"""

from dataclasses import dataclass
from typing import List, Dict, Callable, Any
import json


@dataclass
class EvalCase:
    """A single evaluation test case."""
    input_vars: Dict[str, str]
    expected_output: str  # Or criteria for judgment
    category: str  # For slice-based analysis


@dataclass
class EvalResult:
    """Result of evaluating one test case."""
    case: EvalCase
    actual_output: str
    score: float  # 0.0 - 1.0
    latency_ms: float
    token_count: int
    notes: str


def evaluate_prompt(
    prompt_template: str,
    eval_cases: List[EvalCase],
    llm_call_fn: Callable,
    scoring_fn: Callable,
    model: str = "claude-sonnet-4-20250514",
) -> Dict[str, Any]:
    """Evaluate a prompt against a golden dataset.

    Args:
        prompt_template: The prompt with {{variable}} placeholders
        eval_cases: Test cases with inputs and expected outputs
        llm_call_fn: Function to call the LLM
        scoring_fn: Function to score output vs expected
        model: Model to evaluate against

    Returns:
        Evaluation report with overall and per-category scores
    """
    results: List[EvalResult] = []

    for case in eval_cases:
        # Render the prompt
        prompt = prompt_template
        for key, value in case.input_vars.items():
            prompt = prompt.replace(f"{{{{{key}}}}}", value)

        # Call the LLM
        import time
        start = time.time()
        output = llm_call_fn(prompt, model=model)
        latency = (time.time() - start) * 1000

        # Score the output
        score = scoring_fn(output, case.expected_output)

        results.append(EvalResult(
            case=case,
            actual_output=output,
            score=score,
            latency_ms=latency,
            token_count=len(output.split()),  # Rough estimate
            notes="",
        ))

    # Aggregate results
    overall_score = sum(r.score for r in results) / len(results)
    by_category: Dict[str, float] = {}
    for cat in set(r.case.category for r in results):
        cat_results = [r for r in results if r.case.category == cat]
        by_category[cat] = sum(
            r.score for r in cat_results
        ) / len(cat_results)

    return {
        "overall_score": round(overall_score, 4),
        "by_category": by_category,
        "total_cases": len(results),
        "pass_rate": sum(
            1 for r in results if r.score >= 0.8
        ) / len(results),
        "avg_latency_ms": sum(
            r.latency_ms for r in results
        ) / len(results),
        "passes_threshold": overall_score >= 0.85,
    }