base_classifier

Classifiers answer typed questions about some state.

A classifier is a plain object. Whoever needs answers creates one, keeps it, and calls it. It does not sit in a pipeline and no frames flow into it. A question is one of YesNoQuestion, ChoiceQuestion or ScoreQuestion. Questions are asked by name, several about one state at once: BaseClassifier.ask() takes any mix of kinds, and BaseClassifier.yes_no(), BaseClassifier.choice() and BaseClassifier.score() take questions of one kind and return typed results.

exception pipecat.classifiers.base_classifier.ClassifierError

Bases: Exception

A classifier could not answer a question.

class pipecat.classifiers.base_classifier.YesNoQuestion(*, instructions: str | dict[str, Any] | list[Any], yes: str | dict[str, Any] | list[Any] | None = None, no: str | dict[str, Any] | list[Any] | None = None)

Bases: BaseModel

Whether the state meets a condition.

Parameters:

• instructions – What is being checked for, as a yes or no question. Text, or structured data holding the question in one field and what it refers to in others.

• yes – What counts as a yes, when the question alone leaves it open.

• no – What counts as a no.

class pipecat.classifiers.base_classifier.ChoiceQuestion(*, instructions: str | dict[str, Any] | list[Any], options: dict[str, str | dict[str, Any] | list[Any] | None])

Bases: BaseModel

Which of several options fits the state.

Parameters:

• instructions – What is being decided.

• options – The options to choose from, each mapped to a description of when it applies, or None when the option itself says enough.

class pipecat.classifiers.base_classifier.ScoreQuestion(*, instructions: str | dict[str, Any] | list[Any], levels: Annotated[list[str | dict[str, Any] | list[Any]], MinLen(min_length=2)])

Bases: BaseModel

Where the state falls on an ordered scale.

Parameters:

• instructions – What is being rated.

• levels – The levels of the scale in order, lowest first, each described in a few words or as structured data. At least two.

class pipecat.classifiers.base_classifier.YesNoResult(*, probability: float)

Bases: BaseModel

Answer to a YesNoQuestion.

Parameters:

probability – How likely the answer is yes, from 0 to 1.

property is_yes: bool

Whether yes is the likelier answer.

Callers that need more certainty than that compare probability with a threshold of their own.

class pipecat.classifiers.base_classifier.ChoiceResult(*, choice: str, probabilities: dict[str, float], confidence: float)

Bases: BaseModel

Answer to a ChoiceQuestion.

Parameters:

• choice – The option that fits best.

• probabilities – How likely each option is, keyed by option.

• confidence – How sure the classifier is of choice, from 0 to 1.

class pipecat.classifiers.base_classifier.ScoreLevel(*, level: str | dict[str, Any] | list[Any], probability: float)

Bases: BaseModel

One level of a ScoreQuestion’s scale and how likely it is.

Parameters:

• level – The level as the question gave it.

• probability – How likely the state is at this level, from 0 to 1.

class pipecat.classifiers.base_classifier.ScoreResult(*, score: float, levels: list[ScoreLevel], confidence: float)

Bases: BaseModel

Answer to a ScoreQuestion.

Parameters:

• score – Where the state falls on the scale, as a position from 0 (the first level) to one less than the number of levels. It is the probability-weighted position, so it may fall between two levels.

• levels – How likely each level is, in the question’s order.

• confidence – How sure the classifier is of score, from 0 to 1.

probability(level: str | dict[str, Any] | list[Any]) → float

How likely one level is.

Parameters:

level – The level as the question gave it.

Returns:

The level’s probability.

Raises:

KeyError – If the scale has no such level.

class pipecat.classifiers.base_classifier.BaseClassifier(**kwargs)

Bases: BaseObject

Answers typed questions about a state.

Every question is about a state: plain text, or structured data such as a transcript with speaker labels or a trimmed screen snapshot. ask() answers any number of questions about one state, by name; the three typed methods are built on it and take questions of one kind. Subclasses implement _ask().

An owner that runs inside a worker calls setup() with its task manager before the first question, and cleanup() when it is done.

Event handlers available:

• on_metrics: Called after every call with its metrics, the time it took and, when the classifier knows it, the tokens it used. A classifier cannot push frames, so the owner is the one to put them in a MetricsFrame.

Example:

await self._classifier.setup(self.task_manager)

@self._classifier.event_handler("on_metrics")
async def on_metrics(classifier, data: list[MetricsData]):
    await self.push_frame(MetricsFrame(data=data))

__init__(**kwargs)

Initialize the classifier.

Parameters:

**kwargs – Additional arguments passed to the parent class.

property model: str | None

The model that answers, named in the metrics.

async ask(state: str | dict[str, Any] | list[Any], questions: Mapping[str, YesNoQuestion | ChoiceQuestion | ScoreQuestion]) → dict[str, YesNoResult | ChoiceResult | ScoreResult]

Answer several questions about one state.

Parameters:

• state – What the questions are about.

• questions – The questions, by name.

Returns:

One result per question, by the same names, each of the type its question calls for.

Raises:

ClassifierError – If the answers could not be produced, or not in time: every classifier answers or raises within a bound of its own, so a caller waiting on it is never left hanging.

async yes_no(state: str | dict[str, Any] | list[Any], questions: Mapping[str, YesNoQuestion]) → dict[str, YesNoResult]

Ask whether the state meets each condition.

Parameters:

• state – What the questions are about.

• questions – The questions, by name.

Returns:

How likely each answer is yes, by the same names.

Raises:

ClassifierError – If the answers could not be produced.

async choice(state: str | dict[str, Any] | list[Any], questions: Mapping[str, ChoiceQuestion]) → dict[str, ChoiceResult]

Ask which option fits the state, for each question.

Parameters:

• state – What the questions are about.

• questions – The questions, by name.

Returns:

The option that fits and how likely each one is, by the same names.

Raises:

ClassifierError – If the answers could not be produced.

async score(state: str | dict[str, Any] | list[Any], questions: Mapping[str, ScoreQuestion]) → dict[str, ScoreResult]

Ask where the state falls on each scale.

Parameters:

• state – What the questions are about.

• questions – The questions, by name.

Returns:

The position on each scale and how likely each level is, by the same names.

Raises:

ClassifierError – If the answers could not be produced.