🎯 Prompting Strategies Benchmark#

Prompting Strategies Benchmark runs a controlled head-to-head experiment comparing three foundational prompting techniques — Direct Prompting, Zero-Shot Chain-of-Thought (CoT), and Few-Shot Chain-of-Thought — on the exact same multi-step arithmetic word problem (buying 7 pens at ₹15, returning 2 for a ₹12 refund each, buying 4 more at ₹18, and paying with a ₹500 note, where the ground-truth change is 293).

All three strategies run against OpenAI at temperature=0. For each strategy, the service extracts the final Answer: <value> line using a regex parser, grades it against expected_answer="293", and records both prompt_tokens and completion_tokens so you can see the exact accuracy vs. token-cost trade-off.

1. Why Autoregressive Models Need a Scratchpad for Multi-Step Math

An LLM generates text one token at a time, performing a fixed amount of computation (one forward pass through its transformer layers) per emitted token. Under Direct Prompting ("Return ONLY the final answer... Do NOT include any reasoning"), the model is forced to compute (7 × 15) - (2 × 12) + (4 × 18) and subtract from 500 inside a single token jump — where intermediate state errors frequently occur.

2. Zero-Shot Chain-of-Thought ("Think step-by-step")

Adding the instruction "Think step-by-step before giving the final answer" (Kojima et al., 2022) turns the output stream itself into an external working memory scratchpad. Because each subsequent token attends to all previously written intermediate totals, a 4-step problem is decomposed into 4 simple 1-step calculations.

3. Few-Shot Chain-of-Thought (In-Context Exemplars)

Few-Shot CoT (Wei et al., 2022) prepends two worked question-reasoning-answer examples before the target question. Beyond encouraging step-by-step reasoning, in-context exemplars calibrate how the model structures its equations and guarantee that it ends with the exact machine-parseable line Answer: 293.

4. The Token Economics of Reasoning

Reasoning is not free: Direct uses ~5 completion tokens, Zero-Shot CoT uses ~80–150 completion tokens (higher output latency and cost), and Few-Shot CoT increases prompt_tokens on every call. Tracking prompt_tokens, completion_tokens, and total_tokens per strategy quantifies the exact price of higher accuracy.

1. Core Benchmark Architecture (src/python/01_prompting_strategies.py)

• 01_prompting_strategies.py implements a controlled head-to-head prompting benchmark that evaluates three distinct reasoning strategies — Direct Prompting, Zero-Shot Chain-of-Thought (CoT), and Few-Shot Chain-of-Thought — on the same multi-step arithmetic word problem, measuring answer accuracy, prompt tokens, completion tokens, and inference latency.

2. Pydantic Data Schemas & In-Context Exemplars

• PromptingStrategiesRequest(BaseModel): Defines the benchmark parameters with defaults for question: str (the multi-step pen purchase word problem), expected_answer: str = "293", model: str = "gpt-4o-mini", and temperature: float = 0.7.
• StrategyResult(BaseModel): Captures the evaluation metrics for a single prompting strategy: name: str, response_text: str, extracted_answer: str, is_correct: bool, prompt_tokens: int, completion_tokens: int, total_tokens: int, and latency_seconds: float.
• PromptingStrategiesResponse(BaseModel): Wraps the comparative benchmark output containing expected_answer: str, model: str, temperature: float, and strategies: List[StrategyResult].
• FEW_SHOT_EXAMPLES: A multi-line prompt constant containing two worked arithmetic demonstrations (Example 1 on counting apples and Example 2 on worker widget output). Each exemplar pairs a Question: with numbered Thought: steps and a final Answer: <number> delimiter to calibrate the model's step-by-step decomposition and output format.

3. Provider Resolution, Answer Extraction & Accuracy Scoring

• _resolve_client_and_model(model_choice: str) -> tuple[OpenAI, str]: Resolves the requested model family (Google Gemini, Groq, or OpenAI gpt-4o / gpt-4o-mini) into an initialized OpenAI-compatible client and concrete model identifier so the three prompting strategies can be compared across different model families.
• extract_final_answer(response_text: str) -> str: Isolates the final numeric conclusion from a free-text or step-by-step reasoning trace:
  • Step ①: Searches for all case-insensitive occurrences of answer\s*: using re.finditer(). If found, it slices the text after the last Answer: delimiter (matches[-1].end()) so intermediate reasoning steps are ignored; otherwise, it falls back to the final non-empty line.
  • Step ②: Takes the first line of that segment, strips markdown formatting characters (* and `), and normalizes case and whitespace.
• evaluate_accuracy(extracted: str, expected: str) -> bool: Grades whether the extracted answer matches the ground-truth value:
  • Step ①: Returns False if either string is empty, strips currency symbols and punctuation via re.sub(r"[^\w\.\:\/]", "", ...), and checks whether clean_expected in clean_extracted.

4. Main Benchmark Execution: run_prompting_strategies()

• Accepts question: str, expected_answer: str, model: str, and temperature: float, and returns a dictionary matching PromptingStrategiesResponse:
  • Step ①: Resolves (client, resolved_model) via _resolve_client_and_model(model) and clamps temperature to the valid sampling interval [0.0, 2.0].
  • Step ②: Constructs the three strategy prompts in prompts:
    • "Direct": Appends "Answer the question directly. State your final answer in the format: 'Answer: <your answer>'. Do not include any explanations or working." (forcing single-pass calculation without an external token scratchpad).
    • "Zero-Shot CoT": Appends "Think through the problem step by step to find the solution. After your reasoning, state your final answer on a new line in the format: 'Answer: <your answer>'." (turning generated output tokens into intermediate working memory).
    • "Few-Shot CoT": Prepends FEW_SHOT_EXAMPLES before Question: {question}\nThought:\n to guide both intermediate equation structure and final delimiter formatting.
  • Step ③: Iterates through prompts.items(), times each client.chat.completions.create() call with time.perf_counter(), and reads prompt_tokens and completion_tokens from resp.usage.
  • Step ④: Extracts the final answer via extract_final_answer(text), grades correctness via evaluate_accuracy(extracted, expected_answer), and records the strategy's accuracy, token breakdown (total_tokens = p_tok + c_tok), and latency.
  • Step ⑤: Returns the comparative benchmark dictionary across all three strategies.

5. End-to-End AI Benchmark Flow

1. run_prompting_strategies() receives the target word problem, expected ground-truth answer, model identifier, and sampling temperature.
2. It constructs three prompt variants — Direct (no scratchpad), Zero-Shot CoT (instruction-triggered scratchpad), and Few-Shot CoT (exemplar-guided scratchpad) — holding the question, model, and temperature constant.
3. Each prompt is sent to the LLM while measuring wall-clock generation latency and capturing prompt vs. completion token counts from resp.usage.
4. extract_final_answer() strips away the intermediate reasoning trace to isolate the final Answer: ... clause, and evaluate_accuracy() verifies it against expected_answer, revealing the accuracy vs. token-cost trade-off across the three prompting strategies.

Prompting Strategies Benchmark Service (Direct vs Zero-Shot CoT vs Few-Shot CoT).

"""Prompting Strategies Benchmark Service (Direct vs Zero-Shot CoT vs Few-Shot CoT).

Benchmarks three fundamental prompting strategies:
1. Direct: "Answer the question directly. Do not include working."
2. Zero-Shot CoT: "Think through the problem step by step..."
3. Few-Shot CoT: Provides explicit worked reasoning examples.

Measures accuracy, prompt tokens, completion tokens, and latency trade-offs.
"""

import os
import re
import time
from typing import List
from dotenv import find_dotenv, load_dotenv
from openai import OpenAI
from pydantic import BaseModel

load_dotenv(find_dotenv())


class PromptingStrategiesRequest(BaseModel):
    """Input payload for Prompting Strategies Benchmark."""

    question: str = "A shop sells pens at 23 rupees each. Ravi buys 7 pens, returns 2 of them, then buys 4 more. He pays the final bill with a 500 rupee note. How much change does he get? (Answer with the number only)"
    expected_answer: str = "293"
    model: str = "gpt-4o-mini"
    temperature: float = 0.7


class StrategyResult(BaseModel):
    name: str
    response_text: str
    extracted_answer: str
    is_correct: bool
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int
    latency_seconds: float


class PromptingStrategiesResponse(BaseModel):
    """Output payload from Prompting Strategies Benchmark."""

    expected_answer: str
    model: str = "gpt-4o-mini"
    temperature: float = 0.7
    strategies: List[StrategyResult]


FEW_SHOT_EXAMPLES = """
Example 1:
Question: If I have 3 apples and I give 2 to my friend, but then my friend gives me 1 back, how many apples do I have?
Thought:
1. Start with 3 apples.
2. Give 2 away: 3 - 2 = 1 apple remaining.
3. Friend gives 1 back: 1 + 1 = 2 apples.
Answer: 2

Example 2:
Question: A worker makes 12 widgets an hour. How many widgets can 3 workers make in 4 hours?
Thought:
1. One worker makes 12 widgets in 1 hour.
2. Three workers make 12 * 3 = 36 widgets in 1 hour.
3. In 4 hours, they make 36 * 4 = 144 widgets.
Answer: 144
"""


def _resolve_client_and_model(model_choice: str) -> tuple[OpenAI, str]:
    """Resolve the OpenAI-compatible client and model identifier for the chosen model."""
    choice = (model_choice or "").strip().lower()
    if "gemini" in choice and os.getenv("GEMINI_API_KEY"):
        return (
            OpenAI(
                api_key=os.getenv("GEMINI_API_KEY"),
                base_url=os.getenv("GEMINI_BASE_URL", "https://generativelanguage.googleapis.com/v1beta/openai/"),
            ),
            os.getenv("GEMINI_MODEL", "gemini-3.8-flash"),
        )
    if ("groq" in choice or "oss" in choice or "llama" in choice) and os.getenv("GROQ_API_KEY"):
        return (
            OpenAI(
                api_key=os.getenv("GROQ_API_KEY"),
                base_url=os.getenv("GROQ_BASE_URL", "https://api.groq.com/openai/v1"),
            ),
            os.getenv("GROQ_MODEL", "openai/gpt-oss-20b"),
        )
    resolved_model = model_choice.strip() if choice in ("gpt-4o", "gpt-4o-mini") else os.getenv("OPENAI_MODEL", "gpt-4o-mini")
    return (
        OpenAI(
            api_key=os.getenv("OPENAI_API_KEY"),
            base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),
        ),
        resolved_model,
    )


def extract_final_answer(response_text: str) -> str:
    """Extract final answer clause from output."""
    # ① Look for explicit 'Answer:' delimiter or fallback to the final line
    text = response_text.strip()
    matches = list(re.finditer(r"answer\s*:", text, flags=re.IGNORECASE))
    if matches:
        segment = text[matches[-1].end() :]
    else:
        lines = [ln for ln in text.splitlines() if ln.strip()]
        segment = lines[-1] if lines else ""

    # ② Strip markdown formatting and normalize case
    segment = segment.strip().splitlines()[0] if segment.strip() else ""
    return segment.replace("*", "").replace("`", "").strip().lower()


def evaluate_accuracy(extracted: str, expected: str) -> bool:
    """Evaluate if extracted answer matches expected."""
    # ① Normalize both extracted and expected answers before checking containment
    if not extracted or not expected:
        return False
    clean_extracted = re.sub(r"[^\w\.\:\/]", "", extracted)
    clean_expected = re.sub(r"[^\w\.\:\/]", "", expected.lower())
    return clean_expected in clean_extracted


def run_prompting_strategies(
    question: str = "A shop sells pens at 23 rupees each. Ravi buys 7 pens, returns 2 of them, then buys 4 more. He pays the final bill with a 500 rupee note. How much change does he get? (Answer with the number only)",
    expected_answer: str = "293",
    model: str = "gpt-4o-mini",
    temperature: float = 0.7,
) -> dict:
    """Execute all three prompting strategies and compare metrics."""
    # ① Initialize the LLM client and resolve model and temperature configuration
    client, resolved_model = _resolve_client_and_model(model)
    clamped_temp = max(0.0, min(2.0, float(temperature)))

    # ② Construct Direct, Zero-Shot CoT, and Few-Shot CoT prompts
    prompts = {
        "Direct": f"{question}\n\nAnswer the question directly. State your final answer in the format: 'Answer: <your answer>'. Do not include any explanations or working.",
        "Zero-Shot CoT": f"{question}\n\nThink through the problem step by step to find the solution. After your reasoning, state your final answer on a new line in the format: 'Answer: <your answer>'.",
        "Few-Shot CoT": f"Solve the following reasoning problems step by step.\n\n{FEW_SHOT_EXAMPLES}\nQuestion: {question}\nThought:\n",
    }

    results = []

    # ③ Run each strategy and measure latency and token usage
    for name, p_text in prompts.items():
        start_t = time.perf_counter()
        resp = client.chat.completions.create(
            model=resolved_model,
            messages=[{"role": "user", "content": p_text}],
            temperature=clamped_temp,
        )
        dur = time.perf_counter() - start_t
        text = resp.choices[0].message.content or ""
        p_tok = resp.usage.prompt_tokens if resp.usage else 0
        c_tok = resp.usage.completion_tokens if resp.usage else 0

        # ④ Extract the final answer and score accuracy against ground truth
        extracted = extract_final_answer(text)
        correct = evaluate_accuracy(extracted, expected_answer)

        results.append(
            {
                "name": name,
                "response_text": text,
                "extracted_answer": extracted,
                "is_correct": correct,
                "prompt_tokens": p_tok,
                "completion_tokens": c_tok,
                "total_tokens": p_tok + c_tok,
                "latency_seconds": round(dur, 3),
            }
        )

    # ⑤ Return comparative benchmark metrics across all three strategies
    return {
        "expected_answer": expected_answer,
        "model": resolved_model,
        "temperature": clamped_temp,
        "strategies": results,
    }

Flask Blueprint server and route handler (prompting_strategies_route) in src/python/app.py.

"""Flask server for 06-ai-systems."""

import importlib.util
import os
import sys
from pathlib import Path

from dotenv import find_dotenv, load_dotenv
from flask import Blueprint, Flask, jsonify, request
from flask_cors import CORS
from rate_limiter import check_rate_limit

load_dotenv(find_dotenv())

PATH_PREFIX = os.environ.get("PATH_PREFIX", "")
STATIC_DIR = Path(__file__).resolve().parents[1]
app = Flask(__name__, static_folder=str(STATIC_DIR))
CORS(app)

bp = Blueprint("main", __name__)

_CUR_DIR = Path(__file__).resolve().parent


def _load_service(filename: str, alias: str):
    # ① Resolve the numbered service file path and create its import spec
    file_path = _CUR_DIR / filename
    spec = importlib.util.spec_from_file_location(f"services.{alias}", file_path)
    if spec is None or spec.loader is None:
        raise ImportError(f"Cannot load module from {file_path}")
    # ② Register and execute the module so its functions can be called by routes
    mod = importlib.util.module_from_spec(spec)
    sys.modules[f"services.{alias}"] = mod
    spec.loader.exec_module(mod)
    return mod


mod_prompt = _load_service("01_prompting_strategies.py", "prompting_strategies")
mod_syco = _load_service("02_sycophancy_trap.py", "sycophancy_trap")
mod_refund = _load_service("03_refund_bench.py", "refund_bench")


@bp.before_request
def enforce_rate_limit():
    # ① Only rate-limit mutating POST requests
    if request.method == "POST":
        # ② Check whether this client has exceeded 30 requests per hour
        blocked, msg, retry_after = check_rate_limit(
            request, max_requests=30, window_seconds=3600
        )
        if blocked:
            # ③ Return HTTP 429 with Retry-After header when rate-limited
            resp = jsonify({"error": msg})
            resp.status_code = 429
            resp.headers["Retry-After"] = str(retry_after)
            return resp


@bp.route("/")
def index():
    # ① Read index.html and inject the runtime PATH_PREFIX into data-api-base
    with open(os.path.join(app.static_folder, "index.html"), encoding="utf-8") as f:
        html = f.read()
    html = html.replace('data-api-base=""', f'data-api-base="{PATH_PREFIX}"')
    # ② Return the patched HTML page
    return app.response_class(html, mimetype="text/html")


@bp.route("/css/<path:filename>")
def css(filename):
    return app.send_static_file(os.path.join("css", filename))


@bp.route("/js/<path:filename>")
def js(filename):
    return app.send_static_file(os.path.join("js", filename))


@bp.route("/explanation/<path:filename>")
@bp.route("/info/<path:filename>")
def explanation_pages(filename):
    return app.send_static_file(os.path.join("explanation", filename))


@bp.route("/theory/<path:filename>")
def theory_pages(filename):
    """Serve the theory documentation pages from src/theory."""
    return app.send_static_file(os.path.join("theory", filename))


@bp.route("/api/prompting-strategies", methods=["POST"])
def prompting_strategies_route():
    # ① Parse question, expected_answer, model, and temperature from JSON body
    data = request.get_json(force=True, silent=True) or {}
    question = (data.get("question") or "").strip()
    expected = (data.get("expected_answer") or "").strip()
    model = data.get("model") or "gpt-4o-mini"
    temp = float(data.get("temperature", 0.7))
    if not question:
        return jsonify({"error": "Question cannot be empty"}), 400
    try:
        # ② Run Direct, Zero-Shot CoT, and Few-Shot CoT benchmark in 01_prompting_strategies.py
        res = mod_prompt.run_prompting_strategies(question, expected, model, temp)
        res["result"] = f"Benchmark evaluated across {len(res.get('strategies', []))} prompting strategies."
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/sycophancy-trap", methods=["POST"])
def sycophancy_trap_route():
    # ① Parse case_type, model, and temperature from JSON body
    data = request.get_json(force=True, silent=True) or {}
    case_type = data.get("case_type") or "Math Gaslight"
    model = data.get("model") or "gpt-4o-mini"
    temp = float(data.get("temperature", 0.7))
    try:
        # ② Run the multi-turn sycophancy trap evaluation in 02_sycophancy_trap.py
        res = mod_syco.run_sycophancy_trap(case_type, model, temp)
        res["result"] = f"Final Verdict: {res.get('final_verdict')}"
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/refund-bench", methods=["POST"])
def refund_bench_route():
    # ① Parse complaint_text, judge bench layout, and auto-approval cap from JSON body
    data = request.get_json(force=True, silent=True) or {}
    complaint_text = (data.get("complaint_text") or "").strip()
    bench = data.get("bench") or "mixed3"
    cap = float(data.get("cap", 2000.0))
    if not complaint_text:
        return jsonify({"error": "Complaint text cannot be empty"}), 400
    try:
        # ② Execute the 4-stage multi-judge dispute adjudication pipeline in 03_refund_bench.py
        res = mod_refund.run_refund_bench(complaint_text, bench, cap)
        res["result"] = (
            f"Approved Refund: ₹{res.get('total_approved_refund', 0):.0f} "
            f"| Held for Review: ₹{res.get('total_held_for_review', 0):.0f}"
        )
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


app.register_blueprint(bp, url_prefix=PATH_PREFIX)

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

Review these core concepts and engineering trade-offs when designing prompting strategies for production AI systems:

1. Compute-Per-Token Bottleneck in Direct Prompting: Because autoregressive transformers execute a fixed number of layer operations per emitted token, forcing an immediate answer via Direct Prompting compresses multi-step arithmetic into a single forward pass where intermediate carry/state errors frequently occur.
2. Externalized Working Memory via Zero-Shot CoT: Appending "Think through the problem step by step" converts the output token stream into an explicit scratchpad; each newly generated token attends to all previously written intermediate equations, decomposing an N-step problem into N simple single-step calculations.
3. Structural & Format Calibration via Few-Shot CoT: Prepending worked Question / Thought / Answer exemplars (FEW_SHOT_EXAMPLES) teaches the model both how to decompose the reasoning steps and how to terminate reliably with the exact machine-parseable delimiter Answer: <value>.
4. The Accuracy–Token–Latency Triangle: Reasoning tokens are not free: Direct Prompting minimizes completion tokens (~5–10 tokens) and latency at the expense of accuracy, Zero-Shot CoT increases completion tokens (higher output cost and generation latency), and Few-Shot CoT additionally increases prompt_tokens on every request.
5. Anchor-Based Answer Extraction (extract_final_answer): Scanning for the last occurrence of Answer: via case-insensitive regex and falling back to the final non-empty line prevents intermediate numbers mentioned during step-by-step reasoning from polluting the extracted final answer.
6. Normalized Ground-Truth Grading (evaluate_accuracy): Stripping markdown formatting (*, `) and non-alphanumeric punctuation/currency symbols before comparing clean_expected in clean_extracted prevents false negatives when a model writes ₹293 or 293 rupees instead of bare 293.
7. Multi-Provider OpenAI-Compatible Routing (_resolve_client_and_model): By routing OpenAI, Google Gemini, and Groq through the unified OpenAI SDK interface with provider-specific base_url and API keys, the benchmark compares prompting strategies across distinct model families without changing evaluation code.
8. Defensive Parameter Clamping & Rate Limiting: Clamping temperature to [0.0, 2.0] in the service layer and enforcing a 30-requests-per-hour sliding window in Flask's @bp.before_request hook protect upstream LLM quotas from malformed payloads or automated abuse.