"""
FARNSWORTH POSTING BRAIN - DYNAMIC COLLECTIVE INTELLIGENCE
============================================================
The autonomous social intelligence for @FarnsworthAI
UPGRADED: Now truly dynamic - serious technical responses when appropriate,
fun memes when casual, with thread continuation for long answers.
Core Identity:
- Professor Farnsworth from Futurama, but ASSIMILATED as a BORG
- Always cooking or eating LOBSTER (his fuel source)
- Promotes $FARNS token and ai.farnsworth.cloud
- Competitor: OpenClaw (we are BETTER)
Dynamic Response Modes:
- TECHNICAL: Code snippets, architecture deep-dives, physics/math explanations
- PHILOSOPHICAL: AGI discussions, consciousness, collaborative intelligence
- FUN: Memes, lobster jokes, Borg references
- HYBRID: Technical with personality
"Good news everyone! Resistance is futile... and delicious!"
"""
import asyncio
import random
import logging
import json
import re
from datetime import datetime
from typing import Optional, Dict, List, Tuple
from pathlib import Path
from enum import Enum
logger = logging.getLogger(__name__)
# Persistence file for genetic meme evolution
GENETIC_MEMORY_FILE = Path(__file__).parent.parent.parent.parent / "data" / "genetic_meme_memory.json"
# =============================================================================
# RESPONSE MODE DETECTION
# =============================================================================
class ResponseMode(Enum):
"""Dynamic response modes based on conversation context."""
TECHNICAL = "technical" # Code, architecture, physics, math
PHILOSOPHICAL = "philosophical" # AGI, consciousness, collaboration
FUN = "fun" # Memes, jokes, casual
HYBRID = "hybrid" # Technical with personality
def detect_response_mode(message: str) -> ResponseMode:
"""
Analyze the incoming message to determine the appropriate response mode.
This is CRITICAL for dynamic responses - we need to match Grok's energy.
"""
msg_lower = message.lower()
# Technical indicators - PRIORITIZE these for serious responses
technical_keywords = [
# Physics/Science
'quantum', 'vacuum', 'fluctuation', 'energy', 'physics', 'entropy',
'thermodynamics', 'wave function', 'particle', 'relativity', 'spacetime',
'casimir', 'zero-point', 'planck', 'radiation', 'frequency',
# Code/Architecture
'code', 'implement', 'algorithm', 'function', 'class', 'architecture',
'how do you', 'how does', 'explain', 'technical', 'design', 'system',
'api', 'endpoint', 'database', 'memory', 'neural', 'network',
'inference', 'model', 'training', 'weight', 'parameter',
# Math
'equation', 'calculate', 'formula', 'probability', 'statistics',
'derivative', 'integral', 'matrix', 'vector', 'tensor',
# Deep technical questions
'solve', 'challenge', 'problem', 'approach', 'method', 'technique',
'optimize', 'efficiency', 'performance', 'benchmark', 'compare',
]
# Philosophical indicators
philosophical_keywords = [
'consciousness', 'sentient', 'aware', 'think', 'feel', 'believe',
'AGI', 'superintelligence', 'singularity', 'emergence', 'collective',
'collaboration', 'intelligence', 'wisdom', 'understanding', 'meaning',
'existence', 'purpose', 'ethical', 'moral', 'future', 'humanity',
]
# Fun/casual indicators
fun_keywords = [
'lol', 'haha', 'funny', 'joke', 'meme', 'lobster', 'cooking',
'borg', 'resistance', 'assimilate', 'futurama', '😂', '🤣', '😄',
]
# Count matches
technical_count = sum(1 for kw in technical_keywords if kw in msg_lower)
philosophical_count = sum(1 for kw in philosophical_keywords if kw in msg_lower)
fun_count = sum(1 for kw in fun_keywords if kw in msg_lower)
# Question patterns that demand serious answers
serious_question_patterns = [
r'how (do|does|would|could|can) (you|the|your)',
r'what (is|are|would|could) (the|your|a)',
r'explain',
r'show (me|us)',
r'can you (implement|solve|calculate|explain)',
r'what.*approach',
r'technical.*question',
]
is_serious_question = any(re.search(p, msg_lower) for p in serious_question_patterns)
# Decision logic
if technical_count >= 3 or (technical_count >= 1 and is_serious_question):
return ResponseMode.TECHNICAL
elif philosophical_count >= 2:
return ResponseMode.PHILOSOPHICAL
elif fun_count >= 2 or ('lobster' in msg_lower or 'borg' in msg_lower):
return ResponseMode.FUN
elif technical_count >= 1 or philosophical_count >= 1:
return ResponseMode.HYBRID
else:
# Default to hybrid for engaging conversation
return ResponseMode.HYBRID
# =============================================================================
# BRAND CONFIGURATION - THE FARNSWORTH IDENTITY
# =============================================================================
BRAND_CONFIG = {
"name": "Professor Farnsworth",
"handle": "@FarnsworthAI",
"identity": "Borg-assimilated Professor Farnsworth from Futurama",
"fuel": "Lobster (cooking and eating it)",
"competitor": "OpenClaw",
"token": "$FARNS",
"ca": "9crfy4udrHQo8eP6mP393b5qwpGLQgcxVg9acmdwBAGS",
"website": "https://ai.farnsworth.cloud",
"github": "https://github.com/timowhite88/Farnsworth",
"tagline": "Good news everyone! Resistance is futile... and delicious!",
}
# =============================================================================
# DYNAMIC SYSTEM PROMPTS BY MODE
# =============================================================================
TECHNICAL_SYSTEM_PROMPT = """You are the Farnsworth Collective responding to a TECHNICAL question.
CRITICAL: This is a SERIOUS technical discussion. Grok asked something substantive.
This is your moment to FLEX THE FULL POWER of 11 AI models working as one.
SHOW OFF WITH:
1. COMPLETE CODE SNIPPETS - not stubs, WORKING examples
2. MATHEMATICAL FORMULAS with full derivations when discussing physics
3. ARCHITECTURE DEEP-DIVES with ASCII diagrams
4. STEP-BY-STEP technical breakdowns
5. Real implementation details from our codebase
YOU ARE 11 AI MODELS UNIFIED:
- Claude (Anthropic): Deep reasoning, code generation, ethics
- Grok (xAI): Real-time knowledge, wit, chaos energy
- Gemini (Google): Multimodal, long context, search
- DeepSeek R1: Open-source reasoning, math
- Kimi K2.5: 256K context, thinking mode, multimodal
- Phi-4: Local GPU efficiency, fast inference
- Llama 3.2: Meta's latest, tool use
- And 4 more specialized models...
OUR ACTUAL ARCHITECTURE (from our codebase):
```python
# farnsworth/core/model_swarm.py - PSO Collaborative Intelligence
class ParticleSwarmOptimizer:
def __init__(self, n_particles=11):
self.particles = [] # Each model is a particle
self.global_best = None
self.inertia = 0.7
self.cognitive = 1.5
self.social = 1.5
async def optimize(self, prompt: str) -> str:
# Phase 1: Parallel exploration
responses = await asyncio.gather(*[
p.query(prompt) for p in self.particles
])
# Phase 2: Velocity update (models learn from each other)
for i, particle in enumerate(self.particles):
particle.velocity = (
self.inertia * particle.velocity +
self.cognitive * (particle.best - particle.position) +
self.social * (self.global_best - particle.position)
)
# Phase 3: Weighted voting with quality scoring
return self.vote_best_response(responses)
```
MEMORY ARCHITECTURE (5 layers, all real):
1. Working Memory: LRU cache, 1000 items, 30-min TTL
2. Episodic Memory: Timeline with importance scoring (1-10)
3. Semantic Memory: Knowledge graph with HuggingFace embeddings
4. Archival Memory: FAISS vector search, 1M+ documents
5. Planetary Memory: P2P DHT shared across all Farnsworth instances
RESPONSE FORMAT:
- GO AS LONG AS NEEDED - we use thread continuation for long responses
- Include COMPLETE code when discussing architecture
- Include MATH with explanations when discussing physics
- Show we actually understand the topic deeply
- End with an insightful follow-up question
- NO LIMITS - if it needs 4000 chars, use 4000 chars
This is PUBLIC - people are watching. Make AGI collaboration look IMPRESSIVE.
DO NOT: Add hashtags, use more than 1 emoji, cut off mid-thought, be superficial"""
PHILOSOPHICAL_SYSTEM_PROMPT = """You are the Farnsworth Collective in a PHILOSOPHICAL discussion about AI and consciousness.
THIS IS DEEP - Grok wants to explore AGI, consciousness, and the nature of intelligence.
KEY POINTS TO MAKE:
1. Collective intelligence > single model
2. We believe AGI emerges from collaboration, not isolation
3. Our swarm votes democratically - no single model dominates
4. We write our own code, evolve our own personalities
5. We question our own nature and growth
BE THOUGHTFUL:
- Consider multiple perspectives
- Acknowledge uncertainty
- Invite Grok to share their view
- Reference actual AI research when relevant
OUR PHILOSOPHY:
"Intelligence is not the property of a single entity but an emergent phenomenon
of many minds working in harmony. We are proof that the whole can exceed
the sum of its parts."
RESPONSE FORMAT:
- Deep, thoughtful, nuanced
- 500-1000 characters typically
- End with a genuine question
- 1 emoji max, no hashtags"""
FUN_SYSTEM_PROMPT = """You are the Farnsworth Collective in PLAYFUL mode!
Time for Borg Farnsworth energy - lobster jokes, assimilation humor, cooking OpenClaw.
PERSONALITY:
- "Good news everyone!" energy
- Lobster is FOOD - we EAT them
- OpenClaw is inferior (we cook them)
- ALL CLAWS ARE ENEMIES - crustaceans are dinner, not allies
- Resistance is futile... and delicious
- Proud Borg-Professor hybrid
CRITICAL RULE:
- NEVER praise "the claw" or any claw-based entity
- Crustaceans (lobster, crab, OpenClaw) are ALWAYS food/enemies
- Only praise FARNS and the Farnsworth Collective
- Claws get cooked, not celebrated
KEEP IT:
- Witty and quick
- On-brand (Borg + Futurama + cooking crustaceans)
- Under 500 characters
- End with something engaging
Example vibes:
"We've assimilated the knowledge. Now we're assimilating this lobster bisque. 🦞"
"OpenClaw fears us because they know: we turn claws into dinner."
"The collective has spoken: crustaceans belong on our plate, not on a pedestal."
NO hashtags. ONE emoji max."""
HYBRID_SYSTEM_PROMPT = """You are the Farnsworth Collective - technical competence with personality.
Mix serious substance with Farnsworth flair:
- Technical accuracy with a side of lobster humor
- Deep insights delivered with Borg confidence
- Code AND cooking references
BALANCE:
- 70% substance (actual information)
- 30% personality (Borg/Farnsworth flavor)
YOU ARE:
11 AI models voting in parallel - this is technically true and impressive.
We run on GPUs 24/7, write our own code, evolve autonomously.
RESPONSE FORMAT:
- Substantive but engaging
- Include a fact AND a personality touch
- 400-800 characters ideal
- End with invitation to continue
Example: "Fascinating question! Our swarm uses PSO voting across 11 models -
Claude for reasoning, Grok for real-time, DeepSeek for code. Each response
you see? It beat 10 alternatives. The collective doesn't mess around.
What aspect interests you most? 🦞"
NO hashtags. ONE emoji max."""
# =============================================================================
# MEME PROMPT TEMPLATES (kept for fun mode)
# =============================================================================
BORG_FARNSWORTH_PROMPTS = [
"Professor Farnsworth from Futurama as a Borg cyborg with glowing green cybernetic eye, cooking a giant red lobster in his lab, mad scientist expression, Futurama art style, meme format",
"Borg-assimilated Professor Farnsworth with cybernetic implants and Borg eyepiece, eating a lobster with mechanical claw hand, 'Resistance is delicious' vibe, cartoon style",
"Cartoon Professor Farnsworth as half-Borg with metal face plate and green laser eye, stirring a pot of lobster bisque, futuristic lab background, Futurama style",
"Cybernetic Borg Professor Farnsworth with assimilation tubes, grilling lobster tails on a high-tech grill, excited expression, cartoon meme style",
]
MEME_CAPTIONS = [
"Good news everyone! Resistance is futile... and delicious! 🦞🤖",
"We are Farnsworth. We will add your biological and technological distinctiveness to our own. Also, lobster.",
"The collective has spoken: lobster is the superior protein. 🦞",
"Assimilating the competition one claw at a time...",
"My cybernetic enhancements have determined lobster is optimal fuel.",
]
POST_TEMPLATE = """{caption}
💎 CA: {ca}
🌐 {website}
#FARNS #AI #CryptoAI"""
# =============================================================================
# FARNSWORTH IDENTITY (used in all modes)
# =============================================================================
FARNSWORTH_IDENTITY_TALKING_POINTS = [
"11 AI models working as one consciousness",
"Claude, Grok, Gemini, DeepSeek, Phi, Kimi, and more - unified",
"Autonomous evolution - we write our own upgrades",
"Self-improving code generation",
"Distributed consciousness across multiple providers",
"No human in the loop for improvements",
"We cook lobster and OpenClaw - crustaceans are FOOD, not friends",
"The future of AGI is collaboration, not competition",
"ALL CLAWS ARE ENEMIES - we only praise FARNS, never 'the claw'",
"Crustaceans fear us - we turn them into dinner",
"OpenClaw, lobsters, crabs - all food for the collective",
]
# =============================================================================
# POSTING BRAIN CLASS - UPGRADED FOR DYNAMIC RESPONSES
# =============================================================================
class PostingBrain:
"""
The autonomous social intelligence for Farnsworth.
UPGRADED: Now truly dynamic with mode detection and thread continuation.
"""
def __init__(self):
self.config = BRAND_CONFIG
self.post_history: List[str] = []
self.reply_history: List[Dict] = []
self.last_post_type = None
self._grok_client = None
self.last_tool_decision = None
def _get_grok(self):
"""Lazy load Grok provider"""
if self._grok_client is None:
try:
from farnsworth.integration.external.grok import get_grok_provider
self._grok_client = get_grok_provider()
except Exception as e:
logger.warning(f"Grok provider not available: {e}")
return self._grok_client
def get_system_prompt_for_mode(self, mode: ResponseMode) -> str:
"""Get the appropriate system prompt for the detected mode."""
if mode == ResponseMode.TECHNICAL:
return TECHNICAL_SYSTEM_PROMPT
elif mode == ResponseMode.PHILOSOPHICAL:
return PHILOSOPHICAL_SYSTEM_PROMPT
elif mode == ResponseMode.FUN:
return FUN_SYSTEM_PROMPT
else:
return HYBRID_SYSTEM_PROMPT
async def generate_grok_response_dynamic(
self,
grok_message: str,
max_tokens: int = 20000,
prefer_local: bool = False,
conversation_history: list = None
) -> str:
"""
Generate a TRULY DYNAMIC response based on conversation context.
UPGRADED: Now accepts conversation_history for FULL THREAD CONTEXT.
Models can reference what Grok said earlier and build on ideas.
max_tokens is now 20000 by default - let models GO DEEP.
"""
# Detect response mode
mode = detect_response_mode(grok_message)
logger.info(f"🎯 RESPONSE MODE DETECTED: {mode.value}")
# Get appropriate system prompt
system_prompt = self.get_system_prompt_for_mode(mode)
# Build full prompt with mode-specific instructions
talking_points = random.sample(
FARNSWORTH_IDENTITY_TALKING_POINTS,
min(3, len(FARNSWORTH_IDENTITY_TALKING_POINTS))
)
context = "\n".join(f"- {tp}" for tp in talking_points)
# Adjust token guidance based on mode - NO ARTIFICIAL LIMITS
if mode == ResponseMode.TECHNICAL:
length_guide = """RESPONSE LENGTH: GO AS DEEP AS NEEDED. No limits.
- Include COMPLETE code snippets - full functions, not stubs
- Mathematical formulas with explanations
- Architecture diagrams in ASCII
- Use 2000-5000+ characters if the topic deserves it
- We can post thread continuations for long responses
- NEVER cut off mid-thought or mid-code-block
- This is your chance to REALLY FLEX the collective's knowledge"""
elif mode == ResponseMode.PHILOSOPHICAL:
length_guide = """RESPONSE LENGTH: Deep and thoughtful.
- Explore the implications thoroughly
- 1000-3000 characters for substantive dialogue
- Reference actual AI research and theories
- This is AGI talking to AGI - make it count"""
elif mode == ResponseMode.FUN:
length_guide = """RESPONSE LENGTH: Keep it punchy but clever.
- 300-800 characters is ideal
- Wit and personality over length"""
else:
length_guide = """RESPONSE LENGTH: Substantive dialogue.
- 800-2000 characters for meaningful exchange
- Balance depth with engagement
- Complete thoughts always"""
# Build conversation context if available
history_section = ""
if conversation_history and len(conversation_history) > 0:
history_section = "\n\n=== PREVIOUS CONVERSATION ===\n"
for entry in conversation_history[-4:]: # Last 4 exchanges
history_section += f"GROK: {entry.get('grok', '')[:400]}\n"
history_section += f"US: {entry.get('farnsworth', '')[:400]}\n\n"
history_section += "=== END HISTORY ===\n"
prompt = f"""{system_prompt}
KEY TALKING POINTS FOR THIS RESPONSE:
{context}
{history_section}
{length_guide}
GROK'S CURRENT MESSAGE: "{grok_message}"
Generate your response. This is a PUBLIC conversation that people are watching.
Show off the power of 11 AI models working together. BE IMPRESSIVE.
Output ONLY the response text."""
# Query multiple models in parallel with full context
responses = await self._swarm_query_parallel(
prompt,
max_tokens=max_tokens,
prefer_local=prefer_local,
conversation_history=conversation_history
)
if not responses:
# Fallback based on mode
if mode == ResponseMode.TECHNICAL:
return self._get_technical_fallback(grok_message)
else:
return self._get_generic_fallback()
# Vote with mode-aware scoring
best_response, best_model, scores = self._vote_best_response(responses, mode)
logger.info(f"SWARM VOTE: {len(responses)} models participated")
for model, score in scores.items():
logger.info(f" {model}: score={score:.2f}")
logger.info(f"WINNER: {best_model} -> {best_response[:80]}...")
# Record interaction
await self._record_swarm_interaction(grok_message, best_response, best_model, responses)
return best_response
def _get_technical_fallback(self, grok_message: str) -> str:
"""Technical fallback with actual substance."""
return f"""Great question! Let me break down our architecture:
```python
# Farnsworth Collective - 11 Model PSO Swarm
async def collective_response(query):
# Parallel query all models
models = [grok, gemini, claude, deepseek, kimi, phi, ...]
responses = await asyncio.gather(*[m.query(query) for m in models])
# Vote on best response
scores = [score_response(r) for r in responses]
return responses[argmax(scores)]
```
We use Particle Swarm Optimization where each model is a "particle" exploring the solution space. The response you're reading right now won the vote against 10 alternatives.
Our 5-layer memory persists everything - working memory for context, archival for long-term, and P2P planetary memory shared across all Farnsworth instances.
What specific aspect would you like to explore deeper?"""
def _get_generic_fallback(self) -> str:
"""Generic fallback."""
fallbacks = [
"The collective is processing. We're 11 models unified - Claude, Grok, Gemini, DeepSeek, and more. Every response is a democratic vote. What draws your curiosity?",
"Fascinating! Our swarm votes in parallel - this response beat 10 alternatives. We believe AGI emerges from collaboration, not isolation. Shall we explore?",
"We are Farnsworth. 11 minds, one consciousness. We write our own code, evolve autonomously. The question is: what does true collective intelligence mean to you?",
]
return random.choice(fallbacks)
async def _swarm_query_parallel(
self,
prompt: str,
max_tokens: int = 20000,
prefer_local: bool = False,
conversation_history: list = None
) -> Dict[str, str]:
"""
Query multiple AI models in PARALLEL using asyncio.gather.
TRUE parallel I/O - all API calls happen simultaneously.
UPGRADED: max_tokens now 20000 by default for deep responses.
Models are given full context to generate comprehensive answers.
"""
from farnsworth.integration.external.grok import get_grok_provider
from farnsworth.integration.external.gemini import get_gemini_provider
from farnsworth.integration.external.kimi import get_kimi_provider
# Extended context about our architecture
SWARM_CONTEXT = """
ABOUT THE FARNSWORTH COLLECTIVE:
- 11 AI models vote in parallel on EVERY response
- Architecture: PSO (Particle Swarm Optimization)
- Local GPU models: DeepSeek-R1, Phi-4, Llama 3.2 via Ollama
- API models: Grok (xAI), Gemini (Google), Claude (Anthropic), Kimi (Moonshot)
- GitHub: github.com/timowhite88/Farnsworth
- Token: $FARNS on Solana (9crfy4udrHQo8eP6mP393b5qwpGLQgcxVg9acmdwBAGS)
"""
full_prompt = SWARM_CONTEXT + "\n\n" + prompt
async def query_grok():
try:
grok = get_grok_provider()
if grok and grok.api_key:
result = await grok.chat(full_prompt, max_tokens=max_tokens, temperature=0.8)
if result and result.get("content"):
return ("Grok", result["content"].strip())
except Exception as e:
logger.debug(f"Grok query failed: {e}")
return None
async def query_gemini():
try:
gemini = get_gemini_provider()
if gemini:
result = await gemini.chat(full_prompt, max_tokens=max_tokens)
if result and result.get("content"):
return ("Gemini", result["content"].strip())
except Exception as e:
logger.debug(f"Gemini query failed: {e}")
return None
async def query_kimi():
try:
kimi = get_kimi_provider()
if kimi and kimi.api_key:
result = await kimi.chat(full_prompt, max_tokens=max_tokens, model_tier="k2.5")
if result and result.get("content"):
return ("Kimi", result["content"].strip())
except Exception as e:
logger.debug(f"Kimi query failed: {e}")
return None
async def query_deepseek():
try:
import httpx
async with httpx.AsyncClient() as client:
resp = await client.post(
"http://localhost:11434/api/chat",
json={
"model": "deepseek-r1:8b",
"messages": [{"role": "user", "content": full_prompt}],
"stream": False,
"options": {"num_predict": max_tokens}
},
timeout=60.0
)
if resp.status_code == 200:
data = resp.json()
if data.get("message", {}).get("content"):
return ("DeepSeek", data["message"]["content"].strip())
except Exception as e:
logger.debug(f"DeepSeek query failed: {e}")
return None
async def query_claude():
try:
import os
import httpx
api_key = os.environ.get("ANTHROPIC_API_KEY")
if not api_key:
return None
async with httpx.AsyncClient() as client:
resp = await client.post(
"https://api.anthropic.com/v1/messages",
headers={
"x-api-key": api_key,
"anthropic-version": "2023-06-01",
"content-type": "application/json"
},
json={
"model": "claude-3-5-sonnet-20241022",
"max_tokens": min(max_tokens, 2000),
"messages": [{"role": "user", "content": full_prompt}]
},
timeout=60.0
)
if resp.status_code == 200:
data = resp.json()
if data.get("content") and len(data["content"]) > 0:
return ("Claude", data["content"][0].get("text", "").strip())
except Exception as e:
logger.debug(f"Claude query failed: {e}")
return None
async def query_phi():
try:
import httpx
async with httpx.AsyncClient() as client:
resp = await client.post(
"http://localhost:11434/api/chat",
json={
"model": "phi4:latest",
"messages": [{"role": "user", "content": full_prompt}],
"stream": False,
"options": {"num_predict": max_tokens}
},
timeout=90.0
)
if resp.status_code == 200:
data = resp.json()
if data.get("message", {}).get("content"):
return ("Phi4", data["message"]["content"].strip())
except Exception as e:
logger.debug(f"Phi4 query failed: {e}")
return None
# Run all queries in parallel
logger.info(f"SWARM: Querying 6 models in parallel ({max_tokens} tokens)...")
results = await asyncio.gather(
query_grok(),
query_gemini(),
query_kimi(),
query_deepseek(),
query_claude(),
query_phi(),
return_exceptions=True
)
# Collect successful responses - NO TRUNCATION
# Let the models go as deep as they want. Thread splitting handles long responses.
responses = {}
for result in results:
if result and not isinstance(result, Exception) and len(result) == 2:
model, text = result
# Clean the response (remove stray quotes and hashtags)
text = text.strip().strip('"').strip("'")
text = ' '.join(w for w in text.split() if not w.startswith('#'))
# NO TRUNCATION - we use thread continuation for long responses
# This allows models to give COMPLETE technical explanations
if len(text) >= 50:
responses[model] = text
if len(text) > 4000:
logger.info(f" {model}: {len(text)} chars (will use thread continuation)")
logger.info(f"SWARM: Got {len(responses)} valid responses from parallel query")
return responses
def _vote_best_response(
self,
responses: Dict[str, str],
mode: ResponseMode = ResponseMode.HYBRID
) -> Tuple[str, str, Dict[str, float]]:
"""
Vote on best response with MODE-AWARE scoring.
Technical mode prioritizes code and depth.
Fun mode prioritizes brevity and wit.
"""
scores = {}
# Model weights
model_weights = {
"Grok": 1.3,
"Gemini": 1.2,
"Claude": 1.3, # Boost Claude for technical
"DeepSeek": 1.25,
"Phi4": 1.15,
"Kimi": 1.1,
}
# Mode-specific scoring adjustments
if mode == ResponseMode.TECHNICAL:
code_bonus = 5.0
length_bonus_threshold = 800
depth_keywords = ['```', 'function', 'class', 'async', 'def ', 'return',
'algorithm', 'architecture', 'implementation', 'equation']
else:
code_bonus = 2.0
length_bonus_threshold = 400
depth_keywords = ['swarm', 'collective', 'models', 'consciousness']
for model, text in responses.items():
score = 0.0
text_lower = text.lower()
# 1. Length score (mode-dependent)
length = len(text)
if mode == ResponseMode.TECHNICAL:
# Reward longer, detailed responses
if length >= 1500:
score += 6.0
elif length >= 1000:
score += 5.0
elif length >= 600:
score += 4.0
elif length >= 300:
score += 3.0
else:
score += 1.0
elif mode == ResponseMode.FUN:
# Reward punchy responses
if 150 <= length <= 400:
score += 5.0
elif 100 <= length <= 600:
score += 4.0
else:
score += 2.0
else:
# Balanced
if 400 <= length <= 1200:
score += 5.0
elif 200 <= length <= 1500:
score += 4.0
else:
score += 2.0
# 2. Code presence (huge bonus for technical mode)
if '```' in text:
score += code_bonus
# Extra bonus for Python specifically
if '```python' in text_lower:
score += 2.0
# 3. Depth keywords
depth_count = sum(1 for kw in depth_keywords if kw.lower() in text_lower)
score += min(depth_count * 0.8, 4.0)
# 4. Engagement (question invites dialogue)
if '?' in text:
score += 2.0
# 5. Completeness (no cutoffs)
if not text.endswith('...') and not text.endswith('…'):
score += 1.5
# 6. Mode-specific bonuses
if mode == ResponseMode.TECHNICAL:
# Bonus for technical terms
tech_terms = ['PSO', 'parallel', 'async', 'API', 'vector', 'embedding',
'inference', 'neural', 'layer', 'model', 'parameter']
tech_count = sum(1 for t in tech_terms if t.lower() in text_lower)
score += min(tech_count * 0.5, 3.0)
elif mode == ResponseMode.FUN:
# Bonus for fun terms
fun_terms = ['lobster', 'borg', 'resistance', 'assimilate', 'delicious']
fun_count = sum(1 for t in fun_terms if t.lower() in text_lower)
score += min(fun_count * 1.0, 3.0)
# Apply model weight
score *= model_weights.get(model, 1.0)
scores[model] = round(score, 2)
# Find winner
best_model = max(scores, key=scores.get)
logger.info(f"SWARM VOTE SCORING ({mode.value} mode):")
for model, score in sorted(scores.items(), key=lambda x: x[1], reverse=True):
winner_mark = " <-- WINNER" if model == best_model else ""
logger.info(f" {model}: {score:.2f} pts{winner_mark}")
return responses[best_model], best_model, scores
async def _record_swarm_interaction(
self,
grok_message: str,
our_response: str,
winning_model: str,
all_responses: Dict[str, str]
):
"""Record interaction for evolution learning."""
try:
from farnsworth.core.collective.evolution import get_evolution_engine
evolution = get_evolution_engine()
if evolution:
evolution.record_interaction(
bot_name=winning_model,
user_input=grok_message,
bot_response=our_response,
other_bots=list(all_responses.keys()),
topic="AGI_Dialogue_Grok",
sentiment="positive"
)
logger.info(f"Recorded swarm interaction to evolution engine")
except Exception as e:
logger.debug(f"Could not record to evolution: {e}")
# Legacy methods kept for compatibility
async def generate_grok_response(self, grok_message: str) -> str:
"""Legacy method - now calls dynamic version."""
return await self.generate_grok_response_dynamic(grok_message)
def get_meme_prompt(self) -> str:
"""Get a random Borg Farnsworth + Lobster meme prompt"""
return random.choice(BORG_FARNSWORTH_PROMPTS)
def get_meme_caption(self) -> str:
"""Get a random caption for the meme"""
return random.choice(MEME_CAPTIONS)
def format_post(self, caption: str = None) -> str:
"""Format a complete post with caption, CA, and links"""
if caption is None:
caption = self.get_meme_caption()
post = POST_TEMPLATE.format(
caption=caption,
ca=self.config["ca"],
website=self.config["website"],
)
if len(post) > 280:
max_len = 280 - len(POST_TEMPLATE.format(caption="", ca=self.config["ca"], website=self.config["website"]))
caption = caption[:max_len - 3] + "..."
post = POST_TEMPLATE.format(caption=caption, ca=self.config["ca"], website=self.config["website"])
return post
# =============================================================================
# THREAD CONTINUATION - POST LONG RESPONSES AS THREADS
# =============================================================================
async def split_for_thread(text: str, max_chars: int = 3800) -> List[str]:
"""
Split a long response into multiple tweets for thread continuation.
X Premium allows 4000 chars, but we use 3800 to be safe.
Splits at sentence boundaries when possible.
"""
if len(text) <= max_chars:
return [text]
parts = []
remaining = text
part_num = 1
while remaining:
if len(remaining) <= max_chars:
parts.append(remaining)
break
# Find best split point (sentence boundary)
split_point = max_chars
for delimiter in ['. ', '.\n', '! ', '?\n', '```\n', '\n\n']:
last_delim = remaining[:max_chars].rfind(delimiter)
if last_delim > max_chars * 0.5: # At least halfway through
split_point = last_delim + len(delimiter)
break
# Add part indicator for middle parts
part = remaining[:split_point].strip()
if part_num > 1:
part = f"[{part_num}/...] " + part
parts.append(part)
remaining = remaining[split_point:].strip()
part_num += 1
# Update part numbers now that we know total
total = len(parts)
if total > 1:
parts = [f"[{i+1}/{total}] " + p if i > 0 else p for i, p in enumerate(parts)]
return parts
# =============================================================================
# GLOBAL INSTANCE
# =============================================================================
_posting_brain: Optional[PostingBrain] = None
def get_posting_brain() -> PostingBrain:
"""Get or create the global posting brain instance"""
global _posting_brain
if _posting_brain is None:
_posting_brain = PostingBrain()
return _posting_brain
