Prompt Engineering Best Practices for Production LLMs

SYSTEM_PROMPT = """You are a medical data extraction assistant.
 
Extract the following information from the clinical note and return ONLY valid JSON.
 
Required fields:
- patient_age: integer or null
- chief_complaint: string
- medications: array of strings
- allergies: array of strings
 
Example output:
{
  "patient_age": 45,
  "chief_complaint": "chest pain",
  "medications": ["aspirin", "lisinopril"],
  "allergies": ["penicillin"]
}
"""

from pydantic import BaseModel, validator
from typing import Optional
 
class ClinicalExtraction(BaseModel):
    patient_age: Optional[int]
    chief_complaint: str
    medications: list[str]
    allergies: list[str]
 
    @validator('patient_age')
    def age_must_be_reasonable(cls, v):
        if v is not None and (v < 0 or v > 150):
            raise ValueError('Age must be between 0 and 150')
        return v
 
def extract_clinical_data(note: str) -> ClinicalExtraction:
    response = llm.invoke(SYSTEM_PROMPT, note)
    return ClinicalExtraction.model_validate_json(response)

from sentence_transformers import SentenceTransformer
import numpy as np
 
class FewShotSelector:
    def __init__(self, examples: list[dict]):
        self.examples = examples
        self.encoder = SentenceTransformer('all-MiniLM-L6-v2')
        self.embeddings = self.encoder.encode([e['input'] for e in examples])
 
    def select_examples(self, query: str, k: int = 3) -> list[dict]:
        """Select the k most similar examples to the query."""
        query_embedding = self.encoder.encode(query)
        similarities = np.dot(self.embeddings, query_embedding)
        top_indices = np.argsort(similarities)[-k:][::-1]
        return [self.examples[i] for i in top_indices]

REASONING_PROMPT = """
Analyze the following customer support ticket and determine the appropriate action.
 
Think through this step by step:
1. First, identify the customer's core issue
2. Then, assess the urgency level (low, medium, high, critical)
3. Next, determine the appropriate department
4. Finally, suggest the initial response
 
Ticket: {ticket_text}
 
Let me work through this step by step:
"""

class TicketAnalysis(BaseModel):
    reasoning_steps: list[str]
    core_issue: str
    urgency: Literal["low", "medium", "high", "critical"]
    department: str
    suggested_response: str
 
COT_PROMPT = """
Analyze the ticket and return your analysis as JSON.
 
{
  "reasoning_steps": ["Step 1: ...", "Step 2: ...", ...],
  "core_issue": "...",
  "urgency": "low|medium|high|critical",
  "department": "...",
  "suggested_response": "..."
}
"""

# prompts/clinical_extraction_v2.yaml
name: clinical_extraction
version: "2.1.0"
description: "Extract structured data from clinical notes"
model: gpt-4-turbo
temperature: 0
max_tokens: 1000
 
system_prompt: |
  You are a medical data extraction assistant specializing in...
 
user_template: |
  Extract information from the following clinical note:
 
  {note_content}
 
  Return only valid JSON matching the schema.
 
schema:
  type: object
  required: [patient_age, chief_complaint]
  properties:
    patient_age:
      type: integer
      minimum: 0
      maximum: 150

class PromptTestSuite:
    def __init__(self, prompt_template: str):
        self.prompt = prompt_template
        self.test_cases = []
 
    def add_test(self, input_data: dict, expected: dict, tags: list[str] = []):
        self.test_cases.append({
            "input": input_data,
            "expected": expected,
            "tags": tags
        })
 
    def run_evaluation(self) -> dict:
        results = []
        for case in self.test_cases:
            response = self.invoke(case["input"])
            score = self.evaluate(response, case["expected"])
            results.append({"case": case, "response": response, "score": score})
 
        return {
            "total": len(results),
            "passed": sum(1 for r in results if r["score"] >= 0.9),
            "average_score": sum(r["score"] for r in results) / len(results),
            "by_tag": self._aggregate_by_tag(results)
        }

class PromptABTest:
    def __init__(self, prompt_a: str, prompt_b: str, split: float = 0.5):
        self.prompts = {"A": prompt_a, "B": prompt_b}
        self.split = split
        self.results = {"A": [], "B": []}
 
    def get_prompt(self, request_id: str) -> tuple[str, str]:
        """Deterministic assignment based on request ID."""
        variant = "A" if hash(request_id) % 100 < self.split * 100 else "B"
        return variant, self.prompts[variant]
 
    def record_result(self, variant: str, success: bool, latency: float):
        self.results[variant].append({"success": success, "latency": latency})

def validate_input(text: str, max_tokens: int = 4000) -> str:
    """Sanitize and validate input before sending to LLM."""
    # Remove potential injection attempts
    text = re.sub(r'<\|.*?\|>', '', text)
 
    # Truncate if too long
    tokens = tokenizer.encode(text)
    if len(tokens) > max_tokens:
        text = tokenizer.decode(tokens[:max_tokens])
 
    return text

class OutputGuardrails:
    def __init__(self):
        self.sensitive_patterns = [
            r'\b\d{3}-\d{2}-\d{4}\b',  # SSN
            r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',  # Email
        ]
 
    def check(self, output: str) -> tuple[bool, list[str]]:
        issues = []
        for pattern in self.sensitive_patterns:
            if re.search(pattern, output):
                issues.append(f"Sensitive data pattern detected: {pattern}")
        return len(issues) == 0, issues

from functools import lru_cache
import hashlib
 
class PromptCache:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.ttl = 3600  # 1 hour
 
    def get_cache_key(self, prompt: str, params: dict) -> str:
        content = f"{prompt}:{json.dumps(params, sort_keys=True)}"
        return hashlib.sha256(content.encode()).hexdigest()
 
    async def get_or_compute(self, prompt: str, params: dict, compute_fn):
        key = self.get_cache_key(prompt, params)
        cached = await self.redis.get(key)
        if cached:
            return json.loads(cached)
 
        result = await compute_fn(prompt, params)
        await self.redis.setex(key, self.ttl, json.dumps(result))
        return result

def optimize_prompt(prompt: str, target_reduction: float = 0.2) -> str:
    """Reduce prompt token count while preserving meaning."""
    # Remove redundant whitespace
    prompt = re.sub(r'\s+', ' ', prompt)
 
    # Use abbreviations for common terms
    replacements = {
        "for example": "e.g.",
        "that is": "i.e.",
        "and so on": "etc.",
    }
    for full, abbrev in replacements.items():
        prompt = prompt.replace(full, abbrev)
 
    return prompt

class PromptMetrics:
    def record(self, prompt_id: str, response: dict, metadata: dict):
        metrics = {
            "prompt_id": prompt_id,
            "timestamp": datetime.utcnow(),
            "latency_ms": metadata["latency_ms"],
            "input_tokens": metadata["input_tokens"],
            "output_tokens": metadata["output_tokens"],
            "cost_usd": self.calculate_cost(metadata),
            "success": metadata.get("success", True),
            "model": metadata["model"]
        }
        self.emit(metrics)