Spaces:
Sleeping
Sleeping
File size: 3,934 Bytes
ee2e23e 9f542a1 ee2e23e 6d6b8af |
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 |
"""Lightweight AI core for identity scanning and analysis.
This module provides a compact, dependency-safe `AICore` class used by
other components. The original file contained merge markers and
incomplete code; this replacement focuses on providing functioning
interfaces (async `generate_response`, `shutdown`) and uses existing
local components where available.
"""
from typing import Any, Dict, List, Optional
import asyncio
import json
import logging
import os
from .multimodal_analyzer import MultimodalAnalyzer
from .dynamic_learning import DynamicLearner
from .health_monitor import HealthMonitor
try:
from ..utils.logger import logger
except Exception:
logger = logging.getLogger(__name__)
class AICore:
"""Minimal, safe AICore replacement for identity scanning workflows."""
def __init__(self, config_path: str = "config.json"):
self.config = self._load_config(config_path)
self.multimodal = MultimodalAnalyzer()
self.learner = DynamicLearner()
self.health = HealthMonitor()
self._initialized = False
def _load_config(self, path: str) -> Dict[str, Any]:
if not path or not os.path.exists(path):
return {}
try:
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
except Exception:
return {}
async def initialize(self) -> bool:
"""Initialize async subsystems (e.g., health monitor)."""
try:
ok = await self.health.initialize()
self._initialized = ok
return ok
except Exception as e:
logger.exception("Failed to initialize AICore: %s", e)
return False
async def generate_response(self, user_id: int, query: str, multimodal_input: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Produce a safe, explainable response using local subsystems.
- Runs a lightweight analysis of any multimodal input
- Updates the dynamic learner with a summary of the interaction
- Returns a dict containing analysis and a text response
"""
try:
analyses = {}
if multimodal_input:
analyses = self.multimodal.analyze_content(multimodal_input)
# Simple content-based reply
text_summary = ""
if "text" in analyses:
t = analyses["text"]
text_summary = f"Received text: length={t.get('length')} words={t.get('word_count')}"
else:
text_summary = f"Query received: {query[:200]}"
# Update learner with a compact interaction record
adaptation_score = self.learner.update({
"user_id": user_id,
"query_summary": text_summary,
"multimodal_modalities": list(analyses.keys())
})
# Health snapshot
health_status = self.health.get_health_summary()
response_text = f"Acknowledged. Adaptation score={adaptation_score:.2f}."
return {
"response": response_text,
"analysis": analyses,
"adaptation_score": adaptation_score,
"health": health_status,
}
except Exception as e:
logger.exception("generate_response failed: %s", e)
return {"error": "internal_error", "detail": str(e)}
async def shutdown(self):
"""Clean up resources if necessary."""
# HealthMonitor has no async shutdown but we keep the method for parity.
await asyncio.sleep(0)
# Module quick test when run directly
if __name__ == "__main__":
async def _test():
core = AICore()
await core.initialize()
out = await core.generate_response(1, "Hello world", {"text": "Hello world from test"})
print(out)
await core.shutdown()
asyncio.run(_test()) |