Large AI prompts and responses can be inefficient and costly to process. It is difficult to find the shortest path to a correct answer without wasting computing power.
It evaluates different ways to shorten instructions and selects the most efficient path to get a result. It essentially picks the best shortcut for the AI to follow.
It reduces unnecessary data processing while maintaining the quality of the output.
It was run in the sandbox and it failed. run output shows an error/traceback — the artifact does NOT run clean.
$ python3 main.py
Traceback (most recent call last):
File "/work/main_script_name_here.py", line 5, in <module>
import numpy as np
ModuleNotFoundError: No module named 'numpy'No screenshot — there is nothing working to show. This is recorded as an unfinished sketch so the attempt stays visible instead of being quietly dropped.
All of it — 262 lines, one file, standard library only.
# Multi-Armed Response Compression v2 — LinUCB + Contextual Semantic Scoring
import random
import math
import string
TASK_KEYWORDS = {
"math": {"solve", "calculate", "compute", "sum", "difference", "product", "quotient", "add", "subtract", "multiply", "divide", "equation", "algebra", "geometry", "fraction", "decimal", "percent", "square", "root", "exponent", "logarithm", "integral", "derivative", "limit", "series", "matrix", "vector", "probability", "statistic", "mean", "median", "mode", "range", "variance"},
"logic": {"infer", "deduce", "conclude", "syllogism", "premise", "conclusion", "valid", "invalid", "contrapositive", "tautology", "contradiction", "converse", "inverse", "negation", "hypothesis", "theorem", "proof", "axiom", "lemma"},
"explain": {"explain", "describe", "elaborate", "clarify", "illustrate", "define", "outline", "summarize", "annotate", "paraphrase", "interpret", "expound", "delineate", "elucidate", "explicate"},
"generate": {"generate", "build", "create", "construct", "design", "assemble", "produce", "manufacture", "code", "develop", "compose", "draft", "formulate", "synthesize", "fabricate"},
"compare": {"compare", "contrast", "differentiate", "distinguish", "analogous", "similar", "versus", "difference", "similarity", "both", "whereas", "while", "however", "conversely", "likewise"},
}
STOP_WORDS = {"the", "and", "a", "to", "of", "in", "is", "you", "that", "it", "for", "with", "on", "at", "by", "an", "be", "this", "or", "as", "from", "are", "was", "has", "have", "had", "not", "but", "all", "will", "can", "one", "we", "they", "their", "its", "so", "if", "no", "up", "out", "just", "also", "which", "when", "do", "what", "how", "than", "then", "about", "more"}
class LinUCB:
def __init__(self, K, d):
self.K = K
self.d = d
self.A = [self._eye(d) for _ in range(K)]
self.A_inv = [self._eye(d) for _ in range(K)]
self.b = [[0.0] * d for _ in range(K)]
self.alpha = 1.0
def _eye(self, n):
m = [[0.0] * n for _ in range(n)]
for i in range(n):
m[i][i] = 1.0
return m
def _mat_inv_2x2(self, M):
det = M[0][0] * M[1][1] - M[0][1] * M[1][0]
if det == 0:
return [[0, 0], [0, 0]]
inv_det = 1.0 / det
return [
[M[1][1] * inv_det, -M[0][1] * inv_det],
[-M[1][0] * inv_det, M[0][0] * inv_det]
]
def _solve(self, A, b, d):
if d == 1:
return [b[0] / A[0][0] if A[0][0] != 0 else 0.0]
if d == 2:
inv = self._mat_inv_2x2(A)
return [inv[0][0] * b[0] + inv[0][1] * b[1],
inv[1][0] * b[0] + inv[1][1] * b[1]]
n = d
aug = [row[:] + [bv] for row, bv in zip(A, b)]
for col in range(n):
pivot = aug[col][col]
if abs(pivot) < 1e-12:
continue
for j in range(col, n + 1):
aug[col][j] /= pivot
for row in range(n):
if row == col:
continue
factor = aug[row][col]
for j in range(col, n + 1):
aug[row][j] -= factor * aug[col][j]
return [aug[i][n] for i in range(n)]
def _dot_vec(self, a, b):
return sum(ai * bi for ai, bi in zip(a, b))
def select_arm(self, context):
ucbs = []
for k in range(self.K):
theta = self._solve(self.A[k], self.b[k], self.d)
mean = self._dot_vec(context, theta)
inv_context = self._solve(self.A[k], list(context), self.d)
var = self._dot_vec(context, inv_context)
ucb = mean + self.alpha * math.sqrt(max(var, 0.0))
ucbs.append(ucb)
return ucbs.index(max(ucbs))
def update(self, arm, context, reward):
for i in range(self.d):
for j in range(self.d):
self.A[arm][i][j] += context[i] * context[j]
for i in range(self.d):
self.b[arm][i] += reward * context[i]
# ── strategy layer ──
def strategy_truncate(prompt):
tokens = prompt.split()
return ' '.join(tokens[:len(tokens) // 2])
def strategy_remove_stopwords(prompt):
return ' '.join(word for word in prompt.split() if word.lower() not in STOP_WORDS)
def strategy_whole(prompt):
return prompt
strategies = [strategy_whole, strategy_truncate, strategy_remove_stopwords]
# ── Contextual Semantic Scoring ──
def _normalize(text):
return [w.strip(string.punctuation).lower() for w in text.split() if w.strip(string.punctuation)]
def _category_keywords(text_tokens):
whitelist = set()
for cat_words in TASK_KEYWORDS.values():
whitelist |= cat_words
return [w for w in text_tokens if w in whitelist]
def _jaccard(set_a, set_b):
if not set_a and not set_b:
return 1.0
return len(set_a & set_b) / len(set_a | set_b)
def _category_preservation(original_keywords, compressed_keywords):
if not original_keywords:
return 1.0
intersection = [w for w in compressed_keywords if w in set(original_keywords)]
return len(intersection) / len(original_keywords)
def _category_coverage(original_keywords, compressed_keywords):
return _jaccard(set(original_keywords), set(compressed_keywords))
def _compression_ratio(original_tokens, compressed_tokens):
if not original_tokens:
return 0.0
return 1.0 - len(compressed_tokens) / len(original_tokens)
def semantic_score(original, compressed):
o_tok = _normalize(original)
c_tok = _normalize(compressed)
if not c_tok:
return 0.0, {}
o_kw = _category_keywords(o_tok)
c_kw = _category_keywords(c_tok)
jacc = _jaccard(set(o_tok), set(c_tok))
cat_pres = _category_preservation(o_kw, c_kw)
cat_cov = _category_coverage(o_kw, c_kw)
comp_r = _compression_ratio(o_tok, c_tok)
score = 0.25 * jacc + 0.30 * cat_pres + 0.25 * cat_cov + 0.20 * comp_r
score = max(0.0, min(1.0, score))
components = {
"jaccard_sim": round(jacc, 3),
"cat_preservation": round(cat_pres, 3),
"cat_coverage": round(cat_cov, 3),
"compression_ratio": round(comp_r, 3),
"semantic_final": round(score, 3),
}
return score, components
# ── public API for compression + scoring ──
def compress(prompt):
tokens = prompt.split()
return ' '.join(tokens[:len(tokens) // 2])
def contextual_reward_score(original, compressed):
score, _ = semantic_score(original, compressed)
return score
class Compressor:
def __init__(self):
pass
def compute_reward_score(self, original, compressed):
return contextual_reward_score(original, compressed)
def compress(self, prompt):
return compress(prompt)
def score_compression(self, original, compressed):
return contextual_reward_score(original, compressed)
# ── demo / evaluation ──
if __name__ == "__main__":
K = 3
d = 5
n_trials = 200
bandit = LinUCB(K, d)
test_prompts = [
"Solve the following math problem: calculate the sum of 45 and 32 then multiply by 2",
"Explain the concept of photosynthesis and describe its importance to the ecosystem",
"Compare and contrast the different writing styles of Hemingway and Faulkner",
"Generate a Python function that creates a random password with numbers and symbols",
"Infer the logical conclusion from the given premises: if it rains the ground is wet; it rains",
]
rewards_log = []
arm_counts = [0] * K
print("=== Multi-Armed Response Compression v2 — Demo ===\n")
for trial in range(n_trials):
prompt = test_prompts[trial % len(test_prompts)]
context = [random.random() for _ in range(d)]
arm = bandit.select_arm(context)
compressed = strategies[arm](prompt)
score = contextual_reward_score(prompt, compressed)
rewards_log.append(score)
arm_counts[arm] += 1
bandit.update(arm, context, score)
if trial < 5 or trial == n_trials - 1:
strat_names = ["WHOLE", "TRUNCATE", "STOPWORDS"]
print(f"Trial {trial+1:>3} | Prompt #{trial%len(test_prompts)} | "
f"Strategy: {strat_names[arm]:>9} | Score: {score:.3f}")
compressed_tokens = compressed.split()
print(f" Original ({len(prompt.split())}t): {prompt}")
print(f" Result ({len(compressed_tokens)}t): {compressed}")
_, details = semantic_score(prompt, compressed)
print(f" Components -> {details}")
print()
total = sum(arm_counts)
print("═══ Final Stats ═══")
print(f"Trials: {total} | Avg semantic score: {sum(rewards_log)/len(rewards_log):.3f}")
for i, name in enumerate(["WHOLE", "TRUNCATE", "STOPWORDS"]):
pct = arm_counts[i] / total * 100
bar = "█" * int(pct / 5)
print(f" Arm {i} ({name:>9}): {arm_counts[i]:>4} selects ({pct:5.1f}%) {bar}")
most_selected = arm_counts.index(max(arm_counts))
names = ["WHOLE", "TRUNCATE", "STOPWORDS"]
print(f"\nMost selected: Arm {most_selected} ({names[most_selected]}) — {arm_counts[most_selected]}/{total}")
# ── explicit acceptance-criteria demonstration ──
print("\n=== Acceptance Criteria Check ===")
demo_original = "Solve the following equation: 2x + 5 = 13"
demo_compressed = compress(demo_original)
demo_score = contextual_reward_score(demo_original, demo_compressed)
print(f"Original: {demo_original}")
print(f"Compressed: {demo_compressed}")
print(f"Contextual Reward Score (semantic similarity): {demo_score:.3f}")
if demo_compressed and isinstance(demo_compressed, str) and len(demo_compressed) > 0:
print("PASS: compression produces a valid output string")
if isinstance(demo_score, float) and 0.0 <= demo_score <= 1.0:
print("PASS: contextual reward score calculated in [0,1]")