File size: 4,276 Bytes
559af1d |
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 |
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
# Check dependencies before importing AEGIS components
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)
# Register default agents
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}
}
}
# Dispatch to AEGIS council
self.council.dispatch(input_data)
# Get analysis results
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
"""
# Apply wisdom score to improve clarity
if virtue_profile.get("wisdom", 0) < 0.5:
response = self.ai_core.self_refine_response(response)
# Add empathetic framing for low compassion
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() |