plexus.reports.blocks package
- class plexus.reports.blocks.AcceptanceRate(config: Dict[str, Any], params: Dict[str, Any] | None, api_client: PlexusDashboardClient)
Bases:
FeedbackRatesBaseReport block that measures item-level and score-result-level acceptance rates.
- DEFAULT_DESCRIPTION: str | None = 'Score result acceptance metrics'
- DEFAULT_MAX_ITEMS = 0
- DEFAULT_NAME: str | None = 'Acceptance Rate'
- async generate() Tuple[Dict[str, Any] | None, str | None]
Generates the data for this report block asynchronously.
Access block configuration via self.config and report parameters via self.params. Use self.api_client for data fetching. Use self._log(“message”) to record log information.
- Returns:
- A tuple containing:
A dictionary containing the generated data (JSON-serializable), or None on failure.
A string containing concatenated log messages, or None if no logs.
- class plexus.reports.blocks.AcceptanceRateTimeline(config: Dict[str, Any], params: Dict[str, Any] | None, api_client: PlexusDashboardClient)
Bases:
FeedbackRatesBaseReport block that measures score-result acceptance rate over time (bucketed).
This is the same core metric as AcceptanceRate (score_result_acceptance_rate), but computed per time bucket instead of as a single aggregate.
- DEFAULT_DESCRIPTION: str | None = 'Score result acceptance rate over time'
- DEFAULT_NAME: str | None = 'Acceptance Rate Timeline'
- TRAILING_BUCKET_DAYS: Dict[str, int] = {'trailing_14d': 14, 'trailing_1d': 1, 'trailing_30d': 30, 'trailing_7d': 7}
- async generate() Tuple[Dict[str, Any] | None, str | None]
Generates the data for this report block asynchronously.
Access block configuration via self.config and report parameters via self.params. Use self.api_client for data fetching. Use self._log(“message”) to record log information.
- Returns:
- A tuple containing:
A dictionary containing the generated data (JSON-serializable), or None on failure.
A string containing concatenated log messages, or None if no logs.
- class plexus.reports.blocks.ActionItems(config: Dict[str, Any], params: Dict[str, Any] | None, api_client: PlexusDashboardClient)
Bases:
BaseReportBlockGenerate a prioritised action-item list from a sibling FeedbackAlignment block.
- DEFAULT_DESCRIPTION: str | None = 'Prioritised list of classifier improvement tasks from feedback root-cause analysis.'
- DEFAULT_NAME: str | None = 'Action Items'
- async generate() Tuple[Dict[str, Any] | None, str | None]
Generates the data for this report block asynchronously.
Access block configuration via self.config and report parameters via self.params. Use self.api_client for data fetching. Use self._log(“message”) to record log information.
- Returns:
- A tuple containing:
A dictionary containing the generated data (JSON-serializable), or None on failure.
A string containing concatenated log messages, or None if no logs.
- class plexus.reports.blocks.BaseReportBlock(config: Dict[str, Any], params: Dict[str, Any] | None, api_client: PlexusDashboardClient)
Bases:
ABCAbstract base class for all report blocks.
Each subclass is responsible for generating a specific section or data point within a report.
- DEFAULT_DESCRIPTION: str | None = None
- DEFAULT_NAME: str | None = None
- __init__(config: Dict[str, Any], params: Dict[str, Any] | None, api_client: PlexusDashboardClient)
- attach_detail_file(report_block_id: str, file_name: str, content: bytes, content_type: str | None = None) str
Attach a detail file to this report block.
This uploads the file to S3 and adds its path to the ReportBlock’s attachedFiles field. The Amplify Gen2 storage expects paths to be stored, not complex objects.
- Args:
report_block_id: ID of the report block to attach the file to file_name: Name of the file to create content: Bytes content of the file (changed from str to bytes) content_type: Optional MIME type for the file
- Returns:
The S3 path to the file
- abstractmethod async generate() Tuple[Dict[str, Any] | None, str | None]
Generates the data for this report block asynchronously.
Access block configuration via self.config and report parameters via self.params. Use self.api_client for data fetching. Use self._log(“message”) to record log information.
- Returns:
- A tuple containing:
A dictionary containing the generated data (JSON-serializable), or None on failure.
A string containing concatenated log messages, or None if no logs.
- get_resolved_dataset_id() str | None
Get the resolved dataset ID that this block actually used.
- Returns:
The dataset ID if one was resolved, None otherwise
- set_resolved_dataset_id(dataset_id: str)
Set the resolved dataset ID that this block actually used.
This should be called by blocks after they have resolved and used a dataset, to record which specific dataset was actually processed.
- Args:
dataset_id: The ID of the dataset that was resolved and used
- class plexus.reports.blocks.CorrectionRate(config: Dict[str, Any], params: Dict[str, Any] | None, api_client: PlexusDashboardClient)
Bases:
FeedbackRatesBaseReport block that measures score-result correction behavior from feedback edits.
- DEFAULT_DESCRIPTION: str | None = 'Per-item and corpus-level correction rates from feedback edits'
- DEFAULT_MAX_ITEMS = 200
- DEFAULT_NAME: str | None = 'Correction Rate'
- async generate() Tuple[Dict[str, Any] | None, str | None]
Generates the data for this report block asynchronously.
Access block configuration via self.config and report parameters via self.params. Use self.api_client for data fetching. Use self._log(“message”) to record log information.
- Returns:
- A tuple containing:
A dictionary containing the generated data (JSON-serializable), or None on failure.
A string containing concatenated log messages, or None if no logs.
- class plexus.reports.blocks.CostAnalysis(config: Dict[str, Any], params: Dict[str, Any] | None, api_client: PlexusDashboardClient)
Bases:
BaseReportBlockReportBlock: Cost analysis over ScoreResults with summary or detail outputs.
- Config:
- scorecard (optional): id/key/name/externalId (any identifier)
Use “all” to analyze all scorecards in the account
score (optional): id/key/name/externalId (resolved within scorecard when provided)
hours (int, default 1): preferred time window
days (int, default 0): used when hours not provided
start_date (str, optional): YYYY-MM-DD or ISO8601 (overrides hours/days)
end_date (str, optional): YYYY-MM-DD or ISO8601 (overrides hours/days)
limit (int, optional): max cost-bearing ScoreResults to include
group_by (str|None): None | ‘scorecard’ | ‘score’ | ‘scorecard_score’
mode (str): ‘summary’ (default) or ‘detail’
breakdown (bool): when summary mode, include grouped summaries
- DEFAULT_DESCRIPTION: str | None = 'ScoreResult cost metrics and breakdowns'
- DEFAULT_NAME: str | None = 'Cost Analysis'
- async generate() Tuple[Dict[str, Any] | None, str | None]
Generates the data for this report block asynchronously.
Access block configuration via self.config and report parameters via self.params. Use self.api_client for data fetching. Use self._log(“message”) to record log information.
- Returns:
- A tuple containing:
A dictionary containing the generated data (JSON-serializable), or None on failure.
A string containing concatenated log messages, or None if no logs.
- class plexus.reports.blocks.ExplanationAnalysis(config: Dict[str, Any], params: Dict[str, Any] | None, api_client: PlexusDashboardClient)
Bases:
FeedbackAlignmentSemantic reinforcement-memory analysis over normal production ScoreResult explanations.
This is intentionally a separate report block from FeedbackAlignment.
- DEFAULT_DESCRIPTION: str | None = 'Semantic reinforcement-memory analysis over ScoreResult explanations'
- DEFAULT_NAME: str | None = 'Explanation Analysis'
- async generate() Tuple[Dict[str, Any] | None, str | None]
Fetches feedback data and performs agreement analysis.
- class plexus.reports.blocks.FeedbackAlignment(config: Dict[str, Any], params: Dict[str, Any] | None, api_client: PlexusDashboardClient)
Bases:
BaseReportBlockAnalyzes feedback data using Gwet’s AC1 agreement coefficient.
This block retrieves FeedbackItem records and compares initial and final answer values to calculate agreement scores using Gwet’s AC1.
If a specific ‘score_id’ is provided in the config, it analyzes only that score. Otherwise, it analyzes all scores associated with the provided ‘scorecard’ that have a mapping to a Plexus Score with an externalId.
- Config:
scorecard (str): Scorecard identifier. This is REQUIRED. days (int, optional): Number of days in the past to analyze (default: 14).
FeedbackItems updated within this period will be considered.
- start_date (str, optional): Start date for analysis in YYYY-MM-DD format.
If provided, overrides ‘days’.
- end_date (str, optional): End date for analysis in YYYY-MM-DD format.
Defaults to today if not specified.
- score_id (str, optional): Specific score to analyze. May be Plexus score UUID or
external ID (e.g. Call Criteria question ID). Also read from report params as ‘score’ or ‘score_id’ when not in config. If specified, only this score will be analyzed.
- DEFAULT_DESCRIPTION: str | None = 'Inter-rater Reliability Assessment'
- DEFAULT_NAME: str | None = 'Feedback Alignment'
- async generate() Tuple[Dict[str, Any] | None, str | None]
Fetches feedback data and performs agreement analysis.
- class plexus.reports.blocks.FeedbackAlignmentTimeline(config: Dict[str, Any], params: Dict[str, Any] | None, api_client: PlexusDashboardClient)
Bases:
BaseReportBlockReport block for visualizing feedback alignment change over time.
Supports: - scorecard only: all scores on the scorecard - scorecard + score/score_id: single-score mode
Bucket policy supports: - complete historical buckets (default when no explicit window is provided) - exact-window buckets (when days or start_date/end_date is provided)
- CALENDAR_BUCKET_TYPES = {'calendar_biweek', 'calendar_day', 'calendar_month', 'calendar_week'}
- DEFAULT_DAYS = 30
- DEFAULT_DESCRIPTION: str | None = 'Alignment metrics over time'
- DEFAULT_NAME: str | None = 'Feedback Alignment Timeline'
- TRAILING_BUCKET_DAYS: Dict[str, int] = {'trailing_14d': 14, 'trailing_1d': 1, 'trailing_30d': 30, 'trailing_7d': 7}
- WEEK_START_INDEX = {'monday': 0, 'sunday': 6}
- async generate() Tuple[Dict[str, Any] | None, str | None]
Generates the data for this report block asynchronously.
Access block configuration via self.config and report parameters via self.params. Use self.api_client for data fetching. Use self._log(“message”) to record log information.
- Returns:
- A tuple containing:
A dictionary containing the generated data (JSON-serializable), or None on failure.
A string containing concatenated log messages, or None if no logs.
- class plexus.reports.blocks.FeedbackAnalysis(config: Dict[str, Any], params: Dict[str, Any] | None, api_client: PlexusDashboardClient)
Bases:
FeedbackAlignmentBackward-compatible alias for FeedbackAlignment.
Report configurations that reference class: FeedbackAnalysis should execute with the same behavior and config contract as FeedbackAlignment, including support for days and explicit start_date / end_date windows.
- async generate()
Fetches feedback data and performs agreement analysis.
- class plexus.reports.blocks.FeedbackContradictions(config: Dict[str, Any], params: Dict[str, Any] | None, api_client: PlexusDashboardClient)
Bases:
BaseReportBlockGuideline-vetting report block with contradiction and aligned modes.
- DEFAULT_DESCRIPTION: str | None = 'Guideline Contradiction Analysis'
- DEFAULT_NAME: str | None = 'Feedback Contradictions'
- async generate() Tuple[Dict[str, Any] | None, str | None]
Generates the data for this report block asynchronously.
Access block configuration via self.config and report parameters via self.params. Use self.api_client for data fetching. Use self._log(“message”) to record log information.
- Returns:
- A tuple containing:
A dictionary containing the generated data (JSON-serializable), or None on failure.
A string containing concatenated log messages, or None if no logs.
- class plexus.reports.blocks.FeedbackVolumeTimeline(config: Dict[str, Any], params: Dict[str, Any] | None, api_client: PlexusDashboardClient)
Bases:
FeedbackAlignmentTimelineReport block for visualizing feedback-item volume over time.
Supports: - scorecard only: all scores on the scorecard - scorecard + score/score_id: single-score mode
Bucket policy matches FeedbackAlignmentTimeline so volume charts can share the same time-window controls and report-level parameters.
- DEFAULT_DESCRIPTION: str | None = 'Feedback item volume over time'
- DEFAULT_NAME: str | None = 'Feedback Volume Timeline'
- async generate() Tuple[Dict[str, Any] | None, str | None]
Generates the data for this report block asynchronously.
Access block configuration via self.config and report parameters via self.params. Use self.api_client for data fetching. Use self._log(“message”) to record log information.
- Returns:
- A tuple containing:
A dictionary containing the generated data (JSON-serializable), or None on failure.
A string containing concatenated log messages, or None if no logs.
- class plexus.reports.blocks.RecentFeedback(config: Dict[str, Any], params: Dict[str, Any] | None, api_client: PlexusDashboardClient)
Bases:
FeedbackRatesBaseReport block that lists recently collected feedback rows for a scorecard/score.
- DEFAULT_DESCRIPTION: str | None = 'Recent feedback activity in the selected window'
- DEFAULT_NAME: str | None = 'Recent Feedback'
- async generate() Tuple[Dict[str, Any] | None, str | None]
Generates the data for this report block asynchronously.
Access block configuration via self.config and report parameters via self.params. Use self.api_client for data fetching. Use self._log(“message”) to record log information.
- Returns:
- A tuple containing:
A dictionary containing the generated data (JSON-serializable), or None on failure.
A string containing concatenated log messages, or None if no logs.
- class plexus.reports.blocks.ScoreChampionVersionTimeline(config: Dict[str, Any], params: Dict[str, Any] | None, api_client: PlexusDashboardClient)
Bases:
FeedbackRatesBaseReport block for visualizing score champion changes over time.
Champion transitions are read only from ScoreVersion.metadata.championHistory. Older versions without championHistory are intentionally not inferred.
- DEFAULT_DESCRIPTION: str | None = 'Champion version changes and associated evaluation metrics'
- DEFAULT_NAME: str | None = 'Score Champion Version Timeline'
- async generate() Tuple[Dict[str, Any] | None, str | None]
Generates the data for this report block asynchronously.
Access block configuration via self.config and report parameters via self.params. Use self.api_client for data fetching. Use self._log(“message”) to record log information.
- Returns:
- A tuple containing:
A dictionary containing the generated data (JSON-serializable), or None on failure.
A string containing concatenated log messages, or None if no logs.
- class plexus.reports.blocks.ScoreInfo(config: Dict[str, Any], params: Dict[str, Any] | None, api_client: PlexusDashboardClient)
Bases:
BaseReportBlockGenerates a block of information about a specific Score.
- Config:
- score (str): The ID or name of the Score to display.
(Note: Current mock implementation uses this as ID)
- include_variant (bool, optional): Whether to include variant details.
Defaults to False.
- async generate() Tuple[Dict[str, Any] | None, str | None]
Fetches (mock) Score data and returns it with logs.
- class plexus.reports.blocks.ScorecardHistory(config: Dict[str, Any], params: Dict[str, Any] | None, api_client: PlexusDashboardClient)
Bases:
BaseReportBlockReport block summarizing featured ScoreVersion changes over a date window.
Scope: - scorecard only: all scores on the scorecard - scorecard + score/score_id: a single score on the scorecard
- DEFAULT_DAYS = 30
- DEFAULT_DESCRIPTION: str | None = 'Featured score-version changes and champion promotion status'
- DEFAULT_NAME: str | None = 'Scorecard History'
- REQUIRED_SUMMARY_SECTION_HEADINGS = ('What changed', 'Guideline / rubric changes', 'Scoring behavior changes', 'Questions for SMEs / stakeholders', 'Rollout and evidence')
- SUMMARY_DIFF_CHAR_LIMIT = 1500
- async generate() Tuple[Dict[str, Any] | None, str | None]
Generates the data for this report block asynchronously.
Access block configuration via self.config and report parameters via self.params. Use self.api_client for data fetching. Use self._log(“message”) to record log information.
- Returns:
- A tuple containing:
A dictionary containing the generated data (JSON-serializable), or None on failure.
A string containing concatenated log messages, or None if no logs.
- class plexus.reports.blocks.TopicAnalysis(config: Dict[str, Any], params: Dict[str, Any] | None, api_client: PlexusDashboardClient)
Bases:
BaseReportBlock- DEFAULT_NAME: str | None = 'Topic Analysis'
- DEFAULT_PROMPT = '\n I have a topic from call center transcripts that is described by the following keywords: [KEYWORDS]\n In this topic, these customer-agent conversations are representative examples:\n [DOCUMENTS]\n\n Based on the keywords and representative examples above, provide a short, descriptive label for this topic in customer service context. Return only the label, no other text or formatting.\n '
- __init__(config: Dict[str, Any], params: Dict[str, Any] | None, api_client: PlexusDashboardClient)
- async generate() Tuple[Dict[str, Any] | None, str | None]
Generates the data for this report block asynchronously.
Access block configuration via self.config and report parameters via self.params. Use self.api_client for data fetching. Use self._log(“message”) to record log information.
- Returns:
- A tuple containing:
A dictionary containing the generated data (JSON-serializable), or None on failure.
A string containing concatenated log messages, or None if no logs.
- class plexus.reports.blocks.VectorTopicMemory(config: Dict[str, Any], params: Dict[str, Any] | None, api_client: PlexusDashboardClient)
Bases:
BaseReportBlockReportBlock that rebuilds a full topic memory view by re-indexing datasets into an AWS S3 Vectors index.
- Config:
scorecard (str): Scorecard identifier. When provided with days, uses feedback items (same as FeedbackAlignment). days (int): Number of days of feedback to include. Used with scorecard. data: { source?: str, dataset?: str, content_column?: str, fresh?: bool, content_source?: str }
- content_source:
edit_comment (default): FeedbackItem.editCommentValue for mismatches
transcript: Item transcript text linked from feedback items
score_result_no_explanation: ScoreResult.explanation for normal production predictions with value=’No’
s3_vectors: { bucket_name: str, index_name: str, index_arn?: str, region?: str } embedding: { model_id?: str, preprocessing_version?: str } clustering: { min_topic_size?: int }
coarse_min_topic_fraction?: float # default 0.02 when min_topic_fraction is not set coarse_target_max_topics_per_score?: int # default 12 when target_max_topics_per_score is not set
mode: “full” | “incremental” # incremental = KNN assign only, no re-cluster label:
batch_one_pass?: bool # default true: one LLM call for all selected buckets batch_model?: str # default gpt-4o batch_prompt?: str # optional user prompt override; supports {{topic_bundles_json}} request_timeout_seconds?: int # default 60 batch_request_timeout_seconds?: int # default request_timeout_seconds
- DEFAULT_DESCRIPTION: str | None = 'Persistent vector-based topic memory from S3 Vectors index'
- DEFAULT_NAME: str | None = 'Vector Topic Memory'
- MEDIUM_TERM_DAYS = 30
- SHORT_TERM_DAYS = 14
- async generate() Tuple[Dict[str, Any] | None, str | None]
Orchestrates: resolve dataset -> embed -> index -> cluster -> memory weights.
Submodules
- plexus.reports.blocks.acceptance_rate module
- plexus.reports.blocks.acceptance_rate_timeline module
- plexus.reports.blocks.action_items_block module
- plexus.reports.blocks.base module
- plexus.reports.blocks.correction_rate module
- plexus.reports.blocks.cost_analysis module
- plexus.reports.blocks.data_utils module
- plexus.reports.blocks.explanation_analysis module
- plexus.reports.blocks.feedback_alignment module
- plexus.reports.blocks.feedback_alignment_timeline module
- plexus.reports.blocks.feedback_analysis module
- plexus.reports.blocks.feedback_contradictions module
- plexus.reports.blocks.feedback_rates_base module
- plexus.reports.blocks.feedback_scope_resolver module
- plexus.reports.blocks.feedback_utils module
- plexus.reports.blocks.feedback_volume_timeline module
- plexus.reports.blocks.guideline_vetting module
- plexus.reports.blocks.identifier_utils module
- plexus.reports.blocks.recent_feedback module
- plexus.reports.blocks.reinforcement_helpers module
- plexus.reports.blocks.report_block_orm module
- plexus.reports.blocks.score_champion_version_timeline module
- plexus.reports.blocks.score_info module
- plexus.reports.blocks.score_resolution module
- plexus.reports.blocks.scorecard_history module
- plexus.reports.blocks.topic_analysis module
- plexus.reports.blocks.vector_topic_memory module