client

HTTP client for Jev, TypeSafe’s hosted classification model.

One JevClient holds one HTTP/2 connection pool, adds the auth header, retries when Jev is busy, and counts the tokens every request used. Several JevClassifier instances can share one.

pipecat.classifiers.jev.client.DEFAULT_BASE_URL = 'https://api.typesafe.ai'

Where the API is served.

pipecat.classifiers.jev.client.DEFAULT_MODEL = 'jev-1.13.0'

The Jev model to ask. Pinned so thresholds tuned against it hold.

pipecat.classifiers.jev.client.DEFAULT_TIMEOUT = 10.0

Seconds to wait for a reply before giving up.

class pipecat.classifiers.jev.client.JevUsage(*, input_tokens: int = 0, output_tokens: int = 0)[source]

Bases: BaseModel

Tokens Jev used: for one request, or over every request in JevClient.usage.

Parameters:

  • input_tokens – Tokens sent.

  • output_tokens – Tokens received.

class pipecat.classifiers.jev.client.JevClient(*, api_key: str, base_url: str = 'https://api.typesafe.ai', model: str = 'jev-1.13.0', timeout: float = 10.0, max_retries: int = 3)[source]

Bases: object

HTTP client for Jev’s systemone endpoint.

Holds one HTTP/2 connection pool, so many small requests share a connection and can be in flight at the same time. The connection is kept open between questions, so a gap between them does not cost a new TLS handshake. Retries with backoff when Jev answers 429 or 529. Counts the tokens every request used in usage.

connect() opens the connection ahead of the first question, and close() releases it.

__init__(*, api_key: str, base_url: str = 'https://api.typesafe.ai', model: str = 'jev-1.13.0', timeout: float = 10.0, max_retries: int = 3)[source]

Initialize the client.

Parameters:

  • api_key – Jev API key.

  • base_url – Where the API is served.

  • model – The Jev model to ask. Pinned so thresholds tuned against it keep holding; jev-latest follows TypeSafe’s newest release.

  • timeout – Seconds to wait for a reply before giving up.

  • max_retries – How many times to retry a request Jev refused because it was busy.

property model: str

The Jev model the questions go to.

property usage: JevUsage

Tokens used so far, over every request.

async connect()[source]

Open the connection to Jev.

Lists the models, the cheapest request there is, so the first question finds the connection already open. Connects once: a client shared by several classifiers is asked by each of them, and only the first call sends anything.

Raises:

ClassifierError – If Jev could not be reached or refused the request.

async close()[source]

Close the connection pool.

async ask(state: str | dict[str, Any] | list[Any], questions: Mapping[str, dict[str, Any]]) → tuple[dict[str, dict[str, Any]], JevUsage][source]

Send questions about one state and return Jev’s answers.

Parameters:

  • state – What the questions are about.

  • questions – The questions by name, each in Jev’s own format: a type of noul, choice or score, instructions, and criteria.

Returns:

The answers by the same names, in Jev’s own format, and the tokens the request used.

Raises:

ClassifierError – If Jev rejected the request, kept refusing it because it was busy, could not be reached, or left a question unanswered.