AGInt (Augmentic Intelligence) represents the cognitive heart of the mindX system - a sophisticated artificial intelligence core that implements the classical P-O-D-A (Perceive-Orient-Decide-Act) cognitive cycle with modern AI capabilities. As the central orchestrator of intelligent behavior, AGInt bridges the gap between high-level strategic reasoning and operational execution through dynamic model selection, adaptive decision-making, and seamless integration with the BDI (Belief-Desire-Intention) agent framework.
AGInt embodies the concept of "Augmentic Intelligence" - intelligence that augments and amplifies human cognitive capabilities. Combining systematic observation, analysis, decision-making, and action execution. This sequencial approach creates a robust, adaptable AGInt system capable of autonomous operation while maintaining transparency and interpretability in its cognitive processes.
The AGInt system is built on the foundational principle that effective artificial intelligence requires a structured cognitive architecture that mirrors proven decision-making frameworks. The P-O-D-A cycle, originally developed for military and strategic applications, provides this structure:
AGInt operates as the central cognitive processor within the MindX ecosystem, managing:
Cognitive Loop Management: Continuous P-O-D-A cycle execution Dynamic Model Selection: Intelligent routing to appropriate LLM models based on task requirements Decision Tree Processing: Rule-based and AI-assisted decision making BDI Integration: Seamless delegation to Belief-Desire-Intention agents Self-Repair Capabilities: Autonomous system health monitoring and recovery Memory Integration: Persistent learning and experience storage
class AGInt:
"""
Augmentic Intelligence Core - P-O-D-A Cognitive Processor
Implements the Perceive-Orient-Decide-Act cognitive cycle with:
- Dynamic model selection and routing
- Rule-based and AI-assisted decision making
- BDI agent integration and task delegation
- Self-repair and health monitoring capabilities
- Memory integration for persistent learning
"""
class AgentStatus(Enum):
INACTIVE = "INACTIVE"
RUNNING = "RUNNING"
AWAITING_DIRECTIVE = "AWAITING_DIRECTIVE"
FAILED = "FAILED"
class DecisionType(Enum):
BDI_DELEGATION = "BDI_DELEGATION" # Delegate to BDI agent
RESEARCH = "RESEARCH" # Perform web research
COOLDOWN = "COOLDOWN" # Pause and recover
SELF_REPAIR = "SELF_REPAIR" # Execute system repair
IDLE = "IDLE" # No action required
PERFORM_TASK = "PERFORM_TASK" # Direct task execution
SELF_IMPROVEMENT = "SELF_IMPROVEMENT" # Learning and optimization
STRATEGIC_EVOLUTION = "STRATEGIC_EVOLUTION" # System evolution
self.state_summary = {
"llm_operational": True,
"awareness": "",
"last_decision": None,
"cycle_count": 0
}
_perceive())async def _perceive(self) -> Dict[str, Any]:
"""
Gathers environmental information and system state.
Includes failure context from previous actions for adaptive learning.
"""
perception_data = {"timestamp": time.time()}
if self.last_action_context and not self.last_action_context.get('success'):
perception_data['last_action_failure_context'] = self.last_action_context.get('result')
return perception_data
Key Features:
_orient_and_decide())async def _orient_and_decide(self, perception: Dict[str, Any]) -> Dict[str, Any]:
"""
Analyzes perception data and formulates decision strategy.
Combines rule-based logic with AI-powered situational analysis.
"""
decision_type = await self._decide_rule_based(perception)
# AI-powered situational awareness and decision detail formulation
prompt = f"As an AI core, your directive is '{self.primary_directive}'..."
response = await self._execute_cognitive_task(prompt, TaskType.REASONING, json_mode=True)
Key Features:
_decide_rule_based())async def _decide_rule_based(self, perception: Dict[str, Any]) -> DecisionType:
"""
Implements deterministic decision tree for system reliability.
Prioritizes system health and failure recovery.
"""
if not self.state_summary.get("llm_operational", True):
return DecisionType.SELF_REPAIR
elif perception.get('last_action_failure_context'):
return DecisionType.RESEARCH
else:
return DecisionType.BDI_DELEGATION
Decision Priority Matrix: System Health: Self-repair if LLM systems are non-operational Failure Recovery: Research and analysis if previous action failed Normal Operation: Delegate to BDI agent for task execution
_act())async def _act(self, decision: Dict[str, Any]) -> Tuple[bool, Any]:
"""
Routes decisions to appropriate execution functions.
Provides comprehensive action logging and result tracking.
"""
action_map = {
DecisionType.BDI_DELEGATION: lambda: self._delegate_task_to_bdi(details.get("task_description")),
DecisionType.RESEARCH: lambda: self._execute_research(details.get("search_query")),
DecisionType.SELF_REPAIR: self._execute_self_repair,
DecisionType.COOLDOWN: self._execute_cooldown,
}
AGInt implements sophisticated model selection logic that:
async def _execute_cognitive_task(self, prompt: str, task_type: TaskType, kwargs) -> Optional[str]:
"""
Executes cognitive tasks with intelligent model selection and fallback.
"""
all_capabilities = list(self.model_registry.capabilities.values())
ranked_models = self.model_registry.model_selector.select_model(all_capabilities, task_type)
for model_id in valid_models:
try:
handler = self.model_registry.get_handler(self.model_registry.capabilities[model_id].provider)
response_str = await handler.generate_text(prompt, model=model_id, kwargs)
if kwargs.get("json_mode"):
json.loads(response_str) # Validate JSON format
return response_str
except Exception as e:
continue # Try next model
self.state_summary["llm_operational"] = False
return None
AGInt includes sophisticated self-repair capabilities:
async def _execute_self_repair(self) -> Tuple[bool, Any]:
"""
Executes comprehensive self-repair with mandatory verification.
"""
# Coordinate system analysis
interaction = await self.coordinator_agent.create_interaction(
InteractionType.SYSTEM_ANALYSIS,
"Automated self-repair triggered."
)
# Force model registry reload
await self.model_registry.force_reload()
# Verify LLM connectivity
verification_result = await self._execute_cognitive_task(
"Status check. Respond ONLY with 'OK'.",
TaskType.HEALTH_CHECK
)
# Update system state
if verification_result and "OK" in verification_result:
self.state_summary["llm_operational"] = True
return True, {"message": "Self-repair verification successful."}
AGInt maintains comprehensive memory integration:
AGInt represents a breakthrough in cognitive AI architecture, combining the proven P-O-D-A decision-making framework with modern AI capabilities. Its sophisticated design enables:
The system's strength lies in its hybrid approach - combining deterministic rule-based logic for critical system functions with AI-powered reasoning for complex situational analysis. This creates a robust, reliable, and intelligent system capable of autonomous operation while maintaining transparency and interpretability.
from core.agint import AGInt
from core.bdi_agent import BDIAgent
from llm.model_registry import ModelRegistry
from utils.config import Config
Initialize core components
config = Config()
model_registry = ModelRegistry(config)
bdi_agent = BDIAgent(agent_id="bdi_main", config=config)
Create AGInt instance
agint = AGInt(
agent_id="agint_main",
bdi_agent=bdi_agent,
model_registry=model_registry,
config=config,
coordinator_agent=coordinator,
memory_agent=memory_agent,
web_search_tool=web_search
)
# Start AGInt with a primary directive
agint.start("Optimize system performance and handle user requests efficiently")
The cognitive loop will begin automatically:
Perceive current system state
Orient and analyze situation
Decide on optimal action
Act on the decision
Repeat cycle
# Configure cognitive cycle timing
config.set("agint.cycle_delay_seconds", 3.0)
Configure learning parameters
config.set("agint.learning.alpha", 0.1) # Learning rate
config.set("agint.learning.gamma", 0.9) # Discount factor
Configure self-repair settings
config.set("agint.llm_failure_cooldown_seconds", 30)
Configure model selection preferences
config.set("agint.preferred_models", ["gemini/gemini-1.5-flash", "groq/llama3-70b"])
# Check AGInt status
print(f"Status: {agint.status}")
print(f"Cycle Count: {agint.state_summary['cycle_count']}")
print(f"LLM Operational: {agint.state_summary['llm_operational']}")
Access current awareness
print(f"Situational Awareness: {agint.state_summary['awareness']}")
Review decision history
print(f"Last Decision: {agint.state_summary['last_decision']}")
Stop the cognitive loop
await agint.stop()
# AGInt automatically delegates complex tasks to BDI agent
The BDI agent handles belief-desire-intention reasoning
Results are fed back to AGInt for learning and adaptation
# AGInt logs all cognitive processes to memory
Memory entries include:
- Perception data and timestamps
- Decision rationale and details
- Action outcomes and success/failure status
- Learning updates and Q-value changes
Access memory logs
memory_logs = await memory_agent.get_agent_memories("agint_main")
for log in memory_logs:
print(f"Process: {log['content']['process_name']}")
print(f"Data: {log['content']['data']}")
# AGInt can autonomously perform web research
When RESEARCH decision is made:
Formulates search query based on context
Executes web search using integrated tools
Analyzes results and updates knowledge base
Feeds insights back into decision-making process
# AGInt includes comprehensive error handling:
LLM Failure Recovery
- Detects non-responsive models
- Attempts fallback to alternative models
- Triggers self-repair if all models fail
Decision Validation
- Validates AI responses for proper format
- Falls back to rule-based decisions if AI fails
- Logs all failures for analysis
System Health Monitoring
- Continuously monitors system components
- Triggers coordinator-based system analysis
- Performs verification after repairs
# AGInt includes several performance optimization features:
Model Caching
- Caches model handlers for faster access
- Reuses connections where possible
Asynchronous Processing
- Non-blocking cognitive operations
- Concurrent task execution where appropriate
Adaptive Timing
- Configurable cycle delays
- Dynamic adjustment based on system load
Memory Efficiency
- Structured logging with importance levels
- Automatic cleanup of old entries
- Efficient state representation
# AGInt can be extended with custom decision types and actions:
class CustomDecisionType(Enum):
CUSTOM_ACTION = "CUSTOM_ACTION"
Add custom action handler
async def _execute_custom_action(self) -> Tuple[bool, Any]:
# Custom action implementation
return True, {"message": "Custom action completed"}
Extend action map
agint.action_map[CustomDecisionType.CUSTOM_ACTION] = agint._execute_custom_action
Directive Clarity: Provide clear, specific primary directives. Configuration Tuning: Adjust cycle timing based on use case requirements Memory Management: Monitor memory usage and configure cleanup policies Model Selection: Configure preferred models based on task requirements Error Monitoring: Regularly review error logs and failure patterns Performance Tracking: Monitor cycle times and decision success rates Integration Testing: Verify proper integration with all system components
Common Issues and Solutions:
LLM Connectivity Issues - Check model registry configuration - Verify API keys and endpoints - Review network connectivity
Slow Cognitive Cycles - Adjust cycle delay settings - Optimize model selection preferences - Review system resource usage
Decision Making Problems - Review rule-based logic - Check AI prompt formulation - Verify BDI agent integration
Memory Integration Issues - Verify memory agent configuration - Check logging permissions - Review memory storage paths
AGInt represents the pinnacle of cognitive AI architecture within the MindX system, providing robust, reliable, and intelligent decision-making capabilities forming the foundation for all higher-level AI operations.