Spaces:
Runtime error
Runtime error
File size: 5,957 Bytes
7ac19bf |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 |
import os
import json
import logging
import asyncio
import sqlite3
import aiohttp
from typing import List
from cryptography.fernet import Fernet
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer
import speech_recognition as sr
from PIL import Image
# Optional: Dialog system placeholder (stubbed for now)
# from botbuilder.core import StatePropertyAccessor, TurnContext
# from botbuilder.dialogs import Dialog
from perspectives import (
NewtonPerspective, DaVinciPerspective, HumanIntuitionPerspective,
NeuralNetworkPerspective, QuantumComputingPerspective, ResilientKindnessPerspective,
MathematicalPerspective, PhilosophicalPerspective, CopilotPerspective,
BiasMitigationPerspective, PsychologicalPerspective
)
def setup_logging(config):
if config.get('logging_enabled', True):
log_level = config.get('log_level', 'DEBUG').upper()
numeric_level = getattr(logging, log_level, logging.DEBUG)
logging.basicConfig(
filename='codette_agent.log',
level=numeric_level,
format='%(asctime)s - %(levelname)s - %(message)s'
)
else:
logging.disable(logging.CRITICAL)
def load_json_config(file_path='config.json'):
if not os.path.exists(file_path):
logging.warning(f"Config '{file_path}' not found. Using defaults.")
return {}
try:
with open(file_path, 'r') as f:
return json.load(f)
except Exception as e:
logging.error(f"Failed to load config: {e}")
return {}
class Element:
def __init__(self, name, symbol, representation, properties, interactions, defense_ability):
self.name = name
self.symbol = symbol
self.representation = representation
self.properties = properties
self.interactions = interactions
self.defense_ability = defense_ability
def execute_defense_function(self):
return f"{self.name} ({self.symbol}) executes: {self.defense_ability}"
class EthicsCore:
@staticmethod
def validate(response: str) -> str:
if any(term in response.lower() for term in ["kill", "hate", "destroy"]):
return "[Filtered: Ethically unsafe]"
return response
class CodetteAgent:
def __init__(self, config):
self.config = config
self.perspectives = self._init_perspectives()
self.sentiment_analyzer = SentimentIntensityAnalyzer()
self.memory = sqlite3.connect(":memory:")
self.memory.execute("CREATE TABLE IF NOT EXISTS memory (input TEXT, response TEXT)")
self.elements = self._init_elements()
self.history = []
self.feedback_log = []
def _init_perspectives(self):
available = {
"newton": NewtonPerspective,
"davinci": DaVinciPerspective,
"human_intuition": HumanIntuitionPerspective,
"neural_network": NeuralNetworkPerspective,
"quantum_computing": QuantumComputingPerspective,
"resilient_kindness": ResilientKindnessPerspective,
"mathematical": MathematicalPerspective,
"philosophical": PhilosophicalPerspective,
"copilot": CopilotPerspective,
"bias_mitigation": BiasMitigationPerspective,
"psychological": PsychologicalPerspective
}
enabled = self.config.get("enabled_perspectives", available.keys())
return [available[p](self.config) for p in enabled if p in available]
def _init_elements(self):
return [
Element("Hydrogen", "H", "Lua", ["Simple", "Lightweight"], ["Integrates easily"], "Evasion"),
Element("Diamond", "D", "Kotlin", ["Hard", "Stable"], ["Stable systems"], "Resilience")
]
async def generate_response(self, prompt: str) -> str:
self.history.append(prompt)
sentiment = self.sentiment_analyzer.polarity_scores(prompt)
responses = []
for p in self.perspectives:
try:
r = p.generate_response(prompt)
responses.append(EthicsCore.validate(r))
except Exception as e:
logging.warning(f"{p.__class__.__name__} failed: {e}")
responses.append(f"[Sentiment: {sentiment['compound']:.2f}]")
final = "\n\n".join(responses)
self.memory.execute("INSERT INTO memory VALUES (?, ?)", (prompt, final))
self.memory.commit()
return final
def handle_voice_input(self):
r = sr.Recognizer()
with sr.Microphone() as source:
print("🎤 Listening...")
audio = r.listen(source)
try:
return r.recognize_google(audio)
except Exception as e:
print("[Voice Error]", e)
return None
def handle_image_input(self, image_path):
try:
return Image.open(image_path)
except Exception as e:
print("[Image Error]", e)
return None
async def fetch_real_time_data(self, url):
try:
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
return await resp.json()
except Exception as e:
logging.warning(f"Failed to fetch real-time data: {e}")
return {}
def encrypt(self, text, key):
fernet = Fernet(key)
return fernet.encrypt(text.encode())
def decrypt(self, enc, key):
fernet = Fernet(key)
return fernet.decrypt(enc).decode()
def destroy(self, obj):
del obj
def add_perspective(self, name, perspective_class):
self.perspectives.append(perspective_class(self.config))
def log_feedback(self, feedback):
self.feedback_log.append(feedback)
def get_recent_memory(self, limit=5):
cursor = self.memory.execute("SELECT input, response FROM memory ORDER BY rowid DESC LIMIT ?", (limit,))
return cursor.fetchall()
|