plexus.agentic.alignment_optimizer module

Alignment Optimizer - Programmatic interface for feedback evaluation optimization with RCA

This module provides a clean Python API for running the Tactus-based feedback alignment optimizer procedure, enabling integration into automated systems and agentic applications.

class plexus.agentic.alignment_optimizer.AlignmentOptimizer(client, mcp_server)

Bases: object

Programmatic interface to feedback alignment optimization using Tactus procedures.

This class wraps the feedback_alignment_optimizer Tactus procedure, providing a simple Python API for running iterative score optimization with RCA-based analysis.

Example:

```python from plexus.agentic import AlignmentOptimizer from plexus.dashboard.api.client import PlexusDashboardClient

client = PlexusDashboardClient(api_key=”…”) optimizer = AlignmentOptimizer(client, mcp_server)

result = await optimizer.optimize(

scorecard=”customer-service”, score=”empathy”, days=90, max_iterations=10

)

print(f”Improved AC1 by {result[‘improvement’]:.4f}”) print(f”Completed {len(result[‘iterations’])} iterations”) ```

Initialize the AlignmentOptimizer.

Args:

client: PlexusDashboardClient instance with valid credentials mcp_server: MCP server instance for tool execution

__init__(client, mcp_server)

Initialize the AlignmentOptimizer.

Args:

client: PlexusDashboardClient instance with valid credentials mcp_server: MCP server instance for tool execution

async optimize(scorecard: str, score: str, days: int = 90, max_iterations: int = 10, improvement_threshold: float = 0.02, dry_run: bool = False, on_iteration: Callable[[int, Dict[str, Any]], None] | None = None) → Dict[str, Any]

Run feedback evaluation optimization for a score.

This method executes the complete optimization workflow: 1. Run baseline feedback evaluation with RCA 2. Iteratively analyze RCA, propose changes, evaluate, and compare 3. Return comprehensive results with all iteration data

Args:

scorecard: Scorecard identifier (name, key, or ID) score: Score identifier (name, key, or ID) days: Feedback window in days (default: 90) max_iterations: Maximum optimization iterations (default: 10) improvement_threshold: Minimum AC1 improvement to continue (default: 0.02 = 2%) dry_run: If True, run analysis only without making score updates (default: False) on_iteration: Optional callback called after each iteration with (iteration_num, iteration_data)

Returns:
Dict containing:

  • success: bool - Whether optimization completed successfully

  • status: str - Completion status (converged, max_iterations, user_stopped, etc.)

  • message: str - Human-readable summary message

  • baseline_evaluation_id: str - Initial baseline evaluation ID

  • final_evaluation_id: str - Final evaluation ID

  • iterations: list - Array of iteration results with metrics and deltas

  • improvement: float - Total AC1 improvement from baseline to final

  • scorecard_id: str - Resolved scorecard ID

  • score_id: str - Resolved score ID

Raises:

FileNotFoundError: If procedure YAML cannot be found Exception: If procedure execution fails

async run_baseline_only(scorecard: str, score: str, days: int = 90) → Dict[str, Any]

Run only the baseline evaluation without optimization iterations.

This is useful for getting an initial RCA analysis before deciding whether to proceed with full optimization.

Args:

scorecard: Scorecard identifier score: Score identifier days: Feedback window in days

Returns:

Dict with baseline evaluation results and RCA

class plexus.agentic.alignment_optimizer.OptimizationMonitor

Bases: object

Helper class for monitoring optimization progress in real-time.

Example:

```python monitor = OptimizationMonitor()

result = await optimizer.optimize(

scorecard=”customer-service”, score=”empathy”, on_iteration=monitor.on_iteration

)

monitor.print_summary() ```

__init__()
on_iteration(iteration_num: int, iteration_data: Dict[str, Any])

Callback for each completed iteration.

print_summary()

Print a summary of all iterations.