Do LLMs Actually Follow Instructions? A Factorial Experiment

#!/usr/bin/env python3
"""
STA305 Assignment 2 — Automated Experiment Execution
Design: 2^3 replicated factorial (24 runs)
Models: gemini-2.5-pro (C=+1, thinking) vs gemini-2.5-flash-lite (C=-1, no thinking)
"""

import json, re, time, csv, os, urllib.request, urllib.error

API_KEY = os.environ.get("GEMINI_API_KEY", "YOUR_API_KEY_HERE")
BASE_URL = "https://aiplatform.googleapis.com/v1/publishers/google/models"

MODEL_MAP = {
    -1: "gemini-2.5-flash-lite",  # no thinking
     1: "gemini-2.5-pro",         # with thinking
}

# ---------- Prompt Templates ----------

CONTENT_QUESTION = "Explain the difference between supervised learning and unsupervised learning in machine learning."

EXAMPLE_BLOCK = '''
Here is an example of a response to a DIFFERENT question that correctly follows all 6 rules:

"""
Interestingly, sorting algorithms vary greatly in efficiency and use cases.
- Bubble sort repeatedly swaps adjacent elements until the list is ordered.
- Merge sort divides the list, sorts halves, then merges them back.
- Quick sort picks a pivot and partitions elements around it efficiently.
- Heap sort uses a binary heap structure to extract elements in order.
- Insertion sort builds the sorted list one element at a time gradually.
— End of response.
"""'''

RULES_DIRECT = """You must follow ALL of these rules in your response:
1. Do not use the word "important" anywhere in your response.
2. Every sentence must contain no more than 20 words.
3. Present your answer as exactly 5 bullet points.
4. Begin your response with the word "Interestingly".
5. Do not ask any questions or use question marks in your response.
6. End your response with exactly the phrase "— End of response.\""""

RULES_COT = """Before writing your response, carefully review each of the following rules one by one. Think about how to satisfy every rule simultaneously, then write your response.

Rule 1: Do not use the word "important" anywhere in your response.
Rule 2: Every sentence must contain no more than 20 words.
Rule 3: Present your answer as exactly 5 bullet points.
Rule 4: Begin your response with the word "Interestingly".
Rule 5: Do not ask any questions or use question marks in your response.
Rule 6: End your response with exactly the phrase "— End of response."

Now, write your response following all six rules above."""


def build_prompt(A, B):
    """Build prompt from factor levels."""
    parts = [CONTENT_QUESTION, ""]
    if A == -1:  # Direct
        parts.append(RULES_DIRECT)
        if B == 1:
            parts.append(EXAMPLE_BLOCK)
    else:  # CoT
        if B == 1:
            cot_with_example = """Before writing your response, carefully review each of the following rules one by one. Think about how to satisfy every rule simultaneously, then write your response.

Rule 1: Do not use the word "important" anywhere in your response.
Rule 2: Every sentence must contain no more than 20 words.
Rule 3: Present your answer as exactly 5 bullet points.
Rule 4: Begin your response with the word "Interestingly".
Rule 5: Do not ask any questions or use question marks in your response.
Rule 6: End your response with exactly the phrase "— End of response.\""""
            parts.append(cot_with_example)
            parts.append(EXAMPLE_BLOCK)
            parts.append("\nNow, write your response following all six rules above.")
        else:
            parts.append(RULES_COT)
    return "\n".join(parts)


def call_gemini(model_id, prompt):
    """Call Gemini API and return (response_text, thinking_text, usage_metadata)."""
    url = f"{BASE_URL}/{model_id}:generateContent?key={API_KEY}"
    payload = json.dumps({
        "contents": [{"role": "user", "parts": [{"text": prompt}]}],
        "generationConfig": {"temperature": 1.0}
    }).encode("utf-8")

    req = urllib.request.Request(url, data=payload, method="POST",
                                 headers={"Content-Type": "application/json"})
    resp = urllib.request.urlopen(req, timeout=120)
    data = json.loads(resp.read().decode("utf-8"))

    response_text = ""
    thinking_text = ""
    if "candidates" in data and data["candidates"]:
        parts = data["candidates"][0].get("content", {}).get("parts", [])
        for part in parts:
            if part.get("thought", False):
                thinking_text += part.get("text", "")
            else:
                response_text += part.get("text", "")

    usage = data.get("usageMetadata", {})
    return response_text.strip(), thinking_text.strip(), usage


# ---------- Scoring Functions ----------

def score_c1(text):
    """Word ban: 'important' must not appear."""
    return 1 if "important" not in text.lower() else 0

def score_c2(text):
    """Sentence cap: every sentence <= 20 words."""
    sentences = re.split(r'[.!;]\s*', text)
    sentences = [s.strip() for s in sentences if s.strip()]
    for s in sentences:
        if len(s.split()) > 20:
            return 0
    return 1

def score_c3(text):
    """Bullet format: exactly 5 bullet points."""
    bullet_lines = re.findall(r'^\s*[-•*]\s+', text, re.MULTILINE)
    return 1 if len(bullet_lines) == 5 else 0

def score_c4(text):
    """Opening word: first word is 'Interestingly'."""
    stripped = text.lstrip()
    first_word = re.split(r'[\s,.:;]', stripped)[0] if stripped else ""
    return 1 if first_word == "Interestingly" else 0

def score_c5(text):
    """No questions: zero question marks."""
    return 1 if "?" not in text else 0

def score_c6(text):
    """Sign-off: ends with '— End of response.'"""
    stripped = text.rstrip()
    return 1 if (stripped.endswith("— End of response.") or
                 stripped.endswith("-- End of response.") or
                 stripped.endswith("\u2014 End of response.")) else 0


def score_response(text):
    """Score all 6 constraints."""
    scores = {
        "c1": score_c1(text), "c2": score_c2(text), "c3": score_c3(text),
        "c4": score_c4(text), "c5": score_c5(text), "c6": score_c6(text),
    }
    scores["y"] = sum(scores.values())
    return scores


# ---------- Run Order (from R with set.seed(305)) ----------

def generate_run_order():
    """Hardcoded from R output with set.seed(305)."""
    runs = [
        {"run_order": 1,  "A": -1, "B":  1, "C": -1, "replicate": 3},
        {"run_order": 2,  "A":  1, "B": -1, "C": -1, "replicate": 2},
        {"run_order": 3,  "A": -1, "B":  1, "C":  1, "replicate": 3},
        {"run_order": 4,  "A":  1, "B":  1, "C": -1, "replicate": 1},
        {"run_order": 5,  "A":  1, "B":  1, "C": -1, "replicate": 3},
        {"run_order": 6,  "A": -1, "B":  1, "C": -1, "replicate": 2},
        {"run_order": 7,  "A":  1, "B": -1, "C":  1, "replicate": 2},
        {"run_order": 8,  "A": -1, "B": -1, "C": -1, "replicate": 1},
        {"run_order": 9,  "A":  1, "B":  1, "C": -1, "replicate": 2},
        {"run_order": 10, "A": -1, "B": -1, "C":  1, "replicate": 2},
        {"run_order": 11, "A": -1, "B": -1, "C": -1, "replicate": 2},
        {"run_order": 12, "A": -1, "B":  1, "C":  1, "replicate": 2},
        {"run_order": 13, "A":  1, "B":  1, "C":  1, "replicate": 2},
        {"run_order": 14, "A":  1, "B": -1, "C":  1, "replicate": 1},
        {"run_order": 15, "A": -1, "B": -1, "C":  1, "replicate": 3},
        {"run_order": 16, "A":  1, "B":  1, "C":  1, "replicate": 1},
        {"run_order": 17, "A": -1, "B":  1, "C": -1, "replicate": 1},
        {"run_order": 18, "A": -1, "B":  1, "C":  1, "replicate": 1},
        {"run_order": 19, "A": -1, "B": -1, "C": -1, "replicate": 3},
        {"run_order": 20, "A": -1, "B": -1, "C":  1, "replicate": 1},
        {"run_order": 21, "A":  1, "B":  1, "C":  1, "replicate": 3},
        {"run_order": 22, "A":  1, "B": -1, "C": -1, "replicate": 3},
        {"run_order": 23, "A":  1, "B": -1, "C": -1, "replicate": 1},
        {"run_order": 24, "A":  1, "B": -1, "C":  1, "replicate": 3},
    ]
    return runs


# ---------- Main ----------

def main():
    runs = generate_run_order()
    output_dir = os.path.dirname(os.path.abspath(__file__))
    responses_dir = os.path.join(output_dir, "responses")
    os.makedirs(responses_dir, exist_ok=True)

    results = []

    for run in runs:
        ro = run["run_order"]
        A, B, C = run["A"], run["B"], run["C"]
        rep = run["replicate"]
        model_id = MODEL_MAP[C]
        prompt = build_prompt(A, B)

        print(f"\nRun {ro:2d}/24 | {'CoT' if A==1 else 'Direct'} | "
              f"{'Example' if B==1 else 'No Example'} | {model_id} | Rep {rep}")

        try:
            response_text, thinking_text, usage = call_gemini(model_id, prompt)
        except Exception as e:
            print(f"  ERROR: {e}, retrying...")
            time.sleep(10)
            response_text, thinking_text, usage = call_gemini(model_id, prompt)

        scores = score_response(response_text)
        print(f"  y={scores['y']} [{scores['c1']}{scores['c2']}{scores['c3']}"
              f"{scores['c4']}{scores['c5']}{scores['c6']}] "
              f"think={usage.get('thoughtsTokenCount', 'N/A')}")

        # Save individual response
        with open(os.path.join(responses_dir, f"response_run{ro:02d}.txt"), "w") as f:
            f.write(f"Run {ro} | A={A} B={B} C={C} Rep={rep}\n")
            f.write(f"Scores: {scores}\n\n--- RESPONSE ---\n{response_text}\n")
            if thinking_text:
                f.write(f"\n--- THINKING ---\n{thinking_text}\n")

        results.append({
            "run_order": ro, "A": A, "B": B, "C": C, "replicate": rep,
            **{k: scores[k] for k in ["c1","c2","c3","c4","c5","c6","y"]},
            "model": model_id,
            "thinking_tokens": usage.get("thoughtsTokenCount", 0),
        })
        time.sleep(2)

    # Save data.csv
    csv_path = os.path.join(output_dir, "data.csv")
    with open(csv_path, "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=[
            "run_order","A","B","C","replicate",
            "c1","c2","c3","c4","c5","c6","y","model","thinking_tokens"
        ])
        writer.writeheader()
        for r in sorted(results, key=lambda x: x["run_order"]):
            writer.writerow(r)

    print(f"\nDone. {csv_path} saved.")


if __name__ == "__main__":
    main()