src.ai.providers.llm_abstraction module#

LLM Abstraction Layer for Marcus AI.

Provides a unified interface across different LLM providers (Anthropic, OpenAI, local models) with intelligent fallback and provider switching capabilities.

This module implements the strategy pattern for LLM providers, allowing seamless switching between providers and automatic fallback on failures.

Classes#

LLMAbstraction: Multi-provider LLM abstraction with intelligent fallback

Notes

Provider selection is controlled by the MARCUS_LLM_PROVIDER environment variable. The system automatically falls back to alternative providers on failure.

Examples

>>> llm = LLMAbstraction()
>>> analysis = await llm.analyze_task_semantics(task, context)
>>> if analysis.fallback_used:
...     print("Using fallback provider")

class src.ai.providers.llm_abstraction.LLMAbstraction[source]#

Bases: object

Multi-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.

providers#

Available LLM provider instances

Type:: dict

current_provider#

Name of the primary provider to use

Type:: str

fallback_providers#

Ordered list of providers to try on failure

Type:: list of str

provider_stats#

Performance statistics for each provider

Type:: dict

analyze_task_semantics(task, context)[source]#

Analyze task meaning and intent

Parameters:

task (Task)
context (Dict[str, Any])

Return type:

SemanticAnalysis

infer_dependencies_semantic(tasks)[source]#

Infer logical dependencies between tasks

Parameters:: tasks (List[Task])
Return type:: List[SemanticDependency]

generate_enhanced_description(task, context)[source]#

Create improved task descriptions

Parameters:

task (Task)
context (Dict[str, Any])

Return type:

str

estimate_effort_intelligently(task, context)[source]#

AI-powered effort estimation

Parameters:

task (Task)
context (Dict[str, Any])

Return type:

EffortEstimate

analyze_blocker_and_suggest_solutions(task, blocker, severity, agent)[source]#

Analyze blockers and provide solutions

Parameters:

task (Task)
blocker_description (str)
severity (str)
agent (Dict[str, Any] | None)

Return type:

List[str]

Notes

Providers are initialized lazily to avoid circular imports. Statistics are tracked for intelligent provider selection.

__init__()[source]#

Return type:: None

async analyze_task_semantics(task, context)[source]#

Analyze task semantics using the best available provider.

Parameters:

task (Task) – Task to analyze for semantic meaning
context (Dict[str, Any]) – Project context including related tasks

Returns:

Comprehensive semantic analysis including intent and risks

Return type:

SemanticAnalysis

Notes

Automatically falls back to alternative providers on failure. Tagged analyze_task_semantics via _tagged_operation() so the resulting token_events.operation row carries that label instead of the provider’s default.

async infer_dependencies_semantic(tasks)[source]#

Infer semantic dependencies between tasks.

Parameters:: tasks (List[Task]) – All tasks to analyze for dependencies
Returns:: Inferred logical relationships between tasks
Return type:: List[SemanticDependency]

Notes

Complements rule-based dependency detection with semantic understanding. Tagged infer_dependencies via _tagged_operation().

async generate_enhanced_description(task, context)[source]#

Generate enhanced task description.

Parameters:

task (Task) – Task needing clearer description
context (Dict[str, Any]) – Project context for better understanding

Returns:

Enhanced description with more detail and clarity

Return type:

str

async estimate_effort_intelligently(task, context)[source]#

Estimate task effort using AI.

Parameters:

task (Task) – Task to estimate completion time for
context (Dict[str, Any]) – Project context with historical performance data

Returns:

AI-powered time estimate with confidence and factors

Return type:

EffortEstimate

async analyze_blocker_and_suggest_solutions(task, blocker_description, severity, agent)[source]#

Analyze a blocker and suggest solutions.

Parameters:

task (Task) – The blocked task
blocker_description (str) – Detailed description of the blocker
severity (str) – Severity level: ‘low’, ‘medium’, or ‘high’
agent (Optional[Dict[str, Any]]) – Agent information for context

Returns:

Prioritized list of solution suggestions

Return type:

List[str]

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 carry max_tokens override).
operation (Optional[str]) – Logical operation label to attach to the cost event. When provided, the recorder’s active PlannerContext is shadowed with an operation_override for the duration of the call, so token_events.operation records operation instead of whatever default the provider stamps. Used for per-call drill-down in the Cato cost dashboard. See src/cost_tracking/operations.py for the canonical taxonomy and human-readable descriptions.

Returns:

Analysis result as string.

Return type:

str

async switch_provider(provider_name)[source]#

Switch to a different provider.

Parameters:: provider_name (str)
Return type:: bool
Returns:: True if switch successful, False otherwise

get_provider_stats()[source]#

Get performance statistics for all providers.

Return type:: Dict[str, Any]

get_best_provider()[source]#

Determine the best performing provider based on success rate.

Return type:: str
Returns:: Name of best performing provider

async health_check()[source]#

Check health of all providers.

Return type:: Dict[str, Any]
Returns:: Health status for each provider