src.ai.providers package#

AI Provider Components.

class src.ai.providers.BaseLLMProvider[source]#

Bases: ABC

Base class for LLM providers.

Defines the interface that all LLM providers must implement to integrate with Marcus AI system.

analyze_task(task, context)[source]#

Perform semantic analysis on a task

Parameters:
Return type:

SemanticAnalysis

infer_dependencies(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:
Return type:

str

estimate_effort(task, context)[source]#

Estimate task completion time

Parameters:
Return type:

EffortEstimate

analyze_blocker(task, blocker, context)[source]#

Analyze blockers and suggest solutions

Parameters:
Return type:

List[str]

Notes

All methods are async to support various LLM APIs. Implementations should handle rate limiting and retries.

Examples

>>> provider = OpenAIProvider(api_key="...")
>>> analysis = await provider.analyze_task(task, context)
>>> print(f"Task intent: {analysis.task_intent}")
abstractmethod async analyze_task(task, context)[source]#

Analyze task semantics using LLM.

Parameters:

  • task (Task) – Task to analyze

  • context (Dict[str, Any]) – Project context including related tasks

Returns:

Comprehensive semantic analysis of the task

Return type:

SemanticAnalysis

abstractmethod async infer_dependencies(tasks)[source]#

Infer semantic dependencies between tasks.

Parameters:

tasks (List[Task]) – All tasks to analyze for dependencies

Returns:

Inferred dependency relationships

Return type:

List[SemanticDependency]

Notes

This supplements rule-based dependency detection with semantic understanding of task relationships.

abstractmethod async generate_enhanced_description(task, context)[source]#

Generate enhanced task description.

Parameters:

  • task (Task) – Task needing description enhancement

  • context (Dict[str, Any]) – Project context for better understanding

Returns:

Enhanced description with more clarity and detail

Return type:

str

abstractmethod async estimate_effort(task, context)[source]#

Estimate task effort using AI analysis.

Parameters:

  • task (Task) – Task to estimate

  • context (Dict[str, Any]) – Historical data and project context

Returns:

AI-powered effort estimation with confidence

Return type:

EffortEstimate

abstractmethod async analyze_blocker(task, blocker, context)[source]#

Analyze blocker and suggest solutions.

Parameters:

  • task (Task) – The blocked task

  • blocker (str) – Description of what’s blocking progress

  • context (Dict[str, Any]) – Additional context about the situation

Returns:

Prioritized list of solution suggestions

Return type:

List[str]

class src.ai.providers.SemanticAnalysis[source]#

Bases: object

Result of semantic task analysis.

Contains LLM’s understanding of task meaning and implications.

task_intent#

LLM’s interpretation of what the task aims to achieve

Type:

str

semantic_dependencies#

Tasks that logically should complete before this one

Type:

list of str

risk_factors#

Identified risks or complexity factors

Type:

list of str

suggestions#

LLM-generated improvement suggestions

Type:

list of str

confidence#

Model’s confidence in analysis (0.0-1.0)

Type:

float

reasoning#

Explanation of the analysis logic

Type:

str

risk_assessment#

Detailed risk breakdown by category

Type:

dict

fallback_used#

Whether a fallback model was used

Type:

bool

task_intent: str#
semantic_dependencies: List[str]#
risk_factors: List[str]#
suggestions: List[str]#
confidence: float#
reasoning: str#
risk_assessment: Dict[str, Any]#
fallback_used: bool = False#
__init__(task_intent, semantic_dependencies, risk_factors, suggestions, confidence, reasoning, risk_assessment, fallback_used=False)#
Parameters:
Return type:

None

class src.ai.providers.SemanticDependency[source]#

Bases: object

Semantic dependency relationship between tasks.

Represents LLM-inferred dependencies based on task meaning.

dependent_task_id#

ID of task that depends on another

Type:

str

dependency_task_id#

ID of task that must complete first

Type:

str

confidence#

Confidence in this dependency (0.0-1.0)

Type:

float

reasoning#

Explanation of why dependency exists

Type:

str

dependency_type#

Type of dependency: ‘logical’, ‘technical’, or ‘temporal’

Type:

str

Examples

>>> dep = SemanticDependency(
...     dependent_task_id="deploy-api",
...     dependency_task_id="test-api",
...     confidence=0.95,
...     reasoning="Deployment requires successful tests",
...     dependency_type="logical"
... )
dependent_task_id: str#
dependency_task_id: str#
confidence: float#
reasoning: str#
dependency_type: str#
__init__(dependent_task_id, dependency_task_id, confidence, reasoning, dependency_type)#
Parameters:

  • dependent_task_id (str)

  • dependency_task_id (str)

  • confidence (float)

  • reasoning (str)

  • dependency_type (str)

Return type:

None

class src.ai.providers.EffortEstimate[source]#

Bases: object

AI-powered effort estimation.

Provides intelligent time estimates based on task analysis.

estimated_hours#

Estimated hours to complete the task

Type:

float

confidence#

Confidence in the estimate (0.0-1.0)

Type:

float

factors#

Factors considered in the estimation

Type:

list of str

similar_tasks#

IDs of similar historical tasks used for reference

Type:

list of str

risk_multiplier#

Risk adjustment factor (1.0 = no risk, >1.0 = added risk)

Type:

float

Notes

The final estimate should be: estimated_hours * risk_multiplier

estimated_hours: float#
confidence: float#
factors: List[str]#
similar_tasks: List[str]#
risk_multiplier: float#
__init__(estimated_hours, confidence, factors, similar_tasks, risk_multiplier)#
Parameters:
Return type:

None

class src.ai.providers.CloudLLMProvider[source]#

Bases: LocalLLMProvider

Generic cloud LLM provider for OpenAI-compatible hosted APIs.

Inherits all semantic-analysis business logic from LocalLLMProvider and overrides the transport layer only: the HTTP client is pointed at an explicit remote URL with a required API key. The Ollama native-API fallback present in the parent is intentionally omitted — cloud endpoints do not implement it.

Parameters:

  • model (str) – Full model identifier as expected by the remote service (e.g. "accounts/fireworks/models/qwen2p5-coder-7b-instruct").

  • api_key (str) – Bearer token for the remote service.

  • url (str) – Base URL of the OpenAI-compatible inference endpoint (e.g. "https://api.fireworks.ai/inference/v1").

Raises:

ValueError – If model, api_key, or url is empty.

__init__(model, api_key, url)[source]#

Initialize cloud LLM provider.

Parameters:

  • model (str) – Model identifier for the remote service.

  • api_key (str) – Bearer token — must not be empty.

  • url (str) – Base URL for the OpenAI-compatible API — must not be empty.

Return type:

None

class src.ai.providers.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:
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:
Return type:

str

estimate_effort_intelligently(task, context)[source]#

AI-powered effort estimation

Parameters:
Return type:

EffortEstimate

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

Analyze blockers and provide solutions

Parameters:
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

Submodules#

On this page
Edit on GitHub
Show Source