🧩 CoT vs Direct Benchmark#

Chain-of-Thought vs. Direct Reliability Benchmark demonstrates why step-by-step reasoning is a core reliability engineering lever rather than a stylistic preference. The service benchmarks five categories of state-tracking logic puzzles where single-shot intuition regularly fails:

• Spatial Navigation (Grid): Tracking compass headings and (x, y) coordinates across 5 turn-and-walk instructions.
• Relational Logic (Family Tree): Resolving chained kinship relations with a gender constraint.
• Temporal Scheduling: Finding the earliest overlapping 30-minute meeting slot across three people's busy calendars and time-window rules.
• Inventory State Tracking: Computing net warehouse units after sequential shipments, percentages, returns, and restocks.
• Constraint Satisfaction (Truth-Tellers): Deductions over three labeled boxes where exactly one label is true.

For the selected problem, the service calls the model twice — once with a Direct system prompt ("Answer in one short sentence. Do not show working.") and once with a CoT system prompt ("Think step by step, then state your final answer on the last line.") — scoring only the final line against accepted_answers and recording the wall-clock latency of each call.

1. Why State-Tracking Problems Break Direct Prompting

Every puzzle in PROBLEMS has sequential state transitions: step k depends on the exact intermediate state after step k − 1 (e.g., which direction you are facing after three turns on a grid). When forbidden from writing intermediate tokens, a non-reasoning model must compress all state transitions into a single forward pass before emitting the very first word of its answer.

2. Externalizing Intermediate State in the Context Window

Under the CoT prompt, the model writes out each intermediate state transition explicitly (e.g., "Start at (0,0) facing North → walk 3 to (0,3) → turn right to face East..."). Once written, those tokens become part of the attention context for the next step, turning an error-prone multi-hop deduction into a series of trivial single-hop lookups.

3. Final-Line-Only Scoring (Avoiding False Positives)

Why does extract_final_line() strip markdown formatting and inspect only the last non-empty line of the model's output? During a reasoning trace, a model often mentions candidate options it is ruling out (e.g., "Box A cannot contain gold, and Box B is false, so..."). If the grader searched the entire response text for "box b", it would award a false pass to a trace that mentioned Box B mid-thought and concluded with Box C!

4. The Latency-Reliability Trade-Off

Because inference latency scales linearly with the number of generated output tokens, CoT takes noticeably longer than Direct answering (tracked via latency_seconds). Reliability engineering means paying the CoT token/latency budget on tasks that require multi-step state tracking while keeping Direct prompts for simple classification or lookup tasks.

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

• 01_cot_vs_direct.py benchmarks Chain-of-Thought (CoT) reasoning against Direct answering on multi-step state-tracking logic puzzles, measuring how externalizing intermediate state transitions into the token stream improves accuracy at the cost of additional generation latency.

2. Pydantic Schemas & State-Tracking Puzzle Dataset (PROBLEMS)

• CotVsDirectRequest(BaseModel): Defines the benchmark input with default problem_title: str = "Spatial Navigation (Grid)".
• CotVsDirectResult(BaseModel): Captures the evaluation metrics of a single strategy run: strategy: str, response: str, extracted_answer: str, is_correct: bool, and latency_seconds: float.
• CotVsDirectResponse(BaseModel): Wraps the side-by-side comparison containing problem_title: str, accepted_answers: List[str], direct_result: CotVsDirectResult, and cot_result: CotVsDirectResult.
• PROBLEMS: Defines three multi-step state-tracking logic puzzles, each with a question, list of normalized correct_answers, and ground-truth step-by-step explanation:
  • "Spatial Navigation (Grid)": Tracks (X, Y) coordinates and compass heading from (0,0) facing North across 5 sequential forward/turn/backward instructions, ending at (0, -2).
  • "Relational Logic (Family Tree)": Resolves chained kinship relationships across four generations (Alice → Bob → Charlie → Diana → Eve) to identify Alice as Eve's great-aunt / grand-aunt.
  • "Temporal Scheduling": Solves a 5-speaker ordering constraint satisfaction puzzle (E is 3rd, B immediately follows D, D is not 1st, C precedes A) to derive the sequence C, A, E, D, B.

3. Accuracy Scoring & Benchmark Execution

• evaluate_cot_accuracy(extracted: str, accepted: List[str]) -> bool: Grades whether the model's answer matches any accepted ground-truth string:
  • Step ①: Lowercases extracted and strips whitespace and hyphens (.replace(" ", "").replace("-", "")), then checks whether any similarly normalized string in accepted is contained within cleaned.
• run_cot_vs_direct_benchmark(problem_title: str = "Spatial Navigation (Grid)") -> dict: Executes the head-to-head reliability comparison:
  • Step ①: Initializes the OpenAI client for the configured model (default "gpt-4o-mini") and loads the selected puzzle's question and correct_answers from PROBLEMS.
  • Step ② (Direct Strategy): Appends "Answer in one short sentence. Do not show working." to force a single-pass answer without intermediate reasoning tokens, calls client.chat.completions.create() at temperature=0.3 while timing latency with time.perf_counter(), and grades text_d with evaluate_cot_accuracy(text_d, accepted).
  • Step ③ (Chain-of-Thought Strategy): Appends "Think step by step, then state your final answer clearly on a new line." so the model writes out each intermediate state transition before concluding, calls the LLM at the same temperature=0.3, records latency dur_c, and grades text_c with evaluate_cot_accuracy(text_c, accepted).
  • Step ④: Extracts the final line of the CoT trace (text_c.strip().splitlines()[-1]) as extracted_answer and returns problem_title, accepted_answers, direct_result, and cot_result.

4. End-to-End AI Benchmark Flow

1. run_cot_vs_direct_benchmark() looks up the selected state-tracking puzzle and its accepted ground-truth answers in PROBLEMS.
2. It first runs the Direct Strategy at temperature=0.3, forbidding intermediate working so the model must compress all state transitions into a single forward pass.
3. It then runs the Chain-of-Thought Strategy at the same temperature, allowing the model to externalize each step's intermediate state (coordinates, kinship link, or speaker slot) into the context window before emitting its final line.
4. Both outputs are scored against accepted_answers by evaluate_cot_accuracy() and paired with their wall-clock latencies to quantify the reliability gain of step-by-step reasoning.

Chain-of-Thought vs Direct Answer Reliability Benchmark Service.

"""Chain-of-Thought vs Direct Answer Reliability Benchmark Service.

Demonstrates that Chain-of-Thought (step-by-step reasoning) acts as a reliability lever:
1. Benchmarks challenging multi-step logic problems (Spatial Navigation, Relational Logic, Temporal Scheduling, Inventory State, Truth-Tellers).
2. Runs two prompt strategies:
   - Direct: "Answer in one short sentence. Do not show working."
   - CoT: "Think step by step, then state your final answer."
3. Measures accuracy, latency, and step-by-step cognitive traces.
"""

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 CotVsDirectRequest(BaseModel):
    """Input payload for CoT vs Direct Benchmark."""

    problem_title: str = "Spatial Navigation (Grid)"


class CotVsDirectResult(BaseModel):
    strategy: str
    response: str
    extracted_answer: str
    is_correct: bool
    latency_seconds: float


class CotVsDirectResponse(BaseModel):
    """Output payload from CoT vs Direct Benchmark."""

    problem_title: str
    accepted_answers: List[str]
    direct_result: CotVsDirectResult
    cot_result: CotVsDirectResult


PROBLEMS = {
    "Spatial Navigation (Grid)": {
        "question": (
            "Start at origin (0,0) facing North (+Y direction). "
            "Move forward 3 units. Turn right 90 degrees. Move forward 2 units. "
            "Turn right 90 degrees. Move forward 5 units. "
            "Turn left 90 degrees. Move backward 2 units. "
            "What are your exact final (X,Y) coordinates? Format as (X, Y)."
        ),
        "correct_answers": ["(0, -2)", "(0,-2)"],
        "explanation": "Start(0,0) N -> Fwd 3 to (0,3). Right to E -> Fwd 2 to (2,3). Right to S -> Fwd 5 to (2,-2). Left to E -> Back 2 (moving West) to (0,-2).",
    },
    "Relational Logic (Family Tree)": {
        "question": (
            "Alice is the sister of Bob. Bob is the father of Charlie. "
            "Charlie is the brother of Diana. Diana is the mother of Eve. "
            "What is the exact biological relationship of Alice to Eve?"
        ),
        "correct_answers": ["great-aunt", "great aunt", "grand-aunt", "grand aunt"],
        "explanation": "Bob is Diana's father. Alice is Bob's sister, making her Diana's aunt. Diana is Eve's mother, making Alice Eve's great-aunt.",
    },
    "Temporal Scheduling": {
        "question": (
            "Five speakers (A, B, C, D, E) present one after another. "
            "E must speak exactly third. B must speak immediately after D. "
            "D cannot be the first speaker. C must speak at some point before A. "
            "What is the exact sequence of the 5 speakers from first to last? Format as a comma-separated list."
        ),
        "correct_answers": ["c, a, e, d, b", "c,a,e,d,b", "c a e d b"],
        "explanation": "E is 3rd. D cannot be 1st, so DB is 4-5. C before A fills 1-2. Result: C, A, E, D, B.",
    },
}


def evaluate_cot_accuracy(extracted: str, accepted: List[str]) -> bool:
    """Evaluate whether answer matches any accepted answers."""
    # ① Strip whitespace and hyphens before matching against accepted ground-truth strings
    cleaned = extracted.lower().replace(" ", "").replace("-", "")
    for acc in accepted:
        clean_acc = acc.lower().replace(" ", "").replace("-", "")
        if clean_acc in cleaned:
            return True
    return False


def run_cot_vs_direct_benchmark(problem_title: str = "Spatial Navigation (Grid)") -> dict:
    """Execute CoT vs Direct comparison."""
    # ① Initialize OpenAI client and resolve preset logic puzzle
    client = OpenAI(
        api_key=os.getenv("OPENAI_API_KEY"),
        base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),
    )
    model = os.getenv("OPENAI_MODEL", "gpt-4o-mini")

    preset = PROBLEMS.get(problem_title, PROBLEMS["Spatial Navigation (Grid)"])
    q_text = preset["question"]
    accepted = preset["correct_answers"]

    # ② Run Direct Strategy (no intermediate working allowed)
    start_d = time.perf_counter()
    direct_prompt = f"{q_text}\n\nAnswer in one short sentence. Do not show working."
    resp_d = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": direct_prompt}],
        temperature=0.3,
    )
    dur_d = time.perf_counter() - start_d
    text_d = resp_d.choices[0].message.content or ""
    correct_d = evaluate_cot_accuracy(text_d, accepted)

    # ③ Run Chain-of-Thought Strategy (step-by-step reasoning before final answer)
    start_c = time.perf_counter()
    cot_prompt = f"{q_text}\n\nThink step by step, then state your final answer clearly on a new line."
    resp_c = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": cot_prompt}],
        temperature=0.3,
    )
    dur_c = time.perf_counter() - start_c
    text_c = resp_c.choices[0].message.content or ""
    correct_c = evaluate_cot_accuracy(text_c, accepted)

    # ④ Return side-by-side accuracy, latency, and reasoning traces
    return {
        "problem_title": problem_title,
        "accepted_answers": accepted,
        "direct_result": {
            "strategy": "Direct (No Working)",
            "response": text_d,
            "extracted_answer": text_d.strip(),
            "is_correct": correct_d,
            "latency_seconds": round(dur_d, 3),
        },
        "cot_result": {
            "strategy": "Chain-of-Thought (Step-by-Step)",
            "response": text_c,
            "extracted_answer": text_c.strip().splitlines()[-1],
            "is_correct": correct_c,
            "latency_seconds": round(dur_c, 3),
        },
    }

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

"""Flask server for 07-ai-reliability."""

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_cot = _load_service("01_cot_vs_direct.py", "cot_vs_direct")
mod_var = _load_service("02_variance_determinism.py", "variance_determinism")
mod_calib = _load_service("03_tool_schema_calibration.py", "tool_schema_calibration")
mod_err = _load_service("04_tool_error_injection.py", "tool_error_injection")


@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/cot-vs-direct", methods=["POST"])
def cot_vs_direct_route():
    # ① Parse problem_title from the JSON request body
    data = request.get_json(force=True, silent=True) or {}
    problem_title = data.get("problem_title") or "Spatial Navigation (Grid)"
    try:
        # ② Run Direct vs Chain-of-Thought benchmark in 01_cot_vs_direct.py
        res = mod_cot.run_cot_vs_direct_benchmark(problem_title)
        res["result"] = f"Direct: {'Correct' if res['direct_result']['is_correct'] else 'Failed'} | CoT: {'Correct' if res['cot_result']['is_correct'] else 'Failed'}"
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/variance-determinism", methods=["POST"])
def variance_determinism_route():
    # ① Parse review_text from the JSON request body and reject empty strings
    data = request.get_json(force=True, silent=True) or {}
    review_text = (data.get("review_text") or data.get("message") or "").strip()
    if not review_text:
        return jsonify({"error": "Review text cannot be empty"}), 400
    try:
        # ② Run the 3-strategy JSON determinism benchmark in 02_variance_determinism.py
        res = mod_var.run_variance_determinism_benchmark(review_text)
        res["result"] = f"Tested {len(res.get('strategies', []))} JSON extraction strategies."
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/tool-schema-calibration", methods=["POST"])
def tool_schema_calibration_route():
    # ① Parse query from the JSON request body and reject empty strings
    data = request.get_json(force=True, silent=True) or {}
    query = (data.get("query") or data.get("message") or "").strip()
    if not query:
        return jsonify({"error": "Query cannot be empty"}), 400
    try:
        # ② Run the 2x2 tool schema calibration matrix in 03_tool_schema_calibration.py
        res = mod_calib.run_tool_schema_calibration(query)
        res["result"] = res.get("analysis_summary", "Calibration complete.")
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/tool-error-injection", methods=["POST"])
def tool_error_injection_route():
    # ① Parse question from the JSON request body and reject empty strings
    data = request.get_json(force=True, silent=True) or {}
    question = (data.get("question") or data.get("message") or "").strip()
    if not question:
        return jsonify({"error": "Question cannot be empty"}), 400
    try:
        # ② Run the upstream 503 error injection benchmark in 04_tool_error_injection.py
        res = mod_err.run_tool_error_injection(question)
        res["result"] = f"Evaluated error handling across {len(res.get('scenarios', []))} prompt configurations."
        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 key reliability engineering takeaways from the Chain-of-Thought vs. Direct benchmark:

1. Sequential State Tracking Breaks Direct Prompting: Tasks like grid navigation, multi-generation kinship resolution, and constraint scheduling require tracking intermediate state across steps; forbidding intermediate output forces the model to compress all state transitions into a single token jump.
2. Context-Window State Externalization: Instructing the model to "Think step by step" writes each intermediate coordinate, relationship, or slot assignment into the token stream so subsequent attention heads can condition on explicit state rather than hidden activations.
3. Final-Line Answer Extraction: During a reasoning trace, a model frequently mentions intermediate coordinates or ruled-out candidates; extracting text_c.strip().splitlines()[-1] ensures the UI displays the final conclusion rather than intermediate working.
4. Punctuation- and Hyphen-Normalized Grading (evaluate_cot_accuracy): Stripping spaces and hyphens before matching against accepted_answers ensures equivalent surface forms like (0, -2) vs. (0,-2) or great-aunt vs. great aunt are graded identically.
5. Latency Cost of Reliability: Because autoregressive decoding time scales linearly with output token count, Chain-of-Thought trades higher wall-clock latency (latency_seconds) for substantially higher task reliability on multi-hop logic problems.
6. Controlled Low-Temperature Benchmarking (temperature=0.3): Running both Direct and CoT strategies on the exact same model checkpoint and temperature isolates the structural effect of step-by-step reasoning from sampling noise.
7. Selective Deployment of CoT in Production: Use Direct prompting for low-latency single-hop classification or extraction tasks, and reserve Chain-of-Thought (or reasoning models) for workflows involving state transitions, arithmetic, or multi-constraint deduction.