plexus.rca_analysis module

Shared RCA analysis utilities for score result investigation.

Used by both the feedback evaluation RCA pipeline (Evaluation.py) and the on-demand plexus_score_result_investigate MCP tool.

plexus.rca_analysis.analyze_score_result(primary_input: str, predicted: str, correct: str, explanation: str, topic_label: str = 'Score result investigation', score_guidelines: str = '', score_yaml_code: str = '', feedback_context: str = '') → tuple

Run a two-turn GPT-5 mini conversation to analyze a misclassification.

Returns (detailed_cause, suggested_fix):

detailed_cause: why the AI prediction was wrong (2-4 sentences)
suggested_fix: one concrete score code change to prevent this error

Args:

primary_input: Primary input artifact text/context excerpt predicted: The AI’s predicted value correct: The correct/human-labeled value explanation: The AI’s reasoning for its prediction topic_label: Topic or cluster context label score_guidelines: Score guidelines text (truncated to 2000 chars) score_yaml_code: Score YAML configuration (truncated to 4000 chars) feedback_context: Additional context about reviewer feedback

plexus.rca_analysis.build_feedback_context(feedback_comment: str = '', feedback_initial: str = '', feedback_final: str = '', predicted_value: str = '') → str

Build a feedback context string for the analysis prompt that handles the confusing case where a reviewer “agreed” with the production result but the evaluation produced a different result.

Returns a string to include in the analysis prompt.

plexus.rca_analysis.build_misclassification_analysis_summary(topics: list, *, item_classifications_all: List[Dict[str, Any]], analysis_scope: Dict[str, Any] | None = None, max_category_summary_items: int = 20) → dict: Build structured triage output including topic/category summaries and next-action guidance.

plexus.rca_analysis.build_misclassification_classification_contract() → dict: Build the full operator-facing misclassification taxonomy/evidence contract.

plexus.rca_analysis.build_misclassification_item_context(*, feedback_item_id: str, item_id: str, score_id: str, scorecard_id: str, score_version_id: str, predicted_value: str, correct_value: str, score_explanation: str, edit_comment: str, initial_comment: str, final_comment: str, score_guidelines_text: str, score_yaml_configuration: str, scorecard_guidance_text: str, primary_input_text: str, primary_input_modality: str, metadata_snapshot: str, label_provenance_source: str, resolved_final_classes: List[str] | None = None, class_resolution_source: str = '', primary_input_fetch_error: bool = False, missing_required_context_keys: List[str] | None = None, processed_input_text: str = '', processors_config_summary: str = '', rubric_memory_context: Dict[str, Any] | None = None) → dict: Build standardized item context/provenance payload for misclassification analysis.

plexus.rca_analysis.build_rca_analysis_failure_classification(*, item_context: dict, stage: str, exc: BaseException, evidence_flags: Dict[str, Any] | None = None) → dict: Preserve an RCA item as a mechanical failure row when RCA analysis itself fails.

plexus.rca_analysis.build_rca_analysis_failure_details(stage: str, exc: BaseException) → dict: Build a serializable RCA item-stage failure diagnostic.

plexus.rca_analysis.classify_misclassification_item(item_context: dict, evidence_flags: Dict[str, Any]) → dict: Classify one misclassification using standardized context and extracted evidence flags.

plexus.rca_analysis.explain_misclassification_item_classification(*, item_context: Dict[str, Any], classification: Dict[str, Any]) → Dict[str, str]: Generate a short, operator-focused explanation for an assigned misclassification category. Raises ValueError when the LLM output is invalid.

plexus.rca_analysis.extract_misclassification_evidence_flags(*, item_context: Dict[str, Any]) → Dict[str, Any]

Extract item-level triage evidence flags from full context (including feedback comments).

The returned flags are inputs to deterministic category assignment.

plexus.rca_analysis.get_misclassification_classifier_output_contract() → dict: Return the per-item classifier output contract for misclassification analysis.

plexus.rca_analysis.get_misclassification_explainer_output_contract() → dict: Return contract for per-item triage explainer output.

plexus.rca_analysis.get_misclassification_item_context_contract() → dict: Return the canonical per-item context/provenance contract for classification.

plexus.rca_analysis.get_misclassification_item_scope_evidence_contract() → dict: Return the item-scope evidence contract for misclassification classification.

plexus.rca_analysis.normalize_best_evidence_source(raw_source: str) → str

plexus.rca_analysis.resolve_final_output_classes_from_yaml_text(score_yaml_configuration: str) → dict: Resolve final output classes from score YAML using the canonical final-output order: 1) parameters.validation.value.valid_classes 2) classes[].name 3) graph[-1].valid_classes 4) graph[-1].conditions[].output.value 5) graph[-1].LogicalClassifier.code (Score.Result(value=”…”))

plexus.rca_analysis.rubric_memory_state_from_item_context(item_context: dict) → dict: Return compact rubric-memory availability/provenance state for RCA artifacts.