Chain of Thought Prompting — How We Lost $12k Overnight Because Our LLM Didn't Show Its Work

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer

model = AutoModelForCausalLM.from_pretrained("gpt2", output_attentions=True)
tokenizer = AutoTokenizer.from_pretrained("gpt2")

# Input with and without CoT
input_no_cot = "The odd numbers in this group add up to an even number: 4, 8, 9, 15, 12, 2, 1."
input_with_cot = input_no_cot + " Let's think step by step. The odd numbers are 9, 15, 1. Their sum is 25. 25 is odd, so the statement is false."

# Tokenize and forward pass
tokens_no_cot = tokenizer(input_no_cot, return_tensors="pt")
tokens_with_cot = tokenizer(input_with_cot, return_tensors="pt")

with torch.no_grad():
    outputs_no_cot = model(**tokens_no_cot)
    outputs_with_cot = model(**tokens_with_cot)

# Compare attention patterns: look at the last token's attention to all previous tokens
last_token_attn_no_cot = outputs_no_cot.attentions[-1][0, :, -1, :].mean(dim=0)  # average over heads
last_token_attn_with_cot = outputs_with_cot.attentions[-1][0, :, -1, :].mean(dim=0)

print(f"Attention spread (no CoT): {last_token_attn_no_cot.std():.4f}")  # higher variance = focused on few tokens
print(f"Attention spread (with CoT): {last_token_attn_with_cot.std():.4f}")

# With CoT, attention is more distributed across the reasoning chain, not just the input

import openai  # openai>=1.0
import time
import json
from typing import Optional

client = openai.OpenAI(api_key="sk-your-key-here")

class CoTPipeline:
    def __init__(self, model="gpt-4", max_reasoning_tokens=200, temperature=0.0):
        self.model = model
        self.max_reasoning_tokens = max_reasoning_tokens
        self.temperature = temperature

    def query(self, user_input: str, num_chains: int = 1) -> dict:
        """
        Returns: {
            'final_answer': str,
            'reasoning_chains': [str],
            'latency_ms': int,
            'total_tokens': int
        }
        """
        start = time.time()
        chains = []
        for _ in range(num_chains):
            response = client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "You are a helpful assistant. Reason step by step, then provide the answer."},
                    {"role": "user", "content": user_input + " Let's think step by step."}
                ],
                max_tokens=self.max_reasoning_tokens + 50,  # extra for the answer
                temperature=self.temperature
            )
            full_output = response.choices[0].message.content
            chains.append(full_output)

        # Parse reasoning and answer (simple heuristic: last sentence is answer)
        answers = []
        for chain in chains:
            # Assume the last sentence is the answer
            sentences = chain.split(". ")
            answers.append(sentences[-1] if sentences else chain)

        # Majority vote for final answer
        from collections import Counter
        final_answer = Counter(answers).most_common(1)[0][0]

        return {
            "final_answer": final_answer,
            "reasoning_chains": chains,
            "latency_ms": int((time.time() - start) * 1000),
            "total_tokens": sum(len(c.split()) for c in chains)  # rough estimate
        }

# Usage
pipeline = CoTPipeline()
result = pipeline.query("The odd numbers in this group add up to an even number: 4, 8, 9, 15, 12, 2, 1.", num_chains=3)
print(json.dumps(result, indent=2))

import openai
import time

client = openai.OpenAI(api_key="sk-your-key-here")

# Test on factual recall
test_questions = [
    "What is the capital of France?",
    "Who wrote the novel '1984'?",
    "What is the chemical symbol for gold?"
]

def query_with_cot(question):
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "user", "content": question + " Let's think step by step."}
        ],
        max_tokens=100
    )
    return response.choices[0].message.content

def query_direct(question):
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "user", "content": question}
        ],
        max_tokens=50
    )
    return response.choices[0].message.content

for q in test_questions:
    start = time.time()
    cot_answer = query_with_cot(q)
    cot_latency = (time.time() - start) * 1000

    start = time.time()
    direct_answer = query_direct(q)
    direct_latency = (time.time() - start) * 1000

    print(f"Q: {q}")
    print(f"  CoT ({cot_latency:.0f}ms): {cot_answer}")
    print(f"  Direct ({direct_latency:.0f}ms): {direct_answer}")
    print()

# Expected: Direct is faster and equally correct for factual recall. CoT adds latency and sometimes error.

import openai
import redis
import hashlib
import json

# openai>=1.0, redis>=5.0
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
client = openai.OpenAI(api_key="sk-your-key-here")

def get_cot_answer(user_input: str) -> str:
    # Check cache
    input_hash = hashlib.sha256(user_input.encode()).hexdigest()
    cached = r.get(f"cot:{input_hash}")
    if cached:
        return cached

    # Use small model for reasoning chain
    chain_response = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[
            {"role": "user", "content": user_input + " Let's think step by step."}
        ],
        max_tokens=200,
        temperature=0.0
    )
    reasoning_chain = chain_response.choices[0].message.content

    # Use large model for final answer, given the reasoning
    answer_response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "user", "content": f"Given this reasoning: {reasoning_chain}\n\nProvide the final answer to the question: {user_input}"}
        ],
        max_tokens=50,
        temperature=0.0
    )
    final_answer = answer_response.choices[0].message.content

    # Cache the final answer (not the chain, to save space)
    r.setex(f"cot:{input_hash}", 86400, final_answer)  # 24-hour TTL

    return final_answer

# Example usage
print(get_cot_answer("What is the capital of France?"))  # Cached after first call

import openai

client = openai.OpenAI(api_key="sk-your-key-here")

def cot_with_validation(user_input: str) -> str:
    # Step 1: Generate CoT chain
    cot_response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "user", "content": user_input + " Let's think step by step."}
        ],
        max_tokens=200
    )
    chain = cot_response.choices[0].message.content

    # Step 2: Validate the chain
    validation_response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "user", "content": f"Given this reasoning chain: '{chain}'\n\nIs the reasoning logically consistent? Answer 'Yes' or 'No' and explain why."}
        ],
        max_tokens=100
    )
    validation = validation_response.choices[0].message.content

    if "No" in validation[:10]:  # Simple check
        # Chain is invalid, regenerate with different prompt
        cot_response = client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "user", "content": user_input + " Let's think step by step, and double-check each step before moving to the next."}
            ],
            max_tokens=200
        )
        chain = cot_response.choices[0].message.content

    # Step 3: Extract final answer from chain
    # Assume last sentence is answer
    sentences = chain.split(". ")
    return sentences[-1] if sentences else chain

print(cot_with_validation("The odd numbers in this group add up to an even number: 4, 8, 9, 15, 12, 2, 1."))

import openai
import time

client = openai.OpenAI(api_key="sk-your-key-here")

test_input = "A bat and a ball cost $1.10 in total. The bat costs $1.00 more than the ball. How much does the ball cost?"

def direct_prompt():
    start = time.time()
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": test_input}],
        max_tokens=50
    )
    return response.choices[0].message.content, (time.time()-start)*1000

def zero_shot_cot():
    start = time.time()
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": test_input + " Let's think step by step."}],
        max_tokens=200
    )
    return response.choices[0].message.content, (time.time()-start)*1000

def few_shot_cot():
    prompt = """Question: If there are 3 apples and you eat 2, how many are left?
Reasoning: Start with 3 apples. Eat 2. 3 - 2 = 1. Answer: 1.

Question: A bat and a ball cost $1.10 in total. The bat costs $1.00 more than the ball. How much does the ball cost?
Reasoning:"""
    start = time.time()
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=200
    )
    return response.choices[0].message.content, (time.time()-start)*1000

print("Direct:", direct_prompt())
print("Zero-shot CoT:", zero_shot_cot())
print("Few-shot CoT:", few_shot_cot())

# Expected: Direct gives wrong answer ($0.10). Zero-shot and few-shot CoT give correct answer ($0.05).

import openai
import time
import logging
import json

# Configure structured logging
logging.basicConfig(level=logging.INFO, format='%(message)s')
logger = logging.getLogger(__name__)

client = openai.OpenAI(api_key="sk-your-key-here")

def monitored_cot_query(user_input: str, test_set: list = None) -> dict:
    start = time.time()

    try:
        response = client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": user_input + " Let's think step by step."}],
            max_tokens=200,
            temperature=0.0
        )
        chain = response.choices[0].message.content
        latency_ms = (time.time() - start) * 1000
        token_count = len(chain.split())

        # Log metrics
        log_entry = {
            "input_length": len(user_input),
            "chain_length_tokens": token_count,
            "latency_ms": latency_ms,
            "success": True
        }
        logger.info(json.dumps(log_entry))

        return {"chain": chain, "latency_ms": latency_ms, "token_count": token_count}

    except Exception as e:
        log_entry = {
            "input_length": len(user_input),
            "error": str(e),
            "latency_ms": (time.time() - start) * 1000,
            "success": False
        }
        logger.error(json.dumps(log_entry))
        return {"chain": None, "latency_ms": (time.time() - start) * 1000, "error": str(e)}

# Example usage
result = monitored_cot_query("What is the capital of France?")
print(json.dumps(result, indent=2))

// io.thecodeforge
// structured_cot.py — enforce JSON schema on each CoT step

def chain_of_thought(prompt: str, steps: list[dict]) -> list[dict]:
    """Each step dict must have 'name', 'instruction', 'schema' keys."""
    results = []
    context = {"original_prompt": prompt}

    for step in steps:
        # Build a strict prompt: 'Respond ONLY with JSON matching this schema'
        schema_hint = f"""Step: {step['name']}
Instruction: {step['instruction']}
Context: {context}
Respond with JSON: {step['schema']}"""
        raw = model_call(schema_hint)  # your LLM call

        # Validate immediately — no parsing later
        parsed = validate_json(raw, step['schema'])
        if not parsed:
            raise StepValidationError(step['name'], raw)

        results.append(parsed)
        context[step['name']] = parsed  # carry state forward

    return results

// io.thecodeforge
// anchor_injection.py — re-anchor every 3 steps to prevent drift

import hashlib

def anchor_summarizer(task: str) -> str:
    # Lightweight model: returns 10-word summary
    return lightweight_summarize(task, max_words=10)

def run_anchored_chain(original_task: str, steps: list) -> list:
    anchor = anchor_summarizer(original_task)
    results = []

    for i, step in enumerate(steps):
        if i > 0 and i % 3 == 0:
            # Inject anchor as first sentence of step instruction
            step['instruction'] = f"[ANCHOR: {anchor}] {step['instruction']}"

        step_result = execute_step(step)
        results.append(step_result)

        # Optional: log drift score (cosine sim between anchor & last output)
        drift_score = cosine_similarity(embed(anchor), embed(step_result['output']))
        if drift_score < 0.5:
            logger.warning(f"Drift detected at step {i}: score {drift_score}")

    return results

# Usage:
# run_anchored_chain("Refund order #12345", steps)