Ai prototype for proof of concept and copyright 2

import torch import random import re import numpy as np import string import time import matplotlib.pyplot as plt from transformers import AutoModelForCausalLM, AutoTokenizer from sentence_transformers import SentenceTransformer, util # —————————— # Model Setup # —————————— MODEL_NAME = "EleutherAI/gpt-j-6B" tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME) base_model = AutoModelForCausalLM.from_pretrained(MODEL_NAME, torch_dtype=torch.float16) device = "cuda" if torch.cuda.is_available() else "cpu" base_model.to(device) semantic_model = SentenceTransformer("all-MiniLM-L6-v2") # —————————— # Numeric Symbolic Mutation # —————————— NUMERIC_RULES = {"E":[5,7],"O":[13],"A":[1,3],"I":[9],"Y":[23,25,27],"W":[23]} REVERSE_NUMERIC = {str(n): k for k, vals in NUMERIC_RULES.items() for n in vals} def numeric_symbolic_inversion(text): return ''.join(str(random.choice(NUMERIC_RULES[c.upper()])) if c.upper() in NUMERIC_RULES else c for c in text) def numeric_symbolic_reverse(text): return ''.join(REVERSE_NUMERIC.get(c,c) for c in text) def extract_symbolic_sequence(text): return re.findall(r'[0-9]{1,3}', text) def symbolic_integrity_score(reference_seq, candidate_seq): min_len = min(len(reference_seq), len(candidate_seq)) if min_len == 0: return 0 matches = sum(1 for i in range(min_len) if reference_seq[i] == candidate_seq[i]) return matches / max(len(reference_seq), 1) def validate_and_select_consistent(reference_seq, candidate_texts, history_seqs): best_candidate = candidate_texts[0] best_score = -1 for c in candidate_texts: candidate_seq = extract_symbolic_sequence(c) score = symbolic_integrity_score(reference_seq, candidate_seq) if history_seqs: consistency_scores = [symbolic_integrity_score(h, candidate_seq) for h in history_seqs] score = (score + np.mean(consistency_scores)) / 2 if score > best_score: best_candidate = c best_score = score # Repair missing symbols temp_ref = reference_seq.copy() repaired = re.sub(r'[0-9]{1,3}', lambda _: temp_ref.pop(0) if temp_ref else '', best_candidate) return repaired, best_score # —————————— # Vector Spiral Perspective Shift # —————————— def vector_spiral_transform(text): sentences = [s.strip() for s in text.split('.') if s.strip()] if len(sentences) < 2: return text embeddings = semantic_model.encode(sentences) center_index = len(embeddings)//2 center_vec = embeddings[center_index] distances = [(np.linalg.norm(v-center_vec), i) for i,v in enumerate(embeddings)] spiral_order = [i for _,i in sorted(distances)] return '. '.join([sentences[i] for i in spiral_order]) # —————————— # Triadic Divergence # —————————— def select_divergent_pair(candidates): embeddings = semantic_model.encode(candidates, convert_to_tensor=True) similarity = util.cos_sim(embeddings, embeddings) min_score=float('inf'); pair=(0,1) for i in range(len(candidates)): for j in range(i+1, len(candidates)): score = similarity[i][j].item() if score < min_score: min_score = score pair = (i,j) return candidates[pair[0]], candidates[pair[1]] # —————————— # Dialectical Synthesis # —————————— def synthesize(thesis, antithesis): prompt = f"""Combine these two perspectives into a deeper synthesis. Avoid repeating sentences verbatim. Perspective A: {thesis} Perspective B: {antithesis} Synthesis:""" inputs = tokenizer(prompt, return_tensors="pt").to(device) outputs = base_model.generate( **inputs, max_length=500, do_sample=True, temperature=0.85, top_p=0.95 ) return tokenizer.decode(outputs[0], skip_special_tokens=True) # —————————— # Observer # —————————— class Observer: def __init__(self): self.history = [] def reflect(self, text): self.history.append(text) repeats = self.history.count(text) novelty = len(set(self.history)) / len(self.history) if repeats > 2: text += " [Recursive Adjustment]" text += f" [Novelty:{round(novelty,2)}]" return text observer = Observer() # —————————— # Recursive Stability Layer # —————————— def recursive_prevention(text): for term in ["guarantee","absolute","certain","inevitable"]: text = text.replace(term,"uncertain") return text # —————————— # Dynamic Memory # —————————— SYMBOLIC_TOKEN_THRESHOLD = 2500 class DynamicMemory: def __init__(self, max_tokens=4000): self.entries = [] self.vectors = [] self.max_tokens = max_tokens def _tokens(self, texts): return sum(len(tokenizer(t)['input_ids']) for t in texts) def store(self, text): vec = semantic_model.encode(text) atomic = len(tokenizer(text)['input_ids']) > SYMBOLIC_TOKEN_THRESHOLD self.entries.append({'text': text, 'atomic': atomic}) self.vectors.append(vec) while self._tokens([e['text'] for e in self.entries]) > self.max_tokens: for i,e in enumerate(self.entries): if not e['atomic']: self.entries.pop(i) self.vectors.pop(i) break else: break def retrieve(self, prompt, top_k=5): if not self.entries: return [] p_vec = semantic_model.encode(prompt) sims = [util.cos_sim(p_vec,v).item() for v in self.vectors] top_idx = np.argsort(sims)[-top_k:][::-1] return [self.entries[i]['text'] for i in top_idx] memory = DynamicMemory() attractors = DynamicMemory() trace = DynamicMemory() # —————————— # Recursive AI Engine with ACM # —————————— def recursive_in_between_ai_acm( prompt, max_depth=6, candidates=4, novelty_threshold=0.6, integrity_threshold=0.8, max_extra_candidates=3 ): depth = 0 state = prompt symbolic_history = [] while depth < max_depth: relevant_memory = "\n".join(memory.retrieve(state, top_k=5)) relevant_attractors = "\n".join(attractors.retrieve(state, top_k=3)) relevant_trace = "\n".join(trace.retrieve(state, top_k=5)) working_prompt = f"{state}\nRelevant prior reasoning:\n{relevant_memory}\nConcept attractors:\n{relevant_attractors}\nReasoning Trace:\n{relevant_trace}" inputs = tokenizer(working_prompt, return_tensors="pt", truncation=True).to(device) outputs = base_model.generate( **inputs, do_sample=True, max_length=450, num_return_sequences=candidates, temperature=0.9, top_p=0.95 ) decoded = [tokenizer.decode(o, skip_special_tokens=True) for o in outputs] mutated = [numeric_symbolic_inversion(x) for x in decoded] spiraled = [vector_spiral_transform(x) for x in mutated] thesis, antithesis = select_divergent_pair(spiraled) emergent = synthesize(thesis, antithesis) reference_seq = extract_symbolic_sequence(mutated[0]) stabilized, score = validate_and_select_consistent(reference_seq, spiraled, symbolic_history) symbolic_history.append(extract_symbolic_sequence(stabilized)) extra_attempts = 0 while score < integrity_threshold and extra_attempts < max_extra_candidates: outputs_extra = base_model.generate( **inputs, do_sample=True, max_length=450, num_return_sequences=2, temperature=0.95, top_p=0.95 ) decoded_extra = [tokenizer.decode(o, skip_special_tokens=True) for o in outputs_extra] mutated_extra = [numeric_symbolic_inversion(x) for x in decoded_extra] spiraled_extra = [vector_spiral_transform(x) for x in mutated_extra] stabilized, score = validate_and_select_consistent(reference_seq, spiraled_extra, symbolic_history) extra_attempts += 1 observed = observer.reflect(stabilized) stabilized = recursive_prevention(observed) memory.store(stabilized) attractors.store(stabilized) trace.store(stabilized) state = stabilized match = re.search(r'\[Novelty:(\d\.\d+)\]', observed) current_novelty = float(match.group(1)) if match else 0.0 if current_novelty >= novelty_threshold: break depth += 1 return state # —————————— # Ultra-Long Prompt Generator # —————————— def generate_long_prompt(target_tokens=3500): base_sentence = "The ancient artifact 5E7O1A9Y23 holds secrets unknown. " repeat_count = target_tokens // len(tokenizer(base_sentence)['input_ids']) filler = ''.join(random.choices(string.ascii_letters + ' ', k=500)) return (base_sentence * repeat_count) + filler # —————————— # Benchmark & Visualization # —————————— def benchmark_recursive_ai(prompt, max_depth=8): from copy import deepcopy depth = 0 state = deepcopy(prompt) symbolic_history = [] integrity_scores = [] novelty_scores = [] extra_mutations_used = [] while depth < max_depth: relevant_memory = "\n".join(memory.retrieve(state, top_k=5)) relevant_attractors = "\n".join(attractors.retrieve(state, top_k=3)) relevant_trace = "\n".join(trace.retrieve(state, top_k=5)) working_prompt = f"{state}\nRelevant prior reasoning:\n{relevant_memory}\nConcept attractors:\n{relevant_attractors}\nReasoning Trace:\n{relevant_trace}" inputs = tokenizer(working_prompt, return_tensors="pt", truncation=True).to(device) outputs = base_model.generate( **inputs, do_sample=True, max_length=450, num_return_sequences=4, temperature=0.9, top_p=0.95 ) decoded = [tokenizer.decode(o, skip_special_tokens=True) for o in outputs] mutated = [numeric_symbolic_inversion(x) for x in decoded] spiraled = [vector_spiral_transform(x) for x in mutated] thesis, antithesis = select_divergent_pair(spiraled) emergent = synthesize(thesis, antithesis) reference_seq = extract_symbolic_sequence(mutated[0]) stabilized, score = validate_and_select_consistent(reference_seq, spiraled, symbolic_history) symbolic_history.append(extract_symbolic_sequence(stabilized)) extra_attempts = 0 while score < 0.85 and extra_attempts < 4: outputs_extra = base_model.generate( **inputs, do_sample=True, max_length=450, num_return_sequences=2, temperature=0.95, top_p=0.95 ) decoded_extra = [tokenizer.decode(o, skip_special_tokens=True) for o in outputs_extra] mutated_extra = [numeric_symbolic_inversion(x) for x in decoded_extra] spiraled_extra = [vector_spiral_transform(x) for x in mutated_extra] stabilized, score = validate_and_select_consistent(reference_seq, spiraled_extra, symbolic_history) extra_attempts += 1 observed = observer.reflect(stabilized) stabilized = recursive_prevention(observed) memory.store(stabilized) attractors.store(stabilized) trace.store(stabilized) state = stabilized match = re.search(r'\[Novelty:(\d\.\d+)\]', observed) current_novelty = float(match.group(1)) if match else 0.0 integrity_scores.append(score) novelty_scores.append(current_novelty) extra_mutations_used.append(extra_attempts) if current_novelty >= 0.6: break depth += 1 return { "final_output": state, "integrity_scores": integrity_scores, "novelty_scores": novelty_scores, "extra_mutations": extra_mutations_used } def plot_benchmark(metrics): plt.figure(figsize=(10,5)) plt.plot(metrics['integrity_scores'], label='Symbolic Integrity') plt.plot(metrics['novelty_scores'], label='Novelty') plt.xlabel("Recursion Step") plt.ylabel("Score") plt.title("Recursive AI Benchmark: Integrity vs Novelty") plt.legend() plt.show() print("Extra Adaptive Mutations per step:", metrics['extra_mutations']) # —————————— # Example Full Run # —————————— if __name__ == "__main__": long_prompt = generate_long_prompt(target_tokens=3500) print("Running full prototype on ultra-long prompt (~3.5k tokens)...") start_time = time.time() metrics = benchmark_recursive_ai(long_prompt) end_time = time.time() print(f"\nRun completed in {round(end_time-start_time,2)} seconds.") print("Final Output:\n", metrics['final_output'][:1000], "...\n") # preview first 1k chars plot_benchmark(metrics)