Learning Module (APLS)

The learning module implements the Automatic Pattern Learning System (APLS) for cascade optimization. It extracts patterns from stage executions, generates routing rules, analyzes migration costs, and manages the rule proposal workflow.

PatternExtractor

Extracts patterns from cascade stage failures for optimization.

The PatternExtractor analyzes stage results to identify recurring patterns that could potentially be moved to cheaper stages. For example, if an AI stage consistently catches a specific type of input based on certain feature combinations, that pattern could be encoded as a rule.

Example

extractor = PatternExtractor() pattern = extractor.learn_from_failure(context, "ai_analysis", result) if pattern: ... print(f"Learned pattern: {pattern.id}") candidates = extractor.get_migration_candidates(min_confidence=0.9) insights = extractor.get_insights()

__init__(config: Optional[PatternConfig] = None)

Initialize the pattern extractor.

Parameters:

Name	Type	Description	Default
config	Optional[PatternConfig]	Optional configuration for pattern extraction.	None

learn_from_failure(context: ExecutionContext, stage: str, result: StageResult) -> Optional[StageFailurePattern]

Learn patterns from a stage failure or detection.

Analyzes the execution context and stage result to extract patterns that characterize this type of detection.

Parameters:

Name	Type	Description	Default
context	ExecutionContext	The execution context containing input data.	required
stage	str	Name of the stage that detected/flagged the input.	required
result	StageResult	The stage result containing detection details.	required

Returns:

Type	Description
Optional[StageFailurePattern]	The extracted or updated pattern, or None if no pattern found.

get_migration_candidates(min_confidence: float = 0.8, min_samples: int = 10) -> List[StageFailurePattern]

Get patterns that are candidates for migration to cheaper stages.

Returns patterns that have high confidence and sufficient samples to warrant consideration for migration.

Parameters:

Name	Type	Description	Default
min_confidence	float	Minimum confidence threshold.	0.8
min_samples	int	Minimum number of samples.	10

Returns:

Type	Description
List[StageFailurePattern]	List of patterns suitable for migration.

get_insights() -> PatternLearningInsight

Get aggregated learning insights.

Computes overall statistics and recommendations from all learned patterns.

Returns:

Type	Description
PatternLearningInsight	PatternLearningInsight with analysis and recommendations.

clear_patterns() -> None

Clear all learned patterns.

get_pattern(pattern_id: str) -> Optional[StageFailurePattern]

Get a specific pattern by ID.

Parameters:

Name	Type	Description	Default
pattern_id	str	The pattern identifier.	required

Returns:

Type	Description
Optional[StageFailurePattern]	The pattern if found, None otherwise.

get_patterns_by_stage(stage: str) -> List[StageFailurePattern]

Get all patterns for a specific stage.

Parameters:

Name	Type	Description	Default
stage	str	The stage name.	required

Returns:

Type	Description
List[StageFailurePattern]	List of patterns for the stage.

get_patterns_by_type(pattern_type: str) -> List[StageFailurePattern]

Get all patterns of a specific type.

Parameters:

Name	Type	Description	Default
pattern_type	str	The pattern type.	required

Returns:

Type	Description
List[StageFailurePattern]	List of patterns of the specified type.

export_patterns() -> List[Dict[str, Any]]

Export all patterns as dictionaries.

Returns:

Type	Description
List[Dict[str, Any]]	List of pattern dictionaries.

import_patterns(patterns: List[Dict[str, Any]]) -> int

Import patterns from dictionaries.

Parameters:

Name	Type	Description	Default
patterns	List[Dict[str, Any]]	List of pattern dictionaries.	required

Returns:

Type	Description
int	Number of patterns imported.

StageFailurePattern

Represents a pattern extracted from stage failures.

Patterns capture recurring characteristics in data that cause specific stages to trigger. By identifying these patterns, we can potentially move detection to cheaper, earlier stages.

Attributes:

Name	Type	Description
id	str	Unique identifier for this pattern.
stage	str	Name of the cascade stage where this pattern was detected.
pattern_type	str	Type of pattern (threshold, correlation, reasoning, temporal, behavioral).
features	Dict[str, Any]	Extracted feature values that define this pattern.
confidence	float	Confidence score (0-1) in this pattern's reliability.
sample_count	int	Number of samples that matched this pattern.
first_seen	datetime	Timestamp when pattern was first observed.
last_seen	datetime	Timestamp when pattern was last observed.
metadata	Dict[str, Any]	Additional pattern metadata.

to_dict() -> Dict[str, Any]

Convert pattern to dictionary representation.

from_dict(data: Dict[str, Any]) -> StageFailurePattern classmethod

Create pattern from dictionary representation.

RuleGenerator

Generates routing rules from learned patterns.

This class analyzes patterns extracted from execution history and generates appropriate routing rules. It supports multiple generation strategies based on pattern type and can optionally use an LLM for complex rule generation.

Attributes:

Name	Type	Description
llm_client		Optional LLM client for generating complex rules.
min_confidence		Minimum confidence threshold for generating rules.
min_coverage		Minimum estimated coverage for generating rules.

Example

generator = RuleGenerator(min_confidence=0.7) pattern = StageFailurePattern(...) rule = generator.generate_from_pattern(pattern) if rule: ... yaml_output = generator.to_yaml([rule])

__init__(llm_client: Optional[Any] = None, min_confidence: float = 0.5, min_coverage: float = 0.01) -> None

Initialize the rule generator.

Parameters:

Name	Type	Description	Default
llm_client	Optional[Any]	Optional LLM client for generating complex rules. Should have a method like generate(prompt: str) -> str.	None
min_confidence	float	Minimum confidence threshold for rule generation.	0.5
min_coverage	float	Minimum estimated coverage threshold.	0.01

generate_from_pattern(pattern: StageFailurePattern) -> Optional[GeneratedRule]

Generate a routing rule from a pattern.

This method selects the appropriate generation strategy based on the pattern type and attempts to create a rule.

Parameters:

Name	Type	Description	Default
pattern	StageFailurePattern	The pattern to generate a rule from.	required

Returns:

Type	Description
Optional[GeneratedRule]	A GeneratedRule if successful, None if the pattern cannot be
Optional[GeneratedRule]	converted to a rule or doesn't meet thresholds.

to_routing_rule(generated: GeneratedRule) -> RoutingRule

Convert a GeneratedRule to cascade's RoutingRule format.

This method transforms the generated rule into the native RoutingRule format used by the cascade engine.

Parameters:

Name	Type	Description	Default
generated	GeneratedRule	The GeneratedRule to convert.	required

Returns:

Type	Description
RoutingRule	A RoutingRule ready for use in cascade configuration.

to_yaml(rules: List[GeneratedRule]) -> str

Export generated rules as YAML.

This method exports rules in a human-readable YAML format suitable for review and manual editing.

Parameters:

Name	Type	Description	Default
rules	List[GeneratedRule]	List of GeneratedRule objects to export.	required

Returns:

Type	Description
str	YAML string representation of the rules.

Raises:

Type	Description
ImportError	If PyYAML is not installed.

generate_batch(patterns: List[StageFailurePattern]) -> List[GeneratedRule]

Generate rules from a batch of patterns.

Convenience method for processing multiple patterns at once.

Parameters:

Name	Type	Description	Default
patterns	List[StageFailurePattern]	List of patterns to generate rules from.	required

Returns:

Type	Description
List[GeneratedRule]	List of successfully generated rules.

GeneratedRule

A routing rule generated from a learned pattern.

This dataclass represents a machine-generated rule that can be proposed for human review before being deployed.

Attributes:

Name	Type	Description
rule_id	str	Unique identifier for this rule.
name	str	Human-readable rule name.
description	str	Detailed description of what this rule does.
template	RuleTemplate	The template type used to generate this rule.
conditions	List[Dict[str, Any]]	List of condition specifications.
action	str	The action to take when rule matches (APPROVE, REJECT, ESCALATE, FLAG).
target_stage	str	The stage to route to when the rule matches.
source_pattern_id	str	ID of the pattern this rule was generated from.
confidence	float	Confidence score from the source pattern (0.0 to 1.0).
estimated_coverage	float	Estimated percentage of cases this rule would handle.
created_at	datetime	Timestamp when this rule was generated.

to_dict() -> Dict[str, Any]

Convert rule to dictionary representation.

Returns:

Type	Description
Dict[str, Any]	Dictionary representation of the generated rule.

from_dict(data: Dict[str, Any]) -> 'GeneratedRule' classmethod

Create rule from dictionary representation.

Parameters:

Name	Type	Description	Default
data	Dict[str, Any]	Dictionary containing rule data.	required

Returns:

Type	Description
'GeneratedRule'	GeneratedRule instance.

CostAnalyzer

Analyzes migration costs and ROI for cascade pattern optimization.

The CostAnalyzer evaluates whether patterns should be migrated between processing stages based on cost efficiency, detection rates, and risk factors. It uses configurable stage costs to model the tradeoffs between earlier (cheaper but less sophisticated) and later (expensive but more capable) stages.

Example

analyzer = CostAnalyzer()

Calculate ROI for a single pattern

roi = analyzer.calculate_migration_roi(pattern, "RULES", volume=10000) print(f"Recommendation: {roi.recommendation}") print(f"Savings: {roi.cost_reduction_absolute:.2f} per period")

Attributes:

Name	Type	Description
stage_costs		Dictionary mapping stage names to their cost configurations.

__init__(stage_costs: Optional[Dict[str, StageCost]] = None) -> None

Initialize the cost analyzer.

Parameters:

Name	Type	Description	Default
stage_costs	Optional[Dict[str, StageCost]]	Optional custom stage cost configuration. If not provided, uses DEFAULT_STAGE_COSTS.	None

set_stage_cost(stage: str, cost: StageCost) -> None

Set or update the cost configuration for a stage.

Parameters:

Name	Type	Description	Default
stage	str	Stage identifier name.	required
cost	StageCost	Cost configuration for the stage.	required

calculate_migration_roi(pattern: 'StageFailurePattern', target_stage: str, volume: int, migration_effort: Optional[float] = None) -> MigrationROI

Calculate ROI for migrating a pattern to a different stage.

Parameters:

Name	Type	Description	Default
pattern	'StageFailurePattern'	The stage failure pattern to analyze.	required
target_stage	str	Stage to potentially migrate the pattern to.	required
volume	int	Estimated number of items per period.	required
migration_effort	Optional[float]	Cost units for migration effort (default: 100.0).	None

Returns:

Type	Description
MigrationROI	MigrationROI with cost analysis and recommendation.

Raises:

Type	Description
ValueError	If source or target stage is not configured.

analyze_all_candidates(candidates: List['StageFailurePattern'], target_stage: str = 'RULES', volume: int = 1000) -> List[MigrationROI]

Analyze all candidate patterns for migration to a target stage.

Parameters:

Name	Type	Description	Default
candidates	List['StageFailurePattern']	List of stage failure patterns to analyze.	required
target_stage	str	Target stage for migration (default: RULES).	'RULES'
volume	int	Estimated items per period for each pattern.	1000

Returns:

Type	Description
List[MigrationROI]	List of MigrationROI objects for each candidate.

get_total_potential_savings(roi_list: List[MigrationROI]) -> Dict[str, Any]

Calculate total potential savings from a list of ROI analyses.

Parameters:

Name	Type	Description	Default
roi_list	List[MigrationROI]	List of MigrationROI objects to aggregate.	required

Returns:

Type

Description

Dict[str, Any]

Dictionary containing: - total_absolute_savings: Sum of all cost reductions. - total_volume: Sum of all estimated volumes. - average_reduction_percentage: Weighted average cost reduction. - migrate_count: Number of MIGRATE recommendations. - monitor_count: Number of MONITOR recommendations. - reject_count: Number of REJECT recommendations. - patterns_by_recommendation: Pattern IDs grouped by recommendation.

rank_by_roi(roi_list: List[MigrationROI], ascending: bool = False) -> List[MigrationROI]

Rank migration candidates by ROI.

Patterns are ranked primarily by cost reduction absolute value, with recommendation and false positive risk as secondary factors.

Parameters:

Name	Type	Description	Default
roi_list	List[MigrationROI]	List of MigrationROI objects to rank.	required
ascending	bool	If True, rank from lowest to highest ROI.	False

Returns:

Type	Description
List[MigrationROI]	Sorted list of MigrationROI objects.

MigrationROI

ROI calculation for migrating a pattern to a different stage.

Attributes:

Name	Type	Description
pattern_id	str	Unique identifier for the pattern being analyzed.
source_stage	str	Current stage where pattern is being handled.
target_stage	str	Proposed stage to migrate pattern to.
current_cost_per_item	float	Cost per item at current stage.
projected_cost_per_item	float	Expected cost per item at target stage.
estimated_volume	int	Expected items per period to process.
cost_reduction_absolute	float	Total cost savings per period.
cost_reduction_percentage	float	Percentage reduction in costs.
detection_rate_change	float	Change in detection rate (positive = improvement).
false_positive_risk	float	Risk of false positives at target stage (0.0-1.0).
payback_items	int	Number of items needed to break even on migration effort.
recommendation	str	Migration recommendation (MIGRATE, MONITOR, REJECT).

__post_init__() -> None

Validate recommendation value.

ProposalManager

Manages the lifecycle of rule proposals through the approval workflow.

This class provides methods for creating, reviewing, testing, and activating rule proposals. It supports optional file-based persistence for proposals.

Attributes:

Name	Type	Description
storage_path		Optional path for persisting proposals to disk.
_proposals	Dict[str, RuleProposal]	In-memory dictionary of proposals keyed by proposal_id.

Example

manager = ProposalManager(storage_path=Path("./proposals")) proposal = manager.create_proposal(rule, roi) pending = manager.get_pending_proposals() manager.approve(proposal.proposal_id, "reviewer@example.com")

__init__(storage_path: Optional[Path] = None) -> None

Initialize the proposal manager.

Parameters:

Name	Type	Description	Default
storage_path	Optional[Path]	Optional path for file-based persistence. If provided, proposals will be saved to and loaded from this directory.	None

create_proposal(rule: Any, roi: Any) -> RuleProposal

Create a new rule proposal for review.

Parameters:

Name	Type	Description	Default
rule	Any	The machine-generated rule to propose.	required
roi	Any	ROI analysis supporting this proposal.	required

Returns:

Type	Description
RuleProposal	The created RuleProposal with PENDING_REVIEW status.

get_pending_proposals() -> List[RuleProposal]

Get all proposals pending review.

Returns:

Type	Description
List[RuleProposal]	List of proposals with PENDING_REVIEW status, sorted by creation time.

approve(proposal_id: str, reviewer: str, notes: Optional[str] = None) -> RuleProposal

Approve a pending proposal.

Parameters:

Name	Type	Description	Default
proposal_id	str	ID of the proposal to approve.	required
reviewer	str	Identifier of the person approving.	required
notes	Optional[str]	Optional notes explaining the approval decision.	None

Returns:

Type	Description
RuleProposal	The updated RuleProposal with APPROVED status.

Raises:

Type	Description
KeyError	If proposal_id not found.
ValueError	If proposal is not in PENDING_REVIEW status.

reject(proposal_id: str, reviewer: str, notes: str) -> RuleProposal

Reject a pending proposal.

Parameters:

Name	Type	Description	Default
proposal_id	str	ID of the proposal to reject.	required
reviewer	str	Identifier of the person rejecting.	required
notes	str	Required notes explaining the rejection reason.	required

Returns:

Type	Description
RuleProposal	The updated RuleProposal with REJECTED status.

Raises:

Type	Description
KeyError	If proposal_id not found.
ValueError	If proposal is not in PENDING_REVIEW status.

start_testing(proposal_id: str) -> RuleProposal

Start A/B testing for an approved proposal.

Parameters:

Name	Type	Description	Default
proposal_id	str	ID of the proposal to begin testing.	required

Returns:

Type	Description
RuleProposal	The updated RuleProposal with TESTING status.

Raises:

Type	Description
KeyError	If proposal_id not found.
ValueError	If proposal is not in APPROVED status.

record_test_results(proposal_id: str, results: Dict[str, Any]) -> RuleProposal

Record test results for a proposal in testing.

Parameters:

Name	Type	Description	Default
proposal_id	str	ID of the proposal being tested.	required
results	Dict[str, Any]	Dictionary of test results and metrics.	required

Returns:

Type	Description
RuleProposal	The updated RuleProposal with recorded test results.

Raises:

Type	Description
KeyError	If proposal_id not found.
ValueError	If proposal is not in TESTING status.

activate(proposal_id: str) -> RuleProposal

Activate a tested proposal for production use.

Parameters:

Name	Type	Description	Default
proposal_id	str	ID of the proposal to activate.	required

Returns:

Type	Description
RuleProposal	The updated RuleProposal with ACTIVE status.

Raises:

Type	Description
KeyError	If proposal_id not found.
ValueError	If proposal is not in TESTING or APPROVED status.

deprecate(proposal_id: str, reason: str) -> RuleProposal

Deprecate an active proposal.

Parameters:

Name	Type	Description	Default
proposal_id	str	ID of the proposal to deprecate.	required
reason	str	Reason for deprecation.	required

Returns:

Type	Description
RuleProposal	The updated RuleProposal with DEPRECATED status.

Raises:

Type	Description
KeyError	If proposal_id not found.
ValueError	If proposal is not in ACTIVE status.

get_proposal(proposal_id: str) -> Optional[RuleProposal]

Get a proposal by ID.

Parameters:

Name	Type	Description	Default
proposal_id	str	ID of the proposal to retrieve.	required

Returns:

Type	Description
Optional[RuleProposal]	The RuleProposal if found, None otherwise.

get_active_rules() -> List[RuleProposal]

Get all active rule proposals.

Returns:

Type	Description
List[RuleProposal]	List of proposals with ACTIVE status, sorted by activation time.

export_proposals(path: Path) -> None

Export all proposals to a JSON file.

Parameters:

Name	Type	Description	Default
path	Path	Path to write the JSON file.	required

import_proposals(path: Path) -> None

Import proposals from a JSON file.

Parameters:

Name	Type	Description	Default
path	Path	Path to the JSON file to import.	required

Raises:

Type	Description
FileNotFoundError	If the file does not exist.
JSONDecodeError	If the file is not valid JSON.

RuleProposal

A proposal for a machine-generated routing rule.

This dataclass captures all metadata about a rule proposal throughout its lifecycle, from initial creation through review, testing, and activation.

Attributes:

Name	Type	Description
proposal_id	str	Unique identifier for this proposal.
generated_rule	Any	The machine-generated rule being proposed.
roi_analysis	Any	ROI analysis supporting this proposal.
status	ProposalStatus	Current status in the approval workflow.
created_at	datetime	Timestamp when proposal was created.
reviewed_at	Optional[datetime]	Timestamp when proposal was reviewed (approved/rejected).
reviewer	Optional[str]	Identifier of the person who reviewed the proposal.
review_notes	Optional[str]	Notes from the reviewer explaining their decision.
test_results	Optional[Dict[str, Any]]	Results from A/B testing phase.
activated_at	Optional[datetime]	Timestamp when proposal was activated in production.

to_dict() -> Dict[str, Any]

Convert proposal to dictionary for serialization.

Returns:

Type	Description
Dict[str, Any]	Dictionary representation of the proposal.

from_dict(data: Dict[str, Any]) -> 'RuleProposal' classmethod

Create proposal from dictionary.

Parameters:

Name	Type	Description	Default
data	Dict[str, Any]	Dictionary containing proposal data.	required

Returns:

Type	Description
'RuleProposal'	RuleProposal instance.