#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""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 :class:`YesNoQuestion`, :class:`ChoiceQuestion` or
:class:`ScoreQuestion`. Questions are asked by name, several about one state
at once: :meth:`BaseClassifier.ask` takes any mix of kinds, and
:meth:`BaseClassifier.yes_no`, :meth:`BaseClassifier.choice` and
:meth:`BaseClassifier.score` take questions of one kind and return typed
results.
"""
import time
from abc import abstractmethod
from collections.abc import Mapping
from typing import Any, TypeAlias, TypeVar
from pydantic import BaseModel, Field
from pipecat.metrics.metrics import (
LLMTokenUsage,
LLMUsageMetricsData,
MetricsData,
ProcessingMetricsData,
)
from pipecat.utils.base_object import BaseObject
[docs]
class ClassifierError(Exception):
"""A classifier could not answer a question."""
[docs]
class YesNoQuestion(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.
"""
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
[docs]
class ChoiceQuestion(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.
"""
instructions: str | dict[str, Any] | list[Any]
options: dict[str, str | dict[str, Any] | list[Any] | None]
[docs]
class ScoreQuestion(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.
"""
instructions: str | dict[str, Any] | list[Any]
levels: list[str | dict[str, Any] | list[Any]] = Field(min_length=2)
ClassifierQuestion: TypeAlias = YesNoQuestion | ChoiceQuestion | ScoreQuestion
[docs]
class YesNoResult(BaseModel):
"""Answer to a :class:`YesNoQuestion`.
Parameters:
probability: How likely the answer is yes, from 0 to 1.
"""
probability: float
@property
def is_yes(self) -> bool:
"""Whether yes is the likelier answer.
Callers that need more certainty than that compare ``probability``
with a threshold of their own.
"""
return self.probability >= 0.5
[docs]
class ChoiceResult(BaseModel):
"""Answer to a :class:`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.
"""
choice: str
probabilities: dict[str, float]
confidence: float
[docs]
class ScoreLevel(BaseModel):
"""One level of a :class:`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.
"""
level: str | dict[str, Any] | list[Any]
probability: float
[docs]
class ScoreResult(BaseModel):
"""Answer to a :class:`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.
"""
score: float
levels: list[ScoreLevel]
confidence: float
[docs]
def probability(self, level: str | dict[str, Any] | list[Any]) -> float:
"""How likely one level is.
Args:
level: The level as the question gave it.
Returns:
The level's probability.
Raises:
KeyError: If the scale has no such level.
"""
for item in self.levels:
if item.level == level:
return item.probability
raise KeyError(f"not a level of this scale: {level!r}")
ClassifierResult: TypeAlias = YesNoResult | ChoiceResult | ScoreResult
R = TypeVar("R", YesNoResult, ChoiceResult, ScoreResult)
[docs]
class BaseClassifier(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.
:meth:`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 :meth:`_ask`.
An owner that runs inside a worker calls :meth:`setup` with its task
manager before the first question, and :meth:`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 :class:`~pipecat.frames.frames.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))
"""
[docs]
def __init__(self, **kwargs):
"""Initialize the classifier.
Args:
**kwargs: Additional arguments passed to the parent class.
"""
super().__init__(**kwargs)
self._register_event_handler("on_metrics")
@property
def model(self) -> str | None:
"""The model that answers, named in the metrics."""
return None
[docs]
async def ask(
self, state: str | dict[str, Any] | list[Any], questions: Mapping[str, ClassifierQuestion]
) -> dict[str, ClassifierResult]:
"""Answer several questions about one state.
Args:
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.
"""
started = time.perf_counter()
results, usage = await self._ask(state, questions)
await self._call_event_handler(
"on_metrics", self._metrics(time.perf_counter() - started, usage)
)
return results
[docs]
async def yes_no(
self, state: str | dict[str, Any] | list[Any], questions: Mapping[str, YesNoQuestion]
) -> dict[str, YesNoResult]:
"""Ask whether the state meets each condition.
Args:
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.
"""
return self._typed(await self.ask(state, questions), YesNoResult)
[docs]
async def choice(
self, state: str | dict[str, Any] | list[Any], questions: Mapping[str, ChoiceQuestion]
) -> dict[str, ChoiceResult]:
"""Ask which option fits the state, for each question.
Args:
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.
"""
return self._typed(await self.ask(state, questions), ChoiceResult)
[docs]
async def score(
self, state: str | dict[str, Any] | list[Any], questions: Mapping[str, ScoreQuestion]
) -> dict[str, ScoreResult]:
"""Ask where the state falls on each scale.
Args:
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.
"""
return self._typed(await self.ask(state, questions), ScoreResult)
@abstractmethod
async def _ask(
self, state: str | dict[str, Any] | list[Any], questions: Mapping[str, ClassifierQuestion]
) -> tuple[dict[str, ClassifierResult], LLMTokenUsage | None]:
"""Answer the questions, and say what tokens the call used if that is known."""
pass
def _metrics(self, seconds: float, usage: LLMTokenUsage | None) -> list[MetricsData]:
data: list[MetricsData] = [
ProcessingMetricsData(processor=self.name, model=self.model, value=seconds)
]
if usage:
data.append(LLMUsageMetricsData(processor=self.name, model=self.model, value=usage))
return data
def _typed(self, results: dict[str, ClassifierResult], result_type: type[R]) -> dict[str, R]:
typed: dict[str, R] = {}
for name, result in results.items():
if not isinstance(result, result_type):
raise ClassifierError(
f"expected a {result_type.__name__} for {name!r}, got {type(result).__name__}"
)
typed[name] = result
return typed