primus.scores package

Score class namespace.

Scorecard loading resolves score classes dynamically with getattr(primus.scores, class_name). Keep that behavior, but do not import every score implementation at package import time. Many score types require optional training, ML, provider, or workflow dependencies that should only load when that score class is selected.

class primus.scores.AWSComprehendEntityExtractor(**parameters)

Bases: Score

This score uses AWS Comprehend to extract the first named entity from the transcript.

Initialize the Score instance with the given parameters.

Parameters

**parametersdict

Arbitrary keyword arguments that are used to initialize the Parameters instance.

Raises

PydanticValidationError

If the provided parameters do not pass validation.

class Result(*, parameters: Parameters, value: str | bool, explanation: str, confidence: float | None = None, start_time_seconds: float | None = None, end_time_seconds: float | None = None, metadata: dict = {}, error: str | None = None, code: str | None = None)

Bases: Result

Model output data structure.

Attributes

scorestr

The predicted score label.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

explanation: str
model_config = {'protected_namespaces': ()}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

__init__(**parameters)

Initialize the Score instance with the given parameters.

Parameters

**parametersdict

Arbitrary keyword arguments that are used to initialize the Parameters instance.

Raises

PydanticValidationError

If the provided parameters do not pass validation.

extract_first_person_entity(transcript: str) → str
extract_quotes_that_include_first_person_entity(transcript: str, first_person_entity: str) → list[str]
predict(context, model_input: ScoreInput)

Make predictions on the input data.

Parameters

contextAny

Context for the prediction

model_inputScore.Input

The input data for making predictions.

Returns

Union[Score.Result, List[Score.Result]]

Either a single Score.Result or a list of Score.Results

predict_validation()

Placeholder method to satisfy the base class requirement. This validator doesn’t require traditional training.

register_model()

Register the model with MLflow by logging relevant parameters.

save_model()

Save the model to a specified path and log it as an artifact with MLflow.

train_model()

Placeholder method to satisfy the base class requirement. This validator doesn’t require traditional training.

class primus.scores.AWSComprehendSentimentScore(**parameters)

Bases: Score

Score that uses AWS Comprehend to detect sentiment in text.

This score analyzes the sentiment of input text using AWS Comprehend’s detect_sentiment API. It returns one of four sentiment values: - POSITIVE: Text expresses positive sentiment - NEGATIVE: Text expresses negative sentiment - NEUTRAL: Text is neutral or factual - MIXED: Text contains both positive and negative sentiment

The score automatically truncates input text to AWS Comprehend’s limit of 5000 UTF-8 bytes.

Example YAML configuration:

  • name: Customer Sentiment class: AWSComprehendSentimentScore data:

    processors:

    • class: FilterCustomerOnlyProcessor

    • class: RemoveSpeakerIdentifiersTranscriptFilter

Note: Requires AWS credentials to be configured (via environment variables, AWS config file, or IAM role).

Initialize the Score instance with the given parameters.

Parameters

**parametersdict

Arbitrary keyword arguments that are used to initialize the Parameters instance.

Raises

PydanticValidationError

If the provided parameters do not pass validation.

class Result(*, parameters: Parameters, value: str | bool, explanation: str | None = None, confidence: float | None = None, start_time_seconds: float | None = None, end_time_seconds: float | None = None, metadata: dict = {}, error: str | None = None, code: str | None = None)

Bases: Result

Result structure for sentiment classification.

Attributes:

value: Sentiment label (POSITIVE, NEGATIVE, NEUTRAL, or MIXED) explanation: Detailed explanation including confidence scores

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'protected_namespaces': ()}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

__init__(**parameters)

Initialize the Score instance with the given parameters.

Parameters

**parametersdict

Arbitrary keyword arguments that are used to initialize the Parameters instance.

Raises

PydanticValidationError

If the provided parameters do not pass validation.

async classmethod create(**parameters)

Async factory method for creating score instances.

This method is called by Scorecard when instantiating scores from API configurations. Since AWSComprehendSentimentScore doesn’t require async initialization, we just create and return the instance.

async predict(context, model_input: ScoreInput)

Predict sentiment using AWS Comprehend.

Args:

context: Prediction context (unused) model_input: Score.Input containing text to analyze

Returns:

Score.Result with sentiment value and confidence scores

predict_validation()

Placeholder method to satisfy the base class requirement.

This score doesn’t require traditional validation since it uses AWS Comprehend’s pre-trained models.

register_model()

Register the model with MLflow by logging relevant parameters.

AWS Comprehend is a managed service, so there’s no model to register.

save_model()

Save the model to a specified path and log it as an artifact with MLflow.

AWS Comprehend is a managed service, so there’s no model to save.

train_model()

Placeholder method to satisfy the base class requirement.

AWS Comprehend is a pre-trained managed service that doesn’t require training.

class primus.scores.AgenticExtractor(scorecard_name, score_name, **kwargs)

Bases: LangGraphScore

Initialize the LangGraphScore.

This method sets up the score parameters and initializes basic attributes. The language model initialization is deferred to the async setup.

Parameters:

parameters – Configuration parameters for the score and language model.

class Parameters(*, scorecard_name: str | None = None, name: str | None = None, id: str | int | None = None, key: str | None = None, dependencies: List[dict] | None = None, data: dict | None = None, number_of_classes: int | None = None, label_score_name: str | None = None, label_field: str | None = None, validation: ValidationConfig | None = None, model_provider: Literal['ChatOpenAI', 'AzureChatOpenAI', 'BedrockChat', 'ChatVertexAI', 'ChatOllama'] = 'AzureChatOpenAI', model_name: str | None = None, model_region: str | None = None, reasoning_effort: str | None = 'low', verbosity: str | None = 'medium', temperature: float | None = 0, max_tokens: int | None = 500, logprobs: bool | None = False, top_logprobs: int | None = None, graph: list[dict] | None = None, input: dict | None = None, output: dict | None = None, depends_on: List[str] | Dict[str, str | Dict[str, Any]] | None = None, single_line_messages: bool = False, thread_id: str | None = None, prompt: str)

Bases: Parameters

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'protected_namespaces': ()}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

prompt: str
__init__(scorecard_name, score_name, **kwargs)

Initialize the LangGraphScore.

This method sets up the score parameters and initializes basic attributes. The language model initialization is deferred to the async setup.

Parameters:

parameters – Configuration parameters for the score and language model.

build_compiled_workflow(*, model_input: ScoreInput)

Build the LangGraph workflow.

static clean_quote(quote: str) → str
evaluate_model()

This is a placeholder for the validation process. It doesn’t make sense to implement this yet, because we don’t yet have any ground-truth labels to use for validation for any extractor. #YAGNI

load_context(context)
predict(context, model_input: ScoreInput)

Make predictions using the LangGraph workflow.

Parameters

model_inputScore.Input

The input data containing text and metadata

thread_idOptional[str]

Thread ID for checkpointing

batch_dataOptional[Dict[str, Any]]

Additional data for batch processing

**kwargsAny

Additional keyword arguments

Returns

Score.Result

The prediction result with value and explanation

class primus.scores.AgenticValidator(**parameters)

Bases: LangGraphScore

An agentic validator that uses LangGraph and advanced LangChain components to validate education information, specifically for degree, using both transcript and metadata.

This validator uses a language model to analyze transcripts and validate educational claims through a multi-step workflow implemented with LangGraph.

Initialize the AgenticValidator with the given parameters.

Args:

**parameters: Keyword arguments for configuring the validator.

class Input(*, text: str, metadata: Dict[str, ~typing.Any]=<factory>, results: List[Any] | None = None)

Bases: ScoreInput

Model input containing the transcript and metadata.

Attributes:

metadata (Dict[str, Any]): A dictionary containing degree information.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

classmethod handle_nan(v)
metadata: Dict[str, Any]
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class Parameters(*, scorecard_name: str | None = None, name: str | None = None, id: str | int | None = None, key: str | None = None, dependencies: List[dict] | None = None, data: dict | None = None, number_of_classes: int | None = None, label_score_name: str | None = None, label_field: str | None = None, validation: ValidationConfig | None = None, model_provider: Literal['ChatOpenAI', 'AzureChatOpenAI', 'BedrockChat', 'ChatVertexAI', 'ChatOllama']='AzureChatOpenAI', model_name: str | None = None, model_region: str | None = None, reasoning_effort: str | None = 'low', verbosity: str | None = 'medium', temperature: float | None = 0, max_tokens: int | None = 500, logprobs: bool | None = False, top_logprobs: int | None = None, graph: list[dict] | None = None, input: dict | None = None, output: dict | None = None, depends_on: Dict[str, str | ~typing.Dict[str, ~typing.Any]] | None=None, single_line_messages: bool = False, thread_id: str | None = None, labels: List[str] = <factory>, prompt: str = '', dependency: Dict[str, str] | None=None, agent_type: Literal['react', 'langgraph']='react')

Bases: Parameters

Parameters for configuring the AgenticValidator.

Attributes:

labels (List[str]): The labels of the metadata to validate. prompt (str): The custom prompt to use for validation. dependency (Optional[Dict[str, str]]): The dependency configuration. agent_type (Literal): The type of agent to use for validation.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

agent_type: Literal['react', 'langgraph']
dependency: Dict[str, str] | None
labels: List[str]
model_config = {'protected_namespaces': ()}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

prompt: str
class ReActAgentOutputParser(*args: Any, name: str | None = None)

Bases: ReActSingleInputOutputParser

model_config = {'extra': 'ignore', 'protected_namespaces': ()}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

parse(text: str) → AgentAction | AgentFinish

Parse text into agent action/finish.

__init__(**parameters)

Initialize the AgenticValidator with the given parameters.

Args:

**parameters: Keyword arguments for configuring the validator.

async build_compiled_workflow()

Build the compiled workflow for the AgenticValidator.

This overrides the parent class method to properly handle AgenticValidator’s workflow creation logic.

create_lcel_agent()

Create an LCEL-based agent for validation tasks with memory for the transcript.

initialize_validation_workflow()

Initialize the language model and create the workflow.

predict(context, model_input: ScoreInput) → Result

Predict the validity of the education information based on the transcript and metadata.

Args:

model_input (LangGraphScore.Input): The input containing the transcript and metadata.

Returns:

LangGraphScore.Result: The output containing the validation result.

class primus.scores.ExplainableClassifier(**parameters)

Bases: Score

A classifier based on XGBoost that uses n-gram vectorization and produces a ranked list of features for a target class, by importance.

Initialize the Score instance with the given parameters.

Parameters

**parametersdict

Arbitrary keyword arguments that are used to initialize the Parameters instance.

Raises

PydanticValidationError

If the provided parameters do not pass validation.

class Parameters(*, scorecard_name: str | None = None, name: str | None = None, id: str | int | None = None, key: str | None = None, dependencies: List[dict] | None = None, data: dict | None = None, number_of_classes: int | None = None, label_score_name: str | None = None, label_field: str | None = None, validation: ValidationConfig | None = None, top_n_features: int = 10000, leaderboard_n_features: int = 10, target_score_name: str, target_score_value: str, ngram_range: str = '2,3', decision_threshold: float = 0.5, scale_pos_weight_index: float = 0, include_explanations: bool = False, keywords: list = None)

Bases: Parameters

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

decision_threshold: float
include_explanations: bool
keywords: list
leaderboard_n_features: int
model_config = {'protected_namespaces': ()}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

ngram_range: str
scale_pos_weight_index: float
target_score_name: str
target_score_value: str
top_n_features: int
class Result(*, parameters: Parameters, value: str | bool, explanation: str | None = None, confidence: float | None = None, start_time_seconds: float | None = None, end_time_seconds: float | None = None, metadata: dict = {}, error: str | None = None, code: str | None = None)

Bases: Result

ExplainableClassifier result.

Inherits explanation and confidence fields from Score.Result base class.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'protected_namespaces': ()}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

__init__(**parameters)

Initialize the Score instance with the given parameters.

Parameters

**parametersdict

Arbitrary keyword arguments that are used to initialize the Parameters instance.

Raises

PydanticValidationError

If the provided parameters do not pass validation.

evaluate_model()

Evaluate the model on the validation data.

Returns

dict

Dictionary containing evaluation metrics.

explain_model()
predict(context, model_input)

Make predictions on the input data.

Parameters

contextAny

Context for the prediction

model_inputScore.Input

The input data for making predictions.

Returns

Union[Score.Result, List[Score.Result]]

Either a single Score.Result or a list of Score.Results

predict_validation()

Predict on the validation set.

This method should be implemented by subclasses to provide the prediction logic on the validation set.

preprocess_text(text)
register_model()

Register the model with the model registry.

save_model()

Save the model to the model registry.

train_model()

Train the XGBoost model with the specified positive class weight.

Parameters

X_trainnumpy.ndarray

Training data features.

y_trainnumpy.ndarray

Training data labels.

X_valnumpy.ndarray

Validation data features.

y_valnumpy.ndarray

Validation data labels.

vectorize_transcript(transcript: str)
class primus.scores.FastTextClassifier(**parameters)

Bases: Score

Initialize the Score instance with the given parameters.

Parameters

**parametersdict

Arbitrary keyword arguments that are used to initialize the Parameters instance.

Raises

PydanticValidationError

If the provided parameters do not pass validation.

class Parameters(*, scorecard_name: str | None = None, name: str | None = None, id: str | int | None = None, key: str | None = None, dependencies: List[dict] | None = None, data: dict | None = None, number_of_classes: int | None = None, label_score_name: str | None = None, label_field: str | None = None, validation: ValidationConfig | None = None, learning_rate: float = 0.1, dimension: int = 100, window_size: int = 5, number_of_epochs: int = 5, minimum_word_count: int = 1, minimum_label_count: int = 1, minimum_character_ngram_length: int = 0, maximum_character_ngram_length: int = 0, number_of_negative_samples: int = 5, word_ngram_count: int = 1, loss_function: str = 'softmax', bucket_size: int = 2000000, number_of_threads: int = 4, learning_rate_update_rate: int = 100, sampling_threshold: float = 0.0001)

Bases: Parameters

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

bucket_size: int
dimension: int
learning_rate: float
learning_rate_update_rate: int
loss_function: str
maximum_character_ngram_length: int
minimum_character_ngram_length: int
minimum_label_count: int
minimum_word_count: int
model_config = {'protected_namespaces': ()}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

number_of_epochs: int
number_of_negative_samples: int
number_of_threads: int
sampling_threshold: float
window_size: int
word_ngram_count: int
__init__(**parameters)

Initialize the Score instance with the given parameters.

Parameters

**parametersdict

Arbitrary keyword arguments that are used to initialize the Parameters instance.

Raises

PydanticValidationError

If the provided parameters do not pass validation.

data_filename()
get_model_artifact_path()
load_context(context)
load_model(model_path)
predict(model_input, text_column='text')

Make predictions on the input data.

Parameters

contextAny

Context for the prediction

model_inputScore.Input

The input data for making predictions.

Returns

Union[Score.Result, List[Score.Result]]

Either a single Score.Result or a list of Score.Results

predict_validation()

Predict on the validation set.

This method should be implemented by subclasses to provide the prediction logic on the validation set.

process_data()
register_model()

Register the model with the model registry.

save_model()

Save the model to the model registry.

save_model_binary()
train_model()

Train the XGBoost model with the specified positive class weight.

Parameters

X_trainnumpy.ndarray

Training data features.

y_trainnumpy.ndarray

Training data labels.

X_valnumpy.ndarray

Validation data features.

y_valnumpy.ndarray

Validation data labels.

class primus.scores.OpenAIEmbeddingsClassifier(**parameters)

Bases: Score

Initialize the Score instance with the given parameters.

Parameters

**parametersdict

Arbitrary keyword arguments that are used to initialize the Parameters instance.

Raises

PydanticValidationError

If the provided parameters do not pass validation.

class Parameters(*, scorecard_name: str | None = None, name: str | None = None, id: str | int | None = None, key: str | None = None, dependencies: List[dict] | None = None, data: dict | None = None, number_of_classes: int | None = None, label_score_name: str | None = None, label_field: str | None = None, validation: ValidationConfig | None = None, embeddings_model: str, embeddings_model_trainable_layers: int = 3, maximum_tokens_per_window: int = 512, multiple_windows: bool = False, maximum_windows: int = 0, start_from_end: bool = False, number_of_epochs: int, batch_size: int, warmup_learning_rate: float, number_of_warmup_epochs: int, plateau_learning_rate: float, number_of_plateau_epochs: int, learning_rate_decay: float, early_stop_patience: int, l2_regularization_strength: float, dropout_rate: float)

Bases: Parameters

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

batch_size: int
dropout_rate: float
early_stop_patience: int
embeddings_model: str
embeddings_model_trainable_layers: int
l2_regularization_strength: float
learning_rate_decay: float
maximum_tokens_per_window: int
maximum_windows: int
model_config = {'protected_namespaces': ()}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

multiple_windows: bool
number_of_epochs: int
number_of_plateau_epochs: int
number_of_warmup_epochs: int
plateau_learning_rate: float
start_from_end: bool
warmup_learning_rate: float
__init__(**parameters)

Initialize the Score instance with the given parameters.

Parameters

**parametersdict

Arbitrary keyword arguments that are used to initialize the Parameters instance.

Raises

PydanticValidationError

If the provided parameters do not pass validation.

evaluate_model()

Evaluate the model on the validation data.

Returns

dict

Dictionary containing evaluation metrics.

predict(context, model_input)

Make predictions on the input data.

Parameters

contextAny

Context for the prediction

model_inputScore.Input

The input data for making predictions.

Returns

Union[Score.Result, List[Score.Result]]

Either a single Score.Result or a list of Score.Results

predict_validation()

Predict on the validation set.

This method should be implemented by subclasses to provide the prediction logic on the validation set.

process_data(data=None)
train_model()

Train the XGBoost model with the specified positive class weight.

Parameters

X_trainnumpy.ndarray

Training data features.

y_trainnumpy.ndarray

Training data labels.

X_valnumpy.ndarray

Validation data features.

y_valnumpy.ndarray

Validation data labels.

class primus.scores.Score(**parameters)

Bases: ABC

Abstract base class for implementing classification and scoring models in Primus.

Score is the fundamental building block of classification in Primus. Each Score represents a specific classification task and can be implemented using various approaches:

  • Machine learning models (e.g., DeepLearningSemanticClassifier)

  • LLM-based classification (e.g., LangGraphScore)

  • Rule-based systems (e.g., KeywordClassifier)

  • Custom logic (by subclassing Score)

The Score class provides: - Standard input/output interfaces using Pydantic models - Visualization tools for model performance - Cost tracking for API-based models - Metrics computation and logging

Common usage patterns: 1. Creating a custom classifier:

class MyClassifier(Score):
def predict(self, context, model_input: Score.Input) -> Score.Result:

text = model_input.text # Custom classification logic here return Score.Result(

parameters=self.parameters, value=”Yes” if is_positive(text) else “No”

)

  1. Using in a Scorecard:
    scores:
    MyScore:

    class: MyClassifier parameters:

    threshold: 0.8

  2. Training a model:

    classifier = MyClassifier() classifier.train_model() classifier.evaluate_model() classifier.save_model()

  3. Making predictions:
    result = classifier.predict(context, Score.Input(

    text=”content to classify”, metadata={“source”: “email”}

    ))

The Score class is designed to be extended for different classification approaches while maintaining a consistent interface for use in Scorecards and Evaluations.

Initialize the Score instance with the given parameters.

Parameters

**parametersdict

Arbitrary keyword arguments that are used to initialize the Parameters instance.

Raises

PydanticValidationError

If the provided parameters do not pass validation.

class FieldValidation(*, valid_classes: List[str] | None = None, patterns: List[str] | None = None, minimum_length: int | None = None, maximum_length: int | None = None)

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

maximum_length: int | None
minimum_length: int | None
model_config = {'protected_namespaces': ()}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

patterns: List[str] | None
valid_classes: List[str] | None
classmethod validate_patterns(value)
Input

alias of ScoreInput

class Parameters(**data: Any)

Bases: BaseModel

Parameters required for scoring.

Attributes

datadict

Dictionary containing data-related parameters.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

classmethod convert_data_percentage(value)

Convert the percentage value in the data dictionary to a float.

Parameters

valuedict

Dictionary containing data-related parameters.

Returns

dict

Updated dictionary with the percentage value converted to float.

data: dict | None
dependencies: List[dict] | None
id: str | int | None
key: str | None
label_field: str | None
label_score_name: str | None
model_config = {'protected_namespaces': ()}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str | None
number_of_classes: int | None
scorecard_name: str | None
validation: Score.ValidationConfig | None
class Result(*, parameters: Parameters, value: str | bool, explanation: str | None = None, confidence: float | None = None, start_time_seconds: float | None = None, end_time_seconds: float | None = None, metadata: dict = {}, error: str | None = None, code: str | None = None)

Bases: BaseModel

Standard output structure for all Score classifications in Primus.

The Result class provides a consistent way to represent classification outcomes, supporting both simple yes/no results and complex multi-class classifications with explanations. It’s used throughout Primus for: - Individual Score results - Batch processing outputs - Evaluation metrics - Dashboard result tracking

Attributes:

parameters: Configuration used for this classification value: The classification result (e.g., “Yes”/”No” or class label) explanation: Detailed explanation of why this result was chosen confidence: Confidence score for the classification (0.0 to 1.0) metadata: Additional context about the classification error: Optional error message if classification failed

The Result class provides helper methods for common operations: - is_yes(): Check if result is affirmative - is_no(): Check if result is negative - __eq__: Compare results (case-insensitive)

Common usage: 1. Basic classification with explanation:

result = Score.Result(

parameters=self.parameters, value=”Yes”, explanation=”Clear greeting found at beginning of transcript”, confidence=0.95

)

  1. Classification with metadata:
    result = Score.Result(

    parameters=self.parameters, value=”No”, explanation=”No greeting found in transcript”, confidence=0.88, metadata={“source”: “phone_call”}

    )

  2. Error case:
    result = Score.Result(

    parameters=self.parameters, value=”ERROR”, error=”API timeout”

    )

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

code: str | None
confidence: float | None
property confidence_from_metadata: float | None

Backwards compatibility: confidence from metadata

end_time_seconds: float | None
error: str | None
explanation: str | None
property explanation_from_metadata: str | None

Backwards compatibility: explanation from metadata

is_no()
is_yes()
metadata: dict
model_config = {'protected_namespaces': ()}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

parameters: Score.Parameters
start_time_seconds: float | None
validate(validation_config: ValidationConfig)
value: str | bool
exception SkippedScoreException(score_name: str, reason: str)

Bases: Exception

Raised when a score is skipped due to dependency conditions not being met.

__init__(score_name: str, reason: str)
class ValidationConfig(**data: Any)

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

explanation: Score.FieldValidation | None
model_config = {'protected_namespaces': ()}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

value: Score.FieldValidation | None
exception ValidationError

Bases: Exception

Raised when a score result violates configured validation constraints.

__init__(**parameters)

Initialize the Score instance with the given parameters.

Parameters

**parametersdict

Arbitrary keyword arguments that are used to initialize the Parameters instance.

Raises

PydanticValidationError

If the provided parameters do not pass validation.

analyze_dataset()
static apply_processors(score_input: ScoreInput, processors_config: list) → ScoreInput

Apply a list of processors to a Score.Input and return the transformed Score.Input.

Args:

score_input: Input object containing text/metadata/results. processors_config: List of processor configurations.

Returns:

Transformed Score.Input after all configured processors run.

static apply_processors_to_text(text: str, processors_config: list, metadata: dict = None) → str

Apply a list of processors to text for production predictions.

This method applies the same processor pipeline used during training/evaluation to production prediction inputs. It creates a Score.Input, applies all configured processors, and returns the processed text.

Args:

text: Input text to process processors_config: List of processor configurations, each with ‘class’ and optional ‘parameters’

Example: [{‘class’: ‘FilterCustomerOnlyProcessor’},

{‘class’: ‘RemoveSpeakerIdentifiersTranscriptFilter’}]

Returns:

Processed text after applying all processors

Example:
processors = [

{‘class’: ‘FilterCustomerOnlyProcessor’}, {‘class’: ‘RemoveSpeakerIdentifiersTranscriptFilter’}

] processed_text = Score.apply_processors_to_text(raw_text, processors)

evaluate_model()

Evaluate the model on the validation data.

Returns

dict

Dictionary containing evaluation metrics.

classmethod from_name(scorecard, score)
get_accumulated_costs()

Get the expenses that have been accumulated over all the computed elements.

Returns:

dict: Aggregated cost information with totals and components

get_label_score_name()

Determine the appropriate score name based on the parameters.

Returns

str

The determined score name.

property is_multi_class

Determine if the classification problem is multi-class.

This property checks the unique labels in the dataframe to determine if the problem is multi-class.

Returns

bool

True if the problem is multi-class, False otherwise.

is_relevant(text)

Determine if the given text is relevant using the predict method.

Parameters

textstr

The text to be classified.

Returns

bool

True if the text is classified as relevant, False otherwise.

classmethod load(scorecard_identifier: str, score_name: str, use_cache: bool = True, yaml_only: bool = False)

Load a single score configuration with configurable caching behavior.

Args:

scorecard_identifier: A string that identifies the scorecard (ID, name, key, or external ID) score_name: Name of the specific score to load use_cache: If True (default), cache API data to local YAML files. If False, don’t cache. yaml_only: If True, load only from local YAML files without API calls.

Returns:

Score: An initialized Score instance

Raises:

ValueError: If the score cannot be loaded

load_data(*, data=None, excel=None, fresh=False, reload=False)
static log_validation_errors(error: ValidationError)

Log validation errors for the parameters.

Parameters

errorPydanticValidationError

The validation error object containing details about the validation failures.

model_directory_path()
property number_of_classes

Determine the number of classes for the classification problem.

This property checks the unique labels in the dataframe to determine the number of classes.

Returns

int

The number of unique classes.

abstractmethod predict(context, model_input: ScoreInput) → Result | List[Result]

Make predictions on the input data.

Parameters

contextAny

Context for the prediction

model_inputScore.Input

The input data for making predictions.

Returns

Union[Score.Result, List[Score.Result]]

Either a single Score.Result or a list of Score.Results

predict_validation()

Predict on the validation set.

This method should be implemented by subclasses to provide the prediction logic on the validation set.

record_configuration(configuration)

Record the provided configuration dictionary as a JSON file in the appropriate report folder for this model.

Parameters

configurationdict

Dictionary containing the configuration to be recorded.

register_model()

Register the model with the model registry.

report_directory_path()
report_file_name(file_name)

Generate the full path for a report file within the report directory.

Calling this function will implicitly trigger the function to ensure that the report directory exists.

Parameters

file_namestr

The name of the report file.

Returns

str

The full path to the report file with spaces replaced by underscores.

save_model()

Save the model to the model registry.

setup_label_map(labels)

Set up a mapping from labels to integers.

Parameters

labelslist

List of unique labels.

train_model(X_train, y_train, X_val, y_val)

Train the XGBoost model with the specified positive class weight.

Parameters

X_trainnumpy.ndarray

Training data features.

y_trainnumpy.ndarray

Training data labels.

X_valnumpy.ndarray

Validation data features.

y_valnumpy.ndarray

Validation data labels.

class primus.scores.SubjectIdentityScore(scorecard_name=None, score_name=None, findings: List[dict] | None = None, files_scanned: List[str] | None = None, findings_command: str | None = None, source_root: str | None = None, **kwargs)

Bases: Score

Programmatic detector that scores Items by subject identity (metadata.subjectKey) against injected or command-produced findings, independent of source span overlap.

Initialize the Score instance with the given parameters.

Parameters

**parametersdict

Arbitrary keyword arguments that are used to initialize the Parameters instance.

Raises

PydanticValidationError

If the provided parameters do not pass validation.

__init__(scorecard_name=None, score_name=None, findings: List[dict] | None = None, files_scanned: List[str] | None = None, findings_command: str | None = None, source_root: str | None = None, **kwargs)

Initialize the Score instance with the given parameters.

Parameters

**parametersdict

Arbitrary keyword arguments that are used to initialize the Parameters instance.

Raises

PydanticValidationError

If the provided parameters do not pass validation.

async classmethod create(**parameters)

Async factory used by Scorecard when loading YAML/API configurations.

load_context(context=None)
async predict(model_input: ScoreInput, **_kwargs) → Result

Make predictions on the input data.

Parameters

contextAny

Context for the prediction

model_inputScore.Input

The input data for making predictions.

Returns

Union[Score.Result, List[Score.Result]]

Either a single Score.Result or a list of Score.Results

predict_validation()

Predict on the validation set.

This method should be implemented by subclasses to provide the prediction logic on the validation set.

register_model()

Register the model with the model registry.

save_model()

Save the model to the model registry.

class primus.scores.SubjectSpanOverlapScore(scorecard_name=None, score_name=None, findings: List[dict] | None = None, files_scanned: List[str] | None = None, findings_command: str | None = None, source_root: str | None = None, **kwargs)

Bases: Score

Programmatic detector that scores Items when a finding matches both metadata.subjectKey and overlapping source-file spans.

Initialize the Score instance with the given parameters.

Parameters

**parametersdict

Arbitrary keyword arguments that are used to initialize the Parameters instance.

Raises

PydanticValidationError

If the provided parameters do not pass validation.

__init__(scorecard_name=None, score_name=None, findings: List[dict] | None = None, files_scanned: List[str] | None = None, findings_command: str | None = None, source_root: str | None = None, **kwargs)

Initialize the Score instance with the given parameters.

Parameters

**parametersdict

Arbitrary keyword arguments that are used to initialize the Parameters instance.

Raises

PydanticValidationError

If the provided parameters do not pass validation.

async classmethod create(**parameters)

Async factory used by Scorecard when loading YAML/API configurations.

load_context(context=None)
async predict(model_input: ScoreInput, **_kwargs) → Result

Make predictions on the input data.

Parameters

contextAny

Context for the prediction

model_inputScore.Input

The input data for making predictions.

Returns

Union[Score.Result, List[Score.Result]]

Either a single Score.Result or a list of Score.Results

predict_validation()

Predict on the validation set.

This method should be implemented by subclasses to provide the prediction logic on the validation set.

register_model()

Register the model with the model registry.

save_model()

Save the model to the model registry.

class primus.scores.TactusScore(**parameters)

Bases: Score

Score that executes embedded Tactus DSL code for classification.

Uses Tactus runtime with in-process execution (no containers) for high-volume Primus scenarios with trusted code.

The model is specified inside the Lua code via default_model at the procedure level. Individual classifiers inherit it, or can override with their own model parameter.

Example YAML:

class: TactusScore code: |

default_model “openai/gpt-5.4-nano” ClassifyProcedure {

classes = {“YES”, “NO”}, system_message = [[

Classification instructions… ]],

user_message = [[

Analyze: <transcript>{{ text }}</transcript> ]] }

Initialize TactusScore with Tactus code.

class Parameters(*, scorecard_name: str | None = None, name: str | None = None, id: str | int | None = None, key: str | None = None, dependencies: List[dict] | None = None, data: dict | None = None, number_of_classes: int | None = None, label_score_name: str | None = None, label_field: str | None = None, validation: ValidationConfig | None = None, code: str, valid_classes: List[str] | None = None, output: Dict[str, str] | None = None, model_provider: str | None = None, model_name: str | None = None, base_model_name: str | None = None, max_tokens: int | None = None, temperature: float | None = None, top_p: float | None = None, reasoning_effort: str | None = None, verbosity: str | None = None, model_region: str | None = None, logprobs: bool | None = None, top_logprobs: int | None = None, parse_from_start: bool | None = None)

Bases: Parameters

Configuration parameters for TactusScore.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

base_model_name: str | None
code: str
classmethod handle_tactus_code_fallback(data)

Accept ‘tactus_code’ as a fallback for ‘code’ during transition.

logprobs: bool | None
max_tokens: int | None
model_config = {'protected_namespaces': ()}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_name: str | None
model_provider: str | None
model_region: str | None
output: Dict[str, str] | None
parse_from_start: bool | None
reasoning_effort: str | None
temperature: float | None
top_logprobs: int | None
top_p: float | None
valid_classes: List[str] | None
verbosity: str | None
__init__(**parameters)

Initialize TactusScore with Tactus code.

async classmethod create(**parameters) → TactusScore

Factory method for async initialization.

async predict(model_input: ScoreInput, **_kwargs: Any) → Result | List[Result]

Execute Tactus procedure and return classification result.

Parameters

model_inputScore.Input

The input data containing text and metadata

Returns

Score.Result

The prediction result with value and explanation

primus.scores.resolve_score_class(name: str)

Resolve a configured score class without trusting package attributes.

Importing primus.scores.<ClassName> makes Python cache that submodule on this package under ClassName. Looking up the package attribute after that point returns the module instead of invoking __getattr__. Resolve through the explicit module map so registry behavior is import-order safe.

Subpackages

Submodules