src.ai package#
Marcus AI Integration Package.
Provides AI-enhanced intelligence while preserving rule-based safety guarantees. Implements hybrid decision making that combines deterministic logic with semantic understanding and intelligent optimization.
- class src.ai.MarcusAIEngine[source]#
Bases:
objectCentral AI coordination engine implementing hybrid intelligence.
Combines rule-based safety guarantees with AI-powered semantic understanding and intelligent optimization while ensuring rules are never overridden.
The engine follows a strict precedence model: 1. Rule-based validation (mandatory, safety-critical) 2. AI enhancement (optional, additive only) 3. Hybrid confidence calculation (weighted combination)
- Parameters:
None
- llm_client#
AI provider abstraction for semantic analysis
- Type:
- rule_engine#
Rule-based validation engine
- Type:
- hybrid_coordinator#
Coordinates hybrid decision making
- Type:
- analyze_with_hybrid_intelligence(context)[source]#
Perform hybrid analysis on task assignment
- Parameters:
context (AnalysisContext)
- Return type:
- analyze_blocker(task_id, blocker_description, severity, agent, task)[source]#
Analyze blockers and suggest solutions
Examples
>>> engine = MarcusAIEngine() >>> # Analyze task assignment >>> result = await engine.analyze_with_hybrid_intelligence(context) >>> if result.allow_assignment: ... print(f"Assignment allowed with {result.confidence:.0%} confidence") ... >>> # Get AI suggestions for blocker >>> suggestions = await engine.analyze_blocker( ... task_id="task-123", ... blocker_description="Database connection failed", ... severity="high", ... agent=agent_info, ... task=blocked_task ... )
Notes
AI enhancement is controlled by the MARCUS_AI_ENABLED environment variable. The engine always falls back to rule-based decisions if AI fails, ensuring system reliability.
- async analyze_with_hybrid_intelligence(context)[source]#
Perform hybrid analysis combining rule-based safety with AI intelligence.
Executes a multi-step analysis process: 1. Rule-based validation (mandatory) 2. AI enhancement (if enabled and rules pass) 3. Confidence calculation (weighted combination)
- Parameters:
context (
AnalysisContext) – Analysis context containing: - task: Task to analyze for assignment - project_context: Project state and metadata- Returns:
Complete analysis result containing: - allow_assignment: Final decision on assignment - confidence: Combined confidence score (0.0-1.0) - reason: Human-readable explanation - ai_confidence: AI model confidence (if available) - ai_insights: Detailed AI analysis (if available) - fallback_mode: Whether AI was unavailable - confidence_breakdown: Component confidence scores
- Return type:
- Raises:
Exception – Only if AI fails and fallback_on_ai_failure is False
Notes
Rule violations always result in assignment rejection, regardless of AI analysis. AI can only enhance allowed assignments, never override safety rules.
Examples
>>> context = AnalysisContext( ... task=Task(id="1", name="Deploy to production"), ... project_context={"available_tasks": tasks} ... ) >>> result = await engine.analyze_with_hybrid_intelligence(context) >>> if not result.allow_assignment: ... print(f"Blocked: {result.reason}")
- async enhance_task_with_ai(task, context)[source]#
Enhance a task with AI-generated improvements.
Uses AI to generate enhanced descriptions and effort estimates that provide more context and clarity for task execution.
- Parameters:
- Returns:
Enhanced task data containing: - enhanced_description: Improved task description - ai_effort_estimate: AI-based effort estimation - enhancement_confidence: Confidence in enhancements
- Return type:
Notes
Returns empty dict if AI is disabled or enhancement fails. Enhancements are suggestions only and don’t affect task execution.
Examples
>>> enhancements = await engine.enhance_task_with_ai( ... task=Task(name="Setup CI"), ... context=project_context ... ) >>> print(enhancements.get('enhanced_description'))
- async analyze_blocker(task_id, blocker_description, severity, agent, task)[source]#
Analyze a blocker and suggest solutions using AI.
Uses AI to understand the blocker context and generate actionable suggestions for resolution. Falls back to generic suggestions if AI is unavailable.
- Parameters:
- Returns:
Prioritized list of suggested solutions
- Return type:
Notes
Always returns at least 3 suggestions, even if AI fails. High-severity blockers receive more detailed analysis.
Examples
>>> suggestions = await engine.analyze_blocker( ... task_id="task-123", ... blocker_description="PostgreSQL connection timeout", ... severity="high", ... agent={"id": "agent-1", "name": "Backend Dev"}, ... task=database_task ... ) >>> for i, suggestion in enumerate(suggestions, 1): ... print(f"{i}. {suggestion}")
- async get_engine_status()[source]#
Get current AI engine status and configuration.
- Returns:
Engine status containing: - ai_enabled: Whether AI enhancement is active - llm_provider: Current LLM provider name - fallback_mode: Whether fallback is enabled - safety_override_disabled: Confirms rules can’t be overridden - components: Status of each engine component
- Return type:
Examples
>>> status = await engine.get_engine_status() >>> print(f"AI enabled: {status['ai_enabled']}") >>> print(f"Provider: {status['llm_provider']}")
- class src.ai.LLMAbstraction[source]#
Bases:
objectMulti-provider LLM abstraction with intelligent fallback.
Supports multiple LLM providers with automatic fallback when primary fails. Provides a unified interface for all AI operations in Marcus.
- analyze_task_semantics(task, context)[source]#
Analyze task meaning and intent
- Parameters:
- Return type:
- infer_dependencies_semantic(tasks)[source]#
Infer logical dependencies between tasks
- Parameters:
- Return type:
- estimate_effort_intelligently(task, context)[source]#
AI-powered effort estimation
- Parameters:
- Return type:
- analyze_blocker_and_suggest_solutions(task, blocker, severity, agent)[source]#
Analyze blockers and provide solutions
Notes
Providers are initialized lazily to avoid circular imports. Statistics are tracked for intelligent provider selection.
- async analyze_task_semantics(task, context)[source]#
Analyze task semantics using the best available provider.
- Parameters:
- Returns:
Comprehensive semantic analysis including intent and risks
- Return type:
Notes
Automatically falls back to alternative providers on failure. Tagged
analyze_task_semanticsvia_tagged_operation()so the resultingtoken_events.operationrow carries that label instead of the provider’s default.
- async infer_dependencies_semantic(tasks)[source]#
Infer semantic dependencies between tasks.
- Parameters:
- Returns:
Inferred logical relationships between tasks
- Return type:
Notes
Complements rule-based dependency detection with semantic understanding. Tagged
infer_dependenciesvia_tagged_operation().
- async estimate_effort_intelligently(task, context)[source]#
Estimate task effort using AI.
- Parameters:
- Returns:
AI-powered time estimate with confidence and factors
- Return type:
- async analyze_blocker_and_suggest_solutions(task, blocker_description, severity, agent)[source]#
Analyze a blocker and suggest solutions.
- Parameters:
- Returns:
Prioritized list of solution suggestions
- Return type:
Notes
Higher severity blockers receive more detailed analysis.
- async analyze(prompt, context, *, operation=None)[source]#
Analyze content using LLM.
- Parameters:
prompt (
str) – The prompt to analyze.context (
Any) – Analysis context (may carrymax_tokensoverride).operation (
Optional[str]) – Logical operation label to attach to the cost event. When provided, the recorder’s active PlannerContext is shadowed with anoperation_overridefor the duration of the call, sotoken_events.operationrecordsoperationinstead of whatever default the provider stamps. Used for per-call drill-down in the Cato cost dashboard. Seesrc/cost_tracking/operations.pyfor the canonical taxonomy and human-readable descriptions.
- Returns:
Analysis result as string.
- Return type:
- class src.ai.HybridDecisionFramework[source]#
Bases:
objectHybrid decision framework that merges rule-based safety with AI optimization.
Core principle: Rules provide mandatory safety constraints that AI cannot override. AI provides optimization and enhancement when rules allow the assignment.
- async make_assignment_decision(task, context)[source]#
Make hybrid assignment decision combining rules with AI optimization.
- Parameters:
task (
Task)context (
AssignmentContext)
- Return type:
- Returns:
Assignment decision with reasoning and AI enhancements
- async evaluate_assignment_quality(task, agent_id, assignment_outcome)[source]#
Evaluate the quality of a completed assignment for learning.
- class src.ai.HybridAnalysis[source]#
Bases:
objectResult of hybrid rule-based + AI analysis.
Combines deterministic rules with AI enhancement for optimal decisions.
- ai_insights#
Detailed AI analysis if available
- Type:
Examples
>>> result = HybridAnalysis( ... allow_assignment=True, ... confidence=0.85, ... reason="All checks passed with high confidence" ... )
- ai_insights: AIInsights | None = None#
- __init__(allow_assignment, confidence, reason, safety_critical=False, ai_confidence=None, ai_insights=None, fallback_mode=False, confidence_breakdown=None, optimization_score=None)#
- class src.ai.AnalysisContext[source]#
Bases:
objectContext for AI analysis operations.
Provides all necessary information for making intelligent task assignment decisions.
Examples
>>> context = AnalysisContext( ... task=task_to_assign, ... project_context={"available_tasks": tasks}, ... historical_data=performance_history ... )
- class src.ai.AssignmentDecision[source]#
Bases:
objectFinal decision on task assignment.
Represents the complete decision including rule validation, AI enhancement, and audit information.
- ai_suggestions#
AI optimization suggestions if rules passed
- Type:
- timestamp#
When the decision was made (auto-set)
- Type:
datetime
Examples
>>> decision = AssignmentDecision( ... allow=True, ... confidence=0.92, ... reason="All validations passed" ... ) >>> print(f"Decision at {decision.timestamp}: {decision.allow}")
- ai_suggestions: AIOptimizationResult | None = None#
- __init__(allow, confidence, reason, ai_suggestions=None, optimization_score=None, confidence_breakdown=None, safety_critical=False, mandatory_rule_applied=False, timestamp=None)#
Subpackages#
- src.ai.advanced package
- src.ai.core package
MarcusAIEngineMarcusAIEngine.llm_clientMarcusAIEngine.rule_engineMarcusAIEngine.hybrid_coordinatorMarcusAIEngine.ai_enabledMarcusAIEngine.fallback_on_ai_failureMarcusAIEngine.rule_safety_overrideMarcusAIEngine.analyze_with_hybrid_intelligence()MarcusAIEngine.enhance_task_with_ai()MarcusAIEngine.analyze_blocker()MarcusAIEngine.get_engine_status()MarcusAIEngine.__init__()MarcusAIEngine.analyze_with_hybrid_intelligence()MarcusAIEngine.enhance_task_with_ai()MarcusAIEngine.analyze_blocker()MarcusAIEngine.get_engine_status()
HybridAnalysisHybridAnalysis.allow_assignmentHybridAnalysis.confidenceHybridAnalysis.reasonHybridAnalysis.safety_criticalHybridAnalysis.ai_confidenceHybridAnalysis.ai_insightsHybridAnalysis.fallback_modeHybridAnalysis.confidence_breakdownHybridAnalysis.optimization_scoreHybridAnalysis.allow_assignmentHybridAnalysis.confidenceHybridAnalysis.reasonHybridAnalysis.safety_criticalHybridAnalysis.ai_confidenceHybridAnalysis.ai_insightsHybridAnalysis.fallback_modeHybridAnalysis.confidence_breakdownHybridAnalysis.optimization_scoreHybridAnalysis.__init__()
AnalysisContextAnalysisContext.taskAnalysisContext.project_contextAnalysisContext.historical_dataAnalysisContext.team_contextAnalysisContext.constraintsAnalysisContext.taskAnalysisContext.project_contextAnalysisContext.historical_dataAnalysisContext.team_contextAnalysisContext.constraintsAnalysisContext.__init__()
- Submodules
- src.ai.decisions package
HybridDecisionFrameworkAssignmentDecisionAssignmentDecision.allowAssignmentDecision.confidenceAssignmentDecision.reasonAssignmentDecision.ai_suggestionsAssignmentDecision.optimization_scoreAssignmentDecision.confidence_breakdownAssignmentDecision.safety_criticalAssignmentDecision.mandatory_rule_appliedAssignmentDecision.timestampAssignmentDecision.__post_init__()AssignmentDecision.allowAssignmentDecision.confidenceAssignmentDecision.reasonAssignmentDecision.ai_suggestionsAssignmentDecision.optimization_scoreAssignmentDecision.confidence_breakdownAssignmentDecision.safety_criticalAssignmentDecision.mandatory_rule_appliedAssignmentDecision.timestampAssignmentDecision.__post_init__()AssignmentDecision.__init__()
- Submodules
- src.ai.enrichment package
IntelligentTaskEnricherEnhancementResultEnhancementResult.original_taskEnhancementResult.enhanced_descriptionEnhancementResult.suggested_labelsEnhancementResult.estimated_hoursEnhancementResult.suggested_dependenciesEnhancementResult.acceptance_criteriaEnhancementResult.risk_factorsEnhancementResult.confidenceEnhancementResult.reasoningEnhancementResult.changes_madeEnhancementResult.enhancement_timestampEnhancementResult.__post_init__()EnhancementResult.__init__()
- Submodules
- src.ai.learning package
- src.ai.providers package
BaseLLMProviderBaseLLMProvider.analyze_task()BaseLLMProvider.infer_dependencies()BaseLLMProvider.generate_enhanced_description()BaseLLMProvider.estimate_effort()BaseLLMProvider.analyze_blocker()BaseLLMProvider.analyze_task()BaseLLMProvider.infer_dependencies()BaseLLMProvider.generate_enhanced_description()BaseLLMProvider.estimate_effort()BaseLLMProvider.analyze_blocker()
SemanticAnalysisSemanticAnalysis.task_intentSemanticAnalysis.semantic_dependenciesSemanticAnalysis.risk_factorsSemanticAnalysis.suggestionsSemanticAnalysis.confidenceSemanticAnalysis.reasoningSemanticAnalysis.risk_assessmentSemanticAnalysis.fallback_usedSemanticAnalysis.task_intentSemanticAnalysis.semantic_dependenciesSemanticAnalysis.risk_factorsSemanticAnalysis.suggestionsSemanticAnalysis.confidenceSemanticAnalysis.reasoningSemanticAnalysis.risk_assessmentSemanticAnalysis.fallback_usedSemanticAnalysis.__init__()
SemanticDependencySemanticDependency.dependent_task_idSemanticDependency.dependency_task_idSemanticDependency.confidenceSemanticDependency.reasoningSemanticDependency.dependency_typeSemanticDependency.dependent_task_idSemanticDependency.dependency_task_idSemanticDependency.confidenceSemanticDependency.reasoningSemanticDependency.dependency_typeSemanticDependency.__init__()
EffortEstimateEffortEstimate.estimated_hoursEffortEstimate.confidenceEffortEstimate.factorsEffortEstimate.similar_tasksEffortEstimate.risk_multiplierEffortEstimate.estimated_hoursEffortEstimate.confidenceEffortEstimate.factorsEffortEstimate.similar_tasksEffortEstimate.risk_multiplierEffortEstimate.__init__()
CloudLLMProviderLLMAbstractionLLMAbstraction.providersLLMAbstraction.current_providerLLMAbstraction.fallback_providersLLMAbstraction.provider_statsLLMAbstraction.analyze_task_semantics()LLMAbstraction.infer_dependencies_semantic()LLMAbstraction.generate_enhanced_description()LLMAbstraction.estimate_effort_intelligently()LLMAbstraction.analyze_blocker_and_suggest_solutions()LLMAbstraction.__init__()LLMAbstraction.analyze_task_semantics()LLMAbstraction.infer_dependencies_semantic()LLMAbstraction.generate_enhanced_description()LLMAbstraction.estimate_effort_intelligently()LLMAbstraction.analyze_blocker_and_suggest_solutions()LLMAbstraction.analyze()LLMAbstraction.switch_provider()LLMAbstraction.get_provider_stats()LLMAbstraction.get_best_provider()LLMAbstraction.health_check()
- Submodules
Submodules#
- src.ai.types module
- Classes
AnalysisContextAnalysisContext.taskAnalysisContext.project_contextAnalysisContext.historical_dataAnalysisContext.team_contextAnalysisContext.constraintsAnalysisContext.taskAnalysisContext.project_contextAnalysisContext.historical_dataAnalysisContext.team_contextAnalysisContext.constraintsAnalysisContext.__init__()
AIInsightsAIInsights.task_intentAIInsights.semantic_dependenciesAIInsights.risk_factorsAIInsights.suggestionsAIInsights.confidenceAIInsights.reasoningAIInsights.risk_assessmentAIInsights.task_intentAIInsights.semantic_dependenciesAIInsights.risk_factorsAIInsights.suggestionsAIInsights.confidenceAIInsights.reasoningAIInsights.risk_assessmentAIInsights.__init__()
HybridAnalysisHybridAnalysis.allow_assignmentHybridAnalysis.confidenceHybridAnalysis.reasonHybridAnalysis.safety_criticalHybridAnalysis.ai_confidenceHybridAnalysis.ai_insightsHybridAnalysis.fallback_modeHybridAnalysis.confidence_breakdownHybridAnalysis.optimization_scoreHybridAnalysis.allow_assignmentHybridAnalysis.confidenceHybridAnalysis.reasonHybridAnalysis.safety_criticalHybridAnalysis.ai_confidenceHybridAnalysis.ai_insightsHybridAnalysis.fallback_modeHybridAnalysis.confidence_breakdownHybridAnalysis.optimization_scoreHybridAnalysis.__init__()
RuleBasedResultAIOptimizationResultAIOptimizationResult.confidenceAIOptimizationResult.optimization_scoreAIOptimizationResult.improvementsAIOptimizationResult.semantic_confidenceAIOptimizationResult.risk_mitigationAIOptimizationResult.estimated_completion_timeAIOptimizationResult.confidenceAIOptimizationResult.optimization_scoreAIOptimizationResult.improvementsAIOptimizationResult.semantic_confidenceAIOptimizationResult.risk_mitigationAIOptimizationResult.estimated_completion_timeAIOptimizationResult.__init__()
AssignmentDecisionAssignmentDecision.allowAssignmentDecision.confidenceAssignmentDecision.reasonAssignmentDecision.ai_suggestionsAssignmentDecision.optimization_scoreAssignmentDecision.confidence_breakdownAssignmentDecision.safety_criticalAssignmentDecision.mandatory_rule_appliedAssignmentDecision.timestampAssignmentDecision.__post_init__()AssignmentDecision.allowAssignmentDecision.confidenceAssignmentDecision.reasonAssignmentDecision.ai_suggestionsAssignmentDecision.optimization_scoreAssignmentDecision.confidence_breakdownAssignmentDecision.safety_criticalAssignmentDecision.mandatory_rule_appliedAssignmentDecision.timestampAssignmentDecision.__post_init__()AssignmentDecision.__init__()
AssignmentContextAssignmentContext.taskAssignmentContext.agent_idAssignmentContext.agent_statusAssignmentContext.available_tasksAssignmentContext.project_contextAssignmentContext.team_statusAssignmentContext.taskAssignmentContext.agent_idAssignmentContext.agent_statusAssignmentContext.available_tasksAssignmentContext.project_contextAssignmentContext.team_statusAssignmentContext.__init__()