|
|
import importlib
|
|
|
import logging
|
|
|
from typing import Dict, Any, Optional, List
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def check_dependencies() -> List[str]:
|
|
|
"""Check if all required packages are installed."""
|
|
|
required_packages = [
|
|
|
'plotly',
|
|
|
'torch',
|
|
|
'transformers',
|
|
|
'pandas',
|
|
|
'numpy',
|
|
|
'networkx'
|
|
|
]
|
|
|
missing = []
|
|
|
for package in required_packages:
|
|
|
try:
|
|
|
importlib.import_module(package)
|
|
|
except ImportError:
|
|
|
missing.append(package)
|
|
|
return missing
|
|
|
|
|
|
|
|
|
missing_packages = check_dependencies()
|
|
|
if missing_packages:
|
|
|
raise ImportError(
|
|
|
f"AEGIS requires the following packages to be installed: {', '.join(missing_packages)}. "
|
|
|
f"Please install them using: pip install {' '.join(missing_packages)}"
|
|
|
)
|
|
|
|
|
|
from .aegis import AegisCouncil, MetaJudgeAgent, TemporalAgent, VirtueAgent
|
|
|
import sys
|
|
|
from pathlib import Path
|
|
|
sys.path.append(str(Path(__file__).parent.parent))
|
|
|
from ai_core import AICore
|
|
|
|
|
|
class AegisBridge:
|
|
|
def __init__(self, ai_core: AICore, config: Dict[str, Any]):
|
|
|
self.ai_core = ai_core
|
|
|
self.council = AegisCouncil(config)
|
|
|
|
|
|
|
|
|
self.council.register_agent(MetaJudgeAgent("MetaJudgeAgent", self.council.memory, config["meta_judge_weights"]))
|
|
|
self.council.register_agent(TemporalAgent("TemporalAgent", self.council.memory, config["temporal_decay_thresholds"]))
|
|
|
virtue_agent = VirtueAgent("VirtueAgent", self.council.memory, config["virtue_weights"])
|
|
|
virtue_agent.set_federated_trainer(self.council.federated_trainer)
|
|
|
self.council.register_agent(virtue_agent)
|
|
|
|
|
|
def enhance_response(self, prompt: str, response: str) -> Dict[str, Any]:
|
|
|
"""
|
|
|
Enhance Codette's response using AEGIS analysis
|
|
|
"""
|
|
|
input_data = {
|
|
|
"text": response,
|
|
|
"overrides": {
|
|
|
"EthosiaAgent": {"influence": 0.7, "reliability": 0.8, "severity": 0.6},
|
|
|
"AegisCore": {"influence": 0.6, "reliability": 0.9, "severity": 0.7}
|
|
|
}
|
|
|
}
|
|
|
|
|
|
|
|
|
self.council.dispatch(input_data)
|
|
|
|
|
|
|
|
|
reports = self.council.get_reports()
|
|
|
virtue_profile = reports.get("VirtueAgent", {}).get("result", {}).get("virtue_profile", {})
|
|
|
temporal_analysis = reports.get("TemporalAgent", {}).get("result", {})
|
|
|
meta_scores = reports.get("MetaJudgeAgent", {}).get("result", {}).get("scores", [])
|
|
|
|
|
|
return {
|
|
|
"original_response": response,
|
|
|
"virtue_analysis": virtue_profile,
|
|
|
"temporal_analysis": temporal_analysis,
|
|
|
"meta_scores": meta_scores,
|
|
|
"enhanced_response": self._apply_enhancements(response, virtue_profile)
|
|
|
}
|
|
|
|
|
|
def _apply_enhancements(self, response: str, virtue_profile: Dict[str, float]) -> str:
|
|
|
"""
|
|
|
Apply virtue-based enhancements to the response
|
|
|
"""
|
|
|
|
|
|
if virtue_profile.get("wisdom", 0) < 0.5:
|
|
|
response = self.ai_core.self_refine_response(response)
|
|
|
|
|
|
|
|
|
if virtue_profile.get("compassion", 0) < 0.5:
|
|
|
response = self.ai_core.generate_text(
|
|
|
f"Make this response more empathetic while preserving its meaning: {response}",
|
|
|
perspective="human_intuition"
|
|
|
)
|
|
|
|
|
|
return response
|
|
|
|
|
|
def get_analysis_graphs(self) -> Dict[str, str]:
|
|
|
"""
|
|
|
Generate and return analysis visualizations
|
|
|
"""
|
|
|
try:
|
|
|
self.council.draw_explainability_graph("aegis_analysis.html")
|
|
|
return {
|
|
|
"explainability_graph": "aegis_analysis.html"
|
|
|
}
|
|
|
except Exception as e:
|
|
|
return {"error": str(e)}
|
|
|
|
|
|
def get_memory_state(self) -> Dict[str, Any]:
|
|
|
"""
|
|
|
Return the current state of AEGIS memory
|
|
|
"""
|
|
|
return self.council.memory.audit() |
