Building a Prompt Evaluation & Optimization Pipeline with Gemini API — Automated Quality Scoring with LLM-as-Judge

pip install google-genai pandas scipy numpy

import os
from google import genai
 
# Initialize the API client
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
# Models used in the evaluation pipeline
EVAL_MODEL = "gemini-2.5-pro"       # Judge model (evaluator)
TARGET_MODEL = "gemini-2.5-flash"   # Model being evaluated

EVALUATION_RUBRIC = """
You are an expert evaluator of AI assistant outputs.
Score the following response on 4 dimensions, each from 1 to 5.
 
## Evaluation Criteria
 
### 1. Accuracy
- 5: Completely factually accurate. No misinformation
- 4: Mostly accurate. One minor inaccuracy
- 3: Generally accurate but contains several minor inaccuracies
- 2: Contains significant errors
- 1: Mostly inaccurate
 
### 2. Relevance
- 5: Perfectly addresses the user's question
- 4: Mostly addresses the question with minor gaps or extras
- 3: Partially addresses the question
- 2: Mostly off-topic
- 1: Completely irrelevant
 
### 3. Clarity
- 5: Exceptionally clear and well-structured
- 4: Clear with minor room for improvement in structure
- 3: Understandable but some confusing sections
- 2: Difficult to follow
- 1: Nearly incomprehensible
 
### 4. Completeness
- 5: Thoroughly covers all necessary information
- 4: Nearly complete with minor omissions
- 3: Covers key points but has significant gaps
- 2: Mostly incomplete
- 1: Barely any useful information
 
## Output Format
You MUST respond in the following JSON format:
{
  "accuracy": <1-5>,
  "relevance": <1-5>,
  "clarity": <1-5>,
  "completeness": <1-5>,
  "overall": <weighted average of 1-5>,
  "reasoning": "<2-3 sentence justification>"
}
"""

import json
from google.genai import types
 
def evaluate_response(
    user_query: str,
    ai_response: str,
    reference_answer: str | None = None,
) -> dict:
    """
    Use Gemini as a Judge to automatically evaluate AI response quality.
 
    Args:
        user_query: The original user question
        ai_response: The AI output to evaluate
        reference_answer: Optional gold-standard answer for comparison
 
    Returns:
        Dictionary containing scores for each criterion and reasoning
    """
    judge_prompt = f"""{EVALUATION_RUBRIC}
 
## Response to Evaluate
 
**User Query:**
{user_query}
 
**AI Response:**
{ai_response}
"""
    if reference_answer:
        judge_prompt += f"""
**Reference Answer (ideal response):**
{reference_answer}
"""
 
    response = client.models.generate_content(
        model=EVAL_MODEL,
        contents=judge_prompt,
        config=types.GenerateContentConfig(
            temperature=0.0,  # Maximize evaluation consistency
            response_mime_type="application/json",
        ),
    )
 
    try:
        scores = json.loads(response.text)
        # Record token usage for cost tracking
        scores["usage"] = {
            "prompt_tokens": response.usage_metadata.prompt_token_count,
            "output_tokens": response.usage_metadata.candidates_token_count,
        }
        return scores
    except json.JSONDecodeError:
        return {"error": "Failed to parse JSON", "raw": response.text}
 
# Usage example
result = evaluate_response(
    user_query="How do I write a list comprehension in Python?",
    ai_response="A list comprehension is written as [expr for var in iterable]."
)
print(json.dumps(result, indent=2))
# Expected output:
# {
#   "accuracy": 5,
#   "relevance": 5,
#   "clarity": 4,
#   "completeness": 3,
#   "overall": 4.25,
#   "reasoning": "Syntax is correct but lacks examples of conditional
#                 comprehensions and nested expressions."
# }

import random
 
def evaluate_with_debiasing(
    user_query: str,
    response_a: str,
    response_b: str,
    num_rounds: int = 3,
) -> dict:
    """
    Mitigate position bias by randomly swapping response order
    across multiple evaluation rounds and averaging the scores.
    """
    scores_a = []
    scores_b = []
 
    for i in range(num_rounds):
        # Randomly swap presentation order
        if random.random() < 0.5:
            first, second = response_a, response_b
            order = "AB"
        else:
            first, second = response_b, response_a
            order = "BA"
 
        comparison_prompt = f"""
Compare these two AI responses and evaluate each one.
 
**User Query:** {user_query}
 
**Response 1:**
{first}
 
**Response 2:**
{second}
 
Score each response from 1-5 and return JSON:
{{"score_1": <int>, "score_2": <int>, "reasoning": "<str>"}}
"""
        response = client.models.generate_content(
            model=EVAL_MODEL,
            contents=comparison_prompt,
            config=types.GenerateContentConfig(
                temperature=0.0,
                response_mime_type="application/json",
            ),
        )
 
        result = json.loads(response.text)
        if order == "AB":
            scores_a.append(result["score_1"])
            scores_b.append(result["score_2"])
        else:
            scores_a.append(result["score_2"])
            scores_b.append(result["score_1"])
 
    return {
        "response_a_avg": sum(scores_a) / len(scores_a),
        "response_b_avg": sum(scores_b) / len(scores_b),
        "response_a_scores": scores_a,
        "response_b_scores": scores_b,
    }

from dataclasses import dataclass, field
from datetime import datetime
 
@dataclass
class PromptVariant:
    """A prompt variant to test"""
    name: str
    system_instruction: str
    template: str
    model: str = "gemini-2.5-flash"
    temperature: float = 0.7
    max_tokens: int = 2048
 
@dataclass
class TestCase:
    """An evaluation test case"""
    query: str
    reference_answer: str | None = None
    category: str = "general"
    metadata: dict = field(default_factory=dict)
 
@dataclass
class EvalResult:
    """A single evaluation result"""
    variant_name: str
    test_case_query: str
    scores: dict
    latency_ms: float
    token_count: int
    cost_usd: float
    timestamp: str = field(
        default_factory=lambda: datetime.now().isoformat()
    )

import time
import pandas as pd
from scipy import stats
 
# Gemini API pricing (approximate, as of March 2026)
PRICING = {
    "gemini-2.5-pro": {"input": 1.25 / 1_000_000, "output": 10.0 / 1_000_000},
    "gemini-2.5-flash": {"input": 0.15 / 1_000_000, "output": 0.60 / 1_000_000},
}
 
def estimate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
    """Estimate API cost from token usage"""
    pricing = PRICING.get(model, PRICING["gemini-2.5-flash"])
    return (input_tokens * pricing["input"]) + (output_tokens * pricing["output"])
 
def run_ab_test(
    variants: list[PromptVariant],
    test_cases: list[TestCase],
) -> pd.DataFrame:
    """
    Run all prompt variants against all test cases,
    evaluate each response, and return results as a DataFrame.
    """
    results = []
 
    for variant in variants:
        print(f"\n🔄 Testing variant: {variant.name}")
 
        for i, tc in enumerate(test_cases):
            prompt = variant.template.format(query=tc.query)
 
            # Measure latency
            start = time.time()
            response = client.models.generate_content(
                model=variant.model,
                contents=prompt,
                config=types.GenerateContentConfig(
                    system_instruction=variant.system_instruction,
                    temperature=variant.temperature,
                    max_output_tokens=variant.max_tokens,
                ),
            )
            latency_ms = (time.time() - start) * 1000
 
            # Evaluate quality
            scores = evaluate_response(
                user_query=tc.query,
                ai_response=response.text,
                reference_answer=tc.reference_answer,
            )
 
            # Calculate cost
            input_tokens = response.usage_metadata.prompt_token_count
            output_tokens = response.usage_metadata.candidates_token_count
            cost = estimate_cost(variant.model, input_tokens, output_tokens)
 
            results.append(EvalResult(
                variant_name=variant.name,
                test_case_query=tc.query,
                scores=scores,
                latency_ms=latency_ms,
                token_count=input_tokens + output_tokens,
                cost_usd=cost,
            ))
 
            print(f"  [{i+1}/{len(test_cases)}] "
                  f"Score: {scores.get('overall', 'N/A')} | "
                  f"Latency: {latency_ms:.0f}ms | "
                  f"Cost: ${cost:.6f}")
 
    # Convert to DataFrame
    df = pd.DataFrame([
        {
            "variant": r.variant_name,
            "query": r.test_case_query,
            "overall_score": r.scores.get("overall", 0),
            "accuracy": r.scores.get("accuracy", 0),
            "relevance": r.scores.get("relevance", 0),
            "clarity": r.scores.get("clarity", 0),
            "completeness": r.scores.get("completeness", 0),
            "latency_ms": r.latency_ms,
            "tokens": r.token_count,
            "cost_usd": r.cost_usd,
        }
        for r in results
    ])
 
    return df

def analyze_ab_results(df: pd.DataFrame) -> dict:
    """
    Statistically analyze A/B test results using Welch's t-test
    (does not assume equal variance).
    """
    variants = df["variant"].unique()
    if len(variants) < 2:
        return {"error": "Need at least 2 variants for comparison"}
 
    analysis = {}
    baseline = variants[0]
    baseline_scores = df[df["variant"] == baseline]["overall_score"]
 
    for variant in variants[1:]:
        variant_scores = df[df["variant"] == variant]["overall_score"]
 
        # Welch's t-test
        t_stat, p_value = stats.ttest_ind(
            variant_scores, baseline_scores, equal_var=False
        )
 
        # Effect size (Cohen's d)
        pooled_std = (
            (baseline_scores.std() ** 2 + variant_scores.std() ** 2) / 2
        ) ** 0.5
        cohens_d = (
            (variant_scores.mean() - baseline_scores.mean()) / pooled_std
            if pooled_std > 0 else 0
        )
 
        # Cost comparison
        baseline_cost = df[df["variant"] == baseline]["cost_usd"].sum()
        variant_cost = df[df["variant"] == variant]["cost_usd"].sum()
 
        analysis[f"{baseline}_vs_{variant}"] = {
            "baseline_mean": round(baseline_scores.mean(), 3),
            "variant_mean": round(variant_scores.mean(), 3),
            "improvement": round(variant_scores.mean() - baseline_scores.mean(), 3),
            "p_value": round(p_value, 4),
            "significant": p_value < 0.05,
            "cohens_d": round(cohens_d, 3),
            "effect_size": (
                "large" if abs(cohens_d) > 0.8
                else "medium" if abs(cohens_d) > 0.5
                else "small"
            ),
            "cost_ratio": round(variant_cost / baseline_cost, 2) if baseline_cost > 0 else None,
            "recommendation": _get_recommendation(
                p_value, cohens_d, variant_cost / baseline_cost if baseline_cost > 0 else 1
            ),
        }
 
    return analysis
 
def _get_recommendation(p_value: float, cohens_d: float, cost_ratio: float) -> str:
    """Generate an actionable recommendation based on the analysis"""
    if p_value >= 0.05:
        return "No significant difference. Increase sample size or make larger prompt changes."
    if cohens_d > 0.5 and cost_ratio <= 1.2:
        return "✅ Recommended. Significant quality improvement with acceptable cost increase."
    if cohens_d > 0.2 and cost_ratio <= 1.0:
        return "✅ Recommended. Quality improvement with no cost increase."
    if cohens_d > 0.5 and cost_ratio > 1.5:
        return "⚠️ Quality improves but cost increases significantly. Evaluate ROI carefully."
    return "△ Marginal improvement only. Consider alternative approaches."

# Define prompt variants
variant_a = PromptVariant(
    name="baseline",
    system_instruction="You are a helpful AI assistant.",
    template="Please answer the following question.\n\n{query}",
)
 
variant_b = PromptVariant(
    name="structured",
    system_instruction=(
        "You are a technical expert who provides accurate, well-structured answers. "
        "Always state the conclusion first, then provide detailed explanation."
    ),
    template=(
        "Answer the following technical question.\n"
        "Think step by step and include code examples where applicable.\n\n"
        "Question: {query}"
    ),
)
 
# Define test cases
test_cases = [
    TestCase(
        query="How do I make async HTTP requests in Python?",
        category="python",
    ),
    TestCase(
        query="How do I receive streaming responses from Gemini API?",
        category="gemini",
    ),
    TestCase(
        query="What types of authentication are used in REST APIs?",
        category="api",
    ),
    # In production, aim for 30+ test cases
]
 
# Run the A/B test
df = run_ab_test([variant_a, variant_b], test_cases)
 
# Analyze results
analysis = analyze_ab_results(df)
print(json.dumps(analysis, indent=2))
# Expected output:
# {
#   "baseline_vs_structured": {
#     "baseline_mean": 3.667,
#     "variant_mean": 4.333,
#     "improvement": 0.667,
#     "p_value": 0.0312,
#     "significant": true,
#     "cohens_d": 0.89,
#     "effect_size": "large",
#     "cost_ratio": 1.15,
#     "recommendation": "✅ Recommended. Significant quality improvement
#                        with acceptable cost increase."
#   }
# }

def compute_pareto_frontier(df: pd.DataFrame) -> pd.DataFrame:
    """
    Compute the Pareto frontier of quality vs. cost.
    Returns only the Pareto-optimal combinations.
    """
    summary = df.groupby("variant").agg(
        avg_score=("overall_score", "mean"),
        avg_cost=("cost_usd", "mean"),
        avg_latency=("latency_ms", "mean"),
        total_tokens=("tokens", "sum"),
    ).reset_index()
 
    # Extract Pareto-optimal points
    # (higher quality at same cost, or lower cost at same quality)
    pareto = []
    for _, row in summary.iterrows():
        dominated = False
        for _, other in summary.iterrows():
            if (other["avg_score"] >= row["avg_score"] and
                other["avg_cost"] <= row["avg_cost"] and
                (other["avg_score"] > row["avg_score"] or
                 other["avg_cost"] < row["avg_cost"])):
                dominated = True
                break
        if not dominated:
            pareto.append(row)
 
    return pd.DataFrame(pareto)
 
# Example: compare multiple model × prompt combinations
variants_multi = [
    PromptVariant(name="flash-simple", model="gemini-2.5-flash",
                  system_instruction="Be concise.",
                  template="{query}"),
    PromptVariant(name="flash-detailed", model="gemini-2.5-flash",
                  system_instruction="Be thorough and detailed.",
                  template="Please explain in detail.\n{query}"),
    PromptVariant(name="pro-simple", model="gemini-2.5-pro",
                  system_instruction="Be concise.",
                  template="{query}"),
    PromptVariant(name="pro-detailed", model="gemini-2.5-pro",
                  system_instruction="Be thorough and detailed.",
                  template="Please explain in detail.\n{query}"),
]

from datetime import datetime, timedelta
 
class QualityMonitor:
    """
    Continuously monitor prompt quality in production.
    Triggers alerts when the rolling average score drops below a threshold.
    """
 
    def __init__(
        self,
        alert_threshold: float = 3.5,
        window_size: int = 50,
        min_samples: int = 10,
    ):
        self.alert_threshold = alert_threshold
        self.window_size = window_size
        self.min_samples = min_samples
        self.scores: list[dict] = []
 
    def record(self, query: str, response: str, metadata: dict = None):
        """Record a response and run automated evaluation"""
        scores = evaluate_response(user_query=query, ai_response=response)
        entry = {
            "timestamp": datetime.now().isoformat(),
            "query": query,
            "scores": scores,
            "metadata": metadata or {},
        }
        self.scores.append(entry)
        self._check_alerts()
        return scores
 
    def _check_alerts(self):
        """Calculate rolling average and check against threshold"""
        if len(self.scores) < self.min_samples:
            return
 
        recent = self.scores[-self.window_size:]
        avg = sum(
            s["scores"].get("overall", 0) for s in recent
        ) / len(recent)
 
        if avg < self.alert_threshold:
            self._send_alert(avg, len(recent))
 
    def _send_alert(self, avg_score: float, sample_size: int):
        """Send quality degradation alert (connect to Slack, email, etc.)"""
        print(f"⚠️ Quality Alert: Average score over last {sample_size} "
              f"samples dropped to {avg_score:.2f} "
              f"(threshold: {self.alert_threshold})")
 
    def get_daily_report(self) -> dict:
        """Generate a daily quality report"""
        today = datetime.now().date()
        today_scores = [
            s for s in self.scores
            if datetime.fromisoformat(s["timestamp"]).date() == today
        ]
 
        if not today_scores:
            return {"date": str(today), "message": "No data for today"}
 
        overall_scores = [
            s["scores"].get("overall", 0) for s in today_scores
        ]
 
        return {
            "date": str(today),
            "total_evaluations": len(today_scores),
            "average_score": round(sum(overall_scores) / len(overall_scores), 2),
            "min_score": min(overall_scores),
            "max_score": max(overall_scores),
            "below_threshold": sum(
                1 for s in overall_scores if s < self.alert_threshold
            ),
        }
 
# Usage example
monitor = QualityMonitor(alert_threshold=3.5, window_size=50)
 
# Sample a fraction of production requests for evaluation
# (evaluating every request would be too costly — 5-10% is recommended)
import random
SAMPLING_RATE = 0.05  # 5%
 
def handle_request(query: str) -> str:
    """Production request handler with quality sampling"""
    response = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=query,
    )
 
    # Probabilistically sample for quality monitoring
    if random.random() < SAMPLING_RATE:
        monitor.record(query=query, response=response.text)
 
    return response.text